wildduck/webhooks.js

269 lines
8.3 KiB
JavaScript
Raw Normal View History

2020-10-09 16:08:33 +08:00
'use strict';
const log = require('npmlog');
const config = require('wild-config');
const Gelf = require('gelf');
const os = require('os');
const { Queue, Worker } = require('bullmq');
2020-10-09 16:08:33 +08:00
const db = require('./lib/db');
2021-02-26 20:00:13 +08:00
const tools = require('./lib/tools');
const { ObjectId } = require('mongodb');
2020-10-09 16:08:33 +08:00
const axios = require('axios');
2020-10-09 18:43:18 +08:00
const packageData = require('./package.json');
2021-02-26 20:00:13 +08:00
const { MARKED_SPAM, MARKED_HAM } = require('./lib/events');
2020-10-09 16:08:33 +08:00
let loggelf;
let queueWorkers = {};
2020-10-09 16:08:33 +08:00
async function postWebhook(webhook, data) {
2020-10-09 19:01:35 +08:00
let res;
try {
res = await axios.post(webhook.url, data, {
headers: {
'User-Agent': `wildduck/${packageData.version}`
}
});
} catch (err) {
loggelf({
short_message: '[WH] ' + data.ev,
_mail_action: 'webhook',
_wh_id: data.id,
_wh_type: data.ev,
_wh_user: data.user,
_wh_url: webhook.url,
_wh_success: 'no',
_error: err.message
});
throw err;
}
2020-10-09 16:08:33 +08:00
if (!res) {
throw new Error(`Failed to POST request to ${webhook.url}`);
}
2020-10-09 19:01:35 +08:00
loggelf({
short_message: '[WH] ' + data.ev,
_mail_action: 'webhook',
_wh_id: data.id,
_wh_type: data.ev,
_wh_user: data.user,
_wh_url: webhook.url,
_wh_res: res.status,
_wh_success: res.status >= 200 && res.status < 300 ? 'yes' : 'no'
});
2020-10-09 16:08:33 +08:00
log.verbose('Webhooks', 'Posted %s to %s with status %s', data.ev, webhook.url, res.status);
if (res.status === 410) {
// autodelete
try {
await db.users.collection('webhooks').deleteOne({
_id: new ObjectId(webhook._id)
2020-10-09 16:08:33 +08:00
});
} catch (err) {
// ignore
}
2020-10-09 19:01:35 +08:00
return false;
2020-10-09 16:08:33 +08:00
}
if (!res.status || res.status < 200 || res.status >= 300) {
throw new Error(`Invalid response status ${res.status}`);
}
return true;
}
module.exports.start = callback => {
2020-12-28 21:56:56 +08:00
if (!(config.webhooks && config.webhooks.enabled)) {
2020-10-09 16:08:33 +08:00
return setImmediate(() => callback(null, false));
}
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))
};
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);
}
};
const webhooksPostQueue = new Queue('webhooks_post', db.queueConf);
2020-10-09 16:08:33 +08:00
queueWorkers.webhooks = new Worker(
'webhooks',
async job => {
try {
if (!job || !job.data || !job.data.ev) {
return false;
}
2020-10-09 16:08:33 +08:00
const data = job.data;
2020-10-09 16:08:33 +08:00
let evtList = ['*'];
let typeParts = data.ev.split('.');
typeParts.pop();
for (let i = 1; i <= typeParts.length; i++) {
evtList.push(typeParts.slice(0, i) + '.*');
}
evtList.push(data.ev);
2020-10-09 16:08:33 +08:00
const query = { type: { $in: evtList } };
if (data.user) {
query.user = { $in: [new ObjectId(data.user), null] };
}
2020-10-09 19:01:35 +08:00
let whid = new ObjectId();
let count = 0;
2021-02-26 20:00:13 +08:00
let webhooks = await db.users.collection('webhooks').find(query).toArray();
2021-02-26 20:00:13 +08:00
if (!webhooks.length) {
// ignore this event
2021-02-26 20:00:13 +08:00
return;
}
if ([MARKED_SPAM, MARKED_HAM].includes(data.ev)) {
let message = new ObjectId(data.message);
data.message = data.id;
delete data.id;
2021-02-26 20:00:13 +08:00
let messageData = await db.database.collection('messages').findOne(
{ _id: message },
2021-02-26 20:00:13 +08:00
{
projection: {
_id: true,
uid: true,
msgid: true,
subject: true,
mailbox: true,
mimeTree: true,
idate: true
}
2021-02-26 20:00:13 +08:00
}
);
2021-02-26 20:00:13 +08:00
if (!messageData) {
// message already deleted?
return;
}
2021-02-26 20:00:13 +08:00
let parsedHeader = (messageData.mimeTree && messageData.mimeTree.parsedHeader) || {};
2021-02-26 20:00:13 +08:00
let from = parsedHeader.from ||
parsedHeader.sender || [
{
name: '',
address: (messageData.meta && messageData.meta.from) || ''
}
];
let addresses = {
to: [].concat(parsedHeader.to || []),
cc: [].concat(parsedHeader.cc || []),
bcc: [].concat(parsedHeader.bcc || [])
};
tools.decodeAddresses(from);
tools.decodeAddresses(addresses.to);
tools.decodeAddresses(addresses.cc);
tools.decodeAddresses(addresses.bcc);
if (from && from[0]) {
data.from = from[0];
}
for (let addrType of ['to', 'cc', 'bcc']) {
if (addresses[addrType] && addresses[addrType].length) {
data[addrType] = addresses[addrType];
}
2021-02-26 20:00:13 +08:00
}
data.messageId = messageData.msgid;
data.subject = messageData.subject;
data.date = messageData.idate.toISOString();
}
2021-02-26 20:00:13 +08:00
for (let webhook of webhooks) {
count++;
try {
await webhooksPostQueue.add(
'webhook',
{ data: Object.assign({ id: `${whid.toHexString()}:${count}` }, data), webhook },
{
removeOnComplete: true,
removeOnFail: 500,
attempts: 5,
backoff: {
type: 'exponential',
delay: 2000
}
2020-10-09 16:08:33 +08:00
}
);
} catch (err) {
// ignore?
log.error('Events', err);
}
2020-10-09 16:08:33 +08:00
}
} catch (err) {
log.error('Webhooks', err);
throw err;
2020-10-09 16:08:33 +08:00
}
},
Object.assign(
{
concurrency: 1
},
db.queueConf
)
);
queueWorkers.webhooksPost = new Worker(
'webhooks_post',
async job => {
if (!job || !job.data) {
return false;
}
const { data, webhook } = job.data;
return await postWebhook(webhook, data);
},
Object.assign(
{
concurrency: 1
},
db.queueConf
)
);
2020-10-09 16:08:33 +08:00
callback();
};