wildduck/tasks.js

634 lines
21 KiB
JavaScript
Raw Normal View History

2018-10-10 21:19:20 +08:00
'use strict';
const log = require('npmlog');
const config = require('wild-config');
const db = require('./lib/db');
const consts = require('./lib/consts');
const RedFour = require('ioredfour');
const yaml = require('js-yaml');
const fs = require('fs');
const MessageHandler = require('./lib/message-handler');
2019-06-07 20:41:21 +08:00
const MailboxHandler = require('./lib/mailbox-handler');
2021-06-15 15:47:18 +08:00
const CertHandler = require('./lib/cert-handler');
2019-09-28 02:26:17 +08:00
const AuditHandler = require('./lib/audit-handler');
2021-06-20 18:40:04 +08:00
const TaskHandler = require('./lib/task-handler');
2021-06-15 15:47:18 +08:00
2021-06-22 20:10:23 +08:00
const { getCertificate, acquireCert } = require('./lib/acme/certs');
2021-06-15 15:47:18 +08:00
2021-01-04 20:20:48 +08:00
const setupIndexes = yaml.load(fs.readFileSync(__dirname + '/indexes.yaml', 'utf8'));
2018-12-03 17:06:47 +08:00
const Gelf = require('gelf');
const os = require('os');
2018-10-10 21:19:20 +08:00
2018-10-11 16:48:12 +08:00
const taskRestore = require('./lib/tasks/restore');
2018-11-24 02:57:45 +08:00
const taskUserDelete = require('./lib/tasks/user-delete');
const taskQuota = require('./lib/tasks/quota');
2019-09-28 02:26:17 +08:00
const taskAudit = require('./lib/tasks/audit');
2021-06-15 15:47:18 +08:00
const taskAcme = require('./lib/tasks/acme');
2021-06-22 20:10:23 +08:00
const taskAcmeUpdate = require('./lib/tasks/acme-update');
2020-07-16 16:15:04 +08:00
const taskClearFolder = require('./lib/tasks/clear-folder');
2018-10-10 21:19:20 +08:00
let messageHandler;
2019-06-07 20:41:21 +08:00
let mailboxHandler;
2019-09-28 02:26:17 +08:00
let auditHandler;
2021-06-20 18:40:04 +08:00
let taskHandler;
2021-06-15 15:47:18 +08:00
let certHandler;
2018-10-10 21:19:20 +08:00
let gcTimeout;
let taskTimeout;
let gcLock;
2018-12-03 17:06:47 +08:00
let loggelf;
2018-10-10 21:19:20 +08:00
module.exports.start = callback => {
if (!config.tasks.enabled) {
return setImmediate(() => callback(null, false));
}
2018-12-03 17:06:47 +08:00
const component = config.log.gelf.component || 'wildduck';
const hostname = config.log.gelf.hostname || os.hostname();
const gelf =
config.log.gelf && config.log.gelf.enabled
? new Gelf(config.log.gelf.options)
: {
// placeholder
emit: (key, message) => log.info('Gelf', JSON.stringify(message))
2018-12-03 17:06:47 +08:00
};
loggelf = message => {
if (typeof message === 'string') {
message = {
short_message: message
};
}
message = message || {};
if (!message.short_message || message.short_message.indexOf(component.toUpperCase()) !== 0) {
message.short_message = component.toUpperCase() + ' ' + (message.short_message || '');
}
message.facility = component; // facility is deprecated but set by the driver if not provided
message.host = hostname;
message.timestamp = Date.now() / 1000;
message._component = component;
Object.keys(message).forEach(key => {
if (!message[key]) {
delete message[key];
}
});
try {
gelf.emit('gelf.log', message);
} catch (err) {
log.error('Gelf', err);
}
};
2018-10-10 21:19:20 +08:00
gcLock = new RedFour({
redis: db.redis,
namespace: 'wildduck'
});
messageHandler = new MessageHandler({
2018-12-03 19:35:00 +08:00
users: db.users,
2018-10-10 21:19:20 +08:00
database: db.database,
redis: db.redis,
gridfs: db.gridfs,
2019-06-07 20:41:21 +08:00
attachments: config.attachments,
loggelf: message => loggelf(message)
});
mailboxHandler = new MailboxHandler({
database: db.database,
users: db.users,
redis: db.redis,
notifier: messageHandler.notifier,
loggelf: message => loggelf(message)
2018-10-10 21:19:20 +08:00
});
2019-09-28 02:26:17 +08:00
auditHandler = new AuditHandler({
database: db.database,
2019-09-29 20:00:44 +08:00
users: db.users,
2019-09-28 02:26:17 +08:00
gridfs: db.gridfs,
bucket: 'audit',
loggelf: message => loggelf(message)
});
2021-06-20 18:40:04 +08:00
taskHandler = new TaskHandler({
database: db.database
});
2021-06-15 15:47:18 +08:00
certHandler = new CertHandler({
cipher: config.certs && config.certs.cipher,
secret: config.certs && config.certs.secret,
database: db.database,
redis: db.redis
});
2018-10-10 21:19:20 +08:00
let start = () => {
// setup ready
setImmediate(() => {
gcTimeout = setTimeout(clearExpiredMessages, consts.GC_INTERVAL);
gcTimeout.unref();
// start processing pending tasks in 5 minuytes after start
taskTimeout = setTimeout(runTasks, consts.TASK_STARTUP_INTERVAL);
taskTimeout.unref();
});
return callback();
};
let collections = setupIndexes.collections;
let collectionpos = 0;
let ensureCollections = next => {
if (collectionpos >= collections.length) {
2018-10-11 16:48:12 +08:00
log.info('Setup', 'Setup %s collections in MongoDB', collections.length);
2018-10-10 21:19:20 +08:00
return next();
}
let collection = collections[collectionpos++];
db[collection.type || 'database'].createCollection(collection.collection, collection.options, err => {
if (err) {
2018-10-11 16:48:12 +08:00
log.error('Setup', 'Failed creating collection %s %s. %s', collectionpos, JSON.stringify(collection.collection), err.message);
2018-10-10 21:19:20 +08:00
}
ensureCollections(next);
});
};
let indexes = setupIndexes.indexes;
let indexpos = 0;
let ensureIndexes = next => {
if (indexpos >= indexes.length) {
2018-10-11 16:48:12 +08:00
log.info('Setup', 'Setup indexes for %s collections', indexes.length);
2018-10-10 21:19:20 +08:00
return next();
}
let index = indexes[indexpos++];
db[index.type || 'database'].collection(index.collection).createIndexes([index.index], (err, r) => {
if (err) {
2018-10-11 16:48:12 +08:00
log.error('Setup', 'Failed creating index %s %s. %s', indexpos, JSON.stringify(index.collection + '.' + index.index.name), err.message);
2018-10-10 21:19:20 +08:00
} else if (r.numIndexesAfter !== r.numIndexesBefore) {
2018-10-11 16:48:12 +08:00
log.verbose('Setup', 'Created index %s %s', indexpos, JSON.stringify(index.collection + '.' + index.index.name));
2018-10-10 21:19:20 +08:00
} else {
2018-10-11 16:48:12 +08:00
log.verbose(
'Setup',
2018-10-10 21:19:20 +08:00
'Skipped index %s %s: %s',
indexpos,
JSON.stringify(index.collection + '.' + index.index.name),
r.note || 'No index added'
);
}
ensureIndexes(next);
});
};
gcLock.acquireLock('db_indexes', 1 * 60 * 1000, (err, lock) => {
if (err) {
2018-10-11 16:48:12 +08:00
log.error('GC', 'Failed to acquire lock error=%s', err.message);
2018-10-10 21:19:20 +08:00
return start();
} else if (!lock.success) {
return start();
}
ensureCollections(() => {
ensureIndexes(() => {
// Do not release the indexing lock immediatelly
setTimeout(() => {
gcLock.releaseLock(lock, err => {
if (err) {
2021-06-21 15:49:31 +08:00
console.error(lock);
2018-10-11 16:48:12 +08:00
log.error('GC', 'Failed to release lock error=%s', err.message);
2018-10-10 21:19:20 +08:00
}
});
}, 60 * 1000);
return start();
});
});
});
};
function clearExpiredMessages() {
clearTimeout(gcTimeout);
let startTime = Date.now();
// First, acquire the lock. This prevents multiple connected clients for deleting the same messages
gcLock.acquireLock('gc_expired', Math.round(consts.GC_INTERVAL * 1.2) /* Lock expires if not released */, (err, lock) => {
if (err) {
2018-10-11 16:48:12 +08:00
log.error('GC', 'Failed to acquire lock error=%s', err.message);
2018-10-10 21:19:20 +08:00
gcTimeout = setTimeout(clearExpiredMessages, consts.GC_INTERVAL);
gcTimeout.unref();
return;
2021-06-21 15:49:31 +08:00
}
if (!lock.success) {
2018-10-11 16:48:12 +08:00
log.verbose('GC', 'Lock already acquired');
2018-10-10 21:19:20 +08:00
gcTimeout = setTimeout(clearExpiredMessages, consts.GC_INTERVAL);
gcTimeout.unref();
return;
}
2018-10-11 16:48:12 +08:00
log.verbose('GC', 'Got lock for garbage collector');
2018-10-10 21:19:20 +08:00
let done = () => {
gcLock.releaseLock(lock, err => {
if (err) {
2021-06-21 15:49:31 +08:00
console.error(lock);
2018-10-11 16:48:12 +08:00
log.error('GC', 'Failed to release lock error=%s', err.message);
2018-10-10 21:19:20 +08:00
}
gcTimeout = setTimeout(clearExpiredMessages, consts.GC_INTERVAL);
gcTimeout.unref();
});
};
if (config.imap.disableRetention) {
// delete all attachments that do not have any active links to message objects
// do not touch expired messages
return messageHandler.attachmentStorage.deleteOrphaned(() => done(null, true));
}
let deleteOrphaned = next => {
// delete all attachments that do not have any active links to message objects
messageHandler.attachmentStorage.deleteOrphaned(() => {
next(null, true);
});
};
let archiveExpiredMessages = next => {
2018-10-11 16:48:12 +08:00
log.verbose('GC', 'Archiving expired messages');
2018-10-10 21:19:20 +08:00
// find and delete all messages that are expired
// NB! scattered query, searches over all mailboxes and thus over all shards
let cursor = db.database.collection('messages').find({
exp: true,
rdate: {
$lte: Date.now()
}
});
let deleted = 0;
let clear = () =>
cursor.close(() => {
if (deleted) {
2018-10-11 16:48:12 +08:00
log.verbose('GC', 'Deleted %s messages', deleted);
2018-10-10 21:19:20 +08:00
}
return deleteOrphaned(next);
});
let processNext = () => {
if (Date.now() - startTime > consts.GC_INTERVAL * 0.8) {
// deleting expired messages has taken too long time, cancel
return clear();
}
cursor.next((err, messageData) => {
if (err) {
return done(err);
}
if (!messageData) {
return clear();
}
messageHandler.del(
{
messageData,
// do not archive messages of deleted users
2018-12-03 20:50:32 +08:00
archive: !messageData.userDeleted && !messageData.copied
2018-10-10 21:19:20 +08:00
},
err => {
if (err) {
2018-10-11 16:48:12 +08:00
log.error('GC', 'Failed to delete expired message id=%s. %s', messageData._id, err.message);
2018-10-10 21:19:20 +08:00
return cursor.close(() => done(err));
}
2018-10-11 16:48:12 +08:00
log.verbose('GC', 'Deleted expired message id=%s', messageData._id);
2018-10-10 21:19:20 +08:00
deleted++;
if (consts.GC_DELAY_DELETE) {
setTimeout(processNext, consts.GC_DELAY_DELETE);
} else {
setImmediate(processNext);
}
}
);
});
};
processNext();
};
let purgeExpiredMessages = next => {
2018-10-11 16:48:12 +08:00
log.verbose('GC', 'Purging archived messages');
2018-10-10 21:19:20 +08:00
// find and delete all messages that are expired
// NB! scattered query, searches over all mailboxes and thus over all shards
let cursor = db.database
.collection('archived')
.find({
exp: true,
rdate: {
$lte: Date.now()
}
})
.project({
_id: true,
2020-07-16 16:15:04 +08:00
user: true,
2018-10-10 21:19:20 +08:00
mailbox: true,
uid: true,
size: true,
'mimeTree.attachmentMap': true,
'meta.queueId': true,
magic: true,
unseen: true
});
let deleted = 0;
let clear = () =>
cursor.close(() => {
if (deleted) {
2018-10-11 16:48:12 +08:00
log.verbose('GC', 'Purged %s messages', deleted);
2018-10-10 21:19:20 +08:00
}
2020-03-19 19:21:05 +08:00
return deleteOrphaned(() => {
auditHandler
.cleanExpired()
.then(() => {
try {
next();
} catch (err) {
// ignore, only needed to prevent calling next() twice
}
})
.catch(next);
});
2018-10-10 21:19:20 +08:00
});
let processNext = () => {
if (Date.now() - startTime > consts.GC_INTERVAL * 0.8) {
// deleting expired messages has taken too long time, cancel
return clear();
}
cursor.next((err, messageData) => {
if (err) {
return done(err);
}
if (!messageData) {
return clear();
}
db.database.collection('archived').deleteOne({ _id: messageData._id }, err => {
if (err) {
//failed to delete
log.error(
'GC',
2020-07-16 16:15:04 +08:00
'Failed to delete archived message user=%s mailbox=%s uid=%ss id=%s. %s',
messageData.user,
messageData.mailbox,
messageData.uid,
messageData._id,
err.message
);
2018-10-10 21:19:20 +08:00
return cursor.close(() => done(err));
}
log.verbose(
'GC',
2020-07-16 16:15:04 +08:00
'Deleted archived message user=%s mailbox=%s uid=%s id=%s',
messageData.user,
messageData.mailbox,
messageData.uid,
messageData._id
);
loggelf({
short_message: '[DELARCH] Deleted archived message',
_mail_action: 'delete_archived',
_service: 'wd_tasks',
_user: messageData.user,
_mailbox: messageData.mailbox,
_uid: messageData.uid,
_archived_id: messageData._id
});
2018-10-10 21:19:20 +08:00
let attachmentIds = Object.keys(messageData.mimeTree.attachmentMap || {}).map(key => messageData.mimeTree.attachmentMap[key]);
2018-10-19 17:19:43 +08:00
if (!attachmentIds.length) {
// no stored attachments
deleted++;
if (consts.GC_DELAY_DELETE) {
setTimeout(processNext, consts.GC_DELAY_DELETE);
} else {
setImmediate(processNext);
}
return;
}
messageHandler.attachmentStorage.updateMany(attachmentIds, -1, -messageData.magic, err => {
if (err) {
// should we care about this error?
}
deleted++;
if (consts.GC_DELAY_DELETE) {
setTimeout(processNext, consts.GC_DELAY_DELETE);
} else {
setImmediate(processNext);
2018-10-10 21:19:20 +08:00
}
2018-10-19 17:19:43 +08:00
});
2018-10-10 21:19:20 +08:00
});
});
};
processNext();
};
archiveExpiredMessages(() => purgeExpiredMessages(done));
});
}
2021-06-20 18:40:04 +08:00
function timer(ttl) {
return new Promise(done => {
let t = setTimeout(done, ttl);
t.unref();
});
}
async function runTasks() {
2021-06-22 01:32:25 +08:00
let pendingCheckTime = 0;
2018-10-10 21:19:20 +08:00
2021-06-20 18:40:04 +08:00
let done = false;
2021-06-21 19:28:40 +08:00
log.verbose('Tasks', 'Starting task poll loop');
2021-06-20 18:40:04 +08:00
while (!done) {
2021-06-22 01:32:25 +08:00
if (Date.now() - pendingCheckTime > consts.TASK_RELEASE_DELAYED_INTERVAL) {
// Once in a while release pending tasks
try {
await taskHandler.releasePending();
} catch (err) {
log.error('Tasks', 'Failed releasing expired tasks. error=%s', err.message);
await timer(consts.TASK_IDLE_INTERVAL);
}
2021-06-22 20:10:23 +08:00
// and run recurring ACME checks
try {
await new Promise((resolve, reject) => {
// run pseudo task
processTask({ type: 'acme-update', _id: 'acme-update-id', lock: 'acme-update-lock' }, {}, err => {
if (err) {
return reject(err);
} else {
resolve();
}
});
});
} catch (err) {
log.error('Tasks', 'Failed running recurring ACME checks. error=%s', err.message);
await timer(consts.TASK_IDLE_INTERVAL);
}
pendingCheckTime = Date.now();
2021-06-21 15:49:31 +08:00
}
2021-06-20 18:40:04 +08:00
try {
let { data, task } = await taskHandler.getNext();
if (!task) {
await timer(consts.TASK_IDLE_INTERVAL);
continue;
2018-10-10 21:19:20 +08:00
}
2021-06-20 18:40:04 +08:00
try {
await new Promise((resolve, reject) => {
processTask(task, data, err => {
2018-10-10 21:19:20 +08:00
if (err) {
2021-06-20 18:40:04 +08:00
return reject(err);
} else {
resolve();
2018-10-10 21:19:20 +08:00
}
2021-06-20 18:40:04 +08:00
});
});
await taskHandler.release(task, true);
} catch (err) {
await taskHandler.release(task, false);
}
} catch (err) {
2021-06-21 15:49:31 +08:00
log.error('Tasks', 'Failed to process task queue error=%s', err.message);
} finally {
2021-06-21 19:28:40 +08:00
await timer(consts.TASK_IDLE_INTERVAL);
2018-10-10 21:19:20 +08:00
}
2021-06-20 18:40:04 +08:00
}
2021-06-21 15:49:31 +08:00
2021-06-22 20:10:23 +08:00
// probably should never be reached as the loop should take forever
2021-06-21 15:49:31 +08:00
return runTasks();
2018-10-10 21:19:20 +08:00
}
2021-06-20 18:40:04 +08:00
function processTask(task, data, callback) {
log.verbose('Tasks', 'type=%s id=%s data=%s', task.type, task._id, JSON.stringify(data));
2018-10-10 21:19:20 +08:00
2021-06-20 18:40:04 +08:00
switch (task.type) {
2018-10-11 16:48:12 +08:00
case 'restore':
return taskRestore(
2021-06-20 18:40:04 +08:00
task,
data,
2018-10-11 16:48:12 +08:00
{
2018-12-03 17:06:47 +08:00
messageHandler,
2019-06-07 20:41:21 +08:00
mailboxHandler,
2018-12-03 17:06:47 +08:00
loggelf
2018-10-11 16:48:12 +08:00
},
err => {
if (err) {
return callback(err);
}
// release
callback(null, true);
}
);
2019-09-28 02:26:17 +08:00
2018-11-24 02:57:45 +08:00
case 'user-delete':
2021-06-20 18:40:04 +08:00
return taskUserDelete(task, data, { loggelf }, err => {
2018-11-24 02:57:45 +08:00
if (err) {
return callback(err);
}
// release
callback(null, true);
});
2019-09-28 02:26:17 +08:00
2018-11-24 02:57:45 +08:00
case 'quota':
2021-06-20 18:40:04 +08:00
return taskQuota(task, data, { loggelf }, err => {
2018-11-24 02:57:45 +08:00
if (err) {
return callback(err);
}
// release
callback(null, true);
});
2019-09-28 02:26:17 +08:00
case 'audit':
return taskAudit(
2021-06-20 18:40:04 +08:00
task,
data,
2019-09-28 02:26:17 +08:00
{
messageHandler,
auditHandler,
loggelf
},
err => {
if (err) {
return callback(err);
}
// release
callback(null, true);
}
);
2021-06-15 15:47:18 +08:00
case 'acme':
return taskAcme(
2021-06-20 18:40:04 +08:00
task,
data,
2021-06-15 15:47:18 +08:00
{
certHandler,
getCertificate,
loggelf
},
err => {
if (err) {
return callback(err);
}
// release
callback(null, true);
}
);
2021-06-22 20:10:23 +08:00
case 'acme-update':
return taskAcmeUpdate(
task,
data,
{
certHandler,
acquireCert,
loggelf
},
err => {
if (err) {
return callback(err);
}
// release
callback(null, true);
}
);
2020-07-16 16:15:04 +08:00
case 'clear-folder':
return taskClearFolder(
2021-06-20 18:40:04 +08:00
task,
data,
2020-07-16 16:15:04 +08:00
{
messageHandler,
loggelf
},
err => {
if (err) {
return callback(err);
}
// release
callback(null, true);
}
);
2018-10-11 16:48:12 +08:00
default:
// release task by returning true
return callback(null, true);
}
2018-10-10 21:19:20 +08:00
}