feat(job-cleaning): add support for cleaning scheduled jobs (#1020)

* add support for cleaning scheduled jobs

* clean up

Author: benatshippabo

packages/api/src/handlers/cleanJob.ts

@@ -1,17 +1,39 @@
-import {
-  BullBoardRequest,
-  ControllerHandlerReturnType,
-  QueueJob,
-} from '../../typings/app';
+import { BullBoardRequest, ControllerHandlerReturnType, QueueJob } from '../../typings/app';
 import { jobProvider } from '../providers/job';
 import { queueProvider } from '../providers/queue';
+import { BaseAdapter } from '../queueAdapters/base';
+
+function extractRepeatJobKey(job: QueueJob): string | undefined {
+  const key = job.repeatJobKey;
+
+  if (typeof key === 'string' && key.length > 0) {
+    return key;
+  }
+}
 
 async function cleanJob(
   _req: BullBoardRequest,
-  job: QueueJob
+  job: QueueJob,
+  queue: BaseAdapter
 ): Promise<ControllerHandlerReturnType> {
-  await job.remove();
+  const repeatJobKey = extractRepeatJobKey(job);
 
+  if (repeatJobKey) {
+    const removed = await queue.removeJobScheduler(repeatJobKey);
+
+    if (!removed) {
+      throw new Error(
+        `Failed to remove scheduler ${repeatJobKey} for job ${job.toJSON().id ?? 'unknown id'}.`
+      );
+    }
+
+    return {
+      status: 204,
+      body: {},
+    };
+  }
+
+  await job.remove();
   return {
     status: 204,
     body: {},

packages/api/src/queueAdapters/base.ts

@@ -97,6 +97,8 @@ export abstract class BaseAdapter {
 
   public abstract promoteAll(): Promise<void>;
 
+  public abstract removeJobScheduler(id: string): Promise<boolean>;
+
   public abstract getStatuses(): Status[];
 
   public abstract getJobStatuses(): JobStatus[];

packages/api/src/queueAdapters/bull.ts

@@ -1,5 +1,5 @@
-import { Job, Queue } from 'bull';
-import BullQueue from 'bull';
+import BullQueue, { Job, Queue } from 'bull';
+
 import {
   JobCleanStatus,
   JobCounts,
@@ -12,7 +12,10 @@ import { STATUSES } from '../constants/statuses';
 import { BaseAdapter } from './base';
 
 export class BullAdapter extends BaseAdapter {
-  constructor(public queue: Queue, options: Partial<QueueAdapterOptions> = {}) {
+  constructor(
+    public queue: Queue,
+    options: Partial<QueueAdapterOptions> = {}
+  ) {
     super('bull', { ...options, allowCompletedRetries: false });
 
     if (!(queue instanceof BullQueue)) {
@@ -79,6 +82,10 @@ export class BullAdapter extends BaseAdapter {
     await Promise.all(jobs.map((job) => job.promote()));
   }
 
+  public async removeJobScheduler(_id: string): Promise<boolean> {
+    return false;
+  }
+
   public getStatuses(): Status<'bull'>[] {
     return [
       STATUSES.latest,

packages/api/src/queueAdapters/bullMQ.ts

@@ -1,4 +1,5 @@
 import { Job, Queue } from 'bullmq';
+
 import {
   JobCleanStatus,
   JobCounts,
@@ -11,7 +12,10 @@ import { STATUSES } from '../constants/statuses';
 import { BaseAdapter } from './base';
 
 export class BullMQAdapter extends BaseAdapter {
-  constructor(private queue: Queue, options: Partial<QueueAdapterOptions> = {}) {
+  constructor(
+    private queue: Queue,
+    options: Partial<QueueAdapterOptions> = {}
+  ) {
     const libName = 'bullmq';
     super(libName, options);
     if (
@@ -80,6 +84,10 @@ export class BullMQAdapter extends BaseAdapter {
     }
   }
 
+  public removeJobScheduler(id: string): Promise<boolean> {
+    return this.queue.removeJobScheduler(id);
+  }
+
   public getStatuses(): Status[] {
     return [
       STATUSES.latest,