feature/bull_mq_integration

services/workflows-service/.env.example

@@ -35,3 +35,6 @@ WEB_UI_SDK_URL=http://localhost:5202
 #HASHING_KEY_SECRET="$2b$10$FovZTB91/QQ4Yu28nvL8e."
 HASHING_KEY_SECRET_BASE64=JDJiJDEwJDNFeWtwWEs4QkdiczlRaWFwLkM4Vk8=
 NOTION_API_KEY=secret
+REDIS_HOST=localhost
+REDIS_PORT=7381
+REDIS_PASSWORD=password

services/workflows-service/docker-compose.redis.yml

@@ -0,0 +1,22 @@
+version: '3.8'
+services:
+  redis:
+    image: redis:alpine
+    ports:
+      - '${REDIS_PORT}:6379'
+    volumes:
+      - redis-data:/data
+    command: >
+      --requirepass ${REDIS_PASSWORD}
+      --appendonly yes
+    environment:
+      - REDIS_PASSWORD=${REDIS_PASSWORD}
+      - REDIS_PORT=${REDIS_PORT}
+    networks:
+      - app-network
+volumes:
+  redis-data:
+    driver: local
+networks:
+  app-network:
+    driver: bridge

services/workflows-service/package.json

@@ -5,7 +5,7 @@
   "description": "workflow-service",
   "scripts": {
     "spellcheck": "cspell \"*\"",
-    "setup": "npm run docker:db:down && npm run docker:db && wait-on tcp:5432 && npm run db:reset && npm run seed",
+    "setup": "npm run docker:db:down && npm run docker:db && wait-on tcp:5432 && npm run docker:redis:down && npm run docker:redis && npm run db:reset && npm run seed",
     "format": "prettier --write . '!**/*.{md,hbs}'",
     "format:check": "prettier --check . '!**/*.{md,hbs}'",
     "lint": "eslint . --fix",
@@ -30,6 +30,8 @@
     "db:reset:dev:with-data": "npm run db:reset:dev && npm run db:data-migration:migrate && npm run db:data-sync",
     "db:init": "npm run db:migrate-dev -- --name 'initial version' && npm run db:migrate-up seed",
     "prisma:generate": "prisma generate",
+    "docker:redis": "docker compose -f docker-compose.redis.yml up -d --wait",
+    "docker:redis:down": "docker compose -f docker-compose.redis.yml down --volumes",
     "docker:db": "docker compose -f docker-compose.db.yml up -d --wait",
     "docker:db:down": "docker compose -f docker-compose.db.yml down --volumes",
     "docker:build": "docker build .",
@@ -50,8 +52,12 @@
     "@ballerine/common": "0.9.33",
     "@ballerine/workflow-core": "0.6.45",
     "@ballerine/workflow-node-sdk": "0.6.45",
+    "@bull-board/api": "^6.0.0",
+    "@bull-board/express": "^6.0.0",
+    "@bull-board/nestjs": "^6.0.0",
     "@faker-js/faker": "^7.6.0",
     "@nestjs/axios": "^2.0.0",
+    "@nestjs/bullmq": "^10.2.1",
     "@nestjs/common": "^9.3.12",
     "@nestjs/config": "2.3.1",
     "@nestjs/core": "^9.3.12",
@@ -78,6 +84,7 @@
     "ballerine-nestjs-typebox": "3.0.2-next.11",
     "base64-stream": "^1.0.0",
     "bcrypt": "5.1.0",
+    "bullmq": "^5.13.2",
     "class-transformer": "0.5.1",
     "class-validator": "0.14.0",
     "concat-stream": "^2.0.0",

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

@@ -24,6 +24,8 @@ import { ProjectModule } from '@/project/project.module';
 import { UserRepository } from '@/user/user.repository';
 import { AlertDefinitionModule } from '@/alert-definition/alert-definition.module';
 import { SentryModule } from '@/sentry/sentry.module';
+import { BullMqModule } from '@/bull-mq/bull-mq.module';
+import { OutgoingWebhooksModule } from '@/webhooks/outgoing-webhooks/outgoing-webhooks.module';
 
 @Module({
   imports: [
@@ -32,6 +34,8 @@ import { SentryModule } from '@/sentry/sentry.module';
     PrismaModule,
     SentryModule,
     ProjectModule,
+    BullMqModule,
+    OutgoingWebhooksModule,
     HttpModule.register({
       timeout: 5000,
       maxRedirects: 10,
@@ -48,15 +52,14 @@ import { SentryModule } from '@/sentry/sentry.module';
   ],
   controllers: [AlertControllerInternal, AlertControllerExternal],
   providers: [
-    {
-      provide: WebhookHttpService,
-      useExisting: HttpService,
-    },
+    WebhookHttpService,
     AlertService,
     AlertRepository,
     AlertDefinitionRepository,
     WebhookManagerService,
     WebhookEventEmitterService,
+    BullMqModule,
+    OutgoingWebhooksModule,
     // TODO: Export to user module
     UserService,
     UserRepository,

services/workflows-service/src/alert/webhook-manager/webhook-manager.service.ts

@@ -1,27 +1,27 @@
 import { alertWebhookFailure } from '@/events/alert-webhook-failure';
-import { lastValueFrom } from 'rxjs';
 import * as common from '@nestjs/common';
-import { ConfigService } from '@nestjs/config';
 import { ClsService } from 'nestjs-cls';
 import { SentryInterceptor } from '@/sentry/sentry.interceptor';
 import { AppLoggerService } from '@/common/app-logger/app-logger.service';
 import { WebhookEventEmitterService } from './webhook-event-emitter.service';
 import { IWebhookEntityEventData } from './types';
 import { Webhook } from '@/events/get-webhooks';
-import { HttpService } from '@nestjs/axios';
-import { sign } from '@ballerine/common';
+import { AnyRecord } from '@ballerine/common';
+import { env } from '@/env';
+import { OutgoingWebhookQueueService } from '@/bull-mq/outgoing-webhook/outgoing-webhook-queue.service';
+import { OutgoingWebhooksService } from '@/webhooks/outgoing-webhooks/outgoing-webhooks.service';
 
 @common.Injectable()
-export abstract class WebhookHttpService extends HttpService {}
+export class WebhookHttpService {}
 
 @common.Injectable()
 @common.UseInterceptors(SentryInterceptor)
 export class WebhookManagerService {
   constructor(
     private readonly cls: ClsService,
     protected readonly logger: AppLoggerService,
-    protected readonly configService: ConfigService,
-    protected readonly httpService: WebhookHttpService,
+    protected readonly outgoingQueueWebhookService: OutgoingWebhookQueueService,
+    protected readonly outgoingWebhookService: OutgoingWebhooksService,
     protected readonly webhookEventEmitter: WebhookEventEmitterService,
   ) {
     webhookEventEmitter.on('*', async (eventData: any) => {
@@ -45,21 +45,42 @@ export class WebhookManagerService {
     webhook: Webhook;
     webhookSharedSecret: string;
   }) {
-    try {
-      const { id, url, environment, apiVersion } = webhook;
+    const { id, url, environment, apiVersion } = webhook;
+
+    if (env.QUEUE_SYSTEM_ENABLED) {
+      return await this.outgoingQueueWebhookService.addJob({
+        requestConfig: {
+          url,
+          method: 'POST',
+          headers: {},
+          body: data as unknown as AnyRecord,
+          timeout: 15_000,
+        },
+        customerConfig: {
+          webhookSharedSecret,
+        },
+      });
+    }
 
+    try {
       this.logger.log('Sending webhook', { id, url });
 
-      const res = await lastValueFrom(
-        this.httpService.post(url, data, {
-          headers: {
-            'X-HMAC-Signature': sign({
-              payload: data,
-              key: webhookSharedSecret,
-            }),
-          },
-        }),
-      );
+      const response = await this.outgoingWebhookService.invokeWebhook({
+        requestConfig: {
+          url,
+          method: 'POST',
+          headers: {},
+          body: data as unknown as AnyRecord,
+          timeout: 15_000,
+        },
+        customerConfig: {
+          webhookSharedSecret,
+        },
+      });
+
+      if (response.status < 200 || response.status >= 300) {
+        throw new Error(`Webhook failed with status ${response.status} for ${url}`);
+      }
     } catch (error: Error | any) {
       this.logger.error('Webhook error data', {
         data,

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

@@ -39,7 +39,6 @@ import { WorkflowModule } from '@/workflow/workflow.module';
 import { TransactionModule } from '@/transaction/transaction.module';
 import { AlertModule } from '@/alert/alert.module';
 import { SwaggerController } from './swagger/swagger.controller';
-import { WebhooksModule } from '@/webhooks/webhooks.module';
 import { BusinessReportModule } from '@/business-report/business-report.module';
 import { ScheduleModule } from '@nestjs/schedule';
 import { CronModule } from '@/workflow/cron/cron.module';
@@ -48,6 +47,9 @@ import { hashKey } from './customer/api-key/utils';
 import { RuleEngineModule } from './rule-engine/rule-engine.module';
 import { NotionModule } from '@/notion/notion.module';
 import { SecretsManagerModule } from '@/secrets-manager/secrets-manager.module';
+import { BullMqModule } from '@/bull-mq/bull-mq.module';
+import { IncomingWebhooksModule } from '@/webhooks/incoming/incoming-webhooks.module';
+import { OutgoingWebhooksModule } from '@/webhooks/outgoing-webhooks/outgoing-webhooks.module';
 
 export const validate = async (config: Record<string, unknown>) => {
   const zodEnvSchema = z
@@ -85,7 +87,8 @@ export const validate = async (config: Record<string, unknown>) => {
     EventEmitterModule.forRoot(),
     UserModule,
     WorkflowModule,
-    WebhooksModule,
+    IncomingWebhooksModule,
+    OutgoingWebhooksModule,
     UiDefinitionModule,
     StorageModule,
     DataMigrationModule,
@@ -126,6 +129,7 @@ export const validate = async (config: Record<string, unknown>) => {
     RuleEngineModule,
     NotionModule,
     SecretsManagerModule,
+    BullMqModule,
   ],
   providers: [
     {

services/workflows-service/src/bull-mq/base/base-queue-worker.service.ts

@@ -0,0 +1,43 @@
+import { ConnectionOptions, Job, Queue, Worker } from 'bullmq';
+import { Injectable, OnModuleDestroy } from '@nestjs/common';
+import { REDIS_CONFIG } from '@/redis/const/redis-config';
+import { env } from '@/env';
+
+@Injectable()
+export abstract class BaseQueueWorkerService<T = any> implements OnModuleDestroy {
+  protected queue?: Queue;
+  protected worker?: Worker;
+  protected connectionOptions: ConnectionOptions;
+
+  protected constructor(private queueName: string) {
+    this.connectionOptions = {
+      ...REDIS_CONFIG,
+    };
+
+    if (env.QUEUE_SYSTEM_ENABLED !== true) {
+      return;
+    }
+
+    this.queue = new Queue(queueName, { connection: this.connectionOptions });
+  }
+
+  abstract handleJob(job: Job<T>): Promise<void>;
+
+  async addJob(jobData: T, jobOptions = {}): Promise<void> {
+    await this.queue?.add(this.queueName, jobData, jobOptions);
+  }
+
+  protected initializeWorker() {
+    this.worker = new Worker(
+      this.queueName,
+      async (job: Job<T>) => {
+        await this.handleJob(job);
+      },
+      { connection: this.connectionOptions },
+    );
+  }
+
+  async onModuleDestroy() {
+    await Promise.all([this.worker?.close(), this.queue?.close()]);
+  }
+}

services/workflows-service/src/bull-mq/bull-mq.module.ts

@@ -0,0 +1,45 @@
+import { Module } from '@nestjs/common';
+import { BullModule } from '@nestjs/bullmq';
+import { BullBoardModule } from '@bull-board/nestjs';
+import { ExpressAdapter } from '@bull-board/express';
+import { BullAdapter } from '@bull-board/api/bullAdapter';
+import { AppLoggerModule } from '@/common/app-logger/app-logger.module';
+import { QUEUES } from '@/bull-mq/consts';
+import { OutgoingWebhookQueueService } from '@/bull-mq/outgoing-webhook/outgoing-webhook-queue.service';
+import { REDIS_CONFIG } from '@/redis/const/redis-config';
+import { OutgoingWebhooksModule } from '@/webhooks/outgoing-webhooks/outgoing-webhooks.module';
+
+@Module({
+  imports: [
+    AppLoggerModule,
+    OutgoingWebhooksModule,
+    BullModule.forRootAsync({
+      useFactory: () => {
+        return {
+          connection: {
+            ...REDIS_CONFIG,
+          },
+        };
+      },
+    }),
+    BullModule.registerQueue(
+      ...Object.values(QUEUES).map(queue => ({
+        name: queue.name,
+        ...queue.config,
+      })),
+    ),
+    BullBoardModule.forRoot({
+      route: '/queues',
+      adapter: ExpressAdapter,
+    }),
+    ...Object.values(QUEUES).map(queue =>
+      BullBoardModule.forFeature({
+        name: queue.name,
+        adapter: BullAdapter,
+      }),
+    ),
+  ],
+  providers: [OutgoingWebhookQueueService],
+  exports: [BullModule, OutgoingWebhookQueueService],
+})
+export class BullMqModule {}

services/workflows-service/src/bull-mq/consts.ts

@@ -0,0 +1,24 @@
+import { BaseJobOptions } from 'bullmq/dist/esm/interfaces';
+
+export const QUEUES = {
+  INCOMING_WEBHOOKS_QUEUE: {
+    name: 'incoming-webhook-queue',
+    config: {
+      attempts: 10,
+      backoff: {
+        type: 'exponential',
+        delay: 1000,
+      },
+    },
+  },
+  OUTGOING_WEBHOOKS_QUEUE: {
+    name: 'outgoing-webhook-queue',
+    config: {
+      attempts: 10,
+      backoff: {
+        type: 'exponential',
+        delay: 1000,
+      },
+    },
+  },
+} satisfies Record<string, { name: string; config: BaseJobOptions }>;

services/workflows-service/src/bull-mq/incoming-webhook/incoming-webhook-queue.service.ts

@@ -0,0 +1,39 @@
+import { Injectable } from '@nestjs/common';
+import { BaseQueueWorkerService } from '@/bull-mq/base/base-queue-worker.service';
+import { IncomingWebhookData } from '@/bull-mq/incoming-webhook/types/types';
+import { AppLoggerService } from '@/common/app-logger/app-logger.service';
+import { QUEUES } from '@/bull-mq/consts';
+import { Job } from 'bullmq';
+
+@Injectable()
+export class IncomingWebhookQueueService extends BaseQueueWorkerService<IncomingWebhookData> {
+  constructor(protected readonly logger: AppLoggerService) {
+    super(QUEUES.INCOMING_WEBHOOKS_QUEUE.name);
+    this.initializeWorker();
+  }
+
+  async handleJob(job: Job<IncomingWebhookData>) {
+    this.logger.log(`Processing webhook job ${job.id}`);
+
+    const { service: workingService, payload } = job.data;
+    //   TODO - handle the invoking webhook job internally
+  }
+
+  protected initializeWorker() {
+    super.initializeWorker();
+
+    this.worker?.on('completed', (job: Job) => {
+      this.logger.log(`Outgoing Webhook job ${job.id} completed successfully`);
+    });
+
+    this.worker?.on('failed', (job, error, prev) => {
+      this.logger.error(`Outgoing Webhook job ${job?.id} failed after retries: ${error.message}`);
+    });
+
+    this.queue?.on('cleaned', (jobs, type) => {
+      this.logger.log(
+        `${jobs.length} ${type} Outgoing Webhook jobs have been cleaned from the webhook queue`,
+      );
+    });
+  }
+}

services/workflows-service/src/bull-mq/incoming-webhook/types/types.ts

@@ -0,0 +1,5 @@
+export interface IncomingWebhookData {
+  source: string;
+  payload: Record<string, unknown>;
+  service: (payload: Record<string, unknown>) => Promise<void>;
+}