Merge pull request #42 from TNG/schedule-job-with-delay

feat: allow scheduling immediate jobs with delay

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ jobs:
 
     strategy:
       matrix:
-        node: [10, 12, 14, 16]
+        node: [12, 14, 16]
 
     steps:
       - name: Checkout the repository
@@ -42,7 +42,7 @@ jobs:
 
     strategy:
       matrix:
-        node: [10, 12, 14, 16]
+        node: [12, 14, 16]
 
     steps:
       - name: Checkout the repository

package.json

@@ -33,7 +33,7 @@
     "human-interval": "2.0.1",
     "lodash": "4.17.21",
     "luxon": "2.0.2",
-    "mongodb": "4.1.0",
+    "mongodb": "4.1.1",
     "typed-emitter": "1.3.1",
     "uuid": "8.3.2"
   },

src/job/Job.ts

@@ -8,23 +8,24 @@ export type MomoJobStatus = WithoutId<JobEntity>;
 export interface JobDefinition {
   name: string;
   interval: string;
+  firstRunAfter?: number;
   concurrency: number;
   maxRunning: number;
 }
 
 export interface Job extends JobDefinition {
-  immediate: boolean;
   handler: Handler;
 }
 
 export function toJob(job: MomoJob): Job {
-  return { immediate: false, concurrency: 1, maxRunning: 0, ...job };
+  return { concurrency: 1, maxRunning: 0, ...job };
 }
 
