[local-sync, iso-core, cloud-core] feat(send): add multi-send support

Also renames JSONType() -> buildJSONColumnOptions() and
JSONARRAYType() -> buildJSONARRAYColumnOptions() to prevent passing
those return values in as just the type value instead of the entire
options object.

packages/isomorphic-core/src/database-types.js

@@ -1,7 +1,7 @@
 const Sequelize = require('sequelize');
 
 module.exports = {
-  JSONType: (fieldName, {defaultValue = {}} = {}) => ({
+  buildJSONColumnOptions: (fieldName, {defaultValue = {}} = {}) => ({
     type: Sequelize.TEXT,
     get: function get() {
       const val = this.getDataValue(fieldName);
@@ -14,7 +14,7 @@ module.exports = {
       this.setDataValue(fieldName, JSON.stringify(val));
     },
   }),
-  JSONARRAYType: (fieldName) => ({
+  buildJSONARRAYColumnOptions: (fieldName) => ({
     type: Sequelize.TEXT,
     get: function get() {
       const val = this.getDataValue(fieldName);

packages/isomorphic-core/src/models/account.js

@@ -1,5 +1,5 @@
 const crypto = require('crypto');
-const {JSONType, JSONARRAYType} = require('../database-types');
+const {buildJSONColumnOptions, buildJSONARRAYColumnOptions} = require('../database-types');
 
 const {DB_ENCRYPTION_ALGORITHM, DB_ENCRYPTION_PASSWORD} = process.env;
 
@@ -9,16 +9,16 @@ module.exports = (sequelize, Sequelize) => {
     name: Sequelize.STRING,
     provider: Sequelize.STRING,
     emailAddress: Sequelize.STRING,
-    connectionSettings: JSONType('connectionSettings'),
+    connectionSettings: buildJSONColumnOptions('connectionSettings'),
     connectionCredentials: Sequelize.TEXT,
-    syncPolicy: JSONType('syncPolicy'),
-    syncError: JSONType('syncError', {defaultValue: null}),
+    syncPolicy: buildJSONColumnOptions('syncPolicy'),
+    syncError: buildJSONColumnOptions('syncError', {defaultValue: null}),
     firstSyncCompletion: {
       type: Sequelize.STRING(14),
       allowNull: true,
       defaultValue: null,
     },
-    lastSyncCompletions: JSONARRAYType('lastSyncCompletions'),
+    lastSyncCompletions: buildJSONARRAYColumnOptions('lastSyncCompletions'),
   }, {
     indexes: [
       {

packages/isomorphic-core/src/models/transaction.js

@@ -1,4 +1,4 @@
-const {JSONARRAYType} = require('../database-types');
+const {buildJSONARRAYColumnOptions} = require('../database-types');
 
 module.exports = (sequelize, Sequelize) => {
   return sequelize.define('transaction', {
@@ -6,7 +6,7 @@ module.exports = (sequelize, Sequelize) => {
     object: Sequelize.STRING,
     objectId: Sequelize.STRING,
     accountId: Sequelize.STRING,
-    changedFields: JSONARRAYType('changedFields'),
+    changedFields: buildJSONARRAYColumnOptions('changedFields'),
   }, {
     instanceMethods: {
       toJSON: function toJSON() {

packages/local-sync/package.json

@@ -21,6 +21,7 @@
     "rx": "4.1.0",
     "sequelize": "3.27.0",
     "sqlite3": "https://github.com/bengotow/node-sqlite3/archive/bengotow/usleep-v3.1.4.tar.gz",
+    "striptags": "2.1.1",
     "underscore": "1.8.3",
     "utf7": "^1.0.2",
     "vision": "4.1.0"

packages/local-sync/src/local-api/app.js

@@ -11,8 +11,6 @@ const fs = require('fs');
 const path = require('path');
 const LocalDatabaseConnector = require('../shared/local-database-connector')
 
-if (!global.Logger) { global.Logger = console }
-
 const server = new Hapi.Server({
   connections: {
     router: {

packages/local-sync/src/local-api/routes/send.js

@@ -1,36 +1,199 @@
 const Joi = require('joi');
-const nodemailer = require('nodemailer');
 const LocalDatabaseConnector = require('../../shared/local-database-connector');
+const SendingUtils = require('../sending-utils');
+const SendmailClient = require('../sendmail-client');
 
-function toParticipant(payload) {
-  return payload.map((p) => `${p.name} <${p.email}>`).join(',')
+const SEND_TIMEOUT = 1000 * 60; // millliseconds
+
+const recipient = Joi.object().keys({
+  name: Joi.string().required(),
+  email: Joi.string().email().required(),
+  account_id: Joi.string(),
+  client_id: Joi.string(),
+  id: Joi.string(),
+  thirdPartyData: Joi.object(),
+});
+const recipientList = Joi.array().items(recipient);
+
+const respondWithError = (request, reply, error) => {
+  if (!error.httpCode) {
+    error.type = 'apiError';
+    error.httpCode = 500;
+  }
+  request.logger.error('responding with error', error, error.logContext);
+  reply(JSON.stringify(error)).code(error.httpCode);
 }
 
 module.exports = (server) => {
   server.route({
     method: 'POST',
     path: '/send',
-    handler: (request, reply) => { LocalDatabaseConnector.forShared().then((db) => {
-      const accountId = request.auth.credentials.id;
-      db.Account.findById(accountId).then((account) => {
-        const sender = nodemailer.createTransport(account.smtpConfig());
-        const data = request.payload;
+    handler: async (request, reply) => {
+      try {
+        const account = request.auth.credentials;
+        const db = await LocalDatabaseConnector.forAccount(account.id)
+        const draft = await SendingUtils.findOrCreateMessageFromJSON(request.payload, db);
+        // Calculate the response now to prevent errors after the draft has
+        // already been sent.
+        const responseOnSuccess = draft.toJSON();
+        const sender = new SendmailClient(account, request.logger);
+        await sender.send(draft);
+        reply(responseOnSuccess);
+      } catch (err) {
+        respondWithError(request, reply, err);
+      }
+    },
+  });
 
-        const msg = {}
-        for (key of ['from', 'to', 'cc', 'bcc']) {
-          if (data[key]) msg[key] = toParticipant(data[key])
-        }
-        if (!msg.from || msg.from.length === 0) {
-          msg.from = `${account.name} <${account.emailAddress}>`
-        }
-        msg.subject = data.subject,
-        msg.html = data.body,
+  // Initiates a multi-send session by creating a new multi-send draft.
+  server.route({
+    method: 'POST',
+    path: '/send-multiple',
+    config: {
+      validate: {
+        payload: {
+          to: recipientList,
+          cc: recipientList,
+          bcc: recipientList,
+          from: recipientList.length(1).required(),
+          reply_to: recipientList.min(0).max(1),
+          subject: Joi.string().required(),
+          body: Joi.string().required(),
+          thread_id: Joi.string(),
+          reply_to_message_id: Joi.string(),
+          client_id: Joi.string(),
+          account_id: Joi.string(),
+          id: Joi.string(),
+          object: Joi.string(),
+          metadata: Joi.array().items(Joi.string()),
+          date: Joi.number(),
+          files: Joi.array().items(Joi.string()),
+          file_ids: Joi.array().items(Joi.string()),
+          uploads: Joi.array().items(Joi.string()),
+          events: Joi.array().items(Joi.string()),
+          pristine: Joi.boolean(),
+          categories: Joi.array().items(Joi.string()),
+          draft: Joi.boolean(),
+        },
+      },
+    },
+    handler: async (request, reply) => {
+      try {
+        const accountId = request.auth.credentials.id;
+        const db = await LocalDatabaseConnector.forAccount(accountId)
+        const draft = await SendingUtils.findOrCreateMessageFromJSON(request.payload, db, false)
+        await (draft.isSending = true);
+        const savedDraft = await draft.save();
+        reply(savedDraft.toJSON());
+      } catch (err) {
+        respondWithError(request, reply, err);
+      }
+    },
+  });
 
-        sender.sendMail(msg, (error, info) => {
-          if (error) { reply(error).code(400) }
-          else { reply(info.response) }
+  // Performs a single send operation in an individualized multi-send
+  // session. Sends a copy of the draft at draft_id to the specified address
+  // with the specified body, and ensures that a corresponding sent message is
+  // either not created in the user's Sent folder or is immediately
+  // deleted from it.
+  server.route({
+    method: 'POST',
+    path: '/send-multiple/{draftId}',
+    config: {
+      validate: {
+        params: {
+          draftId: Joi.string(),
+        },
+        payload: {
+          send_to: recipient,
+          body: Joi.string(),
+        },
+      },
+    },
+    handler: async (request, reply) => {
+      try {
+        const requestStarted = new Date();
+        const account = request.auth.credentials;
+        const {draftId} = request.params;
+        SendingUtils.validateBase36(draftId, 'draftId')
+        const sendTo = request.payload.send_to;
+        const db = await LocalDatabaseConnector.forAccount(account.id)
+        const draft = await SendingUtils.findMultiSendDraft(draftId, db)
+        const {to, cc, bcc} = draft;
+        const recipients = [].concat(to, cc, bcc);
+        if (!recipients.find(contact => contact.email === sendTo.email)) {
+          throw new SendingUtils.HTTPError(
+            "Invalid sendTo, not present in message recipients",
+            400
+          );
+        }
+
+        const sender = new SendmailClient(account, request.logger);
+
+        if (new Date() - requestStarted > SEND_TIMEOUT) {
+          // Preemptively time out the request if we got stuck doing database work
+          // -- we don't want clients to disconnect and then still send the
+          // message.
+          reply('Request timeout out.').code(504);
+        }
+        const response = await sender.sendCustomBody(draft, request.payload.body, {to: [sendTo]})
+        reply(response);
+      } catch (err) {
+        respondWithError(request, reply, err);
+      }
+    },
+  });
+
+  // Closes out a multi-send session by marking the sending draft as sent
+  // and moving it to the user's Sent folder.
+  server.route({
+    method: 'DELETE',
+    path: '/send-multiple/{draftId}',
+    config: {
+      validate: {
+        params: {
+          draftId: Joi.string(),
+        },
+      },
+    },
+    handler: async (request, reply) => {
+      try {
+        const account = request.auth.credentials;
+        const {draftId} = request.params;
+        SendingUtils.validateBase36(draftId);
+
+        const db = await LocalDatabaseConnector.forAccount(account.id);
+        const draft = await SendingUtils.findMultiSendDraft(draftId, db);
+
+        // gmail creates sent messages for each one, go through and delete them
+        if (account.provider === 'gmail') {
+          try {
+            // TODO: use type: "PermananentDeleteMessage" once it's fully implemented
+            await db.SyncbackRequest.create({
+              type: "DeleteMessage",
+              props: { messageId: draft.id },
+            });
+          } catch (err) {
+            // Even if this fails, we need to finish the multi-send session,
+            request.logger.error(err, err.logContext);
+          }
+        }
+
+        const sender = new SendmailClient(account, request.logger);
+        const rawMime = await sender.buildMime(draft);
+
+        await db.SyncbackRequest.create({
+          accountId: account.id,
+          type: "SaveSentMessage",
+          props: {rawMime},
         });
-      })
-    })},
+
+        await (draft.isSent = true);
+        const savedDraft = await draft.save();
+        reply(savedDraft.toJSON());
+      } catch (err) {
+        respondWithError(request, reply, err);
+      }
+    },
   });
 };

packages/local-sync/src/local-api/sending-utils.js

@@ -0,0 +1,132 @@
+const _ = require('underscore');
+
+const setReplyHeaders = (newMessage, prevMessage) => {
+  if (prevMessage.messageIdHeader) {
+    newMessage.inReplyTo = prevMessage.headerMessageId;
+    if (prevMessage.references) {
+      newMessage.references = prevMessage.references.concat(prevMessage.headerMessageId);
+    } else {
+      newMessage.references = [prevMessage.messageIdHeader];
+    }
+  }
+}
+
+class HTTPError extends Error {
+  constructor(message, httpCode, logContext) {
+    super(message);
+    this.httpCode = httpCode;
+    this.logContext = logContext;
+  }
+}
+
+module.exports = {
+  HTTPError,
+  findOrCreateMessageFromJSON: async (data, db, isDraft) => {
+    const {Thread, Message} = db;
+
+    const existingMessage = await Message.findById(data.id);
+    if (existingMessage) {
+      return existingMessage;
+    }
+
+    const {to, cc, bcc, from, replyTo, subject, body, account_id, date, id} = data;
+
+    const message = Message.build({
+      accountId: account_id,
+      from: from,
+      to: to,
+      cc: cc,
+      bcc: bcc,
+      replyTo: replyTo,
+      subject: subject,
+      body: body,
+      unread: true,
+      isDraft: isDraft,
+      isSent: false,
+      version: 0,
+      date: date,
+      id: id,
+    });
+
+    // TODO
+    // Attach files
+    // Update our contact list
+    // Add events
+    // Add metadata??
+
+    let replyToThread;
+    let replyToMessage;
+    if (data.thread_id != null) {
+      replyToThread = await Thread.find({
+        where: {id: data.thread_id},
+        include: [{
+          model: Message,
+          as: 'messages',
+          attributes: _.without(Object.keys(Message.attributes), 'body'),
+        }],
+      });
+    }
+    if (data.reply_to_message_id != null) {
+      replyToMessage = await Message.findById(data.reply_to_message_id);
+    }
+
+    if (replyToThread && replyToMessage) {
+      if (!replyToThread.messages.find((msg) => msg.id === replyToMessage.id)) {
+        throw new HTTPError(
+          `Message ${replyToMessage.id} is not in thread ${replyToThread.id}`,
+          400
+        )
+      }
+    }
+
+    let thread;
+    if (replyToMessage) {
+      setReplyHeaders(message, replyToMessage);
+      thread = await message.getThread();
+    } else if (replyToThread) {
+      thread = replyToThread;
+      const previousMessages = thread.messages.filter(msg => !msg.isDraft);
+      if (previousMessages.length > 0) {
+        const lastMessage = previousMessages[previousMessages.length - 1]
+        setReplyHeaders(message, lastMessage);
+      }
+    } else {
+      thread = Thread.build({
+        accountId: account_id,
+        subject: message.subject,
+        firstMessageDate: message.date,
+        lastMessageDate: message.date,
+        lastMessageSentDate: message.date,
+      })
+    }
+
+    const savedMessage = await message.save();
+    const savedThread = await thread.save();
+    await savedThread.addMessage(savedMessage);
+
+    return savedMessage;
+  },
+  findMultiSendDraft: async (draftId, db) => {
+    const draft = await db.Message.findById(draftId)
+    if (!draft) {
+      throw new HTTPError(`Couldn't find multi-send draft ${draftId}`, 400);
+    }
+    if (draft.isSent || !draft.isSending) {
+      throw new HTTPError(`Message ${draftId} is not a multi-send draft`, 400);
+    }
+    return draft;
+  },
+  validateRecipientsPresent: (draft) => {
+    const {to, cc, bcc} = draft;
+    const recipients = [].concat(to, cc, bcc);
+    if (recipients.length === 0) {
+      throw new HTTPError("No recipients specified", 400);
+    }
+  },
+  validateBase36: (value, name) => {
+    if (value == null) { return; }
+    if (isNaN(parseInt(value, 36))) {
+      throw new HTTPError(`${name} is not a base-36 integer`, 400)
+    }
+  },
+}

packages/local-sync/src/local-api/sendmail-client.js

@@ -0,0 +1,110 @@
+const nodemailer = require('nodemailer');
+const mailcomposer = require('mailcomposer');
+const {HTTPError} = require('./sending-utils');
+
+const MAX_RETRIES = 1;
+
+const formatParticipants = (participants) => {
+  return participants.map(p => `${p.name} <${p.email}>`).join(',');
+}
+
+class SendmailClient {
+  constructor(account, logger) {
+    this._transporter = nodemailer.createTransport(account.smtpConfig());
+    this._logger = logger;
+  }
+
+  async _send(msgData) {
+    let partialFailure;
+    let error;
+    for (let i = 0; i <= MAX_RETRIES; i++) {
+      try {
+        const results = await this._transporter.sendMail(msgData);
+        const {rejected, pending} = results;
+        if ((rejected && rejected.length > 0) || (pending && pending.length > 0)) {
+          // At least one recipient was rejected by the server,
+          // but at least one recipient got it. Don't retry; throw an
+          // error so that we fail to client.
+          partialFailure = new HTTPError(
+            'Sending to at least one recipient failed', 200, results);
+          throw partialFailure;
+        } else {
+          // Sending was successful!
+          return
+        }
+      } catch (err) {
+        error = err;
+        if (err === partialFailure) {
+          // We don't want to retry in this case, so re-throw the error
+          throw err;
+        }
+        this._logger.error(err);
+      }
+    }
+    this._logger.error('Max sending retries reached');
+
+    // TODO: figure out how to parse different errors, like in cloud-core
+    // https://github.com/nylas/cloud-core/blob/production/sync-engine/inbox/sendmail/smtp/postel.py#L354
+    throw new HTTPError('Sending failed', 500, error)
+  }
+
+  _draftToMsgData(draft) {
+    const msgData = {};
+    for (const field of ['from', 'to', 'cc', 'bcc']) {
+      if (draft[field]) {
+        msgData[field] = formatParticipants(draft[field])
+      }
+    }
+    msgData.subject = draft.subject;
+    msgData.html = draft.body;
+
+    // TODO: attachments
+
+    if (draft.replyTo) {
+      msgData.replyTo = formatParticipants(draft.replyTo);
+    }
+
+    msgData.inReplyTo = draft.inReplyTo;
+    msgData.references = draft.references;
+    msgData.headers = draft.headers;
+    msgData.headers['User-Agent'] = `NylasMailer-K2`
+
+    // TODO: do we want to set messageId or date?
+
+    return msgData;
+  }
+
+  async buildMime(draft) {
+    const builder = mailcomposer(this._draftToMsgData(draft))
+    return new Promise((resolve, reject) => {
+      builder.build((error, result) => {
+        error ? reject(error) : resolve(result)
+      })
+    })
+  }
+
+  async send(draft) {
+    if (draft.isSent) {
+      throw new Error(`Cannot send message ${draft.id}, it has already been sent`);
+    }
+    await this._send(this._draftToMsgData(draft));
+    await (draft.isSent = true);
+    await draft.save();
+  }
+
+  async sendCustomBody(draft, body, recipients) {
+    const origBody = draft.body;
+    draft.body = body;
+    const envelope = {};
+    for (const field of Object.keys(recipients)) {
+      envelope[field] = recipients[field].map(r => r.email);
+    }
+    const raw = await this.buildMime(draft);
+    const responseOnSuccess = draft.toJSON();
+    draft.body = origBody;
+    await this._send({raw, envelope})
+    return responseOnSuccess
+  }
+}
+
+module.exports = SendmailClient;

packages/local-sync/src/local-sync-worker/syncback-task-factory.js

@@ -39,8 +39,12 @@ class SyncbackTaskFactory {
         Task = require('./syncback_tasks/rename-folder.imap'); break;
       case "DeleteFolder":
         Task = require('./syncback_tasks/delete-folder.imap'); break;
+      case "DeleteMessage":
+        Task = require('./syncback_tasks/delete-message.imap'); break;
+      case "SaveSentMessage":
+        Task = require('./syncback_tasks/save-sent-message.imap'); break;
       default:
-        throw new Error(`Invalid Task Type: ${syncbackRequest.type}`)
+        throw new Error(`Task type not defined in syncback-task-factory: ${syncbackRequest.type}`)
     }
     return new Task(account, syncbackRequest)
   }

packages/local-sync/src/local-sync-worker/syncback_tasks/delete-message.imap.js

@@ -0,0 +1,18 @@
+const SyncbackTask = require('./syncback-task')
+const TaskHelpers = require('./task-helpers')
+
+class DeleteMessageIMAP extends SyncbackTask {
+  description() {
+    return `DeleteMessage`;
+  }
+
+  run(db, imap) {
+    const messageId = this.syncbackRequestObject().props.messageId
+
+    return TaskHelpers.openMessageBox({messageId, db, imap})
+      .then(({box, message}) => {
+        return box.addFlags(message.folderImapUID, 'DELETED')
+      })
+  }
+}
+module.exports = DeleteMessageIMAP;

packages/local-sync/src/local-sync-worker/syncback_tasks/perm-delete-message.imap.js

@@ -0,0 +1,24 @@
+const SyncbackTask = require('./syncback-task')
+
+class PermanentlyDeleteMessageIMAP extends SyncbackTask {
+  description() {
+    return `PermanentlyDeleteMessage`;
+  }
+
+  async run(db, imap) {
+    const messageId = this.syncbackRequestObject().props.messageId
+    const message = await db.Message.findById(messageId);
+    const folder = await db.Folder.findById(message.folderId);
+    const box = await imap.openBox(folder.name);
+    const result = await box.addFlags(message.folderImapUID, 'DELETED');
+    return result;
+
+    // TODO: We need to also delete the message from the trash
+    // if (folder.role === 'trash') { return result; }
+    //
+    // const trash = await db.Folder.find({where: {role: 'trash'}});
+    // const trashBox = await imap.openBox(trash.name);
+    // return await trashBox.addFlags(message.folderImapUID, 'DELETED');
+  }
+}
+module.exports = PermanentlyDeleteMessageIMAP;

packages/local-sync/src/local-sync-worker/syncback_tasks/save-sent-message.imap.js

@@ -0,0 +1,15 @@
+const SyncbackTask = require('./syncback-task')
+
+class SaveSentMessageIMAP extends SyncbackTask {
+  description() {
+    return `SaveSentMessage`;
+  }
+
+  async run(db, imap) {
+    // TODO: gmail doesn't have a sent folder
+    const folder = await db.Folder.find({where: {role: 'sent'}});
+    const box = await imap.openBox(folder.name);
+    return box.append(this.syncbackRequestObject().props.rawMime);
+  }
+}
+module.exports = SaveSentMessageIMAP;

packages/local-sync/src/models/folder.js

@@ -1,4 +1,4 @@
-const {DatabaseTypes: {JSONType}} = require('isomorphic-core');
+const {DatabaseTypes: {buildJSONColumnOptions}} = require('isomorphic-core');
 const {formatImapPath} = require('../shared/imap-paths-utils');
 
 module.exports = (sequelize, Sequelize) => {
@@ -8,7 +8,7 @@ module.exports = (sequelize, Sequelize) => {
     version: Sequelize.INTEGER,
     name: Sequelize.STRING,
     role: Sequelize.STRING,
-    syncState: JSONType('syncState'),
+    syncState: buildJSONColumnOptions('syncState'),
   }, {
     indexes: [
       {

packages/local-sync/src/models/message.js

@@ -1,7 +1,19 @@
 const cryptography = require('crypto');
 const {PromiseUtils, IMAPConnection} = require('isomorphic-core')
-const {DatabaseTypes: {JSONType, JSONARRAYType}} = require('isomorphic-core');
+const {DatabaseTypes: {buildJSONColumnOptions, buildJSONARRAYColumnOptions}} = require('isomorphic-core');
+const striptags = require('striptags');
+const SendingUtils = require('../local-api/sending-utils');
 
+const SNIPPET_LENGTH = 191;
+
+const getValidateArrayLength1 = (fieldName) => {
+  return (stringifiedArr) => {
+    const arr = JSON.parse(stringifiedArr);
+    if (arr.length !== 1) {
+      throw new Error(`Value for ${fieldName} must have a length of 1. Value: ${stringifiedArr}`);
+    }
+  };
+}
 
 module.exports = (sequelize, Sequelize) => {
   return sequelize.define('message', {
@@ -10,20 +22,55 @@ module.exports = (sequelize, Sequelize) => {
     version: Sequelize.INTEGER,
     headerMessageId: Sequelize.STRING,
     body: Sequelize.TEXT('long'),
-    headers: JSONType('headers'),
+    headers: buildJSONColumnOptions('headers'),
     subject: Sequelize.STRING(500),
     snippet: Sequelize.STRING(255),
     date: Sequelize.DATE,
+    isDraft: Sequelize.BOOLEAN,
+    isSent: {
+      type: Sequelize.BOOLEAN,
+      set: async function set(val) {
+        if (val) {
+          this.isDraft = false;
+          this.date = (new Date()).getTime();
+          const thread = await this.getThread();
+          await thread.updateFromMessage(this)
+        }
+        this.setDataValue('isSent', val);
+      },
+    },
     unread: Sequelize.BOOLEAN,
     starred: Sequelize.BOOLEAN,
     processed: Sequelize.INTEGER,
-    to: JSONARRAYType('to'),
-    from: JSONARRAYType('from'),
-    cc: JSONARRAYType('cc'),
-    bcc: JSONARRAYType('bcc'),
-    replyTo: JSONARRAYType('replyTo'),
+    to: buildJSONARRAYColumnOptions('to'),
+    from: Object.assign(buildJSONARRAYColumnOptions('from'), {
+      allowNull: true,
+      validate: {validateArrayLength1: getValidateArrayLength1('Message.from')},
+    }),
+    cc: buildJSONARRAYColumnOptions('cc'),
+    bcc: buildJSONARRAYColumnOptions('bcc'),
+    replyTo: Object.assign(buildJSONARRAYColumnOptions('replyTo'), {
+      allowNull: true,
+      validate: {validateArrayLength1: getValidateArrayLength1('Message.replyTo')},
+    }),
+    inReplyTo: { type: Sequelize.STRING, allowNull: true},
+    references: buildJSONARRAYColumnOptions('references'),
     folderImapUID: { type: Sequelize.STRING, allowNull: true},
     folderImapXGMLabels: { type: Sequelize.TEXT, allowNull: true},
+    isSending: {
+      type: Sequelize.BOOLEAN,
+      set: function set(val) {
+        if (val) {
+          if (this.isSent) {
+            throw new Error("Cannot mark a sent message as sending");
+          }
+          SendingUtils.validateRecipientsPresent(this);
+          this.isDraft = false;
+          this.regenerateHeaderMessageId();
+        }
+        this.setDataValue('isSending', val);
+      },
+    },
   }, {
     indexes: [
       {
@@ -69,6 +116,13 @@ module.exports = (sequelize, Sequelize) => {
         })
       },
 
+      // The uid in this header is simply the draft id and version concatenated.
+      // Because this uid identifies the draft on the remote provider, we
+      // regenerate it on each draft revision so that we can delete the old draft
+      // and add the new one on the remote.
+      regenerateHeaderMessageId() {
+        this.headerMessageId = `<${this.id}-${this.version}@mailer.nylas.com>`
+      },
       toJSON() {
         if (this.folder_id && !this.folder) {
           throw new Error("Message.toJSON called on a message where folder were not eagerly loaded.")
@@ -99,5 +153,15 @@ module.exports = (sequelize, Sequelize) => {
         };
       },
     },
+    hooks: {
+      beforeUpdate: (message) => {
+        // Update the snippet if the body has changed
+        if (!message.changed('body')) { return; }
+
+        const plainText = striptags(message.body);
+        // consolidate whitespace groups into single spaces and then truncate
+        message.snippet = plainText.split(/\s+/).join(" ").substring(0, SNIPPET_LENGTH)
+      },
+    },
   });
 };

packages/local-sync/src/models/syncbackRequest.js

@@ -1,4 +1,4 @@
-const {DatabaseTypes: {JSONType}} = require('isomorphic-core');
+const {DatabaseTypes: {buildJSONColumnOptions}} = require('isomorphic-core');
 
 module.exports = (sequelize, Sequelize) => {
   return sequelize.define('syncbackRequest', {
@@ -8,8 +8,8 @@ module.exports = (sequelize, Sequelize) => {
       defaultValue: "NEW",
       allowNull: false,
     },
-    error: JSONType('error'),
-    props: JSONType('props'),
+    error: buildJSONColumnOptions('error'),
+    props: buildJSONColumnOptions('props'),
     accountId: { type: Sequelize.STRING, allowNull: false },
   }, {
     instanceMethods: {

packages/local-sync/src/models/thread.js

@@ -1,4 +1,4 @@
-const {DatabaseTypes: {JSONARRAYType}} = require('isomorphic-core');
+const {DatabaseTypes: {buildJSONARRAYColumnOptions}} = require('isomorphic-core');
 
 module.exports = (sequelize, Sequelize) => {
   return sequelize.define('thread', {
@@ -8,13 +8,19 @@ module.exports = (sequelize, Sequelize) => {
     remoteThreadId: Sequelize.STRING,
     subject: Sequelize.STRING(500),
     snippet: Sequelize.STRING(255),
-    unreadCount: Sequelize.INTEGER,
-    starredCount: Sequelize.INTEGER,
+    unreadCount: {
+      type: Sequelize.INTEGER,
+      get: function get() { return this.getDataValue('unreadCount') || 0 },
+    },
+    starredCount: {
+      type: Sequelize.INTEGER,
+      get: function get() { return this.getDataValue('starredCount') || 0 },
+    },
     firstMessageDate: Sequelize.DATE,
     lastMessageDate: Sequelize.DATE,
     lastMessageReceivedDate: Sequelize.DATE,
     lastMessageSentDate: Sequelize.DATE,
-    participants: JSONARRAYType('participants'),
+    participants: buildJSONARRAYColumnOptions('participants'),
   }, {
     indexes: [
       { fields: ['subject'] },
@@ -56,7 +62,72 @@ module.exports = (sequelize, Sequelize) => {
 
         return this.save();
       },
+      async updateFromMessage(message) {
+        if (message.isDraft) {
+          return this;
+        }
 
+        if (!(message.labels instanceof Array)) {
+          throw new Error("Expected message.labels to be an inflated array.");
+        }
+        if (!message.folder) {
+          throw new Error("Expected message.folder value to be present.");
+        }
+
+        // Update thread participants
+        const {to, cc, bcc} = message;
+        const participantEmails = this.participants.map(contact => contact.email);
+        const newParticipants = to.concat(cc, bcc).filter(contact => {
+          if (participantEmails.includes(contact.email)) {
+            return false;
+          }
+          participantEmails.push(contact.email);
+          return true;
+        })
+        this.participants = this.participants.concat(newParticipants);
+
+        // Update starred/unread counts
+        this.starredCount += message.starred ? 1 : 0;
+        this.unreadCount += message.unread ? 1 : 0;
+
+        // Update dates/snippet
+        if (!this.lastMessageDate || (message.date > this.lastMessageDate)) {
+          this.lastMessageDate = message.date;
+          this.snippet = message.snippet;
+        }
+        if (!this.firstMessageDate || (message.date < this.firstMessageDate)) {
+          this.firstMessageDate = message.date;
+        }
+
+        // Figure out if the message is sent or received and update more dates
+        const isSent = (
+          message.folder.role === 'sent' ||
+          !!message.labels.find(l => l.role === 'sent')
+        );
+
+        if (isSent && ((message.date > this.lastMessageSentDate) || !this.lastMessageSentDate)) {
+          this.lastMessageSentDate = message.date;
+        }
+        if (!isSent && ((message.date > this.lastMessageReceivedDate) || !this.lastMessageReceivedDate)) {
+          this.lastMessageReceivedDate = message.date;
+        }
+
+        const savedThread = await this.save();
+
+        // Update folders/labels
+        // This has to be done after the thread has been saved, because the
+        // thread may not have had an assigned id yet. addFolder()/addLabel()
+        // need an existing thread id to work properly.
+        if (!savedThread.folders.find(f => f.id === message.folderId)) {
+          await savedThread.addFolder(message.folder)
+        }
+        for (const label of message.labels) {
+          if (!savedThread.labels.find(l => l.id === label)) {
+            await savedThread.addLabel(label)
+          }
+        }
+        return savedThread;
+      },
       toJSON() {
         if (!(this.labels instanceof Array)) {
           throw new Error("Thread.toJSON called on a thread where labels were not eagerly loaded.")

packages/local-sync/src/new-message-processor/detect-thread.js

@@ -35,6 +35,7 @@ function emptyThread({Thread, accountId}, options = {}) {
   const t = Thread.build(Object.assign({accountId}, options))
   t.folders = [];
   t.labels = [];
+  t.participants = [];
   return Promise.resolve(t)
 }
 
@@ -95,66 +96,15 @@ function detectThread({db, message}) {
 
     // update the basic properties of the thread
     thread.accountId = message.accountId;
+
     // Threads may, locally, have the ID of any message within the thread
     // (message IDs are globally unique, even across accounts!)
     if (!thread.id) {
       thread.id = `t:${message.id}`
     }
 
-    // update the participants on the thread
-    const threadParticipants = [].concat(thread.participants);
-    const threadEmails = thread.participants.map(p => p.email);
-
-    for (const p of [].concat(message.to, message.cc, message.from)) {
-      if (!threadEmails.includes(p.email)) {
-        threadParticipants.push(p);
-        threadEmails.push(p.email);
-      }
-    }
-    thread.participants = threadParticipants;
-
-    // update starred and unread
-    if (thread.starredCount == null) { thread.starredCount = 0; }
-    thread.starredCount += message.starred ? 1 : 0;
-    if (thread.unreadCount == null) { thread.unreadCount = 0; }
-    thread.unreadCount += message.unread ? 1 : 0;
-
-    // update dates
-    if (!thread.lastMessageDate || (message.date > thread.lastMessageDate)) {
-      thread.lastMessageDate = message.date;
-      thread.snippet = message.snippet;
-      thread.subject = cleanSubject(message.subject);
-    }
-    if (!thread.firstMessageDate || (message.date < thread.firstMessageDate)) {
-      thread.firstMessageDate = message.date;
-    }
-
-    const isSent = (
-      message.folder.role === 'sent' ||
-      !!message.labels.find(l => l.role === 'sent')
-    )
-
-    if (isSent && ((message.date > thread.lastMessageSentDate) || !thread.lastMessageSentDate)) {
-      thread.lastMessageSentDate = message.date;
-    }
-    if (!isSent && ((message.date > thread.lastMessageReceivedDate) || !thread.lastMessageReceivedDate)) {
-      thread.lastMessageReceivedDate = message.date;
-    }
-
-    return thread.save()
-    .then((saved) => {
-      const promises = []
-      // update folders and labels
-      if (!saved.folders.find(f => f.id === message.folderId)) {
-        promises.push(saved.addFolder(message.folder))
-      }
-      for (const label of message.labels) {
-        if (!saved.labels.find(l => l.id === label)) {
-          promises.push(saved.addLabel(label))
-        }
-      }
-      return Promise.all(promises).thenReturn(saved)
-    })
+    thread.subject = cleanSubject(message.subject);
+    return thread.updateFromMessage(message);
   });
 }