Introduce mandatary migration on startup, to minimum version

js/modules/messages_data_migrator.js

@@ -21,6 +21,7 @@ exports.processNext = async ({
   upgradeMessageSchema,
   getMessagesNeedingUpgrade,
   saveMessage,
+  maxVersion = Message.CURRENT_SCHEMA_VERSION,
 } = {}) => {
   if (!isFunction(BackboneMessage)) {
     throw new TypeError(
@@ -49,6 +50,7 @@ exports.processNext = async ({
   const messagesRequiringSchemaUpgrade = await getMessagesNeedingUpgrade(
     numMessagesPerBatch,
     {
+      maxVersion,
       MessageCollection: BackboneMessageCollection,
     }
   );
@@ -56,7 +58,9 @@ exports.processNext = async ({
 
   const upgradeStartTime = Date.now();
   const upgradedMessages = await Promise.all(
-    messagesRequiringSchemaUpgrade.map(upgradeMessageSchema)
+    messagesRequiringSchemaUpgrade.map(message =>
+      upgradeMessageSchema(message, { maxVersion })
+    )
   );
   const upgradeDuration = Date.now() - upgradeStartTime;
 
@@ -87,6 +91,9 @@ exports.dangerouslyProcessAllWithoutIndex = async ({
   numMessagesPerBatch,
   upgradeMessageSchema,
   logger,
+  maxVersion = Message.CURRENT_SCHEMA_VERSION,
+  saveMessage,
+  BackboneMessage,
 } = {}) => {
   if (!isString(databaseName)) {
     throw new TypeError("'databaseName' must be a string");
@@ -99,10 +106,15 @@ exports.dangerouslyProcessAllWithoutIndex = async ({
   if (!isNumber(numMessagesPerBatch)) {
     throw new TypeError("'numMessagesPerBatch' must be a number");
   }
-
   if (!isFunction(upgradeMessageSchema)) {
     throw new TypeError("'upgradeMessageSchema' is required");
   }
+  if (!isFunction(BackboneMessage)) {
+    throw new TypeError("'upgradeMessageSchema' is required");
+  }
+  if (!isFunction(saveMessage)) {
+    throw new TypeError("'upgradeMessageSchema' is required");
+  }
 
   const connection = await database.open(databaseName);
   const databaseVersion = connection.version;
@@ -133,6 +145,9 @@ exports.dangerouslyProcessAllWithoutIndex = async ({
       connection,
       numMessagesPerBatch,
       upgradeMessageSchema,
+      maxVersion,
+      saveMessage,
+      BackboneMessage,
     });
     if (status.done) {
       break;
@@ -162,6 +177,9 @@ exports.processNextBatchWithoutIndex = async ({
   minDatabaseVersion,
   numMessagesPerBatch,
   upgradeMessageSchema,
+  maxVersion,
+  BackboneMessage,
+  saveMessage,
 } = {}) => {
   if (!isFunction(upgradeMessageSchema)) {
     throw new TypeError("'upgradeMessageSchema' is required");
@@ -172,6 +190,9 @@ exports.processNextBatchWithoutIndex = async ({
     connection,
     numMessagesPerBatch,
     upgradeMessageSchema,
+    maxVersion,
+    BackboneMessage,
+    saveMessage,
   });
   return batch;
 };
@@ -203,17 +224,29 @@ const _processBatch = async ({
   connection,
   numMessagesPerBatch,
   upgradeMessageSchema,
+  maxVersion,
+  BackboneMessage,
+  saveMessage,
 } = {}) => {
   if (!isObject(connection)) {
-    throw new TypeError("'connection' must be a string");
+    throw new TypeError('_processBatch: connection must be a string');
   }
 
   if (!isFunction(upgradeMessageSchema)) {
-    throw new TypeError("'upgradeMessageSchema' is required");
+    throw new TypeError('_processBatch: upgradeMessageSchema is required');
   }
 
   if (!isNumber(numMessagesPerBatch)) {
-    throw new TypeError("'numMessagesPerBatch' is required");
+    throw new TypeError('_processBatch: numMessagesPerBatch is required');
+  }
+  if (!isNumber(maxVersion)) {
+    throw new TypeError('_processBatch: maxVersion is required');
+  }
+  if (!isFunction(BackboneMessage)) {
+    throw new TypeError('_processBatch: BackboneMessage is required');
+  }
+  if (!isFunction(saveMessage)) {
+    throw new TypeError('_processBatch: saveMessage is required');
   }
 
   const isAttachmentMigrationComplete = await settings.isAttachmentMigrationComplete(
@@ -241,14 +274,20 @@ const _processBatch = async ({
 
   const upgradeStartTime = Date.now();
   const upgradedMessages = await Promise.all(
-    unprocessedMessages.map(upgradeMessageSchema)
+    unprocessedMessages.map(message =>
+      upgradeMessageSchema(message, { maxVersion })
+    )
   );
   const upgradeDuration = Date.now() - upgradeStartTime;
 
   const saveMessagesStartTime = Date.now();
   const transaction = connection.transaction(MESSAGES_STORE_NAME, 'readwrite');
   const transactionCompletion = database.completeTransaction(transaction);
-  await Promise.all(upgradedMessages.map(_saveMessage({ transaction })));
+  await Promise.all(
+    upgradedMessages.map(message =>
+      saveMessage(message, { Message: BackboneMessage })
+    )
+  );
   await transactionCompletion;
   const saveDuration = Date.now() - saveMessagesStartTime;
 
@@ -281,19 +320,6 @@ const _processBatch = async ({
   };
 };
 
-const _saveMessage = ({ transaction } = {}) => message => {
-  if (!isObject(transaction)) {
-    throw new TypeError("'transaction' is required");
-  }
-
-  const messagesStore = transaction.objectStore(MESSAGES_STORE_NAME);
-  const request = messagesStore.put(message, message.id);
-  return new Promise((resolve, reject) => {
-    request.onsuccess = () => resolve();
-    request.onerror = event => reject(event.target.error);
-  });
-};
-
 // NOTE: Named ‘dangerous’ because it is not as efficient as using our
 // `messages` `schemaVersion` index:
 const _dangerouslyFetchMessagesRequiringSchemaUpgradeWithoutIndex = ({