add message queue integration (#2491)

server/src/integrations/environment/environment.service.ts

@@ -6,6 +6,7 @@ import { AwsRegion } from './interfaces/aws-region.interface';
 import { StorageType } from './interfaces/storage.interface';
 import { SupportDriver } from './interfaces/support.interface';
 import { LoggerDriver } from './interfaces/logger.interface';
+import { MessageQueueType } from './interfaces/message-queue.interface';
 
 @Injectable()
 export class EnvironmentService {
@@ -37,6 +38,14 @@ export class EnvironmentService {
     return this.configService.get<string>('PG_DATABASE_URL')!;
   }
 
+  getRedisHost(): string {
+    return this.configService.get<string>('REDIS_HOST') ?? '127.0.0.1';
+  }
+
+  getRedisPort(): number {
+    return +(this.configService.get<string>('REDIS_PORT') ?? 6379);
+  }
+
   getFrontBaseUrl(): string {
     return this.configService.get<string>('FRONT_BASE_URL')!;
   }
@@ -102,6 +111,13 @@ export class EnvironmentService {
     );
   }
 
+  getMessageQueueType(): MessageQueueType {
+    return (
+      this.configService.get<MessageQueueType>('MESSAGE_QUEUE_TYPE') ??
+      MessageQueueType.PgBoss
+    );
+  }
+
   getStorageS3Region(): AwsRegion | undefined {
     return this.configService.get<AwsRegion>('STORAGE_S3_REGION');
   }

server/src/integrations/environment/interfaces/message-queue.interface.ts

@@ -0,0 +1,4 @@
+export enum MessageQueueType {
+  PgBoss = 'pg-boss',
+  BullMQ = 'bull-mq',
+}

server/src/integrations/integrations.module.ts

@@ -10,6 +10,9 @@ import { StorageType } from './environment/interfaces/storage.interface';
 import { LoggerModule } from './logger/logger.module';
 import { LoggerModuleOptions } from './logger/interfaces';
 import { LoggerDriver } from './environment/interfaces/logger.interface';
+import { MessageQueueModule } from './message-queue/message-queue.module';
+import { MessageQueueModuleOptions } from './message-queue/interfaces';
+import { MessageQueueType } from './environment/interfaces/message-queue.interface';
 
 /**
  * FileStorage Module factory
@@ -84,6 +87,46 @@ const loggerModuleFactory = async (
   }
 };
 
+/**
+ * MessageQueue Module factory
+ * @param environment
+ * @returns MessageQueueModuleOptions
+ */
+const messageQueueModuleFactory = async (
+  environmentService: EnvironmentService,
+): Promise<MessageQueueModuleOptions> => {
+  const type = environmentService.getMessageQueueType();
+
+  switch (type) {
+    case MessageQueueType.PgBoss: {
+      const connectionString = environmentService.getPGDatabaseUrl();
+      return {
+        type: MessageQueueType.PgBoss,
+        options: {
+          connectionString,
+        },
+      };
+    }
+    case MessageQueueType.BullMQ: {
+      const host = environmentService.getRedisHost();
+      const port = environmentService.getRedisPort();
+      return {
+        type: MessageQueueType.BullMQ,
+        options: {
+          connection: {
+            host,
+            port,
+          },
+        },
+      };
+    }
+    default:
+      throw new Error(
+        `Invalid message queue type (${type}), check your .env file`,
+      );
+  }
+};
+
 @Module({
   imports: [
     EnvironmentModule.forRoot({}),
@@ -95,6 +138,10 @@ const loggerModuleFactory = async (
       useFactory: loggerModuleFactory,
       inject: [EnvironmentService],
     }),
+    MessageQueueModule.forRoot({
+      useFactory: messageQueueModuleFactory,
+      inject: [EnvironmentService],
+    }),
   ],
   exports: [],
   providers: [],

server/src/integrations/message-queue/drivers/bullmq.driver.ts

@@ -0,0 +1,60 @@
+import { Queue, QueueOptions, Worker } from 'bullmq';
+
+import { QueueJobOptions } from 'src/integrations/message-queue/drivers/interfaces/job-options.interface';
+
+import { MessageQueues } from 'src/integrations/message-queue/message-queue.constants';
+
+import { MessageQueueDriver } from './interfaces/message-queue-driver.interface';
+
+export type BullMQDriverOptions = QueueOptions;
+
+export class BullMQDriver implements MessageQueueDriver {
+  private queueMap: Record<MessageQueues, Queue> = {} as Record<
+    MessageQueues,
+    Queue
+  >;
+  private workerMap: Record<MessageQueues, Worker> = {} as Record<
+    MessageQueues,
+    Worker
+  >;
+
+  constructor(private options: BullMQDriverOptions) {}
+
+  register(queueName: MessageQueues): void {
+    this.queueMap[queueName] = new Queue(queueName, this.options);
+  }
+
+  async stop() {
+    const workers = Object.values(this.workerMap);
+    const queues = Object.values(this.queueMap);
+    await Promise.all([
+      ...queues.map((q) => q.close()),
+      ...workers.map((w) => w.close()),
+    ]);
+  }
+
+  async work<T>(
+    queueName: MessageQueues,
+    handler: ({ data, id }: { data: T; id: string }) => Promise<void>,
+  ) {
+    const worker = new Worker(queueName, async (job) => {
+      await handler(job as { data: T; id: string });
+    });
+    this.workerMap[queueName] = worker;
+  }
+
+  async add<T>(
+    queueName: MessageQueues,
+    data: T,
+    options?: QueueJobOptions,
+  ): Promise<void> {
+    if (!this.queueMap[queueName]) {
+      throw new Error(
+        `Queue ${queueName} is not registered, make sure you have added it as a queue provider`,
+      );
+    }
+    await this.queueMap[queueName].add(options?.id || '', data, {
+      priority: options?.priority,
+    });
+  }
+}

server/src/integrations/message-queue/drivers/interfaces/job-options.interface.ts

@@ -0,0 +1,4 @@
+export interface QueueJobOptions {
+  id?: string;
+  priority?: number;
+}

server/src/integrations/message-queue/drivers/interfaces/message-queue-driver.interface.ts

@@ -0,0 +1,17 @@
+import { QueueJobOptions } from 'src/integrations/message-queue/drivers/interfaces/job-options.interface';
+
+import { MessageQueues } from 'src/integrations/message-queue/message-queue.constants';
+
+export interface MessageQueueDriver {
+  add<T>(
+    queueName: MessageQueues,
+    data: T,
+    options?: QueueJobOptions,
+  ): Promise<void>;
+  work<T>(
+    queueName: string,
+    handler: ({ data, id }: { data: T; id: string }) => Promise<void> | void,
+  );
+  stop?(): Promise<void>;
+  register?(queueName: MessageQueues): void;
+}

server/src/integrations/message-queue/drivers/pg-boss.driver.ts

@@ -0,0 +1,37 @@
+import PgBoss from 'pg-boss';
+
+import { QueueJobOptions } from 'src/integrations/message-queue/drivers/interfaces/job-options.interface';
+
+import { MessageQueueDriver } from './interfaces/message-queue-driver.interface';
+
+export type PgBossDriverOptions = PgBoss.ConstructorOptions;
+
+export class PgBossDriver implements MessageQueueDriver {
+  private pgBoss: PgBoss;
+
+  constructor(options: PgBossDriverOptions) {
+    this.pgBoss = new PgBoss(options);
+  }
+  async stop() {
+    await this.pgBoss.stop();
+  }
+
+  async init(): Promise<void> {
+    await this.pgBoss.start();
+  }
+
+  async work<T>(
+    queueName: string,
+    handler: ({ data, id }: { data: T; id: string }) => Promise<void>,
+  ) {
+    return this.pgBoss.work(queueName, handler);
+  }
+
+  async add<T>(
+    queueName: string,
+    data: T,
+    options?: QueueJobOptions,
+  ): Promise<void> {
+    await this.pgBoss.send(queueName, data as object, options ? options : {});
+  }
+}

server/src/integrations/message-queue/interfaces/index.ts

@@ -0,0 +1 @@
+export * from './message-queue.interface';

server/src/integrations/message-queue/interfaces/message-queue.interface.ts

@@ -0,0 +1,27 @@
+import { FactoryProvider, ModuleMetadata } from '@nestjs/common';
+
+import { MessageQueueType } from 'src/integrations/environment/interfaces/message-queue.interface';
+
+import { BullMQDriverOptions } from 'src/integrations/message-queue/drivers/bullmq.driver';
+import { PgBossDriverOptions } from 'src/integrations/message-queue/drivers/pg-boss.driver';
+
+export interface PgBossDriverFactoryOptions {
+  type: MessageQueueType.PgBoss;
+  options: PgBossDriverOptions;
+}
+
+export interface BullMQDriverFactoryOptions {
+  type: MessageQueueType.BullMQ;
+  options: BullMQDriverOptions;
+}
+
+export type MessageQueueModuleOptions =
+  | PgBossDriverFactoryOptions
+  | BullMQDriverFactoryOptions;
+
+export type MessageQueueModuleAsyncOptions = {
+  useFactory: (
+    ...args: any[]
+  ) => MessageQueueModuleOptions | Promise<MessageQueueModuleOptions>;
+} & Pick<ModuleMetadata, 'imports'> &
+  Pick<FactoryProvider, 'inject'>;

server/src/integrations/message-queue/message-queue.constants.ts

@@ -0,0 +1,5 @@
+export const QUEUE_DRIVER = Symbol('QUEUE_DRIVER');
+
+export enum MessageQueues {
+  taskAssignedQueue = 'task-assigned-queue',
+}

server/src/integrations/message-queue/message-queue.module.ts

@@ -0,0 +1,51 @@
+import { DynamicModule, Global } from '@nestjs/common';
+
+import { MessageQueueDriver } from 'src/integrations/message-queue/drivers/interfaces/message-queue-driver.interface';
+import { MessageQueueType } from 'src/integrations/environment/interfaces/message-queue.interface';
+
+import { MessageQueueModuleAsyncOptions } from 'src/integrations/message-queue/interfaces';
+import {
+  QUEUE_DRIVER,
+  MessageQueues,
+} from 'src/integrations/message-queue/message-queue.constants';
+import { PgBossDriver } from 'src/integrations/message-queue/drivers/pg-boss.driver';
+import { MessageQueueService } from 'src/integrations/message-queue/services/message-queue.service';
+import { BullMQDriver } from 'src/integrations/message-queue/drivers/bullmq.driver';
+
+@Global()
+export class MessageQueueModule {
+  static forRoot(options: MessageQueueModuleAsyncOptions): DynamicModule {
+    const providers = [
+      {
+        provide: MessageQueues.taskAssignedQueue,
+        useFactory: (driver: MessageQueueDriver) => {
+          return new MessageQueueService(
+            driver,
+            MessageQueues.taskAssignedQueue,
+          );
+        },
+        inject: [QUEUE_DRIVER],
+      },
+      {
+        provide: QUEUE_DRIVER,
+        useFactory: async (...args: any[]) => {
+          const config = await options.useFactory(...args);
+          if (config.type === MessageQueueType.PgBoss) {
+            const boss = new PgBossDriver(config.options);
+            await boss.init();
+            return boss;
+          }
+          return new BullMQDriver(config.options);
+        },
+        inject: options.inject || [],
+      },
+    ];
+
+    return {
+      module: MessageQueueModule,
+      imports: options.imports || [],
+      providers,
+      exports: [MessageQueues.taskAssignedQueue],
+    };
+  }
+}

server/src/integrations/message-queue/services/message-queue-task-assigned.service.spec.ts

@@ -0,0 +1,46 @@
+import { Test, TestingModule } from '@nestjs/testing';
+
+import { MessageQueueDriver } from 'src/integrations/message-queue/drivers/interfaces/message-queue-driver.interface';
+
+import {
+  QUEUE_DRIVER,
+  MessageQueues,
+} from 'src/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/integrations/message-queue/services/message-queue.service';
+
+describe('MessageQueueTaskAssigned queue', () => {
+  let service: MessageQueueService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        {
+          provide: MessageQueues.taskAssignedQueue,
+          useFactory: (driver: MessageQueueDriver) => {
+            return new MessageQueueService(
+              driver,
+              MessageQueues.taskAssignedQueue,
+            );
+          },
+          inject: [QUEUE_DRIVER],
+        },
+        {
+          provide: QUEUE_DRIVER,
+          useValue: {},
+        },
+      ],
+    }).compile();
+
+    service = module.get<MessageQueueService>(MessageQueues.taskAssignedQueue);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+  it('should contain the topic and driver', () => {
+    expect(service).toEqual({
+      driver: {},
+      queueName: MessageQueues.taskAssignedQueue,
+    });
+  });
+});

