feat(queue): implement job scheduler in alert queue service

- Add job scheduling setup for alert-check jobs with specific intervals
- Update relevant imports and refactor the QueueService to utilize RedisService
- Enhance IQueueService interface with additional documentation and job-related methods

Author: OmerHarell

pnpm-lock.yaml

@@ -1,9 +1,8 @@
 import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
 import { AppLoggerService } from '@/common/app-logger/app-logger.service';
 import { AlertService } from './alert.service';
-import type { IQueueService } from '@/common/queue/queue.interface';
+import type { IQueueService, BullBoardInjectedInstance } from '@/common/queue/types';
 import { BULLBOARD_INSTANCE_INJECTION_TOKEN } from '@/common/queue/types';
-import type { BullBoardInjectedInstance } from '@/common/queue/types';
 import { env } from '@/env';
 
 export interface AlertCheckJobData extends Record<string, unknown> {
@@ -43,6 +42,13 @@ export class AlertQueueService implements OnModuleInit {
         },
       });
 
+      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() },
+      });
+
       this.registerWorker();
 
       this.logger.log('Alert queue system setup complete');

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

@@ -1,19 +1,18 @@
-import { MiddlewareConsumer, Module } from '@nestjs/common';
 import { ConfigModule } from '@nestjs/config';
 import { ClsModule } from 'nestjs-cls';
 
 import { AnalyticsModule } from '@/common/analytics-logger/analytics.module';
 import { AppLoggerModule } from '@/common/app-logger/app-logger.module';
 import { QueueModule } from '@/common/queue/queue.module';
-import { configs, env } from '@/env';
+import { configs } from '@/env';
 import { validate } from '@/env-validate';
 import { SecretsManagerModule } from '@/secrets-manager/secrets-manager.module';
 import { SentryModule } from '@/sentry/sentry.module';
 import { WebhooksModule } from '@/webhooks/webhooks.module';
 import { HealthModule } from './health/health.module';
 import { PrismaModule } from './prisma/prisma.module';
 import { AlertModule } from './alert/alert.module';
-import { MetricsAuthMiddleware } from './common/middlewares/metrics-auth.middleware';
+import { Module } from '@nestjs/common';
 
 @Module({
   imports: [

services/workflows-service/src/app.worker.module.ts

@@ -2,20 +2,19 @@ import { Module } from '@nestjs/common';
 import { QueueService } from './queue.service';
 import { QueueBullboardService } from './queue-bullboard.service';
 import { QueueOtelService } from './otel.service';
-import { redisProvider } from './redis.provider';
 import { MonitoringModule } from '@/common/monitoring/monitoring.module';
 import { BULLBOARD_INSTANCE_INJECTION_TOKEN } from './types';
 import { createBullBoard } from '@bull-board/api';
 import { ExpressAdapter } from '@bull-board/express';
+import { RedisModule } from '../redis/redis.module';
 
 @Module({
-  imports: [MonitoringModule],
+  imports: [MonitoringModule, RedisModule],
   providers: [
     QueueService,
     { provide: 'IQueueService', useExisting: QueueService },
     QueueBullboardService,
     QueueOtelService,
-    redisProvider,
     {
       provide: BULLBOARD_INSTANCE_INJECTION_TOKEN,
       useFactory: () => {
@@ -32,7 +31,6 @@ import { ExpressAdapter } from '@bull-board/express';
     'IQueueService',
     QueueBullboardService,
     QueueOtelService,
-    redisProvider,
     BULLBOARD_INSTANCE_INJECTION_TOKEN,
   ],
 })

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

@@ -3,8 +3,7 @@ import { Queue, Worker } from 'bullmq';
 import IORedis from 'ioredis';
 import { AppLoggerService } from '@/common/app-logger/app-logger.service';
 import { env } from '@/env';
-import { QueueOtelService } from './otel.service';
-import { REDIS_CLIENT } from './redis.provider';
+import { RedisService } from '../redis/redis.service';
 import { BullMQPrometheusService } from '@/common/monitoring/bullmq-prometheus.service';
 import type { BullBoardInjectedInstance } from './types';
 import { QueueBullboardService } from './queue-bullboard.service';
@@ -33,15 +32,14 @@ export class QueueService implements OnModuleDestroy {
 
   constructor(
     private readonly logger: AppLoggerService,
-    private readonly queueOtelService: QueueOtelService,
-    @Inject(REDIS_CLIENT) redisClient: IORedis | null,
+    private readonly redisService: RedisService,
     private readonly bullMQPrometheusService: BullMQPrometheusService,
     @Inject('BULLBOARD_INSTANCE') private readonly bullBoard?: BullBoardInjectedInstance,
     private readonly queueBullboardService?: QueueBullboardService,
   ) {
     this.shouldProcessJobs = this.determineIfShouldProcessJobs();
     this.logger.log(`Queue worker mode: ${this.shouldProcessJobs ? 'ENABLED' : 'DISABLED'}`);
-    this.redisClient = redisClient;
+    this.redisClient = this.redisService.client;
   }
 
   private determineIfShouldProcessJobs(): boolean {
@@ -147,7 +145,6 @@ export class QueueService implements OnModuleDestroy {
 
   createQueue<T = any>(queueName: string, options?: QueueOptions<T>): void {
     if (this.queues.has(queueName)) {
-      // Optionally update options if needed
       return;
     }
 
@@ -182,12 +179,51 @@ export class QueueService implements OnModuleDestroy {
     );
 
     await Promise.all([...workerClosePromises, ...queueClosePromises]);
+  }
+
+  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 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,
+            },
+          },
+        },
+      );
+      this.logger.log(`Created job scheduler: ${schedulerId}`, {
+        schedulerId,
+        every: options.every,
+        jobId: firstJob?.id,
+      });
 
-    if (this.redisClient) {
-      await this.redisClient
-        .quit()
-        .catch(err => this.logger.error(`Error closing Redis connection`, { err }));
-      this.redisClient = null;
+      return firstJob;
+    } catch (error) {
+      this.logger.error(`Failed to set up job scheduler: ${schedulerId}`, { error });
+      throw error;
     }
   }
 }

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

@@ -1,9 +1,32 @@
 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>;
+
+  /**
+   * 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;
+}

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

@@ -7,10 +7,13 @@ import { Job } from 'bullmq';
 
 import { AppLoggerService } from '@/common/app-logger/app-logger.service';
 import { BULLBOARD_INSTANCE_INJECTION_TOKEN } from '@/common/queue/types';
-import type { BullBoardInjectedInstance } from '@/common/queue/types';
+import type { BullBoardInjectedInstance, IQueueService } from '@/common/queue/types';
 import { env } from '@/env';
-import { type OutgoingWebhookJobData, type OutgoingWebhookPayloads } from './types/webhook';
-import type { IQueueService } from '@/common/queue/queue.interface';
+import {
+  WebhookError,
+  type OutgoingWebhookJobData,
+  type OutgoingWebhookPayloads,
+} from './types/webhook';
 
 const captureWebhookFailureWithSentry = (errorPayload: Record<string, unknown>) => {
   Sentry.captureException(
@@ -88,6 +91,16 @@ export class WebhooksService implements OnModuleInit {
       return res.data;
     } catch (error) {
       this.handleWebhookJobError(job, error);
+
+      if (isAxiosError(error)) {
+        const webhookError = new WebhookError('Webhook request failed');
+        webhookError.cause = error;
+        webhookError.statusCode = error.response?.status;
+        webhookError.responseData = error.response?.data;
+        webhookError.headers = error.response?.headers;
+        throw webhookError;
+      }
+
       throw error;
     }
   }