Upgrade Message Schema (Data) in Background (#2162)

In order to avoid incurring long startup times, we migrate message schema (data) in the background using `window.requestIdleCallback` API. The migration moves attachment data from IndexedDB to disk and reduces load on our database which may cause data loss (#1589).

On my development profile, this migration reduced the IndexedDB directory from 33.4MB to 4.7MB.

js/background.js

@@ -14,6 +14,7 @@
 ;(function() {
     'use strict';
 
+    const { IdleDetector, MessageDataMigrator } = Signal.Workflow;
     const { Errors, Message } = window.Signal.Types;
     const { upgradeMessageSchema } = window.Signal.Migrations;
 
@@ -75,6 +76,28 @@
 
     storage.fetch();
 
+
+  /* eslint-enable */
+  /* jshint ignore:start */
+  const NUM_MESSAGE_UPGRADES_PER_IDLE = 2;
+  const idleDetector = new IdleDetector();
+  idleDetector.on('idle', async () => {
+    const results = await MessageDataMigrator.processNext({
+      BackboneMessage: Whisper.Message,
+      BackboneMessageCollection: Whisper.MessageCollection,
+      count: NUM_MESSAGE_UPGRADES_PER_IDLE,
+      upgradeMessageSchema,
+      wrapDeferred,
+    });
+    console.log('Upgrade message schema:', results);
+
+    if (!results.hasMore) {
+      idleDetector.stop();
+    }
+  });
+  /* jshint ignore:end */
+  /* eslint-disable */
+
     // We need this 'first' check because we don't want to start the app up any other time
     //   than the first time. And storage.fetch() will cause onready() to fire.
     var first = true;
@@ -85,9 +108,12 @@
         first = false;
 
         ConversationController.load().then(start, start);
+        idleDetector.start();
     });
 
     Whisper.events.on('shutdown', function() {
+      idleDetector.stop();
+
       if (messageReceiver) {
         messageReceiver.close().then(function() {
           Whisper.events.trigger('shutdown-complete');

js/models/messages.js

@@ -148,6 +148,7 @@
           this.revokeImageUrl();
           const attachments = this.get('attachments');
           await Promise.all(attachments.map(deleteAttachmentData));
+          return;
         },
         /* jshint ignore:end */
         /* eslint-disable */

js/modules/idle_detector.js

@@ -0,0 +1,46 @@
+/* eslint-env browser */
+
+const EventEmitter = require('events');
+
+
+const POLL_INTERVAL_MS = 30 * 1000;
+const IDLE_THRESHOLD_MS = 25;
+
+class IdleDetector extends EventEmitter {
+  constructor() {
+    super();
+    this.handle = null;
+    this.timeoutId = null;
+  }
+
+  start() {
+    this._scheduleNextCallback();
+  }
+
+  stop() {
+    if (this.handle) {
+      cancelIdleCallback(this.handle);
+    }
+
+    if (this.timeoutId) {
+      clearTimeout(this.timeoutId);
+    }
+  }
+
+  _scheduleNextCallback() {
+    this.stop();
+    this.handle = window.requestIdleCallback((deadline) => {
+      const { didTimeout } = deadline;
+      const timeRemaining = deadline.timeRemaining();
+      const isIdle = timeRemaining >= IDLE_THRESHOLD_MS;
+      if (isIdle || didTimeout) {
+        this.emit('idle', { timestamp: Date.now(), didTimeout, timeRemaining });
+      }
+      this.timeoutId = setTimeout(() => this._scheduleNextCallback(), POLL_INTERVAL_MS);
+    });
+  }
+}
+
+module.exports = {
+  IdleDetector,
+};

js/modules/messages_data_migrator.js

@@ -0,0 +1,99 @@
+const isNumber = require('lodash/isNumber');
+const isFunction = require('lodash/isFunction');
+const Message = require('./types/message');
+
+
+const processNext = async ({
+  BackboneMessage,
+  BackboneMessageCollection,
+  count,
+  upgradeMessageSchema,
+  wrapDeferred,
+} = {}) => {
+  if (!isFunction(BackboneMessage)) {
+    throw new TypeError('`BackboneMessage` (Whisper.Message) constructor is required');
+  }
+
+  if (!isFunction(BackboneMessageCollection)) {
+    throw new TypeError('`BackboneMessageCollection` (Whisper.MessageCollection)' +
+      ' constructor is required');
+  }
+
+  if (!isNumber(count)) {
+    throw new TypeError('`count` is required');
+  }
+
+  if (!isFunction(upgradeMessageSchema)) {
+    throw new TypeError('`upgradeMessageSchema` is required');
+  }
+
+  if (!isFunction(wrapDeferred)) {
+    throw new TypeError('`wrapDeferred` is required');
+  }
+
+  const startTime = Date.now();
+
+  const startFetchTime = Date.now();
+  const messagesRequiringSchemaUpgrade =
+    await _fetchMessagesRequiringSchemaUpgrade({ BackboneMessageCollection, count });
+  const fetchDuration = Date.now() - startFetchTime;
+
+  const startUpgradeTime = Date.now();
+  const upgradedMessages =
+    await Promise.all(messagesRequiringSchemaUpgrade.map(upgradeMessageSchema));
+  const upgradeDuration = Date.now() - startUpgradeTime;
+
+  const startSaveTime = Date.now();
+  const saveMessage = _saveMessage({ BackboneMessage, wrapDeferred });
+  await Promise.all(upgradedMessages.map(saveMessage));
+  const saveDuration = Date.now() - startSaveTime;
+
+  const totalDuration = Date.now() - startTime;
+  const numProcessed = messagesRequiringSchemaUpgrade.length;
+  const hasMore = numProcessed > 0;
+  return {
+    hasMore,
+    numProcessed,
+    fetchDuration,
+    upgradeDuration,
+    saveDuration,
+    totalDuration,
+  };
+};
+
+const _saveMessage = ({ BackboneMessage, wrapDeferred } = {}) => (message) => {
+  const backboneMessage = new BackboneMessage(message);
+  return wrapDeferred(backboneMessage.save());
+};
+
+const _fetchMessagesRequiringSchemaUpgrade =
+  async ({ BackboneMessageCollection, count } = {}) => {
+    if (!isFunction(BackboneMessageCollection)) {
+      throw new TypeError('`BackboneMessageCollection` (Whisper.MessageCollection)' +
+        ' constructor is required');
+    }
+
+    if (!isNumber(count)) {
+      throw new TypeError('`count` is required');
+    }
+
+    const collection = new BackboneMessageCollection();
+    return new Promise(resolve => collection.fetch({
+      limit: count,
+      index: {
+        name: 'schemaVersion',
+        upper: Message.CURRENT_SCHEMA_VERSION,
+        excludeUpper: true,
+        order: 'desc',
+      },
+    }).always(() => {
+      const models = collection.models || [];
+      const messages = models.map(model => model.toJSON());
+      resolve(messages);
+    }));
+  };
+
+
+module.exports = {
+  processNext,
+};

js/modules/types/attachment.js

@@ -29,8 +29,9 @@ const { migrateDataToFileSystem } = require('./attachment/migrate_data_to_file_s
 //   schemaVersion: integer
 // }
 
-// Returns true if `rawAttachment` is a valid attachment based on our (limited)
+// Returns true if `rawAttachment` is a valid attachment based on our current schema.
-// criteria. Over time, we can expand this definition to become more narrow:
+// Over time, we can expand this definition to become more narrow, e.g. require certain
+// fields, etc.
 exports.isValid = (rawAttachment) => {
   // NOTE: We cannot use `_.isPlainObject` because `rawAttachment` is
   // deserialized by protobuf:
@@ -38,10 +39,7 @@ exports.isValid = (rawAttachment) => {
     return false;
   }
 
-  const hasValidContentType = isString(rawAttachment.contentType);
+  return true;
-  const hasValidFileName =
-    isString(rawAttachment.fileName) || rawAttachment.fileName === null;
-  return hasValidContentType && hasValidFileName;
 };
 
 // Upgrade steps

js/modules/types/message.js

@@ -166,8 +166,13 @@ const toVersion3 = exports._withSchemaVersion(
 );
 
 // UpgradeStep
-exports.upgradeSchema = async (message, { writeAttachmentData } = {}) =>
+exports.upgradeSchema = async (message, { writeAttachmentData } = {}) => {
-  toVersion3(
+  if (!isFunction(writeAttachmentData)) {
+    throw new TypeError('`context.writeAttachmentData` is required');
+  }
+
+  return toVersion3(
     await toVersion2(await toVersion1(await toVersion0(message))),
     { writeAttachmentData }
   );
+};

preload.js

@@ -122,22 +122,28 @@
   const upgradeMessageSchema = message =>
     Message.upgradeSchema(message, upgradeSchemaContext);
 
+  const { IdleDetector} = require('./js/modules/idle_detector');
+
   window.Signal = window.Signal || {};
-  window.Signal.Logs = require('./js/modules/logs');
-  window.Signal.OS = require('./js/modules/os');
   window.Signal.Backup = require('./js/modules/backup');
   window.Signal.Crypto = require('./js/modules/crypto');
+  window.Signal.Logs = require('./js/modules/logs');
   window.Signal.Migrations = {};
   window.Signal.Migrations.loadAttachmentData = Attachment.loadData(readAttachmentData);
   window.Signal.Migrations.deleteAttachmentData = Attachment.deleteData(deleteAttachmentData);
   window.Signal.Migrations.upgradeMessageSchema = upgradeMessageSchema;
   window.Signal.Migrations.V17 = require('./js/modules/migrations/17');
+  window.Signal.OS = require('./js/modules/os');
   window.Signal.Types = window.Signal.Types || {};
   window.Signal.Types.Attachment = Attachment;
   window.Signal.Types.Errors = require('./js/modules/types/errors');
   window.Signal.Types.Message = Message;
   window.Signal.Types.MIME = require('./js/modules/types/mime');
   window.Signal.Types.Settings = require('./js/modules/types/settings');
+  window.Signal.Workflow = {};
+  window.Signal.Workflow.IdleDetector = IdleDetector;
+  window.Signal.Workflow.MessageDataMigrator =
+    require('./js/modules/messages_data_migrator');
 
   // We pull this in last, because the native module involved appears to be sensitive to
   //   /tmp mounted as noexec on Linux.