[local-sync] Synchronize K2 accounts with N1 accounts

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

@@ -1,18 +1,37 @@
-const LocalDatabaseConnector = require('../shared/local-database-connector')
-const os = require('os')
-global.instanceId = os.hostname();
+const {AccountStore} = require('nylas-exports');
 
+const LocalDatabaseConnector = require('../shared/local-database-connector')
 const manager = require('./sync-process-manager')
 
-LocalDatabaseConnector.forShared().then((db) => {
-  const {Account} = db;
-  Account.findAll().then((accounts) => {
-    if (accounts.length === 0) {
-      global.Logger.info(`Couldn't find any accounts to sync. Run this CURL command to auth one!`)
-      global.Logger.info(`curl -X POST -H "Content-Type: application/json" -d '{"email":"inboxapptest1@fastmail.fm", "name":"Ben Gotow", "provider":"imap", "settings":{"imap_username":"inboxapptest1@fastmail.fm","imap_host":"mail.messagingengine.com","imap_port":993,"smtp_host":"mail.messagingengine.com","smtp_port":0,"smtp_username":"inboxapptest1@fastmail.fm", "smtp_password":"trar2e","imap_password":"trar2e","ssl_required":true}}' "http://localhost:2578/auth?client_id=123"`)
+// Right now, it's a bit confusing because N1 has Account objects, and K2 has
+// Account objects. We want to sync all K2 Accounts, but when an N1 Account is
+// deleted, we want to delete the K2 account too.
+
+async function ensureK2Consistency() {
+  const {Account} = await LocalDatabaseConnector.forShared();
+  const k2Accounts = await Account.findAll();
+  const n1Accounts = AccountStore.accounts();
+  const n1Emails = n1Accounts.map(a => a.emailAddress);
+
+  const deletions = [];
+  for (const k2Account of k2Accounts) {
+    const deleted = !n1Emails.includes(k2Account.emailAddress);
+    if (deleted) {
+      console.warn(`Deleting K2 account ID ${k2Account.id} which could not be matched to an N1 account.`)
+      manager.removeWorkerForAccountId(k2Account.id);
+      LocalDatabaseConnector.destroyAccountDatabase(k2Account.id);
+      deletions.push(k2Account.destroy());
     }
-    manager.start();
-  });
+  }
+  return await Promise.all(deletions)
+}
+
+ensureK2Consistency().then(() => {
+  // Step 1: Start all K2 Accounts
+  manager.start();
 });
 
+// Step 2: Watch N1 Accounts, ensure consistency when they change.
+AccountStore.listen(ensureK2Consistency);
+
 global.manager = manager;

packages/local-sync/src/local-sync-worker/sync-process-manager.js

@@ -1,9 +1,6 @@
 const SyncWorker = require('./sync-worker');
 const LocalDatabaseConnector = require('../shared/local-database-connector')
 
-const IDENTITY = `${global.instanceId}-${process.pid}`;
-
-
 /*
 Accounts ALWAYS exist in either `accounts:unclaimed` or an `accounts:{id}` list.
 They are atomically moved between these sets as they are claimed and returned.
@@ -31,50 +28,43 @@ class SyncProcessManager {
     this._workers = {};
     this._listenForSyncsClient = null;
     this._exiting = false;
-    this._logger = global.Logger.child({identity: IDENTITY});
+    this._logger = global.Logger.child();
   }
 
-  start() {
+  async start() {
     this._logger.info(`ProcessManager: Starting with ID`)
 
-    LocalDatabaseConnector.forShared().then(({Account}) =>
-      Account.findAll().then((accounts) => {
-        for (const account of accounts) {
-          this.addWorkerForAccount(account);
-        }
-      }));
+    const {Account} = await LocalDatabaseConnector.forShared();
+    const accounts = await Account.findAll();
+    for (const account of accounts) {
+      this.addWorkerForAccount(account);
+    }
   }
 
   wakeWorkerForAccount(account) {
     this._workers[account.id].syncNow();
   }
 
-  addWorkerForAccount(account) {
-    return LocalDatabaseConnector.ensureAccountDatabase(account.id)
-    .then(() => {
-      return LocalDatabaseConnector.forAccount(account.id).then((db) => {
-        if (this._workers[account.id]) {
-          return Promise.reject(new Error("Local worker already exists"));
-        }
+  async addWorkerForAccount(account) {
+    await LocalDatabaseConnector.ensureAccountDatabase(account.id);
 
-        this._workers[account.id] = new SyncWorker(account, db, () => {
-          this.removeWorkerForAccountId(account.id)
-        });
-        return Promise.resolve();
-      })
-    })
-    .then(() => {
+    try {
+      const db = await LocalDatabaseConnector.forAccount(account.id);
+      if (this._workers[account.id]) {
+        throw new Error("Local worker already exists");
+      }
+      this._workers[account.id] = new SyncWorker(account, db, this);
       this._logger.info({account_id: account.id}, `ProcessManager: Claiming Account Succeeded`)
-    })
-    .catch((err) => {
+    } catch (err) {
       this._logger.error({account_id: account.id, reason: err.message}, `ProcessManager: Claiming Account Failed`)
-    });
+    }
   }
 
   removeWorkerForAccountId(accountId) {
-    const worker = this._workers[accountId];
-    worker.cleanup();
-    this._workers[accountId] = null;
+    if (this._workers[accountId]) {
+      this._workers[accountId].cleanup();
+      this._workers[accountId] = null;
+    }
   }
 
 }

packages/local-sync/src/local-sync-worker/sync-worker.js

@@ -14,13 +14,13 @@ const SyncbackTaskFactory = require('./syncback-task-factory')
 
 
 class SyncWorker {
-  constructor(account, db, onExpired) {
+  constructor(account, db, parentManager) {
     this._db = db;
+    this._manager = parentManager;
     this._conn = null;
     this._account = account;
     this._startTime = Date.now();
     this._lastSyncTime = null;
-    this._onExpired = onExpired;
     this._logger = global.Logger.forAccount(account)
 
     this._destroyed = false;
@@ -153,10 +153,13 @@ class SyncWorker {
       .then(() => this.syncMessagesInAllFolders())
       .then(() => this.onSyncDidComplete())
       .catch((error) => this.onSyncError(error))
-    })
-    .finally(() => {
-      this._lastSyncTime = Date.now()
-      this.scheduleNextSync()
+      .finally(() => {
+        this._lastSyncTime = Date.now()
+        this.scheduleNextSync()
+      })
+    }).catch((err) => {
+      this._logger.error({err}, `SyncWorker: Account could not be loaded. Sync worker will exit.`)
+      this._manager.removeWorkerForAccountId(this._account.id);
     })
   }