wip

Author: arcoraven

package.json

@@ -31,13 +31,15 @@
   },
   "dependencies": {
     "@aws-sdk/client-kms": "^3.398.0",
+    "@bull-board/fastify": "^5.21.1",
     "@fastify/cookie": "^8.3.0",
     "@fastify/express": "^2.3.0",
     "@fastify/swagger": "^8.9.0",
     "@fastify/type-provider-typebox": "^3.2.0",
     "@fastify/websocket": "^8.2.0",
     "@google-cloud/kms": "^4.4.0",
     "@prisma/client": "^5.16.1",
+    "@queuedash/api": "^2.1.0",
     "@sinclair/typebox": "^0.31.28",
     "@t3-oss/env-core": "^0.6.0",
     "@thirdweb-dev/auth": "^4.1.87",

src/db/client.ts

@@ -22,7 +22,7 @@ export const knex = pg({
   acquireConnectionTimeout: 30000,
 } as Knex.Config);
 
-export const isDatabaseHealthy = async (): Promise<boolean> => {
+export const isDatabaseReachable = async () => {
   try {
     await prisma.walletDetails.findFirst();
     return true;

src/server/index.ts

@@ -8,9 +8,9 @@ import { env } from "../utils/env";
 import { logger } from "../utils/logger";
 import { withServerUsageReporting } from "../utils/usage";
 import { updateTxListener } from "./listeners/updateTxListener";
-import { withAuth } from "./middleware/auth";
+import { withAdminRoutes } from "./middleware/adminRoutes";
 import { withCors } from "./middleware/cors";
-import { withEnforceEngineMode } from "./middleware/engine-mode";
+import { withEnforceEngineMode } from "./middleware/engineMode";
 import { withErrorHandler } from "./middleware/error";
 import { withExpress } from "./middleware/express";
 import { withRequestLogs } from "./middleware/logs";
@@ -65,11 +65,13 @@ export const initServer = async () => {
   await withEnforceEngineMode(server);
   await withRateLimit(server);
   await withWebSocket(server);
-  await withAuth(server);
+  // DEBUG
+  // await withAuth(server);
   await withExpress(server);
   await withOpenApi(server);
   await withRoutes(server);
   await withServerUsageReporting(server);
+  await withAdminRoutes(server);
 
   await server.ready();
 

src/server/middleware/adminRoutes.ts

@@ -0,0 +1,22 @@
+import { createBullBoard } from "@bull-board/api";
+import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
+import { FastifyAdapter } from "@bull-board/fastify";
+import { FastifyInstance } from "fastify";
+import { SendTransactionQueue } from "../../worker/queues/sendTransactionQueue";
+
+export const withAdminRoutes = async (server: FastifyInstance) => {
+  const serverAdapter = new FastifyAdapter();
+
+  createBullBoard({
+    queues: [new BullMQAdapter(SendTransactionQueue.q)],
+    serverAdapter,
+  });
+
+  const bullboardPath = "/admin/queues";
+
+  serverAdapter.setBasePath(bullboardPath);
+  await server.register(serverAdapter.registerPlugin(), {
+    basePath: bullboardPath,
+    prefix: bullboardPath,
+  });
+};

src/server/middleware/engine-mode.ts renamed to src/server/middleware/engineMode.ts

@@ -1,9 +1,10 @@
 import { Static, Type } from "@sinclair/typebox";
 import { FastifyInstance } from "fastify";
 import { StatusCodes } from "http-status-codes";
-import { isDatabaseHealthy } from "../../../db/client";
+import { isDatabaseReachable } from "../../../db/client";
 import { env } from "../../../utils/env";
-import { redis } from "../../../utils/redis/redis";
+import { isRedisReachable, redis } from "../../../utils/redis/redis";
+import { createCustomError } from "../../middleware/error";
 
 type EngineFeature = "KEYPAIR_AUTH" | "CONTRACT_SUBSCRIPTIONS" | "IP_ALLOWLIST";
 
@@ -44,11 +45,20 @@ export async function healthCheck(fastify: FastifyInstance) {
       },
     },
     handler: async (req, res) => {
-      const db = await isDatabaseHealthy();
-      if (!db) {
-        return res.status(StatusCodes.SERVICE_UNAVAILABLE).send({
-          error: "The database is unreachable.",
-        });
+      if (!(await isDatabaseReachable())) {
+        throw createCustomError(
+          "The database is unreachable.",
+          StatusCodes.SERVICE_UNAVAILABLE,
+          "FAILED_HEALTHCHECK",
+        );
+      }
+
+      if (!(await isRedisReachable())) {
+        throw createCustomError(
+          "Redis is unreachable.",
+          StatusCodes.SERVICE_UNAVAILABLE,
+          "FAILED_HEALTHCHECK",
+        );
       }
 
       res.status(StatusCodes.OK).send({

src/server/routes/system/health.ts

@@ -30,3 +30,12 @@ redis.on("ready", () => {
     service: "worker",
   });
 });
+
+export const isRedisReachable = async () => {
+  try {
+    await redis.ping();
+    return true;
+  } catch (error) {
+    return false;
+  }
+};

src/utils/redis/redis.ts

@@ -2,15 +2,18 @@ import { Queue } from "bullmq";
 import { redis } from "../../utils/redis/redis";
 import { defaultJobOptions } from "./queues";
 
-export const CANCEL_RECYCLED_NONCES_QUEUE_NAME = "cancel-recycled-nonces";
+const CANCEL_RECYCLED_NONCES_QUEUE_NAME = "cancel-recycled-nonces";
 
-const _queue = new Queue<string>(CANCEL_RECYCLED_NONCES_QUEUE_NAME, {
-  connection: redis,
-  defaultJobOptions,
-});
+export const cancelRecycledNoncesQueue = new Queue<string>(
+  CANCEL_RECYCLED_NONCES_QUEUE_NAME,
+  {
+    connection: redis,
+    defaultJobOptions,
+  },
+);
 
-_queue.setGlobalConcurrency(1);
-_queue.add("hourly-cron", "", {
+cancelRecycledNoncesQueue.setGlobalConcurrency(1);
+cancelRecycledNoncesQueue.add("hourly-cron", "", {
   repeat: { pattern: "* * * * *" },
   // Use a constant jobId to not insert multiple repeatable jobs.
   jobId: "cancel-recycled-nonces-cron",

src/worker/queues/cancelRecycledNoncesQueue.ts

@@ -11,7 +11,7 @@ export type SendTransactionData = {
 export class SendTransactionQueue {
   public static name = "transactions-1-send";
 
-  private static q = new Queue<string>(this.name, {
+  static q = new Queue<string>(this.name, {
     connection: redis,
     defaultJobOptions: {
       ...defaultJobOptions,

src/worker/queues/sendTransactionQueue.ts

@@ -5,7 +5,7 @@ import { isNonceAlreadyUsedError } from "../../utils/error";
 import { logger } from "../../utils/logger";
 import { redis } from "../../utils/redis/redis";
 import { sendCancellationTransaction } from "../../utils/transaction/cancelTransaction";
-import { CANCEL_RECYCLED_NONCES_QUEUE_NAME } from "../queues/cancelRecycledNoncesQueue";
+import { cancelRecycledNoncesQueue } from "../queues/cancelRecycledNoncesQueue";
 import { logWorkerExceptions } from "../queues/queues";
 
 /**
@@ -72,7 +72,7 @@ const getAndDeleteUnusedNonces = async (key: string) => {
 };
 
 // Worker
-const _worker = new Worker(CANCEL_RECYCLED_NONCES_QUEUE_NAME, handler, {
+const _worker = new Worker(cancelRecycledNoncesQueue.name, handler, {
   concurrency: 1,
   connection: redis,
 });