refactor(queue): improve alert queue service and worker registration

- Simplify dependency injection for services in AlertQueueService
- Replace queue setup logic with createQueue method and refactor worker processing
- Eliminate unused code and streamline error handling in webhook job processing

Author: OmerHarell

services/workflows-service/src/alert/alert-queue.service.ts

@@ -1,10 +1,7 @@
 import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
-import { Job } from 'bullmq';
 import { AppLoggerService } from '@/common/app-logger/app-logger.service';
 import { AlertService } from './alert.service';
-import { QueueService } from '@/common/queue/queue.service';
-import { QueueBullboardService } from '@/common/queue/queue-bullboard.service';
-import { BullMQPrometheusService } from '@/common/monitoring/bullmq-prometheus.service';
+import type { IQueueService } from '@/common/queue/queue.interface';
 import { BULLBOARD_INSTANCE_INJECTION_TOKEN } from '@/common/queue/types';
 import type { BullBoardInjectedInstance } from '@/common/queue/types';
 import { env } from '@/env';
@@ -19,11 +16,9 @@ export class AlertQueueService implements OnModuleInit {
   private readonly SCHEDULER_ID = 'transaction-monitoring-alert-check';
 
   constructor(
-    private readonly logger: AppLoggerService,
-    private readonly alertService: AlertService,
-    private readonly queueService: QueueService,
-    private readonly queueBullboardService: QueueBullboardService,
-    private readonly bullMQPrometheusService: BullMQPrometheusService,
+    @Inject(AppLoggerService) private readonly logger: AppLoggerService,
+    @Inject(AlertService) private readonly alertService: AlertService,
+    @Inject('IQueueService') private readonly queueService: IQueueService,
     @Inject(BULLBOARD_INSTANCE_INJECTION_TOKEN)
     private bullBoard: BullBoardInjectedInstance,
   ) {}
@@ -38,32 +33,16 @@ export class AlertQueueService implements OnModuleInit {
 
   private async setupAlertQueue() {
     try {
-      const queue = this.queueService.getQueue<AlertCheckJobData>({
+      this.queueService.createQueue<AlertCheckJobData>(this.QUEUE_NAME, {
         name: this.QUEUE_NAME,
         jobOptions: {
           attempts: 10,
-          backoff: {
-            type: 'exponential',
-            delay: 10_000,
-          },
+          backoff: { type: 'exponential', delay: 10000 },
           removeOnComplete: { count: 100, age: 3600 * 24 },
           removeOnFail: false,
         },
       });
 
-      await this.queueService.setupJobScheduler(queue, this.SCHEDULER_ID, {
-        every: 60 * 60 * 1000,
-        jobName: 'check-transaction-monitoring-alerts',
-        data: { timestamp: Date.now() },
-        jobOptions: {
-          attempts: 10,
-          backoff: {
-            type: 'exponential',
-            delay: 10_000,
-          },
-        },
-      });
-
       this.registerWorker();
 
       this.logger.log('Alert queue system setup complete');
@@ -73,22 +52,21 @@ export class AlertQueueService implements OnModuleInit {
   }
 
   private registerWorker() {
-    this.queueService.registerWorker<AlertCheckJobData>(
-      this.QUEUE_NAME,
-      async (job: Job<AlertCheckJobData>) => {
-        this.logger.log('Processing transaction monitoring alerts check job', {
-          jobId: job.id,
-        });
-
-        await this.alertService.checkAllAlerts();
+    this.queueService.registerWorker(this.QUEUE_NAME, this.processAlertCheckJob.bind(this), {
+      concurrency: 1,
+    });
+  }
 
-        this.logger.log('Completed transaction monitoring alerts check', {
-          jobId: job.id,
-        });
+  private async processAlertCheckJob(job: any) {
+    this.logger.log('Processing transaction monitoring alerts check job', { jobId: job.id });
+    try {
+      await this.alertService.checkAllAlerts();
+      this.logger.log('Completed transaction monitoring alerts check', { jobId: job.id });
 
-        return { success: true, timestamp: Date.now() };
-      },
-      { concurrency: 1 },
-    );
+      return { success: true, timestamp: Date.now() };
+    } catch (error) {
+      this.logger.error('Alert check job failed', { jobId: job.id, error });
+      throw error;
+    }
   }
 }

services/workflows-service/src/common/queue/queue-bullboard.service.ts

@@ -22,14 +22,4 @@ export class QueueBullboardService {
       this.logger.error(`Error registering queue ${queue.name} with BullBoard`, { error });
     }
   }
-
-  registerQueues(bullBoardInstance: any, queues: Queue[]) {
-    try {
-      const adapters = queues.map(queue => new BullMQAdapter(queue));
-      bullBoardInstance.boardInstance.setQueues(adapters);
-      this.logger.log(`Registered ${queues.length} queues with BullBoard`);
-    } catch (error) {
-      this.logger.error('Error registering queues with BullBoard', { error });
-    }
-  }
 }

services/workflows-service/src/common/queue/queue.module.ts

@@ -12,6 +12,7 @@ import { ExpressAdapter } from '@bull-board/express';
   imports: [MonitoringModule],
   providers: [
     QueueService,
+    { provide: 'IQueueService', useExisting: QueueService },
     QueueBullboardService,
     QueueOtelService,
     redisProvider,
@@ -28,6 +29,7 @@ import { ExpressAdapter } from '@bull-board/express';
   ],
   exports: [
     QueueService,
+    'IQueueService',
     QueueBullboardService,
     QueueOtelService,
     redisProvider,

services/workflows-service/src/common/queue/queue.service.ts

@@ -1,5 +1,5 @@
 import { Injectable, OnModuleDestroy, Inject } from '@nestjs/common';
-import { Queue, Worker, Job } from 'bullmq';
+import { Queue, Worker } from 'bullmq';
 import IORedis from 'ioredis';
 import { AppLoggerService } from '@/common/app-logger/app-logger.service';
 import { env } from '@/env';
@@ -9,8 +9,6 @@ import { BullMQPrometheusService } from '@/common/monitoring/bullmq-prometheus.s
 import type { BullBoardInjectedInstance } from './types';
 import { QueueBullboardService } from './queue-bullboard.service';
 
-export type JobProcessor<T = any> = (job: Job<T>) => Promise<any>;
-
 export interface QueueOptions<T = any> {
   name: string;
   concurrency?: number;
@@ -62,16 +60,16 @@ export class QueueService implements OnModuleDestroy {
     return this.shouldProcessJobs;
   }
 
-  getQueue<T = any, R = any, N extends string = string>(options: QueueOptions<T>): Queue<T, R, N> {
+  private getQueue<T = any>(options: QueueOptions<T>): Queue<T> {
     if (!this.redisClient) {
       throw new Error('Redis client not initialized');
     }
 
     if (this.queues.has(options.name)) {
-      return this.queues.get(options.name) as unknown as Queue<T, R, N>;
+      return this.queues.get(options.name) as unknown as Queue<T>;
     }
 
-    const queue = new Queue<T, R, N>(options.name, {
+    const queue = new Queue<T>(options.name, {
       connection: this.redisClient as any,
       defaultJobOptions: {
         attempts: options.jobOptions?.attempts ?? 3,
@@ -98,50 +96,32 @@ export class QueueService implements OnModuleDestroy {
     return queue;
   }
 
-  registerWorker<T = any, R = any, N extends string = string>(
+  registerWorker<T = any>(
     queueName: string,
-    processor: JobProcessor<T>,
-    options: {
-      concurrency?: number;
-      forceLocalProcessing?: boolean;
-    } = {},
-  ): Worker<T, R, N> | null {
+    processor: (job: any) => Promise<any>,
+    options: { concurrency?: number } = {},
+  ): void {
     if (!this.redisClient) {
       throw new Error('Redis client not initialized');
     }
 
-    if (!this.shouldProcessJobs && !options.forceLocalProcessing) {
+    if (!this.shouldProcessJobs) {
       this.logger.debug(
         `Skipping worker registration for queue ${queueName} (not a worker instance)`,
       );
 
-      return null;
+      return;
     }
 
     if (this.workers.has(queueName)) {
-      return this.workers.get(queueName) as unknown as Worker<T, R, N>;
+      return;
     }
 
-    const worker = new Worker<T, R, N>(
-      queueName,
-      async job => {
-        try {
-          return await processor(job);
-        } catch (error) {
-          this.logger.error(`Error processing job ${job.id} in queue ${queueName}`, {
-            error,
-            jobId: job.id,
-            queueName,
-          });
-          throw error;
-        }
-      },
-      {
-        connection: this.redisClient as any,
-        concurrency: options.concurrency ?? 1,
-        autorun: true,
-      },
-    );
+    const worker = new Worker<T, any, string>(queueName, processor, {
+      connection: this.redisClient as any,
+      concurrency: options.concurrency ?? 1,
+      autorun: true,
+    });
 
     worker.on('failed', (job, error) => {
       this.logger.error(`Job ${job?.id} failed in queue ${queueName}`, {
@@ -161,69 +141,34 @@ export class QueueService implements OnModuleDestroy {
       });
     });
 
-    this.workers.set(queueName, worker as unknown as Worker);
+    this.workers.set(queueName, worker as unknown as Worker<T, any, string>);
     this.logger.log(`Worker registered for queue: ${queueName}`);
-
-    return worker;
   }
 
-  async setupJobScheduler<T = any>(
-    queue: Queue,
-    schedulerId: string,
-    options: {
-      every: number;
-      data?: T;
-      jobName?: string;
-      jobOptions?: {
-        attempts?: number;
-        backoff?: {
-          type: 'exponential' | 'fixed';
-          delay: number;
-        };
-      };
-    },
-  ) {
-    try {
-      const schedulers = await queue.getJobSchedulers();
-      const existingScheduler = schedulers.find(s => s.id === schedulerId);
-
-      if (existingScheduler) {
-        this.logger.log(`Job scheduler already exists: ${schedulerId}`, {
-          schedulerId,
-          pattern: existingScheduler.pattern,
-          every: existingScheduler.every,
-        });
-
-        return existingScheduler;
-      }
-
-      const jobName = options.jobName || 'scheduled-job';
-      const firstJob = await queue.upsertJobScheduler(
-        schedulerId,
-        { every: options.every, jobId: schedulerId },
-        {
-          name: jobName,
-          data: options.data || { timestamp: Date.now() },
-          opts: {
-            attempts: options.jobOptions?.attempts || 10,
-            backoff: options.jobOptions?.backoff || {
-              type: 'exponential',
-              delay: 10000,
-            },
-          },
-        },
-      );
+  createQueue<T = any>(queueName: string, options?: QueueOptions<T>): void {
+    if (this.queues.has(queueName)) {
+      // Optionally update options if needed
+      return;
+    }
 
-      this.logger.log(`Created job scheduler: ${schedulerId}`, {
-        schedulerId,
-        every: options.every,
-        firstJobId: firstJob?.id,
-      });
+    const queue = new Queue<T, any, string>(queueName, {
+      connection: this.redisClient as any,
+      defaultJobOptions: options?.jobOptions ?? {
+        attempts: 3,
+        backoff: { type: 'exponential', delay: 5000 },
+        removeOnComplete: { count: 100, age: 3600 * 24 },
+        removeOnFail: false,
+      },
+    });
+    this.queues.set(queueName, queue as Queue);
+    this.logger.log(`Queue created: ${queueName}`);
+
+    if (this.bullMQPrometheusService) {
+      this.bullMQPrometheusService.registerQueue(queue);
+    }
 
-      return firstJob;
-    } catch (error) {
-      this.logger.error(`Failed to set up job scheduler: ${schedulerId}`, { error });
-      throw error;
+    if (this.shouldProcessJobs && this.bullBoard && this.queueBullboardService) {
+      this.queueBullboardService.registerQueue(this.bullBoard, queue);
     }
   }
 

services/workflows-service/src/common/queue/types.ts

@@ -7,14 +7,3 @@ export interface BullBoardInjectedInstance {
   boardInstance: ReturnType<typeof createBullBoard>;
   serverAdapter: ExpressAdapter;
 }
-
-export const bullBoardProvider = {
-  provide: BULLBOARD_INSTANCE_INJECTION_TOKEN,
-  useFactory: (): BullBoardInjectedInstance => {
-    const serverAdapter = new ExpressAdapter();
-    serverAdapter.setBasePath('/api/queues');
-    const boardInstance = createBullBoard({ queues: [], serverAdapter });
-
-    return { boardInstance, serverAdapter };
-  },
-};