fix: Upgrade bullmq and adjust to breaking changes in repeatable jobs (#8446)

packages/modules/workflow-engine-redis/package.json

@@ -52,7 +52,7 @@
     "@mikro-orm/migrations": "5.9.7",
     "@mikro-orm/postgresql": "5.9.7",
     "awilix": "^8.0.0",
-    "bullmq": "^5.4.2",
+    "bullmq": "5.12.0",
     "dotenv": "^16.4.5",
     "ioredis": "^5.3.2",
     "knex": "2.4.2"

packages/modules/workflow-engine-redis/src/utils/workflow-orchestrator-storage.ts

@@ -374,10 +374,7 @@ export class RedisDistributedTransactionStorage
     const jobId =
       typeof jobDefinition === "string" ? jobDefinition : jobDefinition.jobId
 
-    // In order to ensure that the schedule configuration is always up to date, we first cancel an existing job, if there was one
-    // any only then we add the new one.
-    await this.remove(jobId)
-
+    // If it is the same key (eg. the same workflow name), the old one will get overridden.
     await this.queue.add(
       JobType.SCHEDULE,
       {
@@ -388,21 +385,14 @@ export class RedisDistributedTransactionStorage
         repeat: {
           pattern: schedulerOptions.cron,
           limit: schedulerOptions.numberOfExecutions,
+          key: `${JobType.SCHEDULE}_${jobId}`,
         },
-        jobId: `${JobType.SCHEDULE}_${jobId}`,
       }
     )
   }
 
   async remove(jobId: string): Promise<void> {
-    const repeatableJobs = await this.queue.getRepeatableJobs()
-    const job = repeatableJobs.find(
-      (job) => job.id === `${JobType.SCHEDULE}_${jobId}`
-    )
-
-    if (job) {
-      await this.queue.removeRepeatableByKey(job.key)
-    }
+    await this.queue.removeRepeatableByKey(`${JobType.SCHEDULE}_${jobId}`)
   }
 
   async removeAll(): Promise<void> {