refactor(queue): streamline job options and improve queue management

- Remove deprecated job option settings for consistency
- Introduce default job options in queue creation function
- Add validation to ensure valid queue names are provided

Author: OmerHarell

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

@@ -29,15 +29,6 @@ export class AlertQueueService implements OnModuleInit {
 
   private async setupAlertQueue() {
     try {
-      this.queueService.createQueue<AlertCheckJobData>(this.QUEUE_NAME, {
-        name: this.QUEUE_NAME,
-        jobOptions: {
-          attempts: 3,
-          backoff: { type: 'exponential', delay: 10000 },
-          removeOnComplete: { count: 100, age: 3600 * 24 },
-          removeOnFail: false,
-        },
-      });
       await this.queueService.setupJobScheduler(
         this.QUEUE_NAME,
         this.SCHEDULER_ID,
@@ -46,6 +37,12 @@ export class AlertQueueService implements OnModuleInit {
           name: 'alert-check',
           data: { timestamp: Date.now() },
         },
+        {
+          jobOptions: {
+            attempts: 2,
+            backoff: { type: 'exponential', delay: 10000 },
+          },
+        },
       );
 
       this.registerWorker();

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

@@ -8,6 +8,13 @@ import { BullMQPrometheusService } from '@/common/monitoring/bullmq-prometheus.s
 import type { BullBoardInjectedInstance, IQueueService, QueueOptions } from './types';
 import { QueueBullboardService } from './queue-bullboard.service';
 
+const defaultJobOptions = {
+  attempts: 3,
+  backoff: { type: 'exponential', delay: 2000 },
+  removeOnComplete: { count: 100, age: 3600 * 24 * 7 },
+  removeOnFail: false,
+};
+
 @Injectable()
 export class BullMQQueueService implements OnModuleDestroy, IQueueService {
   private redisClient: IORedis | null;
@@ -51,6 +58,12 @@ export class BullMQQueueService implements OnModuleDestroy, IQueueService {
     return this.shouldProcessJobs;
   }
 
+  private validateQueueName(queueName: string): void {
+    if (!queueName || typeof queueName !== 'string' || queueName.trim().length === 0) {
+      throw new Error('Queue name must be a non-empty string');
+    }
+  }
+
   public getQueue(queueName: string): Queue {
     if (!this.redisClient) {
       throw new Error('Redis client not initialized');
@@ -63,11 +76,13 @@ export class BullMQQueueService implements OnModuleDestroy, IQueueService {
     throw new Error(`Queue with name '${queueName}' does not exist. Please create it first.`);
   }
 
-  registerWorker<T = any>(
+  registerWorker(
     queueName: string,
     processor: (job: any) => Promise<any>,
     options: { concurrency?: number } = {},
   ): void {
+    this.validateQueueName(queueName);
+
     if (!this.redisClient) {
       throw new Error('Redis client not initialized');
     }
@@ -113,18 +128,16 @@ export class BullMQQueueService implements OnModuleDestroy, IQueueService {
   }
 
   createQueue(queueName: string, options?: QueueOptions): void {
+    this.validateQueueName(queueName);
+
     if (this.queues.has(queueName)) {
       return;
     }
 
+    const mergedJobOptions = { ...defaultJobOptions, ...(options?.jobOptions || {}) };
     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,
-      },
+      defaultJobOptions: mergedJobOptions,
     });
     this.queues.set(queueName, queue);
     this.logger.log(`Queue created: ${queueName}`);
@@ -157,25 +170,24 @@ export class BullMQQueueService implements OnModuleDestroy, IQueueService {
     jobOpts: {
       name: string;
       data: T;
-      opts?: any;
     },
+    queueOptions?: QueueOptions,
   ): Promise<any> {
+    this.validateQueueName(queueName);
     try {
+      if (!this.queues.has(queueName)) {
+        this.createQueue(queueName, queueOptions);
+      }
+
       const queue = this.getQueue(queueName);
+
       const jobName = jobOpts.name;
       const firstJob = await queue.upsertJobScheduler(
         schedulerId,
         { every: scheduleOpts.every, jobId: schedulerId },
         {
           name: jobName,
           data: jobOpts.data || { timestamp: Date.now() },
-          opts: {
-            attempts: jobOpts.opts?.attempts || 3,
-            backoff: jobOpts.opts?.backoff || {
-              type: 'exponential',
-              delay: 3000,
-            },
-          },
         },
       );
       this.logger.log(`Created job scheduler: ${schedulerId}`, {

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

@@ -11,29 +11,27 @@ export interface BullBoardInjectedInstance {
 export interface IQueueService {
   addJob<T = any>(queueName: string, job_name: string, data: T, opts?: any): Promise<any>;
 
-  registerWorker<T = any>(
+  registerWorker(
     queueName: string,
     processor: (job: any) => Promise<any>,
     options?: { concurrency?: number },
   ): void;
 
   isWorkerEnabled(): boolean;
-  createQueue<T = any>(queueName: string, options?: QueueOptions<T>): void;
+  createQueue(queueName: string, options?: QueueOptions): void;
   setupJobScheduler<T = any>(
     queueName: string,
     schedulerId: string,
     scheduleOpts: { every: number },
     jobOpts: {
       name: string;
       data: T;
-      opts?: any;
     },
+    queueOptions?: QueueOptions,
   ): Promise<any>;
 }
 
-export interface QueueOptions<T = any> {
-  name: string;
-  concurrency?: number;
+export interface QueueOptions {
   jobOptions?: {
     attempts?: number;
     backoff?: {

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

@@ -7,11 +7,15 @@ export const REDIS_CLIENT = Symbol('REDIS_CLIENT');
 
 @Injectable()
 export class RedisService implements OnModuleDestroy {
-  public readonly client: IORedis;
+  public readonly client!: IORedis;
 
   constructor(private readonly logger: AppLoggerService) {
     if (!env.QUEUE_SYSTEM_ENABLED) {
-      this.client = null as any;
+      Object.defineProperty(this, 'client', {
+        get: () => {
+          throw new Error('Redis client is not available when QUEUE_SYSTEM_ENABLED is false');
+        },
+      });
 
       return;
     }

services/workflows-service/src/webhooks/webhooks.service.ts

@@ -56,15 +56,7 @@ export class WebhooksService implements OnModuleInit {
 
   private async setupQueueSystem() {
     try {
-      this.queueService.createQueue<OutgoingWebhookJobData>(this.QUEUE_NAME, {
-        name: this.QUEUE_NAME,
-        jobOptions: {
-          attempts: 3,
-          backoff: { type: 'exponential', delay: 5000 },
-          removeOnComplete: { count: 1000, age: 3600 * 24 * 7 },
-          removeOnFail: false,
-        },
-      });
+      this.queueService.createQueue(this.QUEUE_NAME);
 
       this.registerWorker();
 
@@ -77,11 +69,9 @@ export class WebhooksService implements OnModuleInit {
   }
 
   private registerWorker() {
-    this.queueService.registerWorker<OutgoingWebhookJobData>(
-      this.QUEUE_NAME,
-      this.processWebhookJob.bind(this),
-      { concurrency: 10 },
-    );
+    this.queueService.registerWorker(this.QUEUE_NAME, this.processWebhookJob.bind(this), {
+      concurrency: 10,
+    });
   }
 
   private async processWebhookJob(job: Job<OutgoingWebhookJobData>) {