586 lines
20 KiB
JavaScript
586 lines
20 KiB
JavaScript
const Imap = require('imap');
|
||
const EventEmitter = require('events').EventEmitter;
|
||
const iconv = require('iconv-lite');
|
||
const MailParser = require('mailparser').MailParser;
|
||
const mimelib = require('mimelib');
|
||
|
||
const ImapManager = require('./ImapManager.js');
|
||
const sanitizeHtml = require('./sanitize.js');
|
||
const SQL = require('./select-builder-pgsql.js');
|
||
|
||
class Syncer
|
||
{
|
||
constructor(pg)
|
||
{
|
||
this.syncInProgress = false;
|
||
this.pg = pg;
|
||
this.imap = new ImapManager();
|
||
this.runIdle = this.runIdle.bind(this);
|
||
this.stopIdle = this.stopIdle.bind(this);
|
||
this.events = new EventEmitter();
|
||
}
|
||
|
||
// public
|
||
async init(cfg)
|
||
{
|
||
for (var i = 0; i < cfg.accounts.length; i++)
|
||
{
|
||
await this.addAccount(cfg.accounts[i]);
|
||
}
|
||
await this.loadAccounts();
|
||
}
|
||
|
||
// public
|
||
async syncAll()
|
||
{
|
||
this.syncInProgress = true;
|
||
for (let id in this.accounts)
|
||
{
|
||
await this.syncAccount(this.accounts[id]);
|
||
}
|
||
this.syncInProgress = false;
|
||
this.events.emit('sync', { state: 'complete' });
|
||
}
|
||
|
||
async addAccount(account)
|
||
{
|
||
let row = await SQL.select(this.pg, 'accounts', 'id', { email: account.email }, null, SQL.MS_ROW);
|
||
if (row)
|
||
{
|
||
await SQL.update(this.pg, 'accounts', {
|
||
settings: JSON.stringify({ imap: account.imap, folders: account.folders })
|
||
}, { id: row.id });
|
||
}
|
||
else
|
||
{
|
||
row = (await SQL.insert('accounts', {
|
||
name: account.name,
|
||
email: account.email,
|
||
settings: JSON.stringify({
|
||
imap: account.imap,
|
||
folders: account.folders
|
||
})
|
||
}, { returning: '*' }))[0];
|
||
}
|
||
return row.id;
|
||
}
|
||
|
||
async loadAccounts()
|
||
{
|
||
let rows = await SQL.select(this.pg, 'accounts', '*', []);
|
||
this.accounts = {};
|
||
for (let i = 0; i < rows.length; i++)
|
||
{
|
||
this.accounts[rows[i].id] = rows[i];
|
||
this.imap.setServer(rows[i].id, rows[i].settings.imap);
|
||
}
|
||
}
|
||
|
||
async getSyncConnection(accountId, boxName)
|
||
{
|
||
return await this.imap.getConnection(accountId, null, 'S', this.runIdle, this.stopIdle);
|
||
}
|
||
|
||
idleUidvalidity(accountId, uidvalidity)
|
||
{
|
||
// FIXME uidvalidity changes (FUUUU) remove everything and resync
|
||
}
|
||
|
||
idleMail(accountId, count)
|
||
{
|
||
// <count> new messages arrived while idling, fetch them
|
||
(async () =>
|
||
{
|
||
let srv = await this.getSyncConnection(accountId);
|
||
await this.syncBox(srv, accountId, 'INBOX');
|
||
this.releaseSyncConnection(accountId);
|
||
})().catch(e => console.error(e.stack));
|
||
}
|
||
|
||
idleVanish(accountId, uids)
|
||
{
|
||
// messages expunged by uids
|
||
(async () =>
|
||
{
|
||
let boxId = await SQL.select(
|
||
this.pg, 'folders', 'id', { name: 'INBOX', account_id: accountId }, null, SQL.MS_VALUE
|
||
);
|
||
await this.deleteVanished(boxId, uids);
|
||
})().catch(e => console.error(e.stack));
|
||
}
|
||
|
||
idleExpunge(accountId, seqno)
|
||
{
|
||
// message expunged by (FUUUU) sequence number(s?)
|
||
(async () =>
|
||
{
|
||
let srv = await this.getSyncConnection(accountId);
|
||
await this.syncBox(srv, accountId, 'INBOX');
|
||
this.releaseSyncConnection(accountId);
|
||
})().catch(e => console.error(e.stack));
|
||
}
|
||
|
||
runIdle(accountId, srv)
|
||
{
|
||
if (!srv._idleCallbacks)
|
||
{
|
||
srv._idleCallbacks = {
|
||
uidvalidity: this.idleUidvalidity.bind(this, accountId),
|
||
mail: this.idleMail.bind(this, accountId),
|
||
vanish: this.idleVanish.bind(this, accountId),
|
||
expunge: this.idleExpunge.bind(this, accountId)
|
||
};
|
||
}
|
||
for (let i in srv._idleCallbacks)
|
||
{
|
||
srv.on(i, srv._idleCallbacks[i]);
|
||
}
|
||
srv.openBox('INBOX', true, () => {});
|
||
}
|
||
|
||
stopIdle(accountId, srv)
|
||
{
|
||
for (let i in srv._idleCallbacks)
|
||
{
|
||
srv.removeListener(i, srv._idleCallbacks[i]);
|
||
}
|
||
}
|
||
|
||
releaseSyncConnection(accountId, boxName)
|
||
{
|
||
this.imap.releaseConnection(accountId, 'S');
|
||
}
|
||
|
||
// public
|
||
async syncAccount(account)
|
||
{
|
||
let accountId = await SQL.select(this.pg, 'accounts', 'id', { email: account.email }, null, SQL.MS_VALUE);
|
||
if (!accountId)
|
||
{
|
||
let row = (await SQL.insert(this.pg, 'accounts', {
|
||
name: account.name,
|
||
email: account.email,
|
||
settings: JSON.stringify({
|
||
imap: account.imap
|
||
})
|
||
}, { returning: 'id' }))[0];
|
||
accountId = row.id;
|
||
}
|
||
let srv = await this.getSyncConnection(accountId);
|
||
let boxes = await new Promise((res, err) => srv.getBoxes((e, r) => e ? err(e) : res(r)));
|
||
for (let k in boxes)
|
||
{
|
||
let boxKind = (boxes[k].special_use_attrib || '').replace('\\', '').toLowerCase();
|
||
await this.syncBox(srv, accountId, k, boxKind, true);
|
||
}
|
||
this.releaseSyncConnection(accountId);
|
||
}
|
||
|
||
async syncBox(srv, accountId, boxName, boxKind, doFull)
|
||
{
|
||
let boxStatus = await new Promise((r, e) => srv.openBox(boxName, true, (err, info) => err ? e(err) : r(info)));
|
||
|
||
// IMAP sync: http://tools.ietf.org/html/rfc4549
|
||
let boxRow = await SQL.select(this.pg, 'folders', '*', { account_id: accountId, name: boxStatus.name }, null, SQL.MS_ROW);
|
||
if (boxRow)
|
||
{
|
||
if (boxRow.uidvalidity != boxStatus.uidvalidity)
|
||
{
|
||
await this.deleteMessages({ folder_id: boxRow.id, 'uid is not null': [] });
|
||
boxRow.uidvalidity = boxStatus.uidvalidity;
|
||
}
|
||
}
|
||
else
|
||
{
|
||
boxRow = (await SQL.insert(this.pg, 'folders', {
|
||
name: boxStatus.name,
|
||
uidvalidity: boxStatus.uidvalidity,
|
||
account_id: accountId,
|
||
highestmodseq: 0,
|
||
kind: boxKind||''
|
||
}, { returning: '*' }))[0];
|
||
}
|
||
|
||
// fetch new messages
|
||
let missing = [];
|
||
let maxUid = await SQL.select(this.pg, 'messages', 'MAX(uid)', { folder_id: boxRow.id }, null, SQL.MS_VALUE);
|
||
if (boxRow.highestmodseq)
|
||
{
|
||
this.events.emit('sync', { state: 'start', quick: true, email: this.accounts[accountId].email, folder: boxRow.name });
|
||
process.stderr.write(this.accounts[accountId].email+'/'+boxRow.name+': quick resync\n');
|
||
await this.quickResync(srv, boxRow.id, maxUid, boxRow.highestmodseq, missing);
|
||
boxRow.highestmodseq = boxStatus.highestmodseq;
|
||
}
|
||
else if (doFull && maxUid)
|
||
{
|
||
// list messages, update flags and version tag
|
||
this.events.emit('sync', { state: 'start', email: this.accounts[accountId].email, folder: boxRow.name });
|
||
process.stderr.write(this.accounts[accountId].email+'/'+boxRow.name+': full resync\n');
|
||
await this.fullResync(srv, boxRow.id, maxUid, missing, boxStatus.messages.total);
|
||
}
|
||
|
||
missing.push((maxUid ? maxUid+1 : 1)+':*');
|
||
await this.imap.runFetch(srv, missing, {
|
||
size: true,
|
||
bodies: 'HEADER',
|
||
struct: true,
|
||
}, (messages, state) => this.saveMessages(messages, boxRow.id, state));
|
||
|
||
await SQL.update(this.pg, 'folders', {
|
||
uidvalidity: boxStatus.uidvalidity,
|
||
highestmodseq: boxStatus.highestmodseq||0
|
||
}, { id: boxRow.id });
|
||
}
|
||
|
||
async fullResync(srv, boxId, maxUid, missing, total)
|
||
{
|
||
let flags = await SQL.select(this.pg, 'messages', 'uid, flags', { folder_id: boxId });
|
||
flags = flags.reduce((o, row) => { o[row.uid] = row.flags; return o; }, {});
|
||
|
||
let updateFlags = [];
|
||
|
||
process.stderr.write('\rsynchronizing 0');
|
||
await this.imap.runFetch(
|
||
srv, '1:'+maxUid, {},
|
||
async (messages, state) => this.queueFlags(messages, boxId, state),
|
||
{ flags: flags, updateFlags: updateFlags, missing: missing||[], total: total, nopause: true }
|
||
);
|
||
process.stderr.write('\n');
|
||
this.events.emit('sync', { state: 'finish-box' });
|
||
|
||
await this.updateFlags(boxId, updateFlags);
|
||
|
||
// delete messages removed from IMAP server
|
||
flags = Object.keys(flags);
|
||
if (flags.length)
|
||
{
|
||
await this.deleteMessages({ folder_id: boxId, uid: flags });
|
||
}
|
||
}
|
||
|
||
queueFlags(messages, boxId, fetchState)
|
||
{
|
||
for (let i = 0; i < messages.length; i++)
|
||
{
|
||
let m = messages[i][0];
|
||
if (!fetchState.flags[m.uid])
|
||
{
|
||
fetchState.missing.push(m.uid);
|
||
}
|
||
else
|
||
{
|
||
if (fetchState.flags[m.uid].join(',') != m.flags.join(','))
|
||
{
|
||
fetchState.updateFlags.push({ uid: m.uid, flags: toPgArray(m.flags) });
|
||
}
|
||
delete fetchState.flags[m.uid];
|
||
}
|
||
}
|
||
fetchState.synced += messages.length;
|
||
//this.events.emit('sync', { state: 'progress', done: fetchState.synced, total: fetchState.total });
|
||
process.stderr.write('\rsynchronizing '+fetchState.synced);
|
||
}
|
||
|
||
async updateFlags(boxId, updateFlags, checkMissing)
|
||
{
|
||
if (updateFlags.length)
|
||
{
|
||
let updated = await SQL.update(
|
||
this.pg, { m: 'messages', t: SQL.values(updateFlags) },
|
||
[ 'flags = t.flags::text[]' ],
|
||
{ 'm.folder_id': boxId, 'm.uid = t.uid::int': [] },
|
||
checkMissing ? { returning: 'm.uid' } : null
|
||
);
|
||
if (checkMissing)
|
||
{
|
||
let missing = {};
|
||
for (let i = 0; i < updateFlags.length; i++)
|
||
missing[updateFlags[i].uid] = true;
|
||
for (let i = 0; i < updated.length; i++)
|
||
delete missing[updated[i].uid];
|
||
return Object.keys(missing);
|
||
}
|
||
}
|
||
return [];
|
||
}
|
||
|
||
async quickResync(srv, boxId, maxUid, changedSince, missing)
|
||
{
|
||
let updateFlags = [];
|
||
let vanished = [];
|
||
let onVanish = function(dias)
|
||
{
|
||
vanished = vanished.concat(vanished, dias);
|
||
};
|
||
|
||
srv.on('vanish', onVanish);
|
||
await this.imap.runFetch(
|
||
srv, '1:'+maxUid, { modifiers: { changedsince: changedSince+' VANISHED' } },
|
||
async (messages, state) => this.queueQuickFlags(messages, boxId, state),
|
||
{ updateFlags: updateFlags }
|
||
);
|
||
srv.removeListener('vanish', onVanish);
|
||
let checkedMissing = await this.updateFlags(boxId, updateFlags, missing && true);
|
||
if (missing)
|
||
{
|
||
missing.push.apply(missing, checkedMissing);
|
||
}
|
||
|
||
if (vanished.length)
|
||
{
|
||
await this.deleteVanished(boxId, vanished);
|
||
}
|
||
}
|
||
|
||
async deleteVanished(boxId, vanished)
|
||
{
|
||
let lst = [], dia = [];
|
||
for (let i = 0; i < vanished.length; i++)
|
||
{
|
||
if (vanished[i][1])
|
||
{
|
||
if (Number(vanished[i][1]) > Number(vanished[i][0]) + 1)
|
||
dia.push('uid >= '+vanished[i][0]+' AND uid <= '+vanished[i][1]);
|
||
else
|
||
lst.push(vanished[i][0], vanished[i][1]);
|
||
}
|
||
else
|
||
lst.push(vanished[i][0]);
|
||
}
|
||
if (lst.length)
|
||
{
|
||
dia.push('uid IN ('+lst.join(',')+')');
|
||
}
|
||
await this.deleteMessages({ folder_id: boxId, ['('+dia.join(' OR ')+')']: [] });
|
||
}
|
||
|
||
queueQuickFlags(messages, boxId, fetchState)
|
||
{
|
||
for (let i = 0; i < messages.length; i++)
|
||
{
|
||
let m = messages[i][0];
|
||
fetchState.updateFlags.push({ uid: m.uid, flags: toPgArray(m.flags) });
|
||
}
|
||
}
|
||
|
||
async deleteMessages(where)
|
||
{
|
||
let cond = SQL.where_builder(where);
|
||
let q = SQL.select_builder('messages', 'id', where);
|
||
await this.pg.query(
|
||
SQL.quote_positional(
|
||
'WITH deleting_messages AS (SELECT id FROM messages WHERE '+cond.sql+')'+
|
||
', updated_threads AS ('+
|
||
'UPDATE threads SET first_msg=('+
|
||
'SELECT m.id FROM messages m WHERE m.thread_id=threads.id'+
|
||
' AND m.id NOT IN (SELECT id FROM deleting_messages) ORDER BY time LIMIT 1'+
|
||
') WHERE first_msg IN (SELECT id FROM deleting_messages)'+
|
||
' RETURNING id, first_msg'+
|
||
'), deleted_threads AS ('+
|
||
'DELETE FROM threads WHERE id IN (SELECT id FROM updated_threads WHERE first_msg IS NULL)'+
|
||
' RETURNING id'+
|
||
') DELETE FROM messages WHERE id IN (SELECT id FROM deleting_messages)'
|
||
), cond.bind
|
||
);
|
||
}
|
||
|
||
async saveMessages(messages, boxId)
|
||
{
|
||
let uids = messages.map(m => m[1].uid);
|
||
let exist = await SQL.select(this.pg, 'messages', 'uid, flags', { folder_id: boxId, uid: uids });
|
||
uids = {};
|
||
for (let i = 0; i < exist.length; i++)
|
||
{
|
||
uids[exist[i].uid] = true;
|
||
}
|
||
for (let i = 0; i < messages.length; i++)
|
||
{
|
||
if (!uids[messages[i][1].uid])
|
||
{
|
||
await this.addMessage(boxId, messages[i][0], messages[i][1]);
|
||
}
|
||
}
|
||
}
|
||
|
||
async parseMsg(msg)
|
||
{
|
||
let parser = new MailParser({ streamAttachments: false, defaultCharset: 'windows-1251' });
|
||
return await new Promise((r, j) =>
|
||
{
|
||
parser.once('end', r);
|
||
parser.write(msg);
|
||
parser.end();
|
||
});
|
||
}
|
||
|
||
extractAttachments(struct, attachments)
|
||
{
|
||
attachments = attachments || [];
|
||
for (let i = 0; i < struct.length; i++)
|
||
{
|
||
if (struct[i] instanceof Array)
|
||
{
|
||
this.extractAttachments(struct[i], attachments);
|
||
}
|
||
else if (struct[i].disposition && struct[i].disposition.type == 'attachment')
|
||
{
|
||
attachments.push([
|
||
mimelib.parseMimeWords(struct[i].disposition.params && struct[i].disposition.params.filename || struct[i].description || ''),
|
||
struct[i].type+'/'+struct[i].subtype,
|
||
struct[i].size,
|
||
]);
|
||
}
|
||
}
|
||
return attachments;
|
||
}
|
||
|
||
async addMessage(boxId, msgrow, attrs)
|
||
{
|
||
await this.pg.query('BEGIN');
|
||
try
|
||
{
|
||
await this.addMessageImpl(boxId, msgrow, attrs);
|
||
await this.pg.query('COMMIT');
|
||
}
|
||
catch (e)
|
||
{
|
||
await this.pg.query('ROLLBACK');
|
||
throw e;
|
||
}
|
||
}
|
||
|
||
async addMessageImpl(boxId, msgrow, attrs)
|
||
{
|
||
let header = await this.parseMsg(msgrow.headers);
|
||
header.references = header.references || [];
|
||
if (header.references.length)
|
||
{
|
||
if (!header.inReplyTo || !header.inReplyTo[0])
|
||
header.inReplyTo = [ header.references[header.references.length-1] ];
|
||
else if (header.references[header.references.length-1] != header.inReplyTo[0])
|
||
header.references.push(header.inReplyTo[0]);
|
||
}
|
||
if (!header.date)
|
||
header.date = new Date(attrs.date);
|
||
|
||
if (JSON.stringify(header).indexOf('<27>') >= 0)
|
||
{
|
||
// Charset error!
|
||
console.log(iconv.decode(msgrow.headers, 'cp1251'));
|
||
console.log(header);
|
||
}
|
||
|
||
delete msgrow.headers;
|
||
msgrow.folder_id = boxId;
|
||
msgrow.subject = header.subject || '';
|
||
msgrow.props = JSON.stringify({
|
||
from: ((header.from||[]).map((a) => [ a.name, a.address ]))[0],
|
||
to: (header.to||[]).map((a) => [ a.name, a.address ]),
|
||
cc: (header.cc||[]).map((a) => [ a.name, a.address ]),
|
||
bcc: (header.bcc||[]).map((a) => [ a.name, a.address ]),
|
||
replyto: (header.replyTo||[]).map((a) => [ a.name, a.address ])[0],
|
||
attachments: this.extractAttachments(attrs.struct),
|
||
});
|
||
msgrow.messageid = header.messageId || '';
|
||
msgrow.inreplyto = header.inReplyTo && header.inReplyTo[0] || '';
|
||
msgrow.time = header.date;
|
||
msgrow.size = attrs.size;
|
||
if (!header.headers.received || !header.headers.received.length)
|
||
msgrow.flags.push('out');
|
||
else
|
||
msgrow.flags.push('in');
|
||
msgrow.flags = toPgArray(msgrow.flags);
|
||
msgrow.refs = toPgArray(header.references);
|
||
for (let i in msgrow)
|
||
if (typeof msgrow[i] == 'string')
|
||
msgrow[i] = msgrow[i].replace(/\x00/g, '');
|
||
|
||
let thisIsFirst = false;
|
||
if (header.references.length)
|
||
{
|
||
let threadId = await SQL.select(
|
||
this.pg, 'messages', 'MAX(thread_id)',
|
||
{ messageid: header.references }, null, SQL.MS_VALUE
|
||
);
|
||
if (!threadId)
|
||
{
|
||
threadId = await SQL.select(
|
||
this.pg, 'messages', 'MAX(thread_id)',
|
||
{ 'refs @> array[?]': msgrow.messageid }, null, SQL.MS_VALUE
|
||
);
|
||
if (threadId)
|
||
{
|
||
thisIsFirst = true;
|
||
}
|
||
}
|
||
msgrow.thread_id = threadId;
|
||
}
|
||
console.log(msgrow.time+' '+(header.from && header.from[0] && header.from[0].address || '?')+' '+msgrow.subject);
|
||
msgrow.id = (await SQL.insert(this.pg, 'messages', msgrow, { returning: 'id' }))[0].id;
|
||
if (!msgrow.thread_id)
|
||
{
|
||
msgrow.thread_id = (await SQL.insert(this.pg, 'threads', {
|
||
first_msg: msgrow.id,
|
||
msg_count: 1
|
||
}, { returning: 'id' }))[0].id;
|
||
await SQL.update(this.pg, 'messages', { thread_id: msgrow.thread_id }, { id: msgrow.id });
|
||
}
|
||
else
|
||
{
|
||
let upd = { 'msg_count=msg_count+1': [] };
|
||
if (thisIsFirst)
|
||
{
|
||
upd.first_msg = msgrow.id;
|
||
}
|
||
await SQL.update(this.pg, 'threads', upd, { id: msgrow.thread_id });
|
||
}
|
||
}
|
||
|
||
async fetchFullMessage(account_id, folder_id, folder_name, msg_uid)
|
||
{
|
||
// FIXME: parse and save attachments
|
||
// FIXME: replace inline images
|
||
let srv = await this.imap.getConnection(account_id, folder_name);
|
||
let upd = await this.imap.runFetch(
|
||
srv, msg_uid, { bodies: '' },
|
||
async(messages, state) => await this._parseBody(messages, folder_id)
|
||
);
|
||
this.imap.releaseConnection(account_id);
|
||
return upd;
|
||
}
|
||
|
||
async _parseBody(messages, boxId)
|
||
{
|
||
for (let i = 0; i < messages.length; i++)
|
||
{
|
||
let msg = messages[i];
|
||
let obj = await this.parseMsg(msg[0].headers);
|
||
obj.html = sanitizeHtml(obj.html);
|
||
let upd = { body_text: obj.text||'', body_html: obj.html };
|
||
upd.body_html_text = obj.html.replace(/<style[^>]*>.*<\/style\s*>|<\/?[^>]*>/g, '');
|
||
await SQL.update(this.pg, 'messages m', upd, { folder_id: boxId, uid: msg[0].uid });
|
||
if (messages.length == 1)
|
||
{
|
||
return [ upd ];
|
||
}
|
||
}
|
||
return null;
|
||
}
|
||
}
|
||
|
||
function toPgArray(a)
|
||
{
|
||
a = JSON.stringify(a);
|
||
return '{'+a.substring(1, a.length-1)+'}';
|
||
}
|
||
|
||
function mapToHash(map)
|
||
{
|
||
let h = {};
|
||
for (let v of map)
|
||
h[v[0]] = v[1];
|
||
return h;
|
||
}
|
||
|
||
module.exports = Syncer;
|