feat(#8229): schedule outbound push (#8387)

Added the ability schedule outbound push using cron expression.

#8229

api/package.json

@@ -22,11 +22,13 @@
     "bowser": "^2.11.0",
     "buffer-shims": "^1.0.0",
     "compression": "^1.7.4",
+    "cron-validator": "^1.3.1",
     "express": "^4.18.2",
     "google-libphonenumber": "^3.2.33",
     "gsm": "^0.1.4",
     "helmet": "^3.23.3",
     "http-proxy": "^1.18.1",
+    "later": "^1.2.0",
     "locale": "^0.1.0",
     "lodash": "^4.17.21",
     "moment": "^2.29.1",

sentinel/package.json

@@ -11,6 +11,7 @@
     "async": "^3.2.4",
     "bikram-sambat": "^1.7.0",
     "bikram-sambat-bootstrap": "^1.5.0",
+    "cron-validator": "^1.3.1",
     "epi-week": "^0.0.1",
     "google-libphonenumber": "^3.2.31",
     "gsm": "^0.1.4",

sentinel/src/schedule/index.js

@@ -60,7 +60,7 @@ const runTasks = () => {
       .execute()
       .then(() => logger.info(`Task ${taskName} completed`))
       .catch(err => logger.error(`Task ${taskName} completed with error: %o`, err))
-      .then(() => {
+      .finally(() => {
         ongoingTasks.delete(taskName);
       });
   });

sentinel/src/schedule/outbound.js

@@ -6,8 +6,11 @@ const outbound = require('@medic/outbound')(logger);
 const infodocLib = require('@medic/infodoc');
 infodocLib.initLib(db.medic, db.sentinel);
 
+const transitionsLib = configService.getTransitionsLib();
+
 const CONFIGURED_PUSHES = 'outbound';
 const BATCH_SIZE = 1000;
+const TIME_FRAME_DURATION = 5 * 60 * 1000; // 5 minutes
 
 //
 // Loads all queued tasks and splits them into valid tasks we can work on, and invalid tasks that
@@ -201,13 +204,21 @@ const batch = (configuredPushes, startKey) => {
 };
 
 // Coordinates the attempted pushing of documents that need it
-const execute = () => {
+const execute = async () => {
   const configuredPushes = configService.get(CONFIGURED_PUSHES) || {};
-  if (!Object.keys(configuredPushes).length) {
+  const dueConfiguredPushes = {};
+
+  for (const [key, config] of Object.entries(configuredPushes)) {
+    if (!config.cron || transitionsLib.isWithinTimeFrame(config.cron, TIME_FRAME_DURATION)) {
+      dueConfiguredPushes[key] = config;
+    }
+  }
+
+  if (!Object.keys(dueConfiguredPushes).length) {
     return Promise.resolve();
   }
 
-  return batch(configuredPushes);
+  return batch(dueConfiguredPushes);
 };
 
 module.exports = {

sentinel/tests/unit/schedule/outbound.spec.js

@@ -536,6 +536,7 @@ describe('outbound schedule', () => {
     let removeInvalidTasks;
     let attachInfoDocs;
     let restores;
+    let clock;
 
     beforeEach(() => {
       restores = [];
@@ -552,9 +553,14 @@ describe('outbound schedule', () => {
 
       attachInfoDocs = sinon.stub();
       restores.push(outbound.__set__('attachInfoDocs', attachInfoDocs));
+
+      config.initTransitionLib();
     });
 
-    afterEach(() => restores.forEach(restore => restore()));
+    afterEach(() => {
+      restores.forEach(restore => restore());
+      clock?.restore();
+    });
 
     it('should coordinate finding all queues to process and working through them one by one', () => {
       const config = {
@@ -743,5 +749,40 @@ describe('outbound schedule', () => {
         assert.deepEqual(singlePush.args[3], [task6, doc6, doc6Info, {some: 'config'}, 'test-push-1']);
       });
     });
+
+    it('should only process queued tasks with due "cron" or non-existent "cron" field', () => {
+      const batch = sinon.stub();
+      restores.push(outbound.__set__('batch', batch));
+
+      const VALID_CRON = 'outbound-with-due-cron';
+      const INVALID_CRON = 'outbound-with-undue-cron';
+      const WITHOUT_CRON = 'outbound-without-cron';
+
+      const configs = {
+        [VALID_CRON]: {
+          cron: '5 * * * *'
+        },
+        [INVALID_CRON]: {
+          cron: '15 * * * *'
+        },
+        [WITHOUT_CRON]: {
+          /* test values */
+        }
+      };
+
+      configGet.returns(configs);
+      batch.resolvesArg(0);
+
+      clock = sinon.useFakeTimers(new Date('2023-07-11T03:05:00+0000').getTime());
+
+      return outbound.execute().then((dueConfigs) => {
+        assert.equal(configGet.callCount, 1);
+        assert.equal(batch.callCount, 1);
+        assert.equal(Object.keys(dueConfigs).length, 2);
+        assert.deepEqual(dueConfigs[VALID_CRON], configs[VALID_CRON]);
+        assert.deepEqual(dueConfigs[WITHOUT_CRON], configs[WITHOUT_CRON]);
+        assert.isUndefined(dueConfigs[INVALID_CRON]);
+      });
+    });
   });
 });

shared-libs/transitions/package.json

@@ -26,8 +26,10 @@
     "async": "^3.2.3",
     "bikram-sambat": "^1.7.0",
     "bikram-sambat-bootstrap": "^1.5.0",
+    "cron-validator": "^1.3.1",
     "google-libphonenumber": "^3.2.31",
     "gsm": "^0.1.4",
+    "later": "^1.2.0",
     "lodash": "^4.17.21",
     "moment": "^2.29.1",
     "mustache": "^4.2.0",

shared-libs/transitions/src/index.js

@@ -15,6 +15,8 @@ module.exports = (sourceDb, sourceConfig, sourceLogger) => {
   config.init(sourceConfig);
 
   const transitions = require('./transitions');
+  const utils = require('./lib/utils');
+
   return {
     date: require('./date'),
     dueTasks: require('./schedule/due_tasks'),
@@ -25,5 +27,6 @@ module.exports = (sourceDb, sourceConfig, sourceLogger) => {
     messages: require('./lib/messages'),
     processChange: transitions.processChange,
     processDocs: transitions.processDocs,
+    isWithinTimeFrame: utils.isWithinTimeFrame,
   };
 };