-export function toJobDefinition(job: Job): JobDefinition {
+export function toJobDefinition<T extends JobDefinition>(job: T): JobDefinition {
   return {
     name: job.name,
     interval: job.interval,
+    firstRunAfter: job.firstRunAfter,
     maxRunning: job.maxRunning,
     concurrency: job.concurrency,
   };

src/job/MomoJob.ts

@@ -1,10 +1,10 @@
 export type Handler = () => Promise<string | undefined | void> | string | undefined | void;
 
 export interface MomoJob {
-  immediate?: boolean;
   handler: Handler;
   name: string;
   interval: string;
+  firstRunAfter?: number;
   concurrency?: number;
   maxRunning?: number;
 }

src/job/MomoJobBuilder.ts

@@ -13,8 +13,8 @@ export class MomoJobBuilder {
     return this;
   }
 
-  withImmediate(immediate: boolean): this {
-    this.momoJob.immediate = immediate;
+  withFirstRunAfter(firstRunAfter: number): this {
+    this.momoJob.firstRunAfter = firstRunAfter;
     return this;
   }
 

src/job/validate.ts

@@ -5,7 +5,17 @@ import { Logger } from '../logging/Logger';
 import { MomoErrorType } from '../logging/error/MomoErrorType';
 import { momoError } from '../logging/error/MomoError';
 
-export function validate({ name, interval, concurrency, maxRunning }: Job, logger?: Logger): boolean {
+export function validate({ name, interval, firstRunAfter, concurrency, maxRunning }: Job, logger?: Logger): boolean {
+  if (firstRunAfter !== undefined && firstRunAfter < 0) {
+    logger?.error(
+      'job cannot be defined',
+      MomoErrorType.defineJob,
+      { name, firstRunAfter },
+      momoError.invalidFirstRunAfter
+    );
+    return false;
+  }
+
   if (maxRunning < 0) {
     logger?.error('job cannot be defined', MomoErrorType.defineJob, { name, maxRunning }, momoError.invalidMaxRunning);
     return false;

src/logging/error/MomoError.ts

@@ -3,4 +3,5 @@ export const momoError = {
   invalidConcurrency: new Error('concurrency must be at least 1'),
   invalidMaxRunning: new Error('maxRunning must be at least 0'),
   jobNotFound: new Error('job not found in database'),
+  invalidFirstRunAfter: new Error('firstRunAfter must be at least 0'),
 };

src/repository/JobRepository.ts

@@ -66,15 +66,7 @@ export class JobRepository extends Repository<JobEntity> {
   async list(): Promise<MomoJobStatus[]> {
     const jobs = await this.find();
 
-    return jobs.map((job) => {
-      return {
-        name: job.name,
-        interval: job.interval,
-        concurrency: job.concurrency,
-        maxRunning: job.maxRunning,
-        executionInfo: job.executionInfo,
-      };
-    });
+    return jobs.map((job) => ({ ...toJobDefinition(job), executionInfo: job.executionInfo }));
   }
 
   async updateJob(name: string, update: Partial<JobEntity>): Promise<void> {

src/repository/Repository.ts

@@ -18,11 +18,13 @@ export class Repository<ENTITY extends { _id?: ObjectId }> {
   }
 
   async find(filter: Filter<ENTITY> = {}): Promise<ENTITY[]> {
-    return this.collection.find(filter).toArray();
+    const entities = await this.collection.find(filter).toArray();
+    return entities.map(this.mapNullToUndefined);
   }
 
   async findOne(filter: Filter<ENTITY> = {}): Promise<ENTITY | undefined> {
-    return this.collection.findOne(filter);
+    const entity = await this.collection.findOne(filter);
+    return entity === null ? undefined : this.mapNullToUndefined(entity);
   }
 
   async delete(filter: Filter<ENTITY> = {}): Promise<number> {
@@ -33,4 +35,12 @@ export class Repository<ENTITY extends { _id?: ObjectId }> {
   async deleteOne(filter: Filter<ENTITY>): Promise<void> {
     await this.collection.deleteOne(filter);
   }
+
+  // mongodb returns null instead of undefined for optional fields
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  private mapNullToUndefined(entity: any): ENTITY {
+    const keys = Object.keys(entity);
+    const entries = keys.map((key) => [key, entity[key] ?? undefined]);
+    return Object.fromEntries(entries) as ENTITY;
+  }
 }

src/schedule/Schedule.ts

@@ -1,4 +1,5 @@
 import { sum } from 'lodash';
+
 import { ExecutionInfo, ExecutionStatus, JobResult } from '../job/ExecutionInfo';
 import { ExecutionsRepository } from '../repository/ExecutionsRepository';
 import { JobRepository } from '../repository/JobRepository';

src/scheduler/JobScheduler.ts

@@ -21,7 +21,6 @@ export class JobScheduler {
 
   constructor(
     private readonly jobName: string,
-    private readonly immediate: boolean,
     private readonly jobExecutor: JobExecutor,
     private readonly scheduleId: string,
     private readonly executionsRepository: ExecutionsRepository,
@@ -37,7 +36,7 @@ export class JobScheduler {
     jobRepository: JobRepository
   ): JobScheduler {
     const executor = new JobExecutor(job.handler, scheduleId, executionsRepository, jobRepository, logger);
-    return new JobScheduler(job.name, job.immediate, executor, scheduleId, executionsRepository, jobRepository, logger);
+    return new JobScheduler(job.name, executor, scheduleId, executionsRepository, jobRepository, logger);
   }
 
   getUnexpectedErrorCount(): number {
@@ -87,7 +86,7 @@ export class JobScheduler {
 
     this.interval = jobEntity.interval;
 
-    const delay = calculateDelay(interval, this.immediate, jobEntity);
+    const delay = calculateDelay(interval, jobEntity);
 
     this.jobHandle = setIntervalWithDelay(this.executeConcurrently.bind(this), interval, delay);
 

src/scheduler/calculateDelay.ts

@@ -3,10 +3,10 @@ import { max } from 'lodash';
 
 import { JobEntity } from '../repository/JobEntity';
 
-export function calculateDelay(millisecondsInterval: number, immediate: boolean, job: JobEntity): number {
+export function calculateDelay(millisecondsInterval: number, job: JobEntity): number {
   const nextStart = calculateNextStart(millisecondsInterval, job);
   if (nextStart === undefined) {
-    return immediate ? 0 : millisecondsInterval;
+    return job.firstRunAfter ?? millisecondsInterval;
   }
 
   return max([nextStart - DateTime.now().toMillis(), 0]) ?? 0;

test/executor/JobExecutor.spec.ts

@@ -14,7 +14,6 @@ describe('JobExecutor', () => {
   const job: Job = {
     name: 'test',
     interval: '1 minute',
-    immediate: false,
     concurrency: 1,
     maxRunning: 0,
     handler,

test/job/Job.spec.ts

@@ -5,7 +5,6 @@ describe('fromMomoJob', () => {
     const job = { name: 'test', interval: '1 second', handler: () => undefined };
     expect(toJob(job)).toMatchObject({
       ...job,
-      immediate: false,
       concurrency: 1,
       maxRunning: 0,
     });

test/job/MomoJobBuilder.spec.ts

@@ -5,15 +5,15 @@ describe('MomoJobBuilder', () => {
     const momoJob = new MomoJobBuilder()
       .withName('name')
       .withInterval('one minute')
-      .withImmediate(true)
+      .withFirstRunAfter(0)
       .withConcurrency(1)
       .withMaxRunning(1)
       .withHandler(jest.fn())
       .build();
 
     expect(momoJob.name).toEqual('name');
     expect(momoJob.interval).toEqual('one minute');
-    expect(momoJob.immediate).toEqual(true);
+    expect(momoJob.firstRunAfter).toEqual(0);
     expect(momoJob.concurrency).toEqual(1);
     expect(momoJob.maxRunning).toEqual(1);
     expect(momoJob.handler.toString()).toEqual(jest.fn().toString());
@@ -24,7 +24,7 @@ describe('MomoJobBuilder', () => {
 
     expect(momoJob.name).toEqual('name');
     expect(momoJob.interval).toEqual('one minute');
-    expect(momoJob.immediate).toBeUndefined();
+    expect(momoJob.firstRunAfter).toBeUndefined();
     expect(momoJob.concurrency).toBeUndefined();
     expect(momoJob.maxRunning).toBeUndefined();
     expect(momoJob.handler.toString()).toEqual(jest.fn().toString());

test/job/validate.spec.ts

@@ -42,6 +42,19 @@ describe('validate', () => {
     );
   });
 
+  it('reports error when firstRunAfter is invalid', async () => {
+    const job: Job = toJob({ name: 'test', interval: '1 minute', handler: () => 'finished', firstRunAfter: -1 });
+    expect(validate(job, logger)).toBe(false);
+
+    expect(logger.error).toHaveBeenCalledTimes(1);
+    expect(logger.error).toHaveBeenCalledWith(
+      'job cannot be defined',
+      MomoErrorType.defineJob,
+      { name: job.name, firstRunAfter: -1 },
+      momoError.invalidFirstRunAfter
+    );
+  });
+
   it('reports error when maxRunning is invalid', async () => {
     const job: Job = toJob({ name: 'test', interval: '1 minute', handler: () => 'finished', maxRunning: -1 });
     expect(validate(job, logger)).toBe(false);

test/repository/JobRepository.integration.spec.ts

@@ -92,19 +92,19 @@ describe('JobRepository', () => {
 
   describe('list', () => {
     it('returns jobs', async () => {
-      const job1 = {
+      const job1: JobEntity = {
         name: 'job1',
         interval: '1 minute',
+        firstRunAfter: 0,
         executionInfo: {} as ExecutionInfo,
-        running: 2,
         concurrency: 1,
         maxRunning: 3,
       };
-      const job2 = {
+      const job2: JobEntity = {
         name: 'job2',
         interval: '2 minutes',
+        firstRunAfter: 0,
         executionInfo: {} as ExecutionInfo,
-        running: 0,
         concurrency: 1,
         maxRunning: 0,
       };
@@ -117,13 +117,15 @@ describe('JobRepository', () => {
         {
           name: job1.name,
           interval: job1.interval,
+          firstRunAfter: job1.firstRunAfter,
           concurrency: job1.concurrency,
           maxRunning: job1.maxRunning,
           executionInfo: {},
         },
         {
           name: job2.name,
           interval: job2.interval,
+          firstRunAfter: job1.firstRunAfter,
           concurrency: job2.concurrency,
           maxRunning: job2.maxRunning,
           executionInfo: {},

test/schedule/momo.integration.spec.ts

@@ -104,7 +104,7 @@ describe('Momo', () => {
     });
 
     it('executes an immediate job periodically', async () => {
-      await mongoSchedule.define({ ...momoJob, immediate: true });
+      await mongoSchedule.define({ ...momoJob, firstRunAfter: 0 });
 
       await mongoSchedule.start();