Integrate BullBoard and optimize queue registration

api/.env.example

@@ -9,6 +9,8 @@ S3_BUCKET="0lfyi-v7"
 S3_STORAGE_CLASS="STANDARD"
 S3_USE_SSL=false
 
+NODE_ENV=local
+
 CLICKHOUSE_HOST="127.0.0.1"
 CLICKHOUSE_USERNAME="default"
 CLICKHOUSE_PASSWORD=""

api/package.json

@@ -27,6 +27,9 @@
     "@apollo/server": "^4.10.4",
     "@aptos-labs/ts-sdk": "^1.16.0",
     "@aws-sdk/client-s3": "^3.583.0",
+    "@bull-board/api": "^5.19.2",
+    "@bull-board/express": "^5.19.2",
+    "@bull-board/nestjs": "^5.19.2",
     "@clickhouse/client": "^1.0.2",
     "@nestjs/apollo": "^12.1.0",
     "@nestjs/bullmq": "^10.1.1",

api/src/bullboard/bullboard.service.ts

@@ -0,0 +1,33 @@
+import { Injectable } from '@nestjs/common';
+import { Queue } from 'bullmq';
+import { createBullBoard } from '@bull-board/api';
+import { BullMQAdapter } from '@bull-board/api/bullMQAdapter.js';
+import { ExpressAdapter } from '@bull-board/express';
+
+@Injectable()
+export class BullBoardService {
+
+  private readonly serverAdapter: ExpressAdapter;
+
+  constructor() {
+    this.serverAdapter = new ExpressAdapter();
+  }
+
+  setupBullBoard(queueConfigs: { name: string; connection: any }[]) {
+    const queues: Queue<any, any, string>[] = queueConfigs.map(
+      (config) => new Queue(config.name, { connection: config.connection })
+    );
+
+    createBullBoard({
+      queues: queues.map((queue) => new BullMQAdapter(queue)),
+      serverAdapter: this.serverAdapter,
+    });
+
+    this.serverAdapter.setBasePath('/ui');
+  }
+
+  getServerAdapter() {
+    return this.serverAdapter;
+  }
+
+}

api/src/main.ts

@@ -3,6 +3,7 @@ import { NestFactory } from "@nestjs/core";
 import { NestExpressApplication } from "@nestjs/platform-express";
 import { AppModule } from "./app/app.module.js";
 import getConfig from "./config/config.js";
+import { BullBoardService } from './bullboard/bullboard.service.js';
 
 async function bootstrap() {
   const config = getConfig();
@@ -15,6 +16,13 @@ async function bootstrap() {
   app.set("trust proxy", 1);
   app.enableCors();
 
+  const isProduction = process.env.NODE_ENV === 'production';
+
+  if (!isProduction) {
+    const bullBoardService = app.get(BullBoardService);
+    app.use('/ui', bullBoardService.getServerAdapter().getRouter());
+  }
+
   await app.listen(config.port);
 }
 bootstrap();

api/src/ol/ol.module.ts

@@ -45,6 +45,7 @@ import { TransactionsService } from "./transactions/TransactionsService.js";
 import { Transaction } from "./transactions/Transaction.js";
 import { OnChainTransactionsRepository } from "./transactions/OnChainTransactionsRepository.js";
 import { ExpiredTransactionsProcessor } from "./transactions/ExpiredTransactionsProcessor.js";
+import { BullBoardService } from '../bullboard/bullboard.service.js';
 
 const roles = process.env.ROLES!.split(",");
 
@@ -65,6 +66,16 @@ for (const role of roles) {
   }
 }
 
+// Centralize queue definitions in an array for better reusability
+const queues = [
+  { name: "ol-clickhouse-ingestor", connection: redisClient},
+  { name: "ol-parquet-producer", connection: redisClient },
+  { name: "ol-version-batch", connection: redisClient },
+  { name: "ol-version", connection: redisClient },
+  { name: "expired-transactions", connection: redisClient },
+];
+
+
 @Module({
   imports: [
     S3Module,
@@ -74,30 +85,13 @@ for (const role of roles) {
     OlDbModule,
     WalletSubscriptionModule,
 
-    BullModule.registerQueue({
-      name: "ol-clickhouse-ingestor",
-      connection: redisClient,
-    }),
-
-    BullModule.registerQueue({
-      name: "ol-parquet-producer",
-      connection: redisClient,
-    }),
-
-    BullModule.registerQueue({
-      name: "ol-version-batch",
-      connection: redisClient,
-    }),
-
-    BullModule.registerQueue({
-      name: "ol-version",
-      connection: redisClient,
-    }),
-
-    BullModule.registerQueue({
-      name: "expired-transactions",
-      connection: redisClient,
-    }),
+    // Register queues using a loop to simplify and maintain consistency
+    BullModule.registerQueue(
+      ...queues.map(queue => ({
+        name: queue.name,
+        connection: queue.connection,
+      }))
+    ),
   ],
   providers: [
     UserTransactionsResolver,
@@ -119,6 +113,8 @@ for (const role of roles) {
     MovementsService,
     TransformerService,
 
+    BullBoardService,
+
     // Transactions
     TransactionResolver,
     TransactionsResolver,
@@ -147,6 +143,16 @@ for (const role of roles) {
     ...workers,
   ],
   controllers: [OlController],
-  exports: [OlService, TransformerService, Types.ICommunityWalletsService],
+  exports: [OlService, TransformerService, Types.ICommunityWalletsService, BullBoardService],
 })
-export class OlModule {}
+export class OlModule {
+  isProduction = process.env.NODE_ENV === 'production';
+
+  constructor(private readonly bullBoardService: BullBoardService) {
+    if (!this.isProduction) {
+      // Setup BullBoard for the queues in local development environment
+      // Utilize the same queue array to ensure consistency
+      this.bullBoardService.setupBullBoard(queues);
+    }
+  }
+}