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,26 @@
+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
+    driver_opts:
+      type: tmpfs
+      device: tmpfs
+      o: size=100m
+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/app.module.ts

@@ -11,7 +11,7 @@ import { ServeStaticOptionsService } from './serve-static-options.service';
 import { EndUserModule } from './end-user/end-user.module';
 import { BusinessModule } from './business/business.module';
 import { StorageModule } from './storage/storage.module';
-import { MulterModule } from '@nestjs/platform-express';
+import { ExpressAdapter, MulterModule } from '@nestjs/platform-express';
 import { EventEmitterModule } from '@nestjs/event-emitter';
 import { FilterModule } from '@/filter/filter.module';
 import { configs, env, serverEnvSchema } from '@/env';
@@ -48,6 +48,8 @@ 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 { BullModule } from '@nestjs/bullmq';
+import { BullMqModule } from "@/bull-mq/bull-mq.module";
 
 export const validate = async (config: Record<string, unknown>) => {
   const zodEnvSchema = z
@@ -126,6 +128,7 @@ export const validate = async (config: Record<string, unknown>) => {
     RuleEngineModule,
     NotionModule,
     SecretsManagerModule,
+    BullMqModule
   ],
   providers: [
     {

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

@@ -0,0 +1,48 @@
+import { Module } from '@nestjs/common';
+import { ConfigModule } from '@nestjs/config';
+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 { env } from '@/env';
+import { WebhookService } from '@/bull-mq/webhook/webhook.service';
+import { AppLoggerService } from '@/common/app-logger/app-logger.service';
+import { WebhookProcessor } from '@/bull-mq/webhook/webhook.processor';
+import { AppLoggerModule } from '@/common/app-logger/app-logger.module';
+
+const QUEUE_NAMES = [
+  {
+    name: 'webhook-queue',
+    config: {},
+  },
+];
+
+@Module({
+  imports: [
+    AppLoggerModule,
+    ConfigModule.forRoot({ isGlobal: true }),
+    BullModule.forRootAsync({
+      useFactory: () => ({
+        connection: {
+          password: env.REDIS_PASSWORD,
+          host: env.REDIS_HOST,
+          port: env.REDIS_PORT,
+        },
+      }),
+    }),
+    BullModule.registerQueue(...QUEUE_NAMES.map(({ name }) => ({ name }))),
+    BullBoardModule.forRoot({
+      route: '/queues',
+      adapter: ExpressAdapter,
+    }),
+    ...QUEUE_NAMES.map(({ name }) =>
+      BullBoardModule.forFeature({
+        name,
+        adapter: BullAdapter,
+      }),
+    ),
+  ],
+  providers: [WebhookProcessor, WebhookService, AppLoggerService],
+  exports: [BullModule, WebhookService],
+})
+export class BullMqModule {}

services/workflows-service/src/bull-mq/webhook/webhook.processor.ts

@@ -0,0 +1,67 @@
+import { InjectQueue, Processor, OnQueueEvent } from '@nestjs/bullmq';
+import { Job, Queue } from 'bullmq';
+import axios, { AxiosRequestConfig, Method } from 'axios';
+import { AppLoggerService } from '@/common/app-logger/app-logger.service';
+
+export interface WebhookJobData {
+  url: string;
+  method: Method;
+  headers?: Record<string, string>;
+  body?: Record<string, unknown>;
+  timeout?: number;
+}
+
+@Processor('webhook-queue')
+export class WebhookProcessor {
+  constructor(
+    protected readonly logger: AppLoggerService,
+    @InjectQueue('webhook-queue') private webhookQueue: Queue,
+  ) {}
+
+  async process(job: Job<WebhookJobData>) {
+    this.logger.log(`Processing webhook job ${job.id}`);
+
+    const { url, method, headers, body, timeout } = job.data;
+
+    const config: AxiosRequestConfig = {
+      url,
+      method,
+      headers,
+      data: body,
+      timeout: timeout || 5000, // Default timeout of 5 seconds
+    };
+
+    try {
+      const response = await axios(config);
+
+      return {
+        status: response.status,
+        statusText: response.statusText,
+        headers: response.headers,
+        data: response.data,
+      };
+    } catch (error) {
+      if (axios.isAxiosError(error)) {
+        throw new Error(`Webhook request failed: ${error.message}`);
+      }
+
+      throw error;
+    }
+  }
+
+  @OnQueueEvent('completed')
+  onCompleted(job: Job) {
+    this.logger.log(`Webhook job ${job.id} completed successfully`);
+  }
+
+  @OnQueueEvent('failed')
+  onFailed(job: Job, error: Error) {
+    this.logger.error(`Webhook job ${job.id} failed: ${error.message}`);
+  }
+
+  async onModuleInit() {
+    this.webhookQueue.on('cleaned', (jobs, type) => {
+      this.logger.log(`${jobs.length} ${type} jobs have been cleaned from the webhook queue`);
+    });
+  }
+}

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

@@ -0,0 +1,13 @@
+import { Injectable } from '@nestjs/common';
+import { InjectQueue } from '@nestjs/bullmq';
+import { Queue } from 'bullmq';
+import { WebhookJobData } from './webhook.processor'; // You might want to move this interface to a separate file
+
+@Injectable()
+export class WebhookService {
+  constructor(@InjectQueue('webhook-queue') private webhookQueue: Queue) {}
+
+  async addWebhookJob(data: WebhookJobData) {
+    await this.webhookQueue.add('webhook', data);
+  }
+}

services/workflows-service/src/env.ts

@@ -90,6 +90,9 @@ export const serverEnvSchema = {
   IN_MEMORIES_SECRET_ACQUIRER_ID: z.string().optional(),
   IN_MEMORIES_SECRET_PRIVATE_KEY: z.string().optional(),
   IN_MEMORIES_SECRET_CONSUMER_KEY: z.string().optional(),
+  REDIS_HOST: z.string(),
+  REDIS_PASSWORD: z.string(),
+  REDIS_PORT: z.string().transform(value => Number(value)),
 };
 
 if (!process.env['ENVIRONMENT_NAME'] || process.env['ENVIRONMENT_NAME'] === 'local') {