Merge remote-tracking branch 'private/for-merging' into merge-private

Author: jstejada

packages/client-app/src/flux/attributes/attribute-joined-data.es6

@@ -33,9 +33,25 @@ Section: Database
 export default class AttributeJoinedData extends Attribute {
   static NullPlaceholder = NullPlaceholder;
 
-  constructor({modelKey, jsonKey, modelTable, queryable}) {
+  constructor({modelKey, jsonKey, modelTable, queryable, serializeFn, deserializeFn}) {
     super({modelKey, jsonKey, queryable});
     this.modelTable = modelTable;
+    this.serializeFn = serializeFn;
+    this.deserializeFn = deserializeFn;
+  }
+
+  serialize(thisValue, val) {
+    if (this.serializeFn) {
+      return this.serializeFn.call(thisValue, val);
+    }
+    return val;
+  }
+
+  deserialize(thisValue, val) {
+    if (this.deserializeFn) {
+      return this.deserializeFn.call(thisValue, val);
+    }
+    return val;
   }
 
   toJSON(val) {

packages/client-app/src/flux/models/message.es6

@@ -1,5 +1,6 @@
 import _ from 'underscore'
 import moment from 'moment'
+import {MessageBodyUtils} from 'isomorphic-core'
 
 import File from './file'
 import Utils from './utils'
@@ -104,6 +105,20 @@ export default class Message extends ModelWithMetadata {
     body: Attributes.JoinedData({
       modelTable: 'MessageBody',
       modelKey: 'body',
+      serializeFn: function serializeBody(val) {
+        return MessageBodyUtils.writeBody({
+          msgId: this.id,
+          body: val,
+          forceWrite: false,
+        });
+      },
+      deserializeFn: function deserializeBody(val) {
+        const result = MessageBodyUtils.tryReadBody(val);
+        if (result) {
+          return result;
+        }
+        return val;
+      },
     }),
 
     files: Attributes.Collection({

packages/client-app/src/flux/models/query.es6

@@ -310,7 +310,7 @@ export default class ModelQuery {
           if (value === AttributeJoinedData.NullPlaceholder) {
             value = null;
           }
-          object[attr.modelKey] = value;
+          object[attr.modelKey] = attr.deserialize(object, value);
         }
         return object;
       });

packages/client-app/src/flux/stores/database-writer.es6

@@ -322,8 +322,9 @@ export default class DatabaseWriter {
 
     joinedDataAttributes.forEach((attr) => {
       for (const model of models) {
-        if (model[attr.modelKey] !== undefined) {
-          promises.push(this._query(`REPLACE INTO \`${attr.modelTable}\` (\`id\`, \`value\`) VALUES (?, ?)`, [model.id, model[attr.modelKey]]));
+        const value = model[attr.modelKey];
+        if (value !== undefined) {
+          promises.push(this._query(`REPLACE INTO \`${attr.modelTable}\` (\`id\`, \`value\`) VALUES (?, ?)`, [model.id, attr.serialize(model, value)]));
         }
       }
     });

packages/client-sync/src/message-processor/index.js

@@ -233,7 +233,13 @@ class MessageProcessor {
 
     const thread = await detectThread({db, messageValues});
     messageValues.threadId = thread.id;
-    const createdMessage = await Message.create(messageValues);
+    // The way that sequelize initializes objects doesn't guarantee that the
+    // object will have a value for `id` before initializing the `body` field
+    // (which we now depend on). By using `build` instead of `create`, we can
+    // initialize an object with just the `id` field and then use `update` to
+    // initialize the remaining fields and save the object to the database.
+    const createdMessage = Message.build({id: messageValues.id});
+    await createdMessage.update(messageValues);
 
     if (messageValues.labels) {
       await createdMessage.addLabels(messageValues.labels)

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

@@ -3,14 +3,14 @@ const {
   ExponentialBackoffScheduler,
   IMAPErrors,
   IMAPConnectionPool,
+  MessageBodyUtils,
 } = require('isomorphic-core')
 const {DatabaseTypes: {JSONArrayColumn}} = require('isomorphic-core');
 const {Errors: {APIError}} = require('isomorphic-core')
 const {Actions} = require('nylas-exports')
 
 const MAX_IMAP_TIMEOUT_ERRORS = 5;
 
-
 function validateRecipientsPresent(message) {
   if (message.getRecipients().length === 0) {
     throw new APIError(`No recipients specified`, 400);
@@ -25,7 +25,23 @@ module.exports = (sequelize, Sequelize) => {
     headerMessageId: { type: Sequelize.STRING, allowNull: true },
     gMsgId: { type: Sequelize.STRING, allowNull: true },
     gThrId: { type: Sequelize.STRING, allowNull: true },
-    body: Sequelize.TEXT,
+    body: {
+      type: Sequelize.TEXT,
+      get: function getBody() {
+        const val = this.getDataValue('body');
+        const result = MessageBodyUtils.tryReadBody(val);
+        if (result) {
+          return result;
+        }
+        return val;
+      },
+      set: function setBody(val) {
+        this.setDataValue('body', MessageBodyUtils.writeBody({
+          msgId: this.id,
+          body: val,
+        }));
+      },
+    },
     subject: Sequelize.STRING(500),
     snippet: Sequelize.STRING(255),
     date: Sequelize.DATE,

packages/isomorphic-core/index.js

@@ -14,6 +14,7 @@ module.exports = {
   DatabaseTypes: require('./src/database-types'),
   IMAPConnection: require('./src/imap-connection').default,
   IMAPConnectionPool: require('./src/imap-connection-pool'),
+  MessageBodyUtils: require('./src/message-body-utils'),
   SendmailClient: require('./src/sendmail-client'),
   DeltaStreamBuilder: require('./src/delta-stream-builder'),
   HookTransactionLog: require('./src/hook-transaction-log'),

packages/isomorphic-core/src/message-body-utils.es6

@@ -0,0 +1,92 @@
+const fs = require('fs')
+const mkdirp = require('mkdirp')
+const path = require('path')
+const zlib = require('zlib')
+
+const MAX_PATH_DIRS = 5;
+const FILE_EXTENSION = 'nylasmail'
+
+function baseMessagePath() {
+  return path.join(process.env.NYLAS_HOME, 'messages');
+}
+
+export function tryReadBody(val) {
+  try {
+    const parsed = JSON.parse(val);
+    if (parsed && parsed.path && parsed.path.startsWith(baseMessagePath())) {
+      if (parsed.compressed) {
+        return zlib.gunzipSync(fs.readFileSync(parsed.path)).toString();
+      }
+      return fs.readFileSync(parsed.path, {encoding: 'utf8'});
+    }
+  } catch (err) {
+    console.warn('Got error while trying to parse body path, assuming we need to migrate', err);
+  }
+  return null;
+}
+
+export function pathForBodyFile(msgId) {
+  const pathGroups = [];
+  let remainingId = msgId;
+  while (pathGroups.length < MAX_PATH_DIRS) {
+    pathGroups.push(remainingId.substring(0, 2));
+    remainingId = remainingId.substring(2);
+  }
+  const bodyPath = path.join(...pathGroups);
+  return path.join(baseMessagePath, bodyPath, `${remainingId}.${FILE_EXTENSION}`);
+}
+
+// NB: The return value of this function is what gets written into the database.
+export function writeBody({msgId, body, forceWrite = true} = {}) {
+  const bodyPath = pathForBodyFile(msgId);
+  const bodyDir = path.dirname(bodyPath);
+
+  const compressedBody = zlib.gzipSync(body);
+  const dbEntry = {
+    path: bodyPath,
+    compressed: true,
+  };
+
+  // It's possible that gzipping actually makes the body larger. If that's the
+  // case then just write the uncompressed body instead.
+  let bodyToWrite = compressedBody;
+  if (compressedBody.length >= body.length) {
+    dbEntry.compressed = false;
+    bodyToWrite = body;
+  }
+
+  const result = JSON.stringify(dbEntry);
+  // If the JSON db entry would be longer than the body itself then just write
+  // the body directly into the database.
+  if (result.length > body.length) {
+    return body;
+  }
+
+  try {
+    let exists;
+
+    // If we don't have to write to the file and it already exists then don't.
+    if (!forceWrite) {
+      exists = fs.existsSync(bodyPath);
+      if (exists) {
+        return result;
+      }
+    }
+
+    // We want to minimize the number of times we interact with the filesystem
+    // since it can be slow.
+    if (exists === undefined) {
+      exists = fs.existsSync(bodyPath);
+    }
+    if (!exists) {
+      mkdirp.sync(bodyDir);
+    }
+
+    fs.writeFileSync(bodyPath, bodyToWrite);
+    return result;
+  } catch (err) {
+    // If anything bad happens while trying to write to disk just store the
+    // body in the database.
+    return body;
+  }
+}

scripts/postinstall.es6

@@ -32,7 +32,16 @@ function copyErrorLoggerExtensions(privateDir) {
   fs.copySync(from, to);
 }
 
-async function installPrivateResources() {
+function installClientSyncPackage() {
+  console.log("\n---> Linking client-sync")
+  // link client-sync
+  const clientSyncDir = path.resolve(path.join('packages', 'client-sync'));
+  const destination = path.resolve(path.join('packages', 'client-app', 'internal_packages', 'client-sync'));
+  unlinkIfExistsSync(destination);
+  fs.symlinkSync(clientSyncDir, destination, 'dir');
+}
+
+function installPrivateResources() {
   console.log("\n---> Linking private plugins")
   const privateDir = path.resolve(path.join('packages', 'client-private-plugins'))
   if (!fs.existsSync(privateDir)) {
@@ -49,12 +58,6 @@ async function installPrivateResources() {
     unlinkIfExistsSync(to);
     fs.symlinkSync(from, to, 'dir');
   }
-
-  // link client-sync
-  const clientSyncDir = path.resolve(path.join('packages', 'client-sync'));
-  const destination = path.resolve(path.join('packages', 'client-app', 'internal_packages', 'client-sync'));
-  unlinkIfExistsSync(destination);
-  fs.symlinkSync(clientSyncDir, destination, 'dir');
 }
 
 async function lernaBootstrap(installTarget) {
@@ -124,11 +127,16 @@ function linkJasmineConfigs() {
   console.log("\n---> Linking Jasmine configs");
   const linkToPackages = ['cloud-api', 'cloud-core', 'cloud-workers']
   const from = getJasmineConfigPath('isomorphic-core')
-
   for (const packageName of linkToPackages) {
-    const dir = getJasmineDir(packageName)
-    if (!fs.existsSync(dir)) {
-      fs.mkdirSync(dir)
+    const packageDir = path.join('packages', packageName)
+    if (!fs.existsSync(packageDir)) {
+      console.log("\n---> No cloud packages to link. Moving on")
+      return
+    }
+
+    const jasmineDir = getJasmineDir(packageName)
+    if (!fs.existsSync(jasmineDir)) {
+      fs.mkdirSync(jasmineDir)
     }
     const to = getJasmineConfigPath(packageName)
     unlinkIfExistsSync(to)
@@ -161,7 +169,8 @@ async function main() {
     console.log(`\n---> Installing for target ${installTarget}`);
 
     if ([TARGET_ALL, TARGET_CLIENT].includes(installTarget)) {
-      await installPrivateResources()
+      installPrivateResources()
+      installClientSyncPackage()
     }
 
     await lernaBootstrap(installTarget);