Fix the resumeOnRestart function to handle correctly also the recurring jobs + add unit tests

Author: -

src/pulse/resume-on-restart.ts

@@ -1,5 +1,6 @@
 import createDebugger from 'debug';
 import { Pulse } from '.';
+import { Job } from '../job';
 
 const debug = createDebugger('pulse:resumeOnRestart');
 
@@ -18,6 +19,8 @@ export const resumeOnRestart: ResumeOnRestartMethod = function (this: Pulse, res
 
   if (this._collection && this._resumeOnRestart) {
     const now = new Date();
+
+    // Non-recurring jobs
     this._collection
       .updateMany(
         {
@@ -41,7 +44,48 @@ export const resumeOnRestart: ResumeOnRestartMethod = function (this: Pulse, res
       )
       .then((result) => {
         if (result.modifiedCount > 0) {
-          debug('resuming unfinished %d jobs(%s)', result.modifiedCount, now.toISOString());
+          debug('Resumed %d unfinished standard jobs (%s)', result.modifiedCount, now.toISOString());
+        }
+      });
+
+    // Handling for recurring jobs using repeatInterval or repeatAt
+    this._collection
+      .find({
+        $or: [{ repeatInterval: { $exists: true } }, { repeatAt: { $exists: true } }],
+        nextRunAt: { $exists: false },
+      })
+      .toArray()
+      .then((jobs) => {
+        const updates = jobs.map((jobData) => {
+          const job = new Job({
+            pulse: this,
+            name: jobData.name || '',
+            data: jobData.data || {},
+            type: jobData.type || 'normal',
+            priority: jobData.priority || 'normal',
+            shouldSaveResult: jobData.shouldSaveResult || false,
+            attempts: jobData.attempts || 0,
+            backoff: jobData.backoff,
+            ...jobData,
+          });
+
+          job.computeNextRunAt();
+
+          return this._collection.updateOne(
+            { _id: job.attrs._id },
+            {
+              $set: { nextRunAt: job.attrs.nextRunAt },
+              $unset: { lockedAt: undefined, lastModifiedBy: undefined, lastRunAt: undefined },
+            }
+          );
+        });
+
+        return Promise.all(updates);
+      })
+      .then((results) => {
+        const modifiedCount = results.filter((res) => res.modifiedCount > 0).length;
+        if (modifiedCount > 0) {
+          debug('Resumed %d recurring jobs (%s)', modifiedCount, now.toISOString());
         }
       });
   }

test/unit/pulse.spec.ts

@@ -206,17 +206,51 @@ describe('Test Pulse', () => {
     });
 
     describe('Test resumeOnRestart', () => {
-      test('sets the default resumeOnRestart', () => {
+      test('should enable resumeOnRestart by default', () => {
         expect(globalPulseInstance._resumeOnRestart).toBeTruthy();
       });
 
-      test('sets the custom resumeOnRestart', () => {
+      test('should disable resumeOnRestart when set to false', () => {
         globalPulseInstance.resumeOnRestart(false);
         expect(globalPulseInstance._resumeOnRestart).toBeFalsy();
       });
 
-      test('returns itself', () => {
-        expect(globalPulseInstance.resumeOnRestart(false)).toEqual(globalPulseInstance);
+      test('should resume non-recurring jobs on restart', async () => {
+        const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
+        job.attrs.nextRunAt = new Date(Date.now() - 1000); // Set nextRunAt in the past
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
+        const now = Date.now();
+        expect(updatedJob.attrs.nextRunAt?.getTime()).toBeGreaterThan(now - 100); // Allow a 100ms buffer
+      });
+
+      test('should resume recurring jobs on restart', async () => {
+        const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
+        job.attrs.repeatInterval = '5 minutes';
+        job.attrs.nextRunAt = null;
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
+        const now = Date.now();
+        expect(updatedJob.attrs.nextRunAt).not.toBeNull();
+        expect(updatedJob.attrs.nextRunAt?.getTime()).toBeGreaterThan(now - 100); // Allow a 100ms buffer
+      });
+
+      test('should not modify jobs with existing nextRunAt', async () => {
+        const futureDate = new Date(Date.now() + 60 * 60 * 1000);
+        const job = globalPulseInstance.create('sendEmail', { to: 'user@example.com' });
+        job.attrs.nextRunAt = futureDate;
+        await job.save();
+
+        await globalPulseInstance.resumeOnRestart();
+
+        const updatedJob = (await globalPulseInstance.jobs({ name: 'sendEmail' }))[0];
+        expect(updatedJob.attrs.nextRunAt?.getTime()).toEqual(futureDate.getTime());
       });
     });
   });