server/src/integrations/message-queue/services/message-queue.service.ts

@@ -0,0 +1,37 @@
+import { Inject, Injectable, OnModuleDestroy } from '@nestjs/common';
+
+import { QueueJobOptions } from 'src/integrations/message-queue/drivers/interfaces/job-options.interface';
+import { MessageQueueDriver } from 'src/integrations/message-queue/drivers/interfaces/message-queue-driver.interface';
+
+import {
+  MessageQueues,
+  QUEUE_DRIVER,
+} from 'src/integrations/message-queue/message-queue.constants';
+
+@Injectable()
+export class MessageQueueService implements OnModuleDestroy {
+  constructor(
+    @Inject(QUEUE_DRIVER) protected driver: MessageQueueDriver,
+    protected queueName: MessageQueues,
+  ) {
+    if (typeof this.driver.register === 'function') {
+      this.driver.register(queueName);
+    }
+  }
+
+  async onModuleDestroy() {
+    if (typeof this.driver.stop === 'function') {
+      await this.driver.stop();
+    }
+  }
+
+  add<T>(data: T, options?: QueueJobOptions): Promise<void> {
+    return this.driver.add(this.queueName, data, options);
+  }
+
+  work<T>(
+    handler: ({ data, id }: { data: T; id: string }) => Promise<void> | void,
+  ) {
+    return this.driver.work(this.queueName, handler);
+  }
+}