refactor(queue): simplify queue configuration and job setup

- Remove unnecessary injection of BullBoard instance in AlertQueueService
- Update job options to reduce default attempts for better performance
- Enhance queue service methods for clearer job scheduling and worker registration

Author: OmerHarell

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

@@ -1,8 +1,7 @@
 import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
 import { AppLoggerService } from '@/common/app-logger/app-logger.service';
 import { AlertService } from './alert.service';
-import type { IQueueService, BullBoardInjectedInstance } from '@/common/queue/types';
-import { BULLBOARD_INSTANCE_INJECTION_TOKEN } from '@/common/queue/types';
+import type { IQueueService } from '@/common/queue/types';
 import { env } from '@/env';
 
 export interface AlertCheckJobData extends Record<string, unknown> {
@@ -15,11 +14,9 @@ export class AlertQueueService implements OnModuleInit {
   private readonly SCHEDULER_ID = 'transaction-monitoring-alert-check';
 
   constructor(
-    @Inject(AppLoggerService) private readonly logger: AppLoggerService,
-    @Inject(AlertService) private readonly alertService: AlertService,
+    private readonly logger: AppLoggerService,
+    private readonly alertService: AlertService,
     @Inject('IQueueService') private readonly queueService: IQueueService,
-    @Inject(BULLBOARD_INSTANCE_INJECTION_TOKEN)
-    private bullBoard: BullBoardInjectedInstance,
   ) {}
 
   async onModuleInit() {
@@ -35,19 +32,21 @@ export class AlertQueueService implements OnModuleInit {
       this.queueService.createQueue<AlertCheckJobData>(this.QUEUE_NAME, {
         name: this.QUEUE_NAME,
         jobOptions: {
-          attempts: 10,
+          attempts: 3,
           backoff: { type: 'exponential', delay: 10000 },
           removeOnComplete: { count: 100, age: 3600 * 24 },
           removeOnFail: false,
         },
       });
-
-      const queue = (this.queueService as any).getQueue({ name: this.QUEUE_NAME });
-      await (this.queueService as any).setupJobScheduler(queue, this.SCHEDULER_ID, {
-        every: 60 * 60 * 1000,
-        jobName: 'alert-check',
-        data: { timestamp: Date.now() },
-      });
+      await this.queueService.setupJobScheduler(
+        this.QUEUE_NAME,
+        this.SCHEDULER_ID,
+        { every: 60 * 60 * 1000 },
+        {
+          name: 'alert-check',
+          data: { timestamp: Date.now() },
+        },
+      );
 
       this.registerWorker();
 

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

@@ -1,5 +1,5 @@
 import { Module } from '@nestjs/common';
-import { QueueService } from './queue.service';
+import { BullMQQueueService } from './queue.service';
 import { QueueBullboardService } from './queue-bullboard.service';
 import { QueueOtelService } from './otel.service';
 import { MonitoringModule } from '@/common/monitoring/monitoring.module';
@@ -11,8 +11,8 @@ import { RedisModule } from '../redis/redis.module';
 @Module({
   imports: [MonitoringModule, RedisModule],
   providers: [
-    QueueService,
-    { provide: 'IQueueService', useExisting: QueueService },
+    BullMQQueueService,
+    { provide: 'IQueueService', useExisting: BullMQQueueService },
     QueueBullboardService,
     QueueOtelService,
     {
@@ -27,7 +27,7 @@ import { RedisModule } from '../redis/redis.module';
     },
   ],
   exports: [
-    QueueService,
+    BullMQQueueService,
     'IQueueService',
     QueueBullboardService,
     QueueOtelService,

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

@@ -5,31 +5,24 @@ import { AppLoggerService } from '@/common/app-logger/app-logger.service';
 import { env } from '@/env';
 import { RedisService } from '../redis/redis.service';
 import { BullMQPrometheusService } from '@/common/monitoring/bullmq-prometheus.service';
-import type { BullBoardInjectedInstance } from './types';
+import type { BullBoardInjectedInstance, IQueueService, QueueOptions } from './types';
 import { QueueBullboardService } from './queue-bullboard.service';
 
-export interface QueueOptions<T = any> {
-  name: string;
-  concurrency?: number;
-  jobOptions?: {
-    attempts?: number;
-    backoff?: {
-      type: 'exponential' | 'fixed';
-      delay: number;
-    };
-    removeOnComplete?: boolean | number | { count: number; age: number };
-    removeOnFail?: boolean | number | { count: number; age: number };
-    priority?: number;
-  };
-}
-
 @Injectable()
-export class QueueService implements OnModuleDestroy {
+export class BullMQQueueService implements OnModuleDestroy, IQueueService {
   private redisClient: IORedis | null;
   private queues: Map<string, Queue> = new Map();
   private workers: Map<string, Worker> = new Map();
   private readonly shouldProcessJobs: boolean;
 
+  async addJob<T = any>(queueName: string, job_name: string, data: T, opts?: any): Promise<any> {
+    const queue = this.getQueue(queueName);
+
+    return queue.add(job_name, data, {
+      priority: opts?.priority,
+    });
+  }
+
   constructor(
     private readonly logger: AppLoggerService,
     private readonly redisService: RedisService,
@@ -58,40 +51,16 @@ export class QueueService implements OnModuleDestroy {
     return this.shouldProcessJobs;
   }
 
-  private getQueue<T = any>(options: QueueOptions<T>): Queue<T> {
+  public getQueue(queueName: string): Queue {
     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>;
-    }
-
-    const queue = new Queue<T>(options.name, {
-      connection: this.redisClient as any,
-      defaultJobOptions: {
-        attempts: options.jobOptions?.attempts ?? 3,
-        backoff: options.jobOptions?.backoff ?? {
-          type: options.jobOptions?.backoff?.type ?? 'exponential',
-          delay: options.jobOptions?.backoff?.delay ?? 5000,
-        },
-        removeOnComplete: options.jobOptions?.removeOnComplete ?? { count: 100, age: 3600 * 24 },
-        removeOnFail: options.jobOptions?.removeOnFail ?? false,
-      },
-    });
-
-    this.queues.set(options.name, queue as Queue);
-    this.logger.log(`Queue created: ${options.name}`);
-
-    if (this.bullMQPrometheusService) {
-      this.bullMQPrometheusService.registerQueue(queue);
-    }
-
-    if (this.shouldProcessJobs && this.bullBoard && this.queueBullboardService) {
-      this.queueBullboardService.registerQueue(this.bullBoard, queue);
+    if (this.queues.has(queueName)) {
+      return this.queues.get(queueName) as Queue;
     }
 
-    return queue;
+    throw new Error(`Queue with name '${queueName}' does not exist. Please create it first.`);
   }
 
   registerWorker<T = any>(
@@ -115,8 +84,8 @@ export class QueueService implements OnModuleDestroy {
       return;
     }
 
-    const worker = new Worker<T, any, string>(queueName, processor, {
-      connection: this.redisClient as any,
+    const worker = new Worker(queueName, processor, {
+      connection: this.redisClient,
       concurrency: options.concurrency ?? 1,
       autorun: true,
     });
@@ -139,25 +108,25 @@ export class QueueService implements OnModuleDestroy {
       });
     });
 
-    this.workers.set(queueName, worker as unknown as Worker<T, any, string>);
+    this.workers.set(queueName, worker);
     this.logger.log(`Worker registered for queue: ${queueName}`);
   }
 
-  createQueue<T = any>(queueName: string, options?: QueueOptions<T>): void {
+  createQueue(queueName: string, options?: QueueOptions): void {
     if (this.queues.has(queueName)) {
       return;
     }
 
-    const queue = new Queue<T, any, string>(queueName, {
-      connection: this.redisClient as any,
+    const queue = new Queue(queueName, {
+      connection: this.redisClient as IORedis,
       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.queues.set(queueName, queue);
     this.logger.log(`Queue created: ${queueName}`);
 
     if (this.bullMQPrometheusService) {
@@ -182,41 +151,36 @@ export class QueueService implements OnModuleDestroy {
   }
 
   async setupJobScheduler<T = any>(
-    queue: Queue,
+    queueName: string,
     schedulerId: string,
-    options: {
-      every: number;
-      data?: T;
-      jobName?: string;
-      jobOptions?: {
-        attempts?: number;
-        backoff?: {
-          type: 'exponential' | 'fixed';
-          delay: number;
-        };
-      };
+    scheduleOpts: { every: number },
+    jobOpts: {
+      name: string;
+      data: T;
+      opts?: any;
     },
-  ) {
+  ): Promise<any> {
     try {
-      const jobName = options.jobName || 'scheduled-job';
+      const queue = this.getQueue(queueName);
+      const jobName = jobOpts.name;
       const firstJob = await queue.upsertJobScheduler(
         schedulerId,
-        { every: options.every, jobId: schedulerId },
+        { every: scheduleOpts.every, jobId: schedulerId },
         {
           name: jobName,
-          data: options.data || { timestamp: Date.now() },
+          data: jobOpts.data || { timestamp: Date.now() },
           opts: {
-            attempts: options.jobOptions?.attempts || 10,
-            backoff: options.jobOptions?.backoff || {
+            attempts: jobOpts.opts?.attempts || 3,
+            backoff: jobOpts.opts?.backoff || {
               type: 'exponential',
-              delay: 10000,
+              delay: 3000,
             },
           },
         },
       );
       this.logger.log(`Created job scheduler: ${schedulerId}`, {
         schedulerId,
-        every: options.every,
+        every: scheduleOpts.every,
         jobId: firstJob?.id,
       });
 

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

@@ -1,32 +1,47 @@
 import { createBullBoard } from '@bull-board/api';
 import { ExpressAdapter } from '@bull-board/express';
-import type { QueueOptions } from './queue.service';
 
 export const BULLBOARD_INSTANCE_INJECTION_TOKEN = 'BULLBOARD_INSTANCE';
 
 export interface BullBoardInjectedInstance {
   boardInstance: ReturnType<typeof createBullBoard>;
   serverAdapter: ExpressAdapter;
 }
+
 export interface IQueueService {
-  /**
-   * Add a job to the queue.
-   */
-  addJob<T = any>(queueName: string, data: T, opts?: any): Promise<any>;
+  addJob<T = any>(queueName: string, job_name: string, data: T, opts?: any): Promise<any>;
 
-  /**
-   * Register a worker for a queue. The processor should be a standalone function for testability and abstraction.
-   */
   registerWorker<T = any>(
     queueName: string,
     processor: (job: any) => Promise<any>,
     options?: { concurrency?: number },
   ): void;
 
   isWorkerEnabled(): boolean;
-
-  /**
-   * Explicitly create/configure a queue with options.
-   */
   createQueue<T = any>(queueName: string, options?: QueueOptions<T>): void;
+  setupJobScheduler<T = any>(
+    queueName: string,
+    schedulerId: string,
+    scheduleOpts: { every: number },
+    jobOpts: {
+      name: string;
+      data: T;
+      opts?: any;
+    },
+  ): Promise<any>;
+}
+
+export interface QueueOptions<T = any> {
+  name: string;
+  concurrency?: number;
+  jobOptions?: {
+    attempts?: number;
+    backoff?: {
+      type: 'exponential' | 'fixed';
+      delay: number;
+    };
+    removeOnComplete?: boolean | number | { count: number; age: number };
+    removeOnFail?: boolean | number | { count: number; age: number };
+    priority?: number;
+  };
 }

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

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { RedisService, redisProvider } from './redis.service';
+import { AppLoggerModule } from '@/common/app-logger/app-logger.module';
+
+@Module({
+  imports: [AppLoggerModule],
+  providers: [RedisService, redisProvider],
+  exports: [RedisService, redisProvider],
+})
+export class RedisModule {}

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

@@ -0,0 +1,49 @@
+import { Injectable, OnModuleDestroy } from '@nestjs/common';
+import IORedis from 'ioredis';
+import { env } from '@/env';
+import { AppLoggerService } from '@/common/app-logger/app-logger.service';
+
+export const REDIS_CLIENT = Symbol('REDIS_CLIENT');
+
+@Injectable()
+export class RedisService implements OnModuleDestroy {
+  public readonly client: IORedis;
+
+  constructor(private readonly logger: AppLoggerService) {
+    if (!env.QUEUE_SYSTEM_ENABLED) {
+      this.client = null as any;
+
+      return;
+    }
+
+    const redisConfig = {
+      host: env.REDIS_HOST,
+      port: env.REDIS_PORT,
+      password: env.REDIS_PASSWORD,
+      maxRetriesPerRequest: null,
+      ...(env.ENVIRONMENT_NAME !== 'local' ? { tls: {} } : {}),
+    };
+    this.client = new IORedis(redisConfig);
+
+    this.client.on('error', error => {
+      this.logger.error('Redis connection error', { error });
+    });
+
+    this.client.on('connect', () => {
+      this.logger.log('Redis connected successfully');
+    });
+
+    this.logger.log('Redis client initialized via RedisService.');
+  }
+
+  async onModuleDestroy() {
+    if (this.client) {
+      await this.client.quit();
+    }
+  }
+}
+
+export const redisProvider = {
+  provide: REDIS_CLIENT,
+  useExisting: RedisService,
+};