Escalate sync policy based on number of stream connections

packages/nylas-api/routes/auth.js

@@ -2,7 +2,12 @@ const Joi = require('Joi');
 const _ = require('underscore');
 
 const Serialization = require('../serialization');
-const {IMAPConnection, PubsubConnector, DatabaseConnector} = require('nylas-core');
+const {
+  IMAPConnection,
+  PubsubConnector,
+  DatabaseConnector,
+  SyncPolicy
+} = require('nylas-core');
 
 const imapSmtpSettings = Joi.object().keys({
   imap_host: [Joi.string().ip().required(), Joi.string().hostname().required()],
@@ -22,15 +27,6 @@ const exchangeSettings = Joi.object().keys({
   eas_server_host: [Joi.string().ip().required(), Joi.string().hostname().required()],
 }).required();
 
-const defaultSyncPolicy = {
-  afterSync: 'idle',
-  interval: 30 * 1000,
-  folderSyncOptions: {
-    deepFolderScan: 5 * 60 * 1000,
-  },
-  expiration: Date.now() + 60 * 60 * 1000,
-};
-
 module.exports = (server) => {
   server.route({
     method: 'POST',
@@ -75,7 +71,7 @@ module.exports = (server) => {
           const account = Account.build({
             name: name,
             emailAddress: email,
-            syncPolicy: defaultSyncPolicy,
+            syncPolicy: SyncPolicy.defaultPolicy(),
             connectionSettings: _.pick(settings, [
               'imap_host', 'imap_port',
               'smtp_host', 'smtp_port',

packages/nylas-api/routes/delta.js

@@ -1,6 +1,6 @@
 const Rx = require('rx')
 const _ = require('underscore');
-const {PubsubConnector} = require(`nylas-core`);
+const {PubsubConnector, SyncPolicy} = require(`nylas-core`);
 
 function keepAlive(request) {
   const until = Rx.Observable.fromCallback(request.on)("disconnect")
@@ -49,15 +49,20 @@ module.exports = (server) => {
     path: '/delta/streaming',
     handler: (request, reply) => {
       const outputStream = createOutputStream();
+      const account = request.auth.credentials;
+      PubsubConnector.incrementActivePolicyLockForAccount(account.id)
 
       request.getAccountDatabase().then((db) => {
         const source = Rx.Observable.merge(
-          PubsubConnector.observableForAccountDeltas(db.accountId),
+          PubsubConnector.observableForAccountDeltas(account.id),
           initialTransactions(db, request),
           keepAlive(request)
         ).subscribe(outputStream.pushJSON)
 
-        request.on("disconnect", () => source.dispose());
+        request.on("disconnect", () => {
+          PubsubConnector.decrementActivePolicyLockForAccount(account.id)
+          source.dispose()
+        });
       });
 
       reply(outputStream)

packages/nylas-core/sync-policy.js

@@ -0,0 +1,23 @@
+class SyncPolicy {
+  static defaultPolicy() {
+    return {
+      afterSync: 'close',
+      interval: 120 * 1000,
+      folderSyncOptions: {
+        deepFolderScan: 10 * 60 * 1000,
+      },
+    };
+  }
+
+  static activeUserPolicy() {
+    return {
+      afterSync: 'idle',
+      interval: 30 * 1000,
+      folderSyncOptions: {
+        deepFolderScan: 5 * 60 * 1000,
+      },
+    };
+  }
+}
+
+module.exports = SyncPolicy