Create IndexedDB index from schemaVersion to Message (#2128)

.eslintignore

@@ -16,6 +16,7 @@ test/views/*.js
 
 # ES2015+ files
 !js/background.js
+!js/database.js
 !js/logging.js
 !js/models/conversations.js
 !js/views/attachment_view.js

js/database.js

@@ -8,6 +8,8 @@
 (function () {
   'use strict';
 
+  const { Migrations } = window.Signal;
+
   window.Whisper = window.Whisper || {};
   window.Whisper.Database = window.Whisper.Database || {};
   window.Whisper.Database.id = window.Whisper.Database.id || 'signal';
@@ -127,7 +129,7 @@
     {
       version: '12.0',
       migrate(transaction, next) {
-        console.log('migration 1.0');
+        console.log('migration 12.0');
         console.log('creating object stores');
         const messages = transaction.db.createObjectStore('messages');
         messages.createIndex('conversation', ['conversationId', 'received_at'], {
@@ -233,5 +235,22 @@
         next();
       },
     },
+    {
+      version: 17,
+      async migrate(transaction, next) {
+        console.log('migration 17');
+        console.log('Start migration to database version 17');
+
+        const start = Date.now();
+        await Migrations.V17.run(transaction);
+        const duration = Date.now() - start;
+
+        console.log(
+          'Complete migration to database version 17.',
+          `Duration: ${duration}ms`
+        );
+        next();
+      },
+    },
   ];
 }());

js/models/conversations.js

@@ -617,18 +617,17 @@
           now
         );
 
-        const upgradedAttachments =
+        const messageWithSchema = await Message.upgradeSchema({
-          await Promise.all(attachments.map(Attachment.upgradeSchema));
+          type: 'outgoing',
-        const message = this.messageCollection.add({
           body,
           conversationId: this.id,
-          type: 'outgoing',
+          attachments,
-          attachments: upgradedAttachments,
           sent_at: now,
           received_at: now,
           expireTimer: this.get('expireTimer'),
           recipients: this.getRecipients(),
         });
+        const message = this.messageCollection.add(messageWithSchema);
         if (this.isPrivate()) {
           message.set({ destination: this.id });
         }
@@ -641,7 +640,7 @@
         });
 
         const conversationType = this.get('type');
-        const sendFunc = (() => {
+        const sendFunction = (() => {
           switch (conversationType) {
             case Message.PRIVATE:
               return textsecure.messaging.sendMessageToNumber;
@@ -657,10 +656,10 @@
           profileKey = storage.get('profileKey');
         }
 
-        message.send(sendFunc(
+        message.send(sendFunction(
           this.get('id'),
           body,
-          upgradedAttachments,
+          messageWithSchema.attachments,
           now,
           this.get('expireTimer'),
           profileKey

js/modules/migrations/17/index.js

@@ -0,0 +1,55 @@
+const Message = require('../../types/message');
+
+
+exports.run = async (transaction) => {
+  const messagesStore = transaction.objectStore('messages');
+
+  console.log('Initialize messages schema version');
+  const numUpgradedMessages = await _initializeMessageSchemaVersion(messagesStore);
+  console.log('Complete messages schema version initialization', { numUpgradedMessages });
+
+  console.log('Create index from attachment schema version to attachment');
+  messagesStore.createIndex('schemaVersion', 'schemaVersion', { unique: false });
+};
+
+const _initializeMessageSchemaVersion = messagesStore =>
+  new Promise((resolve, reject) => {
+    const messagePutOperations = [];
+
+    const cursorRequest = messagesStore.openCursor();
+    cursorRequest.onsuccess = async (event) => {
+      const cursor = event.target.result;
+      const hasMoreData = Boolean(cursor);
+      if (!hasMoreData) {
+        await Promise.all(messagePutOperations);
+        return resolve(messagePutOperations.length);
+      }
+
+      const message = cursor.value;
+      const messageWithSchemaVersion = Message.initializeSchemaVersion(message);
+      messagePutOperations.push(putItem(
+        messagesStore,
+        messageWithSchemaVersion,
+        messageWithSchemaVersion.id
+      ));
+
+      return cursor.continue();
+    };
+
+    cursorRequest.onerror = event =>
+      reject(event.target.error);
+  });
+
+//    putItem :: IDBObjectStore -> Item -> Key -> Promise Item
+const putItem = (store, item, key) =>
+  new Promise((resolve, reject) => {
+    try {
+      const request = store.put(item, key);
+      request.onsuccess = event =>
+        resolve(event.target.result);
+      request.onerror = event =>
+        reject(event.target.error);
+    } catch (error) {
+      reject(error);
+    }
+  });

js/modules/types/attachment.js

@@ -1,33 +1,16 @@
-const isFunction = require('lodash/isFunction');
-const isNumber = require('lodash/isNumber');
 const isString = require('lodash/isString');
-const isUndefined = require('lodash/isUndefined');
 
 const MIME = require('./mime');
 const { arrayBufferToBlob, blobToArrayBuffer, dataURLToBlob } = require('blob-util');
 const { autoOrientImage } = require('../auto_orient_image');
 
-// Increment this version number every time we change how attachments are upgraded. This
-// will allow us to retroactively upgrade existing attachments. As we add more upgrade
-// steps, we could design a pipeline that does this incrementally, e.g. from
-// version 0 / unknown -> 1, 1 --> 2, etc., similar to how we do database migrations:
-exports.CURRENT_SCHEMA_VERSION = 2;
-
-// Schema version history
-//
-// Version 1
-//   - Auto-orient JPEG attachments using EXIF `Orientation` data
-//   - Add `schemaVersion` property
-// Version 2
-//   - Sanitize Unicode order override characters
-
 // // Incoming message attachment fields
 // {
 //   id: string
 //   contentType: MIMEType
 //   data: ArrayBuffer
 //   digest: ArrayBuffer
-//   fileName: string
+//   fileName: string | null
 //   flags: null
 //   key: ArrayBuffer
 //   size: integer
@@ -53,76 +36,14 @@ exports.isValid = (rawAttachment) => {
     return false;
   }
 
-  return isString(rawAttachment.contentType) &&
+  const hasValidContentType = isString(rawAttachment.contentType);
-    isString(rawAttachment.fileName);
+  const hasValidFileName =
-};
+    isString(rawAttachment.fileName) || rawAttachment.fileName === null;
-
+  return hasValidContentType && hasValidFileName;
-// Middleware
-// type UpgradeStep = Attachment -> Promise Attachment
-
-// SchemaVersion -> UpgradeStep -> UpgradeStep
-exports.withSchemaVersion = (schemaVersion, upgrade) => {
-  if (!isNumber(schemaVersion)) {
-    throw new TypeError('`schemaVersion` must be a number');
-  }
-  if (!isFunction(upgrade)) {
-    throw new TypeError('`upgrade` must be a function');
-  }
-
-  return async (attachment) => {
-    if (!exports.isValid(attachment)) {
-      console.log('Attachment.withSchemaVersion: Invalid input attachment:', attachment);
-      return attachment;
-    }
-
-    const isAlreadyUpgraded = attachment.schemaVersion >= schemaVersion;
-    if (isAlreadyUpgraded) {
-      return attachment;
-    }
-
-    const expectedVersion = schemaVersion - 1;
-    const isUnversioned = isUndefined(attachment.schemaVersion);
-    const hasExpectedVersion = isUnversioned ||
-      attachment.schemaVersion === expectedVersion;
-    if (!hasExpectedVersion) {
-      console.log(
-        'WARNING: Attachment.withSchemaVersion: Unexpected version:' +
-        ` Expected attachment to have version ${expectedVersion},` +
-        ` but got ${attachment.schemaVersion}.`,
-        attachment
-      );
-      return attachment;
-    }
-
-    let upgradedAttachment;
-    try {
-      upgradedAttachment = await upgrade(attachment);
-    } catch (error) {
-      console.log(
-        'Attachment.withSchemaVersion: error:',
-        error && error.stack ? error.stack : error
-      );
-      return attachment;
-    }
-
-    if (!exports.isValid(upgradedAttachment)) {
-      console.log(
-        'Attachment.withSchemaVersion: Invalid upgraded attachment:',
-        upgradedAttachment
-      );
-      return attachment;
-    }
-
-    return Object.assign(
-      {},
-      upgradedAttachment,
-      { schemaVersion }
-    );
-  };
 };
 
 // Upgrade steps
-const autoOrientJPEG = async (attachment) => {
+exports.autoOrientJPEG = async (attachment) => {
   if (!MIME.isJPEG(attachment.contentType)) {
     return attachment;
   }
@@ -176,10 +97,13 @@ exports._replaceUnicodeOrderOverridesSync = (attachment) => {
 exports.replaceUnicodeOrderOverrides = async attachment =>
   exports._replaceUnicodeOrderOverridesSync(attachment);
 
-// Public API
+exports.removeSchemaVersion = (attachment) => {
-const toVersion1 = exports.withSchemaVersion(1, autoOrientJPEG);
+  if (!exports.isValid(attachment)) {
-const toVersion2 = exports.withSchemaVersion(2, exports.replaceUnicodeOrderOverrides);
+    console.log('Attachment.removeSchemaVersion: Invalid input attachment:', attachment);
+    return attachment;
+  }
 
-// UpgradeStep
+  const attachmentWithoutSchemaVersion = Object.assign({}, attachment);
-exports.upgradeSchema = async attachment =>
+  delete attachmentWithoutSchemaVersion.schemaVersion;
-  toVersion2(await toVersion1(attachment));
+  return attachmentWithoutSchemaVersion;
+};

js/modules/types/message.js

@@ -1,17 +1,165 @@
+const isFunction = require('lodash/isFunction');
+
 const Attachment = require('./attachment');
+const Errors = require('./errors');
+const SchemaVersion = require('./schema_version');
 
 
 const GROUP = 'group';
 const PRIVATE = 'private';
 
+// Schema version history
+//
+// Version 0
+//   - Schema initialized
+// Version 1
+//   - Attachments: Auto-orient JPEG attachments using EXIF `Orientation` data
+// Version 2
+//   - Attachments: Sanitize Unicode order override characters
+const INITIAL_SCHEMA_VERSION = 0;
+
+// Increment this version number every time we add a message schema upgrade
+// step. This will allow us to retroactively upgrade existing messages. As we
+// add more upgrade steps, we could design a pipeline that does this
+// incrementally, e.g. from version 0 / unknown -> 1, 1 --> 2, etc., similar to
+// how we do database migrations:
+exports.CURRENT_SCHEMA_VERSION = 2;
+
+
 // Public API
 exports.GROUP = GROUP;
 exports.PRIVATE = PRIVATE;
 
+// Placeholder until we have stronger preconditions:
+exports.isValid = () =>
+  true;
+
 // Schema
-// Message -> Promise Message
+exports.initializeSchemaVersion = (message) => {
+  const isInitialized = SchemaVersion.isValid(message.schemaVersion) &&
+    message.schemaVersion >= 1;
+  if (isInitialized) {
+    return message;
+  }
+
+  const numAttachments = Array.isArray(message.attachments)
+    ? message.attachments.length
+    : 0;
+  const hasAttachments = numAttachments > 0;
+  if (!hasAttachments) {
+    return Object.assign(
+      {},
+      message,
+      { schemaVersion: INITIAL_SCHEMA_VERSION }
+    );
+  }
+
+  // All attachments should have the same schema version, so we just pick
+  // the first one:
+  const firstAttachment = message.attachments[0];
+  const inheritedSchemaVersion = SchemaVersion.isValid(firstAttachment.schemaVersion)
+    ? firstAttachment.schemaVersion
+    : INITIAL_SCHEMA_VERSION;
+  const messageWithInitialSchema = Object.assign(
+    {},
+    message,
+    {
+      schemaVersion: inheritedSchemaVersion,
+      attachments: message.attachments.map(Attachment.removeSchemaVersion),
+    }
+  );
+
+  return messageWithInitialSchema;
+};
+
+// Middleware
+// type UpgradeStep = Message -> Promise Message
+
+// SchemaVersion -> UpgradeStep -> UpgradeStep
+exports._withSchemaVersion = (schemaVersion, upgrade) => {
+  if (!SchemaVersion.isValid(schemaVersion)) {
+    throw new TypeError('`schemaVersion` is invalid');
+  }
+  if (!isFunction(upgrade)) {
+    throw new TypeError('`upgrade` must be a function');
+  }
+
+  return async (message) => {
+    if (!exports.isValid(message)) {
+      console.log('Message._withSchemaVersion: Invalid input message:', message);
+      return message;
+    }
+
+    const isAlreadyUpgraded = message.schemaVersion >= schemaVersion;
+    if (isAlreadyUpgraded) {
+      return message;
+    }
+
+    const expectedVersion = schemaVersion - 1;
+    const hasExpectedVersion = message.schemaVersion === expectedVersion;
+    if (!hasExpectedVersion) {
+      console.log(
+        'WARNING: Message._withSchemaVersion: Unexpected version:',
+        `Expected message to have version ${expectedVersion},`,
+        `but got ${message.schemaVersion}.`,
+        message
+      );
+      return message;
+    }
+
+    let upgradedMessage;
+    try {
+      upgradedMessage = await upgrade(message);
+    } catch (error) {
+      console.log(
+        'Message._withSchemaVersion: error:',
+        Errors.toLogFormat(error)
+      );
+      return message;
+    }
+
+    if (!exports.isValid(upgradedMessage)) {
+      console.log(
+        'Message._withSchemaVersion: Invalid upgraded message:',
+        upgradedMessage
+      );
+      return message;
+    }
+
+    return Object.assign(
+      {},
+      upgradedMessage,
+      { schemaVersion }
+    );
+  };
+};
+
+
+// Public API
+//      _mapAttachments :: (Attachment -> Promise Attachment) ->
+//                         Message ->
+//                         Promise Message
+exports._mapAttachments = upgradeAttachment => async message =>
+  Object.assign(
+    {},
+    message,
+    {
+      attachments: await Promise.all(message.attachments.map(upgradeAttachment)),
+    }
+  );
+
+const toVersion0 = async message =>
+  exports.initializeSchemaVersion(message);
+
+const toVersion1 = exports._withSchemaVersion(
+  1,
+  exports._mapAttachments(Attachment.autoOrientJPEG)
+);
+const toVersion2 = exports._withSchemaVersion(
+  2,
+  exports._mapAttachments(Attachment.replaceUnicodeOrderOverrides)
+);
+
+// UpgradeStep
 exports.upgradeSchema = async message =>
-  Object.assign({}, message, {
+  toVersion2(await toVersion1(await toVersion0(message)));
-    attachments:
-      await Promise.all(message.attachments.map(Attachment.upgradeSchema)),
-  });

js/modules/types/schema_version.js

@@ -0,0 +1,5 @@
+const isNumber = require('lodash/isNumber');
+
+
+exports.isValid = value =>
+  isNumber(value) && value >= 0;

js/notifications.js

@@ -25,6 +25,7 @@
             this.trigger('click', conversation);
         },
         update: function() {
+            const {isEnabled} = this;
             const isFocused = window.isFocused();
             const isAudioNotificationEnabled = storage.get('audio-notification') || false;
             const isAudioNotificationSupported = Settings.isAudioNotificationSupported();
@@ -33,13 +34,10 @@
             const numNotifications = this.length;
             console.log(
                 'Update notifications:',
-                'isFocused:', isFocused,
+                {isFocused, isEnabled, numNotifications, shouldPlayNotificationSound}
-                'isEnabled:', this.isEnabled,
-                'numNotifications:', numNotifications,
-                'shouldPlayNotificationSound:', shouldPlayNotificationSound
             );
 
-            if (!this.isEnabled) {
+            if (!isEnabled) {
                 return;
             }
 

preload.js

@@ -106,9 +106,10 @@
 
   // ES2015+ modules
   window.Signal = window.Signal || {};
-  window.Signal.OS = require('./js/modules/os');
   window.Signal.Logs = require('./js/modules/logs');
-
+  window.Signal.OS = require('./js/modules/os');
+  window.Signal.Migrations = window.Signal.Migrations || {};
+  window.Signal.Migrations.V17 = require('./js/modules/migrations/17');
   window.Signal.Types = window.Signal.Types || {};
   window.Signal.Types.Attachment = require('./js/modules/types/attachment');
   window.Signal.Types.Errors = require('./js/modules/types/errors');

test/modules/types/attachment_test.js

@@ -5,163 +5,6 @@ const { assert } = require('chai');
 const Attachment = require('../../../js/modules/types/attachment');
 
 describe('Attachment', () => {
-  describe('upgradeSchema', () => {
-    it('should upgrade an unversioned attachment to the latest version', async () => {
-      const input = {
-        contentType: 'application/json',
-        data: null,
-        fileName: 'test\u202Dfig.exe',
-        size: 1111,
-      };
-      const expected = {
-        contentType: 'application/json',
-        data: null,
-        fileName: 'test\uFFFDfig.exe',
-        size: 1111,
-        schemaVersion: Attachment.CURRENT_SCHEMA_VERSION,
-      };
-
-      const actual = await Attachment.upgradeSchema(input);
-      assert.deepEqual(actual, expected);
-    });
-
-    context('with multiple upgrade steps', () => {
-      it('should return last valid attachment when any upgrade step fails', async () => {
-        const input = {
-          contentType: 'application/json',
-          data: null,
-          fileName: 'test\u202Dfig.exe',
-          size: 1111,
-        };
-        const expected = {
-          contentType: 'application/json',
-          data: null,
-          fileName: 'test\u202Dfig.exe',
-          size: 1111,
-          schemaVersion: 1,
-          hasUpgradedToVersion1: true,
-        };
-
-        const v1 = async attachment =>
-          Object.assign({}, attachment, { hasUpgradedToVersion1: true });
-        const v2 = async () => {
-          throw new Error('boom');
-        };
-        const v3 = async attachment =>
-          Object.assign({}, attachment, { hasUpgradedToVersion3: true });
-
-        const toVersion1 = Attachment.withSchemaVersion(1, v1);
-        const toVersion2 = Attachment.withSchemaVersion(2, v2);
-        const toVersion3 = Attachment.withSchemaVersion(3, v3);
-
-        const upgradeSchema = async attachment =>
-          toVersion3(await toVersion2(await toVersion1(attachment)));
-
-        const actual = await upgradeSchema(input);
-        assert.deepEqual(actual, expected);
-      });
-
-      it('should skip out-of-order upgrade steps', async () => {
-        const input = {
-          contentType: 'application/json',
-          data: null,
-          fileName: 'test\u202Dfig.exe',
-          size: 1111,
-        };
-        const expected = {
-          contentType: 'application/json',
-          data: null,
-          fileName: 'test\u202Dfig.exe',
-          size: 1111,
-          schemaVersion: 2,
-          hasUpgradedToVersion1: true,
-          hasUpgradedToVersion2: true,
-        };
-
-        const v1 = async attachment =>
-          Object.assign({}, attachment, { hasUpgradedToVersion1: true });
-        const v2 = async attachment =>
-          Object.assign({}, attachment, { hasUpgradedToVersion2: true });
-        const v3 = async attachment =>
-          Object.assign({}, attachment, { hasUpgradedToVersion3: true });
-
-        const toVersion1 = Attachment.withSchemaVersion(1, v1);
-        const toVersion2 = Attachment.withSchemaVersion(2, v2);
-        const toVersion3 = Attachment.withSchemaVersion(3, v3);
-
-        // NOTE: We upgrade to 3 before 2, i.e. the pipeline should abort:
-        const upgradeSchema = async attachment =>
-          toVersion2(await toVersion3(await toVersion1(attachment)));
-
-        const actual = await upgradeSchema(input);
-        assert.deepEqual(actual, expected);
-      });
-    });
-  });
-
-  describe('withSchemaVersion', () => {
-    it('should require a version number', () => {
-      const toVersionX = () => {};
-      assert.throws(
-        () => Attachment.withSchemaVersion(toVersionX, 2),
-        '`schemaVersion` must be a number'
-      );
-    });
-
-    it('should require an upgrade function', () => {
-      assert.throws(
-        () => Attachment.withSchemaVersion(2, 3),
-        '`upgrade` must be a function'
-      );
-    });
-
-    it('should skip upgrading if attachment has already been upgraded', async () => {
-      const upgrade = async attachment =>
-        Object.assign({}, attachment, { foo: true });
-      const upgradeWithVersion = Attachment.withSchemaVersion(3, upgrade);
-
-      const input = {
-        contentType: 'image/gif',
-        data: null,
-        fileName: 'foo.gif',
-        size: 1111,
-        schemaVersion: 4,
-      };
-      const actual = await upgradeWithVersion(input);
-      assert.deepEqual(actual, input);
-    });
-
-    it('should return original attachment if upgrade function throws', async () => {
-      const upgrade = async () => {
-        throw new Error('boom!');
-      };
-      const upgradeWithVersion = Attachment.withSchemaVersion(3, upgrade);
-
-      const input = {
-        contentType: 'image/gif',
-        data: null,
-        fileName: 'foo.gif',
-        size: 1111,
-      };
-      const actual = await upgradeWithVersion(input);
-      assert.deepEqual(actual, input);
-    });
-
-    it('should return original attachment if upgrade function returns null', async () => {
-      const upgrade = async () => null;
-      const upgradeWithVersion = Attachment.withSchemaVersion(3, upgrade);
-
-      const input = {
-        contentType: 'image/gif',
-        data: null,
-        fileName: 'foo.gif',
-        size: 1111,
-      };
-      const actual = await upgradeWithVersion(input);
-      assert.deepEqual(actual, input);
-    });
-  });
-
   describe('replaceUnicodeOrderOverrides', () => {
     it('should sanitize left-to-right order override character', async () => {
       const input = {
@@ -169,14 +12,12 @@ describe('Attachment', () => {
         data: null,
         fileName: 'test\u202Dfig.exe',
         size: 1111,
-        schemaVersion: 1,
       };
       const expected = {
         contentType: 'image/jpeg',
         data: null,
         fileName: 'test\uFFFDfig.exe',
         size: 1111,
-        schemaVersion: 1,
       };
 
       const actual = await Attachment.replaceUnicodeOrderOverrides(input);
@@ -189,14 +30,12 @@ describe('Attachment', () => {
         data: null,
         fileName: 'test\u202Efig.exe',
         size: 1111,
-        schemaVersion: 1,
       };
       const expected = {
         contentType: 'image/jpeg',
         data: null,
         fileName: 'test\uFFFDfig.exe',
         size: 1111,
-        schemaVersion: 1,
       };
 
       const actual = await Attachment.replaceUnicodeOrderOverrides(input);
@@ -209,14 +48,12 @@ describe('Attachment', () => {
         data: null,
         fileName: 'test\u202e\u202dlol\u202efig.exe',
         size: 1111,
-        schemaVersion: 1,
       };
       const expected = {
         contentType: 'image/jpeg',
         data: null,
         fileName: 'test\uFFFD\uFFFDlol\uFFFDfig.exe',
         size: 1111,
-        schemaVersion: 1,
       };
 
       const actual = await Attachment.replaceUnicodeOrderOverrides(input);
@@ -235,7 +72,6 @@ describe('Attachment', () => {
           data: null,
           fileName,
           size: 1111,
-          schemaVersion: 1,
         };
 
         const actual = Attachment._replaceUnicodeOrderOverridesSync(input);
@@ -243,4 +79,26 @@ describe('Attachment', () => {
       }
     );
   });
+
+  describe('removeSchemaVersion', () => {
+    it('should remove existing schema version', () => {
+      const input = {
+        contentType: 'image/jpeg',
+        data: null,
+        fileName: 'foo.jpg',
+        size: 1111,
+        schemaVersion: 1,
+      };
+
+      const expected = {
+        contentType: 'image/jpeg',
+        data: null,
+        fileName: 'foo.jpg',
+        size: 1111,
+      };
+
+      const actual = Attachment.removeSchemaVersion(input);
+      assert.deepEqual(actual, expected);
+    });
+  });
 });

test/modules/types/message_test.js

@@ -0,0 +1,240 @@
+const { assert } = require('chai');
+
+const Message = require('../../../js/modules/types/message');
+
+
+describe('Message', () => {
+  describe('initializeSchemaVersion', () => {
+    it('should ignore messages with previously inherited schema', () => {
+      const input = {
+        body: 'Imagine there is no heaven…',
+        schemaVersion: 2,
+      };
+      const expected = {
+        body: 'Imagine there is no heaven…',
+        schemaVersion: 2,
+      };
+
+      const actual = Message.initializeSchemaVersion(input);
+      assert.deepEqual(actual, expected);
+    });
+
+    context('for message without attachments', () => {
+      it('should initialize schema version to zero', () => {
+        const input = {
+          body: 'Imagine there is no heaven…',
+          attachments: [],
+        };
+        const expected = {
+          body: 'Imagine there is no heaven…',
+          attachments: [],
+          schemaVersion: 0,
+        };
+
+        const actual = Message.initializeSchemaVersion(input);
+        assert.deepEqual(actual, expected);
+      });
+    });
+
+    context('for message with attachments', () => {
+      it('should inherit existing attachment schema version', () => {
+        const input = {
+          body: 'Imagine there is no heaven…',
+          attachments: [{
+            contentType: 'image/jpeg',
+            fileName: 'lennon.jpg',
+            schemaVersion: 7,
+          }],
+        };
+        const expected = {
+          body: 'Imagine there is no heaven…',
+          attachments: [{
+            contentType: 'image/jpeg',
+            fileName: 'lennon.jpg',
+          }],
+          schemaVersion: 7,
+        };
+
+        const actual = Message.initializeSchemaVersion(input);
+        assert.deepEqual(actual, expected);
+      });
+    });
+  });
+
+  describe('upgradeSchema', () => {
+    it('should upgrade an unversioned message to the latest version', async () => {
+      const input = {
+        attachments: [{
+          contentType: 'application/json',
+          data: null,
+          fileName: 'test\u202Dfig.exe',
+          size: 1111,
+        }],
+        schemaVersion: 0,
+      };
+      const expected = {
+        attachments: [{
+          contentType: 'application/json',
+          data: null,
+          fileName: 'test\uFFFDfig.exe',
+          size: 1111,
+        }],
+        schemaVersion: Message.CURRENT_SCHEMA_VERSION,
+      };
+
+      const actual = await Message.upgradeSchema(input);
+      assert.deepEqual(actual, expected);
+    });
+
+    context('with multiple upgrade steps', () => {
+      it('should return last valid message when any upgrade step fails', async () => {
+        const input = {
+          attachments: [{
+            contentType: 'application/json',
+            data: null,
+            fileName: 'test\u202Dfig.exe',
+            size: 1111,
+          }],
+          schemaVersion: 0,
+        };
+        const expected = {
+          attachments: [{
+            contentType: 'application/json',
+            data: null,
+            fileName: 'test\u202Dfig.exe',
+            size: 1111,
+          }],
+          hasUpgradedToVersion1: true,
+          schemaVersion: 1,
+        };
+
+        const v1 = async message =>
+          Object.assign({}, message, { hasUpgradedToVersion1: true });
+        const v2 = async () => {
+          throw new Error('boom');
+        };
+        const v3 = async message =>
+          Object.assign({}, message, { hasUpgradedToVersion3: true });
+
+        const toVersion1 = Message._withSchemaVersion(1, v1);
+        const toVersion2 = Message._withSchemaVersion(2, v2);
+        const toVersion3 = Message._withSchemaVersion(3, v3);
+
+        const upgradeSchema = async message =>
+          toVersion3(await toVersion2(await toVersion1(message)));
+
+        const actual = await upgradeSchema(input);
+        assert.deepEqual(actual, expected);
+      });
+
+      it('should skip out-of-order upgrade steps', async () => {
+        const input = {
+          attachments: [{
+            contentType: 'application/json',
+            data: null,
+            fileName: 'test\u202Dfig.exe',
+            size: 1111,
+          }],
+          schemaVersion: 0,
+        };
+        const expected = {
+          attachments: [{
+            contentType: 'application/json',
+            data: null,
+            fileName: 'test\u202Dfig.exe',
+            size: 1111,
+          }],
+          schemaVersion: 2,
+          hasUpgradedToVersion1: true,
+          hasUpgradedToVersion2: true,
+        };
+
+        const v1 = async attachment =>
+          Object.assign({}, attachment, { hasUpgradedToVersion1: true });
+        const v2 = async attachment =>
+          Object.assign({}, attachment, { hasUpgradedToVersion2: true });
+        const v3 = async attachment =>
+          Object.assign({}, attachment, { hasUpgradedToVersion3: true });
+
+        const toVersion1 = Message._withSchemaVersion(1, v1);
+        const toVersion2 = Message._withSchemaVersion(2, v2);
+        const toVersion3 = Message._withSchemaVersion(3, v3);
+
+        // NOTE: We upgrade to 3 before 2, i.e. the pipeline should abort:
+        const upgradeSchema = async attachment =>
+          toVersion2(await toVersion3(await toVersion1(attachment)));
+
+        const actual = await upgradeSchema(input);
+        assert.deepEqual(actual, expected);
+      });
+    });
+  });
+
+  describe('_withSchemaVersion', () => {
+    it('should require a version number', () => {
+      const toVersionX = () => {};
+      assert.throws(
+        () => Message._withSchemaVersion(toVersionX, 2),
+        '`schemaVersion` is invalid'
+      );
+    });
+
+    it('should require an upgrade function', () => {
+      assert.throws(
+        () => Message._withSchemaVersion(2, 3),
+        '`upgrade` must be a function'
+      );
+    });
+
+    it('should skip upgrading if message has already been upgraded', async () => {
+      const upgrade = async message =>
+        Object.assign({}, message, { foo: true });
+      const upgradeWithVersion = Message._withSchemaVersion(3, upgrade);
+
+      const input = {
+        id: 'guid-guid-guid-guid',
+        schemaVersion: 4,
+      };
+      const expected = {
+        id: 'guid-guid-guid-guid',
+        schemaVersion: 4,
+      };
+      const actual = await upgradeWithVersion(input);
+      assert.deepEqual(actual, expected);
+    });
+
+    it('should return original message if upgrade function throws', async () => {
+      const upgrade = async () => {
+        throw new Error('boom!');
+      };
+      const upgradeWithVersion = Message._withSchemaVersion(3, upgrade);
+
+      const input = {
+        id: 'guid-guid-guid-guid',
+        schemaVersion: 0,
+      };
+      const expected = {
+        id: 'guid-guid-guid-guid',
+        schemaVersion: 0,
+      };
+      const actual = await upgradeWithVersion(input);
+      assert.deepEqual(actual, expected);
+    });
+
+    it('should return original message if upgrade function returns null', async () => {
+      const upgrade = async () => null;
+      const upgradeWithVersion = Message._withSchemaVersion(3, upgrade);
+
+      const input = {
+        id: 'guid-guid-guid-guid',
+        schemaVersion: 0,
+      };
+      const expected = {
+        id: 'guid-guid-guid-guid',
+        schemaVersion: 0,
+      };
+      const actual = await upgradeWithVersion(input);
+      assert.deepEqual(actual, expected);
+    });
+  });
+});

test/modules/types/schema_version_test.js

@@ -0,0 +1,25 @@
+require('mocha-testcheck').install();
+const { assert } = require('chai');
+
+const SchemaVersion = require('../../../js/modules/types/schema_version');
+
+
+describe('SchemaVersion', () => {
+  describe('isValid', () => {
+    check.it(
+      'should return true for positive integers',
+      gen.posInt,
+      (input) => {
+        assert.isTrue(SchemaVersion.isValid(input));
+      }
+    );
+
+    check.it(
+      'should return false for any other value',
+      gen.primitive.suchThat(value => typeof value !== 'number' || value < 0),
+      (input) => {
+        assert.isFalse(SchemaVersion.isValid(input));
+      }
+    );
+  });
+});