feat(medusa): Option to override existing cron job (#2989)

.changeset/old-scissors-develop.md

@@ -0,0 +1,5 @@
+---
+"@medusajs/medusa": patch
+---
+
+feat(medusa): Option to override existing cron job

packages/medusa/src/services/event-bus.ts

@@ -7,7 +7,7 @@ import { StagedJob } from "../models"
 import { StagedJobRepository } from "../repositories/staged-job"
 import { ConfigModule, Logger } from "../types/global"
 import { sleep } from "../utils/sleep"
-import JobSchedulerService from "./job-scheduler"
+import JobSchedulerService, { CreateJobOptions } from "./job-scheduler"
 
 type InjectedDependencies = {
   manager: EntityManager
@@ -413,12 +413,19 @@ export default class EventBusService {
    * @param handler - the handler to call on each cron job
    * @return void
    */
-  createCronJob<T>(
+  async createCronJob<T>(
     eventName: string,
     data: T,
     cron: string,
-    handler: Subscriber
-  ): void {
-    this.jobSchedulerService_.create(eventName, data, cron, handler)
+    handler: Subscriber,
+    options?: CreateJobOptions
+  ): Promise<void> {
+    await this.jobSchedulerService_.create(
+      eventName,
+      data,
+      cron,
+      handler,
+      options ?? {}
+    )
   }
 }

packages/medusa/src/services/job-scheduler.ts

@@ -13,6 +13,10 @@ type ScheduledJobHandler<T = unknown> = (
   eventName: string
 ) => Promise<void>
 
+export type CreateJobOptions = {
+  keepExisting?: boolean
+}
+
 export default class JobSchedulerService {
   protected readonly config_: ConfigModule
   protected readonly logger_: Logger
@@ -102,21 +106,34 @@ export default class JobSchedulerService {
    * @param handler - the handler to call on the job
    * @return void
    */
-  create<T>(
+  async create<T>(
     eventName: string,
     data: T,
     schedule: string,
-    handler: ScheduledJobHandler
-  ): void {
+    handler: ScheduledJobHandler,
+    options: CreateJobOptions
+  ): Promise<void> {
     this.logger_.info(`Registering ${eventName}`)
     this.registerHandler(eventName, handler)
 
-    this.queue_.add(
-      {
-        eventName,
-        data,
-      },
-      { repeat: { cron: schedule } }
-    )
+    const jobToCreate = {
+      eventName,
+      data,
+    }
+    const repeatOpts = { repeat: { cron: schedule } }
+
+    if (options?.keepExisting) {
+      return await this.queue_.add(eventName, jobToCreate, repeatOpts)
+    }
+
+    const existingJobs = (await this.queue_.getRepeatableJobs()) ?? []
+
+    const existingJob = existingJobs.find((job) => job.name === eventName)
+
+    if (existingJob) {
+      await this.queue_.removeRepeatableByKey(existingJob.key)
+    }
+
+    return await this.queue_.add(eventName, jobToCreate, repeatOpts)
   }
 }