mirror of
https://github.com/nodemailer/wildduck.git
synced 2024-09-20 15:26:03 +08:00
Remove or skip duplicate messages
This commit is contained in:
Andris Reinman
parent
2d43b9c79a
commit
dcc3181891
4
api.js
4
api.js
|
|
@ -1131,7 +1131,9 @@ server.del('/message/:id', (req, res, next) => {
|
|||
query.mailbox = new ObjectID(mailbox);
|
||||
}
|
||||
|
||||
messageHandler.del(query, (err, success) => {
|
||||
messageHandler.del({
|
||||
query
|
||||
}, (err, success) => {
|
||||
if (err) {
|
||||
res.json({
|
||||
error: 'MongoDB Error: ' + err.message,
|
||||
|
|
|
|||
|
|
@ -33,8 +33,8 @@ let queryHandlers = {
|
|||
// matches message header date
|
||||
date(message, query, callback) {
|
||||
let date;
|
||||
if (message.headerdate) {
|
||||
date = message.headerdate;
|
||||
if (message.hdate) {
|
||||
date = message.hdate;
|
||||
} else {
|
||||
let mimeTree = message.mimeTree;
|
||||
if (!mimeTree) {
|
||||
|
|
|
|||
3
imap.js
3
imap.js
|
|
@ -422,6 +422,7 @@ server.onAppend = function (path, flags, date, raw, session, callback) {
|
|||
to: session.user.username,
|
||||
time: Date.now()
|
||||
},
|
||||
session,
|
||||
date,
|
||||
flags,
|
||||
raw
|
||||
|
|
@ -1425,7 +1426,7 @@ server.onSearch = function (path, options, session, callback) {
|
|||
};
|
||||
|
||||
entry = {
|
||||
headerdate: !ne ? entry : {
|
||||
hdate: !ne ? entry : {
|
||||
$not: entry
|
||||
}
|
||||
};
|
||||
|
|
|
|||
|
|
@ -97,10 +97,11 @@
|
|||
"internaldate": 1
|
||||
}
|
||||
}, {
|
||||
"name": "by_headerdate",
|
||||
"name": "by_hdate",
|
||||
"key": {
|
||||
"mailbox": 1,
|
||||
"headerdate": 1
|
||||
"hdate": 1,
|
||||
"msgid": 1
|
||||
}
|
||||
}, {
|
||||
"name": "by_size",
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@ class MessageHandler {
|
|||
let query = {};
|
||||
if (options.mailbox) {
|
||||
if (typeof options.mailbox === 'object' && options.mailbox._id) {
|
||||
return setImmediate(null, options.mailbox);
|
||||
return setImmediate(() => callback(null, options.mailbox));
|
||||
}
|
||||
query._id = options.mailbox;
|
||||
} else {
|
||||
|
|
@ -65,7 +65,16 @@ class MessageHandler {
|
|||
let bodystructure = this.indexer.getBodyStructure(mimeTree);
|
||||
let envelope = this.indexer.getEnvelope(mimeTree);
|
||||
|
||||
let messageId = envelope[9] || ('<' + uuidV1() + '@wildduck.email>');
|
||||
let internaldate = options.date && new Date(options.date) || new Date();
|
||||
let hdate = mimeTree.parsedHeader.date && new Date(mimeTree.parsedHeader.date) || false;
|
||||
|
||||
let flags = [].concat(options.flags || []);
|
||||
|
||||
if (!hdate || hdate.toString() === 'Invalid Date') {
|
||||
hdate = internaldate;
|
||||
}
|
||||
|
||||
let msgid = envelope[9] || ('<' + uuidV1() + '@wildduck.email>');
|
||||
|
||||
let headers = (mimeTree.header || []).map(line => {
|
||||
line = Buffer.from(line, 'binary').toString();
|
||||
|
|
@ -104,164 +113,199 @@ class MessageHandler {
|
|||
return callback(err);
|
||||
}
|
||||
|
||||
this.indexer.processContent(id, mimeTree, (err, maildata) => {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
let internaldate = options.date && new Date(options.date) || new Date();
|
||||
let headerdate = mimeTree.parsedHeader.date && new Date(mimeTree.parsedHeader.date) || false;
|
||||
|
||||
let flags = [].concat(options.flags || []);
|
||||
|
||||
if (!headerdate || headerdate.toString() === 'Invalid Date') {
|
||||
headerdate = internaldate;
|
||||
}
|
||||
|
||||
// prepare message object
|
||||
let message = {
|
||||
_id: id,
|
||||
|
||||
internaldate,
|
||||
headerdate,
|
||||
flags,
|
||||
size,
|
||||
|
||||
meta: options.meta || {},
|
||||
|
||||
headers,
|
||||
mimeTree,
|
||||
envelope,
|
||||
bodystructure,
|
||||
messageId,
|
||||
|
||||
// use boolean for more common flags
|
||||
seen: flags.includes('\\Seen'),
|
||||
flagged: flags.includes('\\Flagged'),
|
||||
deleted: flags.includes('\\Deleted')
|
||||
};
|
||||
|
||||
if (maildata.attachments && maildata.attachments.length) {
|
||||
message.attachments = maildata.attachments;
|
||||
message.hasAttachments = true;
|
||||
} else {
|
||||
message.hasAttachments = false;
|
||||
}
|
||||
|
||||
let maxTextLength = 300 * 1024;
|
||||
|
||||
if (maildata.text) {
|
||||
message.text = maildata.text.replace(/\r\n/g, '\n').trim();
|
||||
message.text = message.text.length <= maxTextLength ? message.text : message.text.substr(0, maxTextLength);
|
||||
message.intro = message.text.replace(/\s+/g, ' ').trim();
|
||||
if (message.intro.length > 128) {
|
||||
message.intro = message.intro.substr(0, 128) + '…';
|
||||
}
|
||||
}
|
||||
|
||||
if (maildata.html && maildata.html.length) {
|
||||
let htmlSize = 0;
|
||||
message.html = maildata.html.map(html => {
|
||||
if (htmlSize >= maxTextLength || !html) {
|
||||
return '';
|
||||
}
|
||||
|
||||
if (htmlSize + Buffer.byteLength(html) <= maxTextLength) {
|
||||
htmlSize += Buffer.byteLength(html);
|
||||
return html;
|
||||
}
|
||||
|
||||
html = html.substr(0, htmlSize + Buffer.byteLength(html) - maxTextLength);
|
||||
htmlSize += Buffer.byteLength(html);
|
||||
return html;
|
||||
}).filter(html => html);
|
||||
}
|
||||
|
||||
// Another server might be waiting for the lock
|
||||
this.redlock.waitAcquireLock(mailbox._id.toString(), 30 * 1000, 10 * 1000, (err, lock) => {
|
||||
// if a similar message already exists then delete the existing one
|
||||
let checkExisting = next => {
|
||||
this.database.collection('messages').findOne({
|
||||
mailbox: mailbox._id,
|
||||
hdate,
|
||||
msgid
|
||||
}, (err, existing) => {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
if (!lock || !lock.success) {
|
||||
// did not get a insert lock in 10 seconds
|
||||
return callback(new Error('The user you are trying to contact is receiving mail at a rate that prevents additional messages from being delivered. Please resend your message at a later time'));
|
||||
if (!existing) {
|
||||
// nothing to do here
|
||||
return next();
|
||||
}
|
||||
|
||||
this.database.collection('users').findOneAndUpdate({
|
||||
_id: mailbox.user
|
||||
}, {
|
||||
$inc: {
|
||||
storageUsed: size
|
||||
}
|
||||
if (options.skipExisting) {
|
||||
// message already exists, just skip it
|
||||
return callback(null, false);
|
||||
}
|
||||
|
||||
// delete existing message
|
||||
this.del({
|
||||
query: {
|
||||
_id: existing._id
|
||||
},
|
||||
mailbox,
|
||||
session: options.session
|
||||
}, err => {
|
||||
if (err) {
|
||||
this.redlock.releaseLock(lock, () => false);
|
||||
return callback(err);
|
||||
}
|
||||
next();
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
checkExisting(() => {
|
||||
this.indexer.processContent(id, mimeTree, (err, maildata) => {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
// prepare message object
|
||||
let message = {
|
||||
_id: id,
|
||||
|
||||
internaldate,
|
||||
hdate,
|
||||
flags,
|
||||
size,
|
||||
|
||||
meta: options.meta || {},
|
||||
|
||||
headers,
|
||||
mimeTree,
|
||||
envelope,
|
||||
bodystructure,
|
||||
msgid,
|
||||
|
||||
// use boolean for more common flags
|
||||
seen: flags.includes('\\Seen'),
|
||||
flagged: flags.includes('\\Flagged'),
|
||||
deleted: flags.includes('\\Deleted')
|
||||
};
|
||||
|
||||
if (maildata.attachments && maildata.attachments.length) {
|
||||
message.attachments = maildata.attachments;
|
||||
message.hasAttachments = true;
|
||||
} else {
|
||||
message.hasAttachments = false;
|
||||
}
|
||||
|
||||
let maxTextLength = 300 * 1024;
|
||||
|
||||
if (maildata.text) {
|
||||
message.text = maildata.text.replace(/\r\n/g, '\n').trim();
|
||||
message.text = message.text.length <= maxTextLength ? message.text : message.text.substr(0, maxTextLength);
|
||||
message.intro = message.text.replace(/\s+/g, ' ').trim();
|
||||
if (message.intro.length > 128) {
|
||||
message.intro = message.intro.substr(0, 128) + '…';
|
||||
}
|
||||
}
|
||||
|
||||
if (maildata.html && maildata.html.length) {
|
||||
let htmlSize = 0;
|
||||
message.html = maildata.html.map(html => {
|
||||
if (htmlSize >= maxTextLength || !html) {
|
||||
return '';
|
||||
}
|
||||
|
||||
if (htmlSize + Buffer.byteLength(html) <= maxTextLength) {
|
||||
htmlSize += Buffer.byteLength(html);
|
||||
return html;
|
||||
}
|
||||
|
||||
html = html.substr(0, htmlSize + Buffer.byteLength(html) - maxTextLength);
|
||||
htmlSize += Buffer.byteLength(html);
|
||||
return html;
|
||||
}).filter(html => html);
|
||||
}
|
||||
|
||||
// Another server might be waiting for the lock
|
||||
this.redlock.waitAcquireLock(mailbox._id.toString(), 30 * 1000, 10 * 1000, (err, lock) => {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
let rollback = err => {
|
||||
this.database.collection('users').findOneAndUpdate({
|
||||
_id: mailbox.user
|
||||
}, {
|
||||
$inc: {
|
||||
storageUsed: -size
|
||||
}
|
||||
}, () => {
|
||||
this.redlock.releaseLock(lock, () => callback(err));
|
||||
});
|
||||
};
|
||||
if (!lock || !lock.success) {
|
||||
// did not get a insert lock in 10 seconds
|
||||
return callback(new Error('The user you are trying to contact is receiving mail at a rate that prevents additional messages from being delivered. Please resend your message at a later time'));
|
||||
}
|
||||
|
||||
// acquire new UID+MODSEQ
|
||||
this.database.collection('mailboxes').findOneAndUpdate({
|
||||
_id: mailbox._id
|
||||
this.database.collection('users').findOneAndUpdate({
|
||||
_id: mailbox.user
|
||||
}, {
|
||||
$inc: {
|
||||
// allocate bot UID and MODSEQ values so when journal is later sorted by
|
||||
// modseq then UIDs are always in ascending order
|
||||
uidNext: 1,
|
||||
modifyIndex: 1
|
||||
storageUsed: size
|
||||
}
|
||||
}, (err, item) => {
|
||||
}, err => {
|
||||
if (err) {
|
||||
return rollback(err);
|
||||
this.redlock.releaseLock(lock, () => false);
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
if (!item || !item.value) {
|
||||
// was not able to acquire a lock
|
||||
let err = new Error('Mailbox is missing');
|
||||
err.imapResponse = 'TRYCREATE';
|
||||
return rollback(err);
|
||||
}
|
||||
let rollback = err => {
|
||||
this.database.collection('users').findOneAndUpdate({
|
||||
_id: mailbox.user
|
||||
}, {
|
||||
$inc: {
|
||||
storageUsed: -size
|
||||
}
|
||||
}, () => {
|
||||
this.redlock.releaseLock(lock, () => callback(err));
|
||||
});
|
||||
};
|
||||
|
||||
let mailbox = item.value;
|
||||
|
||||
// updated message object by setting mailbox specific values
|
||||
message.mailbox = mailbox._id;
|
||||
message.user = mailbox.user;
|
||||
message.uid = mailbox.uidNext;
|
||||
message.modseq = mailbox.modifyIndex + 1;
|
||||
|
||||
this.database.collection('messages').insertOne(message, err => {
|
||||
// acquire new UID+MODSEQ
|
||||
this.database.collection('mailboxes').findOneAndUpdate({
|
||||
_id: mailbox._id
|
||||
}, {
|
||||
$inc: {
|
||||
// allocate bot UID and MODSEQ values so when journal is later sorted by
|
||||
// modseq then UIDs are always in ascending order
|
||||
uidNext: 1,
|
||||
modifyIndex: 1
|
||||
}
|
||||
}, (err, item) => {
|
||||
if (err) {
|
||||
return rollback(err);
|
||||
}
|
||||
|
||||
let uidValidity = mailbox.uidValidity;
|
||||
let uid = message.uid;
|
||||
if (!item || !item.value) {
|
||||
// was not able to acquire a lock
|
||||
let err = new Error('Mailbox is missing');
|
||||
err.imapResponse = 'TRYCREATE';
|
||||
return rollback(err);
|
||||
}
|
||||
|
||||
this.notifier.addEntries(mailbox, false, {
|
||||
command: 'EXISTS',
|
||||
uid: message.uid,
|
||||
message: message._id,
|
||||
modseq: message.modseq
|
||||
}, () => {
|
||||
let mailbox = item.value;
|
||||
|
||||
this.redlock.releaseLock(lock, () => {
|
||||
this.notifier.fire(mailbox.user, mailbox.path);
|
||||
return callback(null, true, {
|
||||
uidValidity,
|
||||
uid
|
||||
// updated message object by setting mailbox specific values
|
||||
message.mailbox = mailbox._id;
|
||||
message.user = mailbox.user;
|
||||
message.uid = mailbox.uidNext;
|
||||
message.modseq = mailbox.modifyIndex + 1;
|
||||
|
||||
this.database.collection('messages').insertOne(message, err => {
|
||||
if (err) {
|
||||
return rollback(err);
|
||||
}
|
||||
|
||||
let uidValidity = mailbox.uidValidity;
|
||||
let uid = message.uid;
|
||||
|
||||
if (options.session && options.session.selected && options.session.selected.mailbox === mailbox.path) {
|
||||
options.session.writeStream.write(options.session.formatResponse('EXISTS', message.uid));
|
||||
}
|
||||
|
||||
this.notifier.addEntries(mailbox, false, {
|
||||
command: 'EXISTS',
|
||||
uid: message.uid,
|
||||
ignore: options.session && options.session.id,
|
||||
message: message._id,
|
||||
modseq: message.modseq
|
||||
}, () => {
|
||||
|
||||
this.redlock.releaseLock(lock, () => {
|
||||
this.notifier.fire(mailbox.user, mailbox.path);
|
||||
return callback(null, true, {
|
||||
uidValidity,
|
||||
uid
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -293,8 +337,8 @@ class MessageHandler {
|
|||
});
|
||||
}
|
||||
|
||||
del(query, callback) {
|
||||
this.database.collection('messages').findOne(query, (err, message) => {
|
||||
del(options, callback) {
|
||||
this.database.collection('messages').findOne(options.query, (err, message) => {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
|
@ -303,17 +347,13 @@ class MessageHandler {
|
|||
return callback(new Error('Message does not exist'));
|
||||
}
|
||||
|
||||
this.database.collection('mailboxes').findOne({
|
||||
_id: message.mailbox
|
||||
this.getMailbox({
|
||||
mailbox: options.mailbox || message.mailbox
|
||||
}, (err, mailbox) => {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
if (!mailbox) {
|
||||
return callback(new Error('Mailbox does not exist'));
|
||||
}
|
||||
|
||||
this.database.collection('messages').deleteOne({
|
||||
_id: message._id
|
||||
}, err => {
|
||||
|
|
@ -338,8 +378,14 @@ class MessageHandler {
|
|||
if (err) {
|
||||
// ignore as we don't really care if we have orphans or not
|
||||
}
|
||||
|
||||
if (options.session && options.session.selected && options.session.selected.mailbox === mailbox.path) {
|
||||
options.session.writeStream.write(options.session.formatResponse('EXPUNGE', message.uid));
|
||||
}
|
||||
|
||||
this.notifier.addEntries(mailbox, false, {
|
||||
command: 'EXPUNGE',
|
||||
ignore: options.session && options.session.id,
|
||||
uid: message.uid,
|
||||
message: message._id
|
||||
}, () => {
|
||||
|
|
|
|||
9
smtp.js
9
smtp.js
|
|
@ -192,14 +192,19 @@ const server = new SMTPServer({
|
|||
},
|
||||
date: false,
|
||||
flags: false,
|
||||
raw: Buffer.concat(chunks, chunklen)
|
||||
}, err => {
|
||||
raw: Buffer.concat(chunks, chunklen),
|
||||
|
||||
// if similar message exists, then skip
|
||||
skipExisting: true
|
||||
}, (err, inserted) => {
|
||||
// remove Delivered-To
|
||||
chunks.shift();
|
||||
chunklen -= header.length;
|
||||
|
||||
if (err) {
|
||||
log.error('SMTP', err);
|
||||
} else if (!inserted) {
|
||||
log.debug('SMTP', 'Message was not inserted');
|
||||
}
|
||||
|
||||
storeNext();
|
||||
|
|
|
|||
Loading…
Reference in a new issue