likeopera-backend/Syncer.js

const Imap = require('imap');
const ImapManager = require('./ImapManager.js');
const EventEmitter = require('events').EventEmitter;
const MailParser = require('mailparser').MailParser;
const iconv = require('iconv-lite');
const mimelib = require('mimelib');

module.exports = Syncer;

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();
    }

    async init(cfg)
    {
        for (var i = 0; i < cfg.accounts.length; i++)
        {
            await this.addAccount(cfg.accounts[i]);
        }
        await this.loadAccounts();
    }

    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: { imap: account.imap, folders: account.folders }
            }, { id: row.id });
        }
        else
        {
            row = (await SQL.insert('accounts', {
                name: account.name,
                email: account.email,
                settings: {
                    imap: account.imap,
                    folders: account.folders
                }
            }, '*'))[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(console.error);
    }

    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(console.error);
    }

    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(console.error);
    }

    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');
    }

    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: {
                    imap: account.imap
                }
            }, 'id'))[0];
            accountId = row.id;
        }
        let srv = await this.getSyncConnection(accountId);
        let boxes = await new Promise((r, e) => srv.getBoxes(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, r));

        // 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||''
            }, '*'))[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('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, {},
            (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::varchar(255)[]' },
                { 'm.folder_id': boxId, 'm.uid=t.uid': [] },
                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' } },
            (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])
                dia.push('uid >= '+vanished[i][0]+' AND uid <= '+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 ')+')': [] });
    }

    // FIXME: async
    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)
    {
        await SQL.update(
            this.pg, 'threads', { first_msg: null },
            { 'first_msg IN ('+SQL.select_builder('messages', 'id', where)+')': [] },
        );
        await SQL.delete(this.pg, 'messages', where);
        await SQL.update(
            this.pg, 'threads',
            { ['first_msg=('+SQL.select_builder(
                'messages', 'id', { 'thread_id=threads.id': [] }, { order_by: 'time', limit: 1 }
            )+')']: [] },
            { first_msg: null }
        );
        await SQL.delete(this.pg, 'threads', { first_msg: null });
    }

    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) =>
        {
            parse.once('end', r);
            parser.write(msg);
        });
    }

    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');
        }
    }

    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.threadId });
        }
    }
}

function toPgArray(a)
{
    a = JSON.stringify(a);
    return '{'+a.substring(1, a.length-1)+'}';
}