nodemailer/wildduck
1
1
Fork 0
mirror of https://github.com/nodemailer/wildduck.git synced 2024-12-27 02:10:52 +08:00
Code Issues Projects Releases Packages Wiki Activity

Changed messages shard key to mailbox+uid

Browse source
This commit is contained in:
Andris Reinman 2017-07-15 19:08:33 +03:00
parent 3f1f0b6fac
commit ed75658f80
28 changed files with 2067 additions and 1991 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -446,7 +446,7 @@ Shard the following collections by these keys:
```javascript
sh.enableSharding('wildduck');
sh.shardCollection('wildduck.messages', { user: 'hashed' });
sh.shardCollection('wildduck.messages', { mailbox: 1, uid: 1 });
sh.shardCollection('wildduck.threads', { user: 'hashed' });
sh.shardCollection('wildduck.attachments.files', { 'metadata.h': 'hashed' });
sh.shardCollection('wildduck.attachments.chunks', { files_id: 'hashed' });

1931
imap.js
View file

File diff suppressed because it is too large Load diff

64
indexes.yaml
View file

@ -55,29 +55,7 @@ indexes:
specialUse: 1
# Indexes for the messages collection
# NB! this is a sharded collection and the shard
# key should be 'user' so keep this field as the first one
# in indexes
- collection: messages
index:
# hashed index needed for sharding
name: messages_shard
key:
user: hashed
- collection: messages
index:
name: mailbox_by_id
key:
_id: 1
user: 1
- collection: messages
index:
name: mailbox_messages
key:
user: 1
mailbox: 1
- collection: messages
index:
name: user_messages_by_thread
@ -86,45 +64,50 @@ indexes:
thread: 1
- collection: messages
index:
# use also as sharding key
name: mailbox_uid
key:
user: 1
mailbox: 1
uid: 1
- collection: messages
index:
# several message related queries include the shard key values
name: mailbox_uid_shard
key:
_id: 1
mailbox: 1
uid: 1
- collection: messages
index:
name: newer_first
key:
mailbox: 1
uid: -1
- collection: messages
index:
name: mailbox_modseq_uid
key:
user: 1
mailbox: 1
modseq: 1
uid: 1
- collection: messages
index:
name: newer_first
key:
user: 1
mailbox: 1
uid: -1
- collection: messages
index:
name: mailbox_flags
key:
user: 1
mailbox: 1
flags: 1
- collection: messages
index:
name: by_modseq
key:
user: 1
mailbox: 1
modseq: 1
- collection: messages
index:
name: by_idate
key:
user: 1
mailbox: 1
idate: 1
_id: 1
@ -132,7 +115,6 @@ indexes:
index:
name: by_idate_newer
key:
user: 1
mailbox: 1
idate: -1
_id: -1
@ -140,7 +122,6 @@ indexes:
index:
name: by_hdate
key:
user: 1
mailbox: 1
hdate: 1
msgid: 1
@ -148,22 +129,18 @@ indexes:
index:
name: by_size
key:
user: 1
mailbox: 1
size: 1
- collection: messages
index:
name: by_headers
key:
user: 1
mailbox: 1
headers.key: 1
headers.value: 1
- collection: messages
index:
# there can be only one $text index per collection, so in order to make
# account wide searches we do not use mailbox as compound key element here.
# IMAP TEXT and BODY searches might be slower though
# there can be only one $text index per collection
name: fulltext
key:
user: 1
@ -176,35 +153,30 @@ indexes:
index:
name: mailbox_seen_flag
key:
user: 1
mailbox: 1
seen: 1
- collection: messages
index:
name: mailbox_deleted_flag
key:
user: 1
mailbox: 1
deleted: 1
- collection: messages
index:
name: mailbox_flagged_flag
key:
user: 1
mailbox: 1
flagged: 1
- collection: messages
index:
name: mailbox_draft_flag
key:
user: 1
mailbox: 1
draft: 1
- collection: messages
index:
name: has_attachment
key:
user: 1
mailbox: 1
ha: 1
- collection: messages

12
lib/consts.js Normal file
View file

@ -0,0 +1,12 @@
'use strict';
module.exports = {
// home many modifications to cache before writing
BULK_BATCH_SIZE: 150,
// how often to clear expired messages
GC_INTERVAL: 10 * 60 * 1000,
// artificail delay between deleting next expired message in ms
GC_DELAY_DELETE: 100
};

56
lib/handlers/on-append.js Normal file
View file

@ -0,0 +1,56 @@
'use strict';
const db = require('../db');
// APPEND mailbox (flags) date message
module.exports = (server, messageHandler) => (path, flags, date, raw, session, callback) => {
server.logger.debug(
{
tnx: 'append',
cid: session.id
},
'[%s] Appending message to "%s"',
session.id,
path
);
db.users.collection('users').findOne({
_id: session.user.id
}, (err, user) => {
if (err) {
return callback(err);
}
if (!user) {
return callback(new Error('User not found'));
}
if (user.quota && user.storageUsed > user.quota) {
return callback(false, 'OVERQUOTA');
}
messageHandler.add(
{
user: session.user.id,
path,
meta: {
source: 'IMAP',
to: session.user.username,
time: Date.now()
},
session,
date,
flags,
raw
},
(err, status, data) => {
if (err) {
if (err.imapResponse) {
return callback(null, err.imapResponse);
}
return callback(err);
}
callback(null, status, data);
}
);
});
};

34
lib/handlers/on-auth.js Normal file
View file

@ -0,0 +1,34 @@
'use strict';
module.exports = (server, userHandler) => (login, session, callback) => {
let username = (login.username || '').toString().trim();
userHandler.authenticate(
username,
login.password,
{
protocol: 'IMAP',
ip: session.remoteAddress
},
(err, result) => {
if (err) {
return callback(err);
}
if (!result) {
return callback();
}
if (result.scope === 'master' && result.enabled2fa) {
// master password not allowed if 2fa is enabled!
return callback();
}
callback(null, {
user: {
id: result.user,
username: result.username
}
});
}
);
};

189
lib/handlers/on-copy.js Normal file
View file

@ -0,0 +1,189 @@
'use strict';
const ObjectID = require('mongodb').ObjectID;
const db = require('../db');
// COPY / UID COPY sequence mailbox
module.exports = server => (path, update, session, callback) => {
server.logger.debug(
{
tnx: 'copy',
cid: session.id
},
'[%s] Copying messages from "%s" to "%s"',
session.id,
path,
update.destination
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
db.database.collection('mailboxes').findOne({
user: session.user.id,
path: update.destination
}, (err, target) => {
if (err) {
return callback(err);
}
if (!target) {
return callback(null, 'TRYCREATE');
}
let cursor = db.database
.collection('messages')
.find({
mailbox: mailbox._id,
uid: {
$in: update.messages
}
})
.sort([['uid', 1]]); // no projection as we need to copy the entire message
let copiedMessages = 0;
let copiedStorage = 0;
let updateQuota = next => {
if (!copiedMessages) {
return next();
}
db.users.collection('users').findOneAndUpdate(
{
_id: mailbox.user
},
{
$inc: {
storageUsed: copiedStorage
}
},
next
);
};
let sourceUid = [];
let destinationUid = [];
let processNext = () => {
cursor.next((err, message) => {
if (err) {
return updateQuota(() => callback(err));
}
if (!message) {
return cursor.close(() => {
updateQuota(() => {
server.notifier.fire(session.user.id, target.path);
return callback(null, true, {
uidValidity: target.uidValidity,
sourceUid,
destinationUid
});
});
});
}
// Copying is not done in bulk to minimize risk of going out of sync with incremental UIDs
sourceUid.unshift(message.uid);
db.database.collection('mailboxes').findOneAndUpdate({
_id: target._id
}, {
$inc: {
uidNext: 1
}
}, {
uidNext: true
}, (err, item) => {
if (err) {
return cursor.close(() => {
updateQuota(() => callback(err));
});
}
if (!item || !item.value) {
// was not able to acquire a lock
return cursor.close(() => {
updateQuota(() => callback(null, 'TRYCREATE'));
});
}
let uidNext = item.value.uidNext;
destinationUid.unshift(uidNext);
message._id = new ObjectID();
message.mailbox = target._id;
message.uid = uidNext;
// retention settings
message.exp = !!target.retention;
message.rdate = Date.now() + (target.retention || 0);
if (!message.meta) {
message.meta = {};
}
message.meta.source = 'IMAPCOPY';
db.database.collection('messages').insertOne(message, err => {
if (err) {
return cursor.close(() => {
updateQuota(() => callback(err));
});
}
copiedMessages++;
copiedStorage += Number(message.size) || 0;
let attachments = Object.keys(message.map || {}).map(key => message.map[key]);
if (!attachments.length) {
return server.notifier.addEntries(
session.user.id,
target.path,
{
command: 'EXISTS',
uid: message.uid,
message: message._id
},
processNext
);
}
// update attachments
db.gridfs.collection('attachments.files').updateMany({
_id: {
$in: attachments
}
}, {
$inc: {
'metadata.c': 1,
'metadata.m': message.magic
}
}, {
multi: true,
w: 1
}, err => {
if (err) {
// should we care about this error?
}
server.notifier.addEntries(
session.user.id,
target.path,
{
command: 'EXISTS',
uid: message.uid,
message: message._id
},
processNext
);
});
});
});
});
};
processNext();
});
});
};

57
lib/handlers/on-create.js Normal file
View file

@ -0,0 +1,57 @@
'use strict';
const db = require('../db');
// CREATE "path/to/mailbox"
module.exports = server => (path, session, callback) => {
server.logger.debug(
{
tnx: 'create',
cid: session.id
},
'[%s] CREATE "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (mailbox) {
return callback(null, 'ALREADYEXISTS');
}
db.users.collection('users').findOne({
_id: session.user.id
}, {
fields: {
retention: true
}
}, (err, user) => {
if (err) {
return callback(err);
}
mailbox = {
user: session.user.id,
path,
uidValidity: Math.floor(Date.now() / 1000),
uidNext: 1,
modifyIndex: 0,
subscribed: true,
flags: [],
retention: user.retention
};
db.database.collection('mailboxes').insertOne(mailbox, err => {
if (err) {
return callback(err);
}
return callback(null, true);
});
});
});
};

109
lib/handlers/on-delete.js Normal file
View file

@ -0,0 +1,109 @@
'use strict';
const db = require('../db');
// DELETE "path/to/mailbox"
module.exports = server => (path, session, callback) => {
server.logger.debug(
{
tnx: 'delete',
cid: session.id
},
'[%s] DELETE "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
if (mailbox.specialUse) {
return callback(null, 'CANNOT');
}
db.database.collection('mailboxes').deleteOne({
_id: mailbox._id
}, err => {
if (err) {
return callback(err);
}
// calculate mailbox size by aggregating the size's of all messages
db.database
.collection('messages')
.aggregate(
[
{
$match: {
mailbox: mailbox._id
}
},
{
$group: {
_id: {
mailbox: '$mailbox'
},
storageUsed: {
$sum: '$size'
}
}
}
],
{
cursor: {
batchSize: 1
}
}
)
.toArray((err, res) => {
if (err) {
return callback(err);
}
let storageUsed = (res && res[0] && res[0].storageUsed) || 0;
db.database.collection('messages').deleteMany({
mailbox: mailbox._id
}, err => {
if (err) {
return callback(err);
}
let done = () => {
db.database.collection('journal').deleteMany({
mailbox: mailbox._id
}, err => {
if (err) {
return callback(err);
}
callback(null, true);
});
};
if (!storageUsed) {
return done();
}
// decrement quota counters
db.users.collection('users').findOneAndUpdate(
{
_id: mailbox.user
},
{
$inc: {
storageUsed: -Number(storageUsed) || 0
}
},
done
);
});
});
});
});
};

146
lib/handlers/on-expunge.js Normal file
View file

@ -0,0 +1,146 @@
'use strict';
const db = require('../db');
// EXPUNGE deletes all messages in selected mailbox marked with \Delete
module.exports = server => (path, update, session, callback) => {
server.logger.debug(
{
tnx: 'expunge',
cid: session.id
},
'[%s] Deleting messages from "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
let cursor = db.database
.collection('messages')
.find({
user: session.user.id,
mailbox: mailbox._id,
deleted: true
})
.project({
_id: true,
uid: true,
size: true,
map: true,
magic: true
})
.sort([['uid', 1]]);
let deletedMessages = 0;
let deletedStorage = 0;
let updateQuota = next => {
if (!deletedMessages) {
return next();
}
db.users.collection('users').findOneAndUpdate(
{
_id: mailbox.user
},
{
$inc: {
storageUsed: -deletedStorage
}
},
next
);
};
let processNext = () => {
cursor.next((err, message) => {
if (err) {
return updateQuota(() => callback(err));
}
if (!message) {
return cursor.close(() => {
updateQuota(() => {
server.notifier.fire(session.user.id, path);
return callback(null, true);
});
});
}
if (!update.silent) {
session.writeStream.write(session.formatResponse('EXPUNGE', message.uid));
}
db.database.collection('messages').deleteOne({
_id: message._id,
mailbox: mailbox._id,
uid: message.uid
}, err => {
if (err) {
return updateQuota(() => cursor.close(() => callback(err)));
}
deletedMessages++;
deletedStorage += Number(message.size) || 0;
let attachments = Object.keys(message.map || {}).map(key => message.map[key]);
if (!attachments.length) {
// not stored attachments
return server.notifier.addEntries(
session.user.id,
path,
{
command: 'EXPUNGE',
ignore: session.id,
uid: message.uid,
message: message._id
},
processNext
);
}
// remove references to attachments (if any exist)
db.gridfs.collection('attachments.files').updateMany({
_id: {
$in: attachments
}
}, {
$inc: {
'metadata.c': -1,
'metadata.m': -message.magic
}
}, {
multi: true,
w: 1
}, err => {
if (err) {
// ignore as we don't really care if we have orphans or not
}
server.notifier.addEntries(
session.user.id,
path,
{
command: 'EXPUNGE',
ignore: session.id,
uid: message.uid,
message: message._id
},
processNext
);
});
});
});
};
processNext();
});
};

207
lib/handlers/on-fetch.js Normal file
View file

@ -0,0 +1,207 @@
'use strict';
const IMAPServerModule = require('../../imap-core');
const imapHandler = IMAPServerModule.imapHandler;
const util = require('util');
const db = require('../db');
const consts = require('../consts');
module.exports = server => (path, options, session, callback) => {
server.logger.debug(
{
tnx: 'fetch',
cid: session.id
},
'[%s] Requested FETCH for "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
let projection = {
uid: true,
modseq: true,
idate: true,
flags: true,
envelope: true,
bodystructure: true,
size: true
};
if (!options.metadataOnly) {
projection.mimeTree = true;
}
let query = {
mailbox: mailbox._id
};
if (options.changedSince) {
query = {
mailbox: mailbox._id,
modseq: {
$gt: options.changedSince
}
};
}
let queryAll = false;
if (options.messages.length !== session.selected.uidList.length) {
// do not use uid selector for 1:*
query.uid = {
$in: options.messages
};
} else {
// 1:*
queryAll = true;
}
let isUpdated = false;
let updateEntries = [];
let notifyEntries = [];
let done = (...args) => {
if (updateEntries.length) {
return db.database.collection('messages').bulkWrite(updateEntries, {
ordered: false,
w: 1
}, () => {
updateEntries = [];
server.notifier.addEntries(session.user.id, path, notifyEntries, () => {
notifyEntries = [];
server.notifier.fire(session.user.id, path);
return callback(...args);
});
});
}
if (isUpdated) {
server.notifier.fire(session.user.id, path);
}
return callback(...args);
};
let cursor = db.database.collection('messages').find(query).project(projection).sort([['uid', 1]]);
let rowCount = 0;
let processNext = () => {
cursor.next((err, message) => {
if (err) {
return done(err);
}
if (!message) {
return cursor.close(() => {
done(null, true);
});
}
if (queryAll && !session.selected.uidList.includes(message.uid)) {
// skip processing messages that we do not know about yet
return processNext();
}
let markAsSeen = options.markAsSeen && !message.flags.includes('\\Seen');
if (markAsSeen) {
message.flags.unshift('\\Seen');
}
let stream = imapHandler.compileStream(
session.formatResponse('FETCH', message.uid, {
query: options.query,
values: session.getQueryResponse(options.query, message, {
logger: server.logger,
fetchOptions: {},
database: db.database,
gridfs: db.gridfs,
acceptUTF8Enabled: session.isUTF8Enabled()
})
})
);
stream.description = util.format('* FETCH #%s uid=%s size=%sB ', ++rowCount, message.uid, message.size);
stream.on('error', err => {
session.socket.write('* BYE INTERNAL ERROR\n');
session.socket.destroy(); // ended up in erroneus state, kill the connection to abort
return cursor.close(() => done(err));
});
// send formatted response to socket
session.writeStream.write(stream, () => {
if (!markAsSeen) {
return processNext();
}
server.logger.debug(
{
tnx: 'flags',
cid: session.id
},
'[%s] UPDATE FLAGS for "%s"',
session.id,
message.uid
);
isUpdated = true;
updateEntries.push({
updateOne: {
filter: {
_id: message._id,
// include sharding key in query
mailbox: mailbox._id,
uid: message.uid
},
update: {
$addToSet: {
flags: '\\Seen'
},
$set: {
seen: true
}
}
}
});
notifyEntries.push({
command: 'FETCH',
ignore: session.id,
uid: message.uid,
flags: message.flags,
message: message._id
});
if (updateEntries.length >= consts.BULK_BATCH_SIZE) {
return db.database.collection('messages').bulkWrite(updateEntries, {
ordered: false,
w: 1
}, err => {
updateEntries = [];
if (err) {
return cursor.close(() => done(err));
}
server.notifier.addEntries(session.user.id, path, notifyEntries, () => {
notifyEntries = [];
server.notifier.fire(session.user.id, path);
processNext();
});
});
} else {
processNext();
}
});
});
};
processNext();
});
};

44
lib/handlers/on-get-quota-root.js Normal file
View file

@ -0,0 +1,44 @@
'use strict';
const db = require('../db');
module.exports = server => (path, session, callback) => {
server.logger.debug(
{
tnx: 'quota',
cid: session.id
},
'[%s] Requested quota root info for "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
db.users.collection('users').findOne({
_id: session.user.id
}, (err, user) => {
if (err) {
return callback(err);
}
if (!user) {
return callback(new Error('User data not found'));
}
return callback(null, {
root: '',
quota: user.quota || server.options.maxStorage || 0,
storageUsed: Math.max(user.storageUsed || 0, 0)
});
});
});
};

36
lib/handlers/on-get-quota.js Normal file
View file

@ -0,0 +1,36 @@
'use strict';
const db = require('../db');
module.exports = server => (quotaRoot, session, callback) => {
server.logger.debug(
{
tnx: 'quota',
cid: session.id
},
'[%s] Requested quota info for "%s"',
session.id,
quotaRoot
);
if (quotaRoot !== '') {
return callback(null, 'NONEXISTENT');
}
db.users.collection('users').findOne({
_id: session.user.id
}, (err, user) => {
if (err) {
return callback(err);
}
if (!user) {
return callback(new Error('User data not found'));
}
return callback(null, {
root: '',
quota: user.quota || server.options.maxStorage || 0,
storageUsed: Math.max(user.storageUsed || 0, 0)
});
});
};

25
lib/handlers/on-list.js Normal file
View file

@ -0,0 +1,25 @@
'use strict';
const db = require('../db');
// LIST "" "*"
// Returns all folders, query is informational
// folders is either an Array or a Map
module.exports = server =>
(server.onList = function(query, session, callback) {
server.logger.debug(
{
tnx: 'list',
cid: session.id
},
'[%s] LIST for "%s"',
session.id,
query
);
db.database
.collection('mailboxes')
.find({
user: session.user.id
})
.toArray(callback);
});

25
lib/handlers/on-lsub.js Normal file
View file

@ -0,0 +1,25 @@
'use strict';
const db = require('../db');
// LSUB "" "*"
// Returns all subscribed folders, query is informational
// folders is either an Array or a Map
module.exports = server => (query, session, callback) => {
server.logger.debug(
{
tnx: 'lsub',
cid: session.id
},
'[%s] LSUB for "%s"',
session.id,
query
);
db.database
.collection('mailboxes')
.find({
user: session.user.id,
subscribed: true
})
.toArray(callback);
};

43
lib/handlers/on-move.js Normal file
View file

@ -0,0 +1,43 @@
'use strict';
// MOVE / UID MOVE sequence mailbox
module.exports = (server, messageHandler) => (path, update, session, callback) => {
server.logger.debug(
{
tnx: 'move',
cid: session.id
},
'[%s] Moving messages from "%s" to "%s"',
session.id,
path,
update.destination
);
messageHandler.move(
{
user: session.user.id,
// folder to move messages from
source: {
user: session.user.id,
path
},
// folder to move messages to
destination: {
user: session.user.id,
path: update.destination
},
session,
// list of UIDs to move
messages: update.messages
},
(...args) => {
if (args[0]) {
if (args[0].imapResponse) {
return callback(null, args[0].imapResponse);
}
return callback(args[0]);
}
callback(...args);
}
);
};

44
lib/handlers/on-open.js Normal file
View file

@ -0,0 +1,44 @@
'use strict';
const db = require('../db');
// SELECT/EXAMINE
module.exports = server => (path, session, callback) => {
server.logger.debug(
{
tnx: 'open',
cid: session.id
},
'[%s] Opening "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
db.database
.collection('messages')
.find({
mailbox: mailbox._id
})
.project({
uid: true
})
.sort([['uid', 1]])
.toArray((err, messages) => {
if (err) {
return callback(err);
}
mailbox.uidList = messages.map(message => message.uid);
callback(null, mailbox);
});
});
};

49
lib/handlers/on-rename.js Normal file
View file

@ -0,0 +1,49 @@
'use strict';
const db = require('../db');
// RENAME "path/to/mailbox" "new/path"
// NB! RENAME affects child and hierarchy mailboxes as well, this example does not do this
module.exports = server => (path, newname, session, callback) => {
server.logger.debug(
{
tnx: 'rename',
cid: session.id
},
'[%s] RENAME "%s" to "%s"',
session.id,
path,
newname
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path: newname
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (mailbox) {
return callback(null, 'ALREADYEXISTS');
}
db.database.collection('mailboxes').findOneAndUpdate({
user: session.user.id,
path
}, {
$set: {
path: newname
}
}, {}, (err, item) => {
if (err) {
return callback(err);
}
if (!item || !item.value) {
// was not able to acquire a lock
return callback(null, 'NONEXISTENT');
}
callback(null, true);
});
});
};

366
lib/handlers/on-search.js Normal file
View file

@ -0,0 +1,366 @@
'use strict';
const db = require('../db');
/**
* Returns an array of matching UID values
*/
module.exports = server => (path, options, session, callback) => {
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
// prepare query
let query = {
mailbox: mailbox._id
};
let walkQuery = (parent, ne, node) => {
node.forEach(term => {
switch (term.key) {
case 'all':
if (ne) {
parent.push({
// should not match anything
_id: -1
});
}
break;
case 'not':
walkQuery(parent, !ne, [].concat(term.value || []));
break;
case 'or': {
let $or = [];
[].concat(term.value || []).forEach(entry => {
walkQuery($or, false, [].concat(entry || []));
});
if ($or.length) {
parent.push({
$or
});
}
break;
}
case 'text': // search over entire email
case 'body': // search over email body
if (term.value && !ne) {
// fulltext can only be in the root of the query, not in $not, $or expressions
// https://docs.mongodb.com/v3.4/tutorial/text-search-in-aggregation/#restrictions
query.user = session.user.id;
query.$text = {
$search: term.value
};
} else {
// can not search by text
parent.push({
// should not match anything
_id: -1
});
}
break;
case 'modseq':
parent.push({
modseq: {
[!ne ? '$gte' : '$lt']: term.value
}
});
break;
case 'uid':
if (Array.isArray(term.value)) {
if (!term.value.length) {
// trying to find a message that does not exist
return callback(null, {
uidList: [],
highestModseq: 0
});
}
if (term.value.length !== session.selected.uidList.length) {
// not 1:*
parent.push({
uid: {
[!ne ? '$in' : '$nin']: term.value
}
});
} else if (ne) {
parent.push({
// should not match anything
_id: -1
});
}
} else {
parent.push({
uid: {
[!ne ? '$eq' : '$ne']: term.value
}
});
}
break;
case 'flag':
{
switch (term.value) {
case '\\Seen':
case '\\Deleted':
case '\\Flagged':
case '\\Draft':
if (term.exists) {
parent.push({
[term.value.toLowerCase().substr(1)]: !ne
});
} else {
parent.push({
[term.value.toLowerCase().substr(1)]: ne
});
}
break;
default:
if (term.exists) {
parent.push({
flags: {
[!ne ? '$eq' : '$ne']: term.value
}
});
} else {
parent.push({
flags: {
[!ne ? '$ne' : '$eq']: term.value
}
});
}
}
}
break;
case 'header':
{
let regex = Buffer.from(term.value, 'binary').toString().replace(/[-/\\^$*+?.()|[\]{}]/g, '\\$&');
let entry = term.value
? {
headers: {
$elemMatch: {
key: term.header,
value: !ne
? {
$regex: regex,
$options: 'i'
}
: {
$not: {
$regex: regex,
$options: 'i'
}
}
}
}
}
: {
'headers.key': !ne
? term.header
: {
$ne: term.header
}
};
parent.push(entry);
}
break;
case 'internaldate':
{
let op = false;
let value = new Date(term.value + ' GMT');
switch (term.operator) {
case '<':
op = '$lt';
break;
case '<=':
op = '$lte';
break;
case '>':
op = '$gt';
break;
case '>=':
op = '$gte';
break;
}
let entry = !op
? [
{
$gte: value
},
{
$lt: new Date(value.getTime() + 24 * 3600 * 1000)
}
]
: {
[op]: value
};
entry = {
idate: !ne
? entry
: {
$not: entry
}
};
parent.push(entry);
}
break;
case 'headerdate':
{
let op = false;
let value = new Date(term.value + ' GMT');
switch (term.operator) {
case '<':
op = '$lt';
break;
case '<=':
op = '$lte';
break;
case '>':
op = '$gt';
break;
case '>=':
op = '$gte';
break;
}
let entry = !op
? [
{
$gte: value
},
{
$lt: new Date(value.getTime() + 24 * 3600 * 1000)
}
]
: {
[op]: value
};
entry = {
hdate: !ne
? entry
: {
$not: entry
}
};
parent.push(entry);
}
break;
case 'size':
{
let op = '$eq';
let value = Number(term.value) || 0;
switch (term.operator) {
case '<':
op = '$lt';
break;
case '<=':
op = '$lte';
break;
case '>':
op = '$gt';
break;
case '>=':
op = '$gte';
break;
}
let entry = {
[op]: value
};
entry = {
size: !ne
? entry
: {
$not: entry
}
};
parent.push(entry);
}
break;
}
});
};
let $and = [];
walkQuery($and, false, options.query);
if ($and.length) {
query.$and = $and;
}
server.logger.info(
{
tnx: 'search',
cid: session.id
},
'[%s] SEARCH %s',
session.id,
JSON.stringify(query)
);
let cursor = db.database.collection('messages').find(query).project({
uid: true,
modseq: true
});
let highestModseq = 0;
let uidList = [];
let processNext = () => {
cursor.next((err, message) => {
if (err) {
server.logger.error(
{
tnx: 'search',
cid: session.id
},
'[%s] SEARCHFAIL %s error="%s"',
session.id,
JSON.stringify(query),
err.message
);
return callback(new Error('Can not make requested search query'));
}
if (!message) {
return cursor.close(() =>
callback(null, {
uidList,
highestModseq
})
);
}
if (highestModseq < message.modseq) {
highestModseq = message.modseq;
}
uidList.push(message.uid);
processNext();
});
};
processNext();
});
};

56
lib/handlers/on-status.js Normal file
View file

@ -0,0 +1,56 @@
'use strict';
const db = require('../db');
// STATUS (X Y X)
module.exports = server => (path, session, callback) => {
server.logger.debug(
{
tnx: 'status',
cid: session.id
},
'[%s] Requested status for "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
db.database
.collection('messages')
.find({
mailbox: mailbox._id
})
.count((err, total) => {
if (err) {
return callback(err);
}
db.database
.collection('messages')
.find({
mailbox: mailbox._id,
seen: false
})
.count((err, unseen) => {
if (err) {
return callback(err);
}
return callback(null, {
messages: total,
uidNext: mailbox.uidNext,
uidValidity: mailbox.uidValidity,
unseen
});
});
});
});
};

355
lib/handlers/on-store.js Normal file
View file

@ -0,0 +1,355 @@
'use strict';
const imapTools = require('../../imap-core/lib/imap-tools');
const db = require('../db');
const consts = require('../consts');
// STORE / UID STORE, updates flags for selected UIDs
module.exports = server => (path, update, session, callback) => {
server.logger.debug(
{
tnx: 'store',
cid: session.id
},
'[%s] Updating messages in "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOne({
user: session.user.id,
path
}, (err, mailbox) => {
if (err) {
return callback(err);
}
if (!mailbox) {
return callback(null, 'NONEXISTENT');
}
let query = {
mailbox: mailbox._id
};
if (update.unchangedSince) {
query = {
mailbox: mailbox._id,
modseq: {
$lte: update.unchangedSince
}
};
}
let queryAll = false;
if (update.messages.length !== session.selected.uidList.length) {
// do not use uid selector for 1:*
query.uid = {
$in: update.messages
};
} else {
// 1:*
queryAll = true;
}
let cursor = db.database
.collection('messages')
.find(query)
.project({
_id: true,
uid: true,
flags: true
})
.sort([['uid', 1]]);
let updateEntries = [];
let notifyEntries = [];
let done = (...args) => {
if (updateEntries.length) {
return db.database.collection('messages').bulkWrite(updateEntries, {
ordered: false,
w: 1
}, () => {
updateEntries = [];
server.notifier.addEntries(session.user.id, path, notifyEntries, () => {
notifyEntries = [];
server.notifier.fire(session.user.id, path);
if (args[0]) {
// first argument is an error
return callback(...args);
} else {
updateMailboxFlags(mailbox, update, () => callback(...args));
}
});
});
}
server.notifier.fire(session.user.id, path);
if (args[0]) {
// first argument is an error
return callback(...args);
} else {
updateMailboxFlags(mailbox, update, () => callback(...args));
}
};
// We have to process all messages one by one instead of just calling an update
// for all messages as we need to know which messages were exactly modified,
// otherwise we can't send flag update notifications and modify modseq values
let processNext = () => {
cursor.next((err, message) => {
if (err) {
return done(err);
}
if (!message) {
return cursor.close(() => done(null, true));
}
if (queryAll && !session.selected.uidList.includes(message.uid)) {
// skip processing messages that we do not know about yet
return processNext();
}
let flagsupdate = false; // query object for updates
let updated = false;
let existingFlags = message.flags.map(flag => flag.toLowerCase().trim());
switch (update.action) {
case 'set':
// check if update set matches current or is different
if (
// if length does not match
existingFlags.length !== update.value.length ||
// or a new flag was found
update.value.filter(flag => !existingFlags.includes(flag.toLowerCase().trim())).length
) {
updated = true;
}
message.flags = [].concat(update.value);
// set flags
if (updated) {
flagsupdate = {
$set: {
flags: message.flags,
seen: message.flags.includes('\\Seen'),
flagged: message.flags.includes('\\Flagged'),
deleted: message.flags.includes('\\Deleted'),
draft: message.flags.includes('\\Draft')
}
};
}
break;
case 'add': {
let newFlags = [];
message.flags = message.flags.concat(
update.value.filter(flag => {
if (!existingFlags.includes(flag.toLowerCase().trim())) {
updated = true;
newFlags.push(flag);
return true;
}
return false;
})
);
// add flags
if (updated) {
flagsupdate = {
$addToSet: {
flags: {
$each: newFlags
}
}
};
if (
newFlags.includes('\\Seen') ||
newFlags.includes('\\Flagged') ||
newFlags.includes('\\Deleted') ||
newFlags.includes('\\Draft')
) {
flagsupdate.$set = {};
if (newFlags.includes('\\Seen')) {
flagsupdate.$set = {
seen: true
};
}
if (newFlags.includes('\\Flagged')) {
flagsupdate.$set = {
flagged: true
};
}
if (newFlags.includes('\\Deleted')) {
flagsupdate.$set = {
deleted: true
};
}
if (newFlags.includes('\\Draft')) {
flagsupdate.$set = {
draft: true
};
}
}
}
break;
}
case 'remove': {
// We need to use the case of existing flags when removing
let oldFlags = [];
let flagsUpdates = update.value.map(flag => flag.toLowerCase().trim());
message.flags = message.flags.filter(flag => {
if (!flagsUpdates.includes(flag.toLowerCase().trim())) {
return true;
}
oldFlags.push(flag);
updated = true;
return false;
});
// remove flags
if (updated) {
flagsupdate = {
$pull: {
flags: {
$in: oldFlags
}
}
};
if (
oldFlags.includes('\\Seen') ||
oldFlags.includes('\\Flagged') ||
oldFlags.includes('\\Deleted') ||
oldFlags.includes('\\Draft')
) {
flagsupdate.$set = {};
if (oldFlags.includes('\\Seen')) {
flagsupdate.$set = {
seen: false
};
}
if (oldFlags.includes('\\Flagged')) {
flagsupdate.$set = {
flagged: false
};
}
if (oldFlags.includes('\\Deleted')) {
flagsupdate.$set = {
deleted: false
};
}
if (oldFlags.includes('\\Draft')) {
flagsupdate.$set = {
draft: false
};
}
}
}
break;
}
}
if (!update.silent) {
// print updated state of the message
session.writeStream.write(
session.formatResponse('FETCH', message.uid, {
uid: update.isUid ? message.uid : false,
flags: message.flags
})
);
}
if (updated) {
updateEntries.push({
updateOne: {
filter: {
_id: message._id,
// include shard key data as well
mailbox: mailbox._id,
uid: message.uid
},
update: flagsupdate
}
});
notifyEntries.push({
command: 'FETCH',
ignore: session.id,
uid: message.uid,
flags: message.flags,
message: message._id
});
if (updateEntries.length >= consts.BULK_BATCH_SIZE) {
return db.database.collection('messages').bulkWrite(updateEntries, {
ordered: false,
w: 1
}, err => {
updateEntries = [];
if (err) {
return cursor.close(() => done(err));
}
server.notifier.addEntries(session.user.id, path, notifyEntries, () => {
notifyEntries = [];
server.notifier.fire(session.user.id, path);
processNext();
});
});
} else {
processNext();
}
} else {
processNext();
}
});
};
processNext();
});
};
function updateMailboxFlags(mailbox, update, callback) {
if (update.action === 'remove') {
// we didn't add any new flags, so there's nothing to update
return callback();
}
let mailboxFlags = imapTools.systemFlags.concat(mailbox.flags || []).map(flag => flag.trim().toLowerCase());
let newFlags = [];
// find flags that are not listed with mailbox
update.value.forEach(flag => {
// limit mailbox flags by 100
if (mailboxFlags.length + newFlags.length >= 100) {
return;
}
// if mailbox does not have such flag, then add it
if (!mailboxFlags.includes(flag.toLowerCase().trim())) {
newFlags.push(flag);
}
});
// nothing new found
if (!newFlags.length) {
return callback();
}
// found some new flags not yet set for mailbox
// FIXME: Should we send unsolicited FLAGS and PERMANENTFLAGS notifications? Probably not
return db.database.collection('mailboxes').findOneAndUpdate(
{
_id: mailbox._id
},
{
$addToSet: {
flags: {
$each: newFlags
}
}
},
{},
callback
);
}

35
lib/handlers/on-subscribe.js Normal file
View file

@ -0,0 +1,35 @@
'use strict';
const db = require('../db');
// SUBSCRIBE "path/to/mailbox"
module.exports = server => (path, session, callback) => {
server.logger.debug(
{
tnx: 'subscribe',
cid: session.id
},
'[%s] SUBSCRIBE to "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOneAndUpdate({
user: session.user.id,
path
}, {
$set: {
subscribed: true
}
}, {}, (err, item) => {
if (err) {
return callback(err);
}
if (!item || !item.value) {
// was not able to acquire a lock
return callback(null, 'NONEXISTENT');
}
callback(null, true);
});
};

35
lib/handlers/on-unsubscribe.js Normal file
View file

@ -0,0 +1,35 @@
'use strict';
const db = require('../db');
// UNSUBSCRIBE "path/to/mailbox"
module.exports = server => (path, session, callback) => {
server.logger.debug(
{
tnx: 'unsubscribe',
cid: session.id
},
'[%s] UNSUBSCRIBE from "%s"',
session.id,
path
);
db.database.collection('mailboxes').findOneAndUpdate({
user: session.user.id,
path
}, {
$set: {
subscribed: false
}
}, {}, (err, item) => {
if (err) {
return callback(err);
}
if (!item || !item.value) {
// was not able to acquire a lock
return callback(null, 'NONEXISTENT');
}
callback(null, true);
});
};

2
lib/imap-notifier.js
View file

@ -213,7 +213,7 @@ class ImapNotifier extends EventEmitter {
_id: {
$in: updated
},
user: mailbox.user
mailbox: mailbox._id
}, {
// only update modseq if the new value is larger than old one
$max: {

119
lib/message-handler.js
View file

@ -317,7 +317,6 @@ class MessageHandler {
checkExistingMessage(user, mailboxId, message, options, callback) {
// if a similar message already exists then update existing one
this.database.collection('messages').findOne({
user,
mailbox: mailboxId,
hdate: message.hdate,
msgid: message.msgid
@ -372,7 +371,9 @@ class MessageHandler {
this.database.collection('messages').findOneAndUpdate({
_id: existing._id,
user: mailbox.user
// hash key
mailbox: mailbox._id,
uid: existing.uid
}, {
$set: {
uid,
@ -466,7 +467,8 @@ class MessageHandler {
this.database.collection('messages').deleteOne({
_id: message._id,
user: mailbox.user
mailbox: mailbox._id,
uid: message.uid
}, err => {
if (err) {
return callback(err);
@ -562,15 +564,11 @@ class MessageHandler {
let cursor = this.database
.collection('messages')
.find({
user: mailbox,
mailbox: mailbox._id,
uid: {
$in: options.messages || []
}
})
.project({
uid: 1
})
.sort([['uid', 1]]);
let sourceUid = [];
@ -615,7 +613,10 @@ class MessageHandler {
return cursor.close(done);
}
sourceUid.unshift(message.uid);
let messageId = message._id;
let messageUid = message.uid;
sourceUid.unshift(messageUid);
this.database.collection('mailboxes').findOneAndUpdate({
_id: target._id
}, {
@ -636,66 +637,74 @@ class MessageHandler {
let uidNext = item.value.uidNext;
destinationUid.unshift(uidNext);
let updateOptions = {
$set: {
mailbox: target._id,
// new mailbox means new UID
uid: uidNext,
// this will be changed later by the notification system
modseq: 0,
// set new mailbox
message.mailbox = target._id;
// retention settings
exp: !!target.retention,
rdate: Date.now() + (target.retention || 0)
}
};
// new mailbox means new UID
message.uid = uidNext;
// this will be changed later by the notification system
message.modseq = 0;
// retention settings
message.exp = !!target.retention;
message.rdate = Date.now() + (target.retention || 0);
if (options.markAsSeen) {
updateOptions.$set.seen = true;
updateOptions.$addToSet = {
flags: '\\Seen'
};
message.seen = true;
if (!message.flags.includes('\\Seen')) {
message.flags.push('\\Seen');
}
}
// update message, change mailbox from old to new one
this.database.collection('messages').findOneAndUpdate({
_id: message._id,
user: mailbox.user
}, updateOptions, err => {
this.database.collection('messages').insertOne(message, (err, r) => {
if (err) {
return cursor.close(() => done(err));
}
if (options.session) {
options.session.writeStream.write(options.session.formatResponse('EXPUNGE', message.uid));
}
let insertId = r.insertedId;
removeEntries.push({
command: 'EXPUNGE',
ignore: options.session && options.session.id,
uid: message.uid
});
// delete old message
this.database.collection('messages').deleteOne({
_id: messageId,
mailbox: mailbox._id,
uid: messageUid
}, err => {
if (err) {
return cursor.close(() => done(err));
}
existsEntries.push({
command: 'EXISTS',
uid: uidNext,
message: message._id
});
if (options.session) {
options.session.writeStream.write(options.session.formatResponse('EXPUNGE', sourceUid));
}
if (existsEntries.length >= BULK_BATCH_SIZE) {
// mark messages as deleted from old mailbox
return this.notifier.addEntries(mailbox, false, removeEntries, () => {
// mark messages as added to new mailbox
this.notifier.addEntries(target, false, existsEntries, () => {
removeEntries = [];
existsEntries = [];
this.notifier.fire(mailbox.user, mailbox.path);
this.notifier.fire(target.user, target.path);
processNext();
});
removeEntries.push({
command: 'EXPUNGE',
ignore: options.session && options.session.id,
uid: sourceUid
});
}
processNext();
existsEntries.push({
command: 'EXISTS',
uid: uidNext,
message: insertId
});
if (existsEntries.length >= BULK_BATCH_SIZE) {
// mark messages as deleted from old mailbox
return this.notifier.addEntries(mailbox, false, removeEntries, () => {
// mark messages as added to new mailbox
this.notifier.addEntries(target, false, existsEntries, () => {
removeEntries = [];
existsEntries = [];
this.notifier.fire(mailbox.user, mailbox.path);
this.notifier.fire(target.user, target.path);
processNext();
});
});
}
processNext();
});
});
});
});

4
lib/pop3-connection.js
View file

@ -626,7 +626,7 @@ class POP3Connection extends EventEmitter {
return next();
}
this._server.onFetchMessage(message.id, this.session, (err, stream) => {
this._server.onFetchMessage(message, this.session, (err, stream) => {
if (err) {
return next(err);
}
@ -676,7 +676,7 @@ class POP3Connection extends EventEmitter {
return next();
}
this._server.onFetchMessage(message.id, this.session, (err, stream) => {
this._server.onFetchMessage(message, this.session, (err, stream) => {
if (err) {
return next(err);
}

2
lib/pop3-server.js
View file

@ -171,7 +171,7 @@ class POP3Server extends EventEmitter {
}
// called when a message body needs to be fetched
onFetchMessage(id, session, callback) {
onFetchMessage(message, session, callback) {
// should return a stream object
return callback(null, false);
}

11
pop3.js
View file

@ -84,12 +84,12 @@ const serverOptions = {
db.database
.collection('messages')
.find({
user: session.user.id,
mailbox: mailbox._id
})
.project({
uid: true,
size: true,
mailbox: true,
// required to decide if we need to update flags after RETR
flags: true,
seen: true
@ -109,6 +109,7 @@ const serverOptions = {
.map(message => ({
id: message._id.toString(),
uid: message.uid,
mailbox: message.mailbox,
size: message.size,
flags: message.flags,
seen: message.seen
@ -120,10 +121,12 @@ const serverOptions = {
});
},
onFetchMessage(id, session, callback) {
onFetchMessage(message, session, callback) {
db.database.collection('messages').findOne({
_id: new ObjectID(id),
user: session.user.id
_id: new ObjectID(message.id),
// shard key
mailbox: message.mailbox,
uid: message.uid
}, {
mimeTree: true,
size: true