explicit worker init step

Author: arcoraven

src/server/index.ts

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

src/server/middleware/adminRoutes.ts

@@ -1,14 +1,31 @@
 import { createBullBoard } from "@bull-board/api";
 import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
 import { FastifyAdapter } from "@bull-board/fastify";
+import { Queue } from "bullmq";
 import { FastifyInstance } from "fastify";
+import { CancelRecycledNoncesQueue } from "../../worker/queues/cancelRecycledNoncesQueue";
+import { MineTransactionQueue } from "../../worker/queues/mineTransactionQueue";
+import { ProcessEventsLogQueue } from "../../worker/queues/processEventLogsQueue";
+import { ProcessTransactionReceiptsQueue } from "../../worker/queues/processTransactionReceiptsQueue";
+import { PruneTransactionsQueue } from "../../worker/queues/pruneTransactionsQueue";
 import { SendTransactionQueue } from "../../worker/queues/sendTransactionQueue";
+import { SendWebhookQueue } from "../../worker/queues/sendWebhookQueue";
 
 export const withAdminRoutes = async (server: FastifyInstance) => {
   const serverAdapter = new FastifyAdapter();
 
+  const queues: Queue[] = [
+    SendWebhookQueue.q,
+    ProcessEventsLogQueue.q,
+    ProcessTransactionReceiptsQueue.q,
+    SendTransactionQueue.q,
+    MineTransactionQueue.q,
+    CancelRecycledNoncesQueue.q,
+    PruneTransactionsQueue.q,
+  ];
+
   createBullBoard({
-    queues: [new BullMQAdapter(SendTransactionQueue.q)],
+    queues: queues.map((q) => new BullMQAdapter(q)),
     serverAdapter,
   });
 

src/utils/transaction/webhook.ts

@@ -1,5 +1,5 @@
 import { WebhooksEventTypes } from "../../schema/webhooks";
-import { enqueueWebhook } from "../../worker/queues/sendWebhookQueue";
+import { SendWebhookQueue } from "../../worker/queues/sendWebhookQueue";
 import { AnyTransaction } from "./types";
 
 export const enqueueTransactionWebhook = async (
@@ -16,9 +16,7 @@ export const enqueueTransactionWebhook = async (
       : status === "errored"
       ? WebhooksEventTypes.ERRORED_TX
       : null;
-
-  if (!type) {
-    return;
+  if (type) {
+    await SendWebhookQueue.enqueueWebhook({ type, queueId });
   }
-  await enqueueWebhook({ type, queueId });
 };

src/worker/index.ts

@@ -7,17 +7,23 @@ import {
   newWebhooksListener,
   updatedWebhooksListener,
 } from "./listeners/webhookListener";
-
-// Init workers.
-import "./tasks/cancelRecycledNoncesWorker";
-import "./tasks/mineTransactionWorker";
-import "./tasks/processEventLogsWorker";
-import "./tasks/processTransactionReceiptsWorker";
-import "./tasks/pruneTransactionsWorker";
-import "./tasks/sendTransactionWorker";
-import "./tasks/sendWebhookWorker";
+import { initCancelRecycledNoncesWorker } from "./tasks/cancelRecycledNoncesWorker";
+import { initMineTransactionWorker } from "./tasks/mineTransactionWorker";
+import { initProcessEventLogsWorker } from "./tasks/processEventLogsWorker";
+import { initProcessTransactionReceiptsWorker } from "./tasks/processTransactionReceiptsWorker";
+import { initPruneTransactionsWorker } from "./tasks/pruneTransactionsWorker";
+import { initSendTransactionWorker } from "./tasks/sendTransactionWorker";
+import { initSendWebhookWorker } from "./tasks/sendWebhookWorker";
 
 export const initWorker = async () => {
+  initCancelRecycledNoncesWorker();
+  initProcessEventLogsWorker();
+  initProcessTransactionReceiptsWorker();
+  initPruneTransactionsWorker();
+  initSendTransactionWorker();
+  initMineTransactionWorker();
+  initSendWebhookWorker();
+
   // Listen for new & updated configuration data.
   await newConfigurationListener();
   await updatedConfigurationListener();

src/worker/queues/cancelRecycledNoncesQueue.ts

@@ -1,20 +1,25 @@
 import { Queue } from "bullmq";
+import { env } from "../../utils/env";
 import { redis } from "../../utils/redis/redis";
 import { defaultJobOptions } from "./queues";
 
-const CANCEL_RECYCLED_NONCES_QUEUE_NAME = "cancel-recycled-nonces";
+export class CancelRecycledNoncesQueue {
+  private static name = "cancel-recycled-nonces";
 
-export const cancelRecycledNoncesQueue = new Queue<string>(
-  CANCEL_RECYCLED_NONCES_QUEUE_NAME,
-  {
+  static q = new Queue<string>(this.name, {
     connection: redis,
     defaultJobOptions,
-  },
-);
+  });
 
-cancelRecycledNoncesQueue.setGlobalConcurrency(1);
-cancelRecycledNoncesQueue.add("hourly-cron", "", {
-  repeat: { pattern: "* * * * *" },
-  // Use a constant jobId to not insert multiple repeatable jobs.
-  jobId: "cancel-recycled-nonces-cron",
-});
+  constructor() {
+    if (env.ENGINE_MODE === "server_only") {
+      return;
+    }
+
+    CancelRecycledNoncesQueue.q.setGlobalConcurrency(1);
+    CancelRecycledNoncesQueue.q.add("cron", "", {
+      repeat: { pattern: "* * * * *" },
+      jobId: "cron",
+    });
+  }
+}

src/worker/queues/mineTransactionQueue.ts

@@ -8,9 +8,9 @@ export type MineTransactionData = {
 };
 
 export class MineTransactionQueue {
-  public static name = "transactions-2-mine";
+  private static name = "transactions-2-mine";
 
-  private static q = new Queue<string>(this.name, {
+  static q = new Queue<string>(this.name, {
     connection: redis,
     defaultJobOptions: {
       ...defaultJobOptions,

src/worker/queues/processEventLogsQueue.ts

@@ -5,14 +5,6 @@ import { getConfig } from "../../utils/cache/getConfig";
 import { redis } from "../../utils/redis/redis";
 import { defaultJobOptions } from "./queues";
 
-export const PROCESS_EVENT_LOGS_QUEUE_NAME = "process-event-logs";
-
-// Queue
-const _queue = new Queue<string>(PROCESS_EVENT_LOGS_QUEUE_NAME, {
-  connection: redis,
-  defaultJobOptions,
-});
-
 // Each job handles a block range for a given chain, filtered by addresses + events.
 export type EnqueueProcessEventLogsData = {
   chainId: number;
@@ -24,28 +16,35 @@ export type EnqueueProcessEventLogsData = {
   toBlock: number; // inclusive
 };
 
-export const enqueueProcessEventLogs = async (
-  data: EnqueueProcessEventLogsData,
-) => {
-  const serialized = SuperJSON.stringify(data);
-  // e.g. 8453:14423125-14423685
-  const jobName = `${data.chainId}:${data.fromBlock}-${data.toBlock}`;
-  const { contractSubscriptionsRequeryDelaySeconds } = await getConfig();
-  const requeryDelays = contractSubscriptionsRequeryDelaySeconds.split(",");
+export class ProcessEventsLogQueue {
+  private static name = "process-event-logs";
 
-  // Enqueue one job immediately and any delayed jobs.
-  await _queue.add(jobName, serialized);
+  static q = new Queue<string>(this.name, {
+    connection: redis,
+    defaultJobOptions,
+  });
 
-  // The last attempt should attempt repeatedly to handle extended RPC issues.
-  // This backoff attempts at intervals:
-  // 30s, 1m, 2m, 4m, 8m, 16m, 32m, ~1h, ~2h, ~4h
-  for (let i = 0; i < requeryDelays.length; i++) {
-    const delay = parseInt(requeryDelays[i]) * 1000;
-    const attempts = i === requeryDelays.length - 1 ? 10 : 0;
-    await _queue.add(jobName, serialized, {
-      delay,
-      attempts,
-      backoff: { type: "exponential", delay: 30_000 },
-    });
-  }
-};
+  static add = async (data: EnqueueProcessEventLogsData) => {
+    const serialized = SuperJSON.stringify(data);
+    // e.g. 8453:14423125-14423685
+    const jobName = `${data.chainId}:${data.fromBlock}-${data.toBlock}`;
+    const { contractSubscriptionsRequeryDelaySeconds } = await getConfig();
+    const requeryDelays = contractSubscriptionsRequeryDelaySeconds.split(",");
+
+    // Enqueue one job immediately and any delayed jobs.
+    await this.q.add(jobName, serialized);
+
+    // The last attempt should attempt repeatedly to handle extended RPC issues.
+    // This backoff attempts at intervals:
+    // 30s, 1m, 2m, 4m, 8m, 16m, 32m, ~1h, ~2h, ~4h
+    for (let i = 0; i < requeryDelays.length; i++) {
+      const delay = parseInt(requeryDelays[i]) * 1000;
+      const attempts = i === requeryDelays.length - 1 ? 10 : 0;
+      await this.q.add(jobName, serialized, {
+        delay,
+        attempts,
+        backoff: { type: "exponential", delay: 30_000 },
+      });
+    }
+  };
+}

src/worker/queues/processTransactionReceiptsQueue.ts

@@ -5,17 +5,6 @@ import { getConfig } from "../../utils/cache/getConfig";
 import { redis } from "../../utils/redis/redis";
 import { defaultJobOptions } from "./queues";
 
-export const PROCESS_TRANSACTION_RECEIPTS_QUEUE_NAME =
-  "process-transaction-receipts";
-
-// Queue
-const _queue = redis
-  ? new Queue<string>(PROCESS_TRANSACTION_RECEIPTS_QUEUE_NAME, {
-      connection: redis,
-      defaultJobOptions,
-    })
-  : null;
-
 // Each job handles a block range for a given chain, filtered by addresses + events.
 export type EnqueueProcessTransactionReceiptsData = {
   chainId: number;
@@ -27,30 +16,35 @@ export type EnqueueProcessTransactionReceiptsData = {
   toBlock: number; // inclusive
 };
 
-export const enqueueProcessTransactionReceipts = async (
-  data: EnqueueProcessTransactionReceiptsData,
-) => {
-  if (!_queue) return;
+export class ProcessTransactionReceiptsQueue {
+  private static name = "process-transaction-receipts";
 
-  const serialized = superjson.stringify(data);
-  // e.g. 8453:14423125-14423685
-  const jobName = `${data.chainId}:${data.fromBlock}-${data.toBlock}`;
-  const { contractSubscriptionsRequeryDelaySeconds } = await getConfig();
-  const requeryDelays = contractSubscriptionsRequeryDelaySeconds.split(",");
+  static q = new Queue<string>(this.name, {
+    connection: redis,
+    defaultJobOptions,
+  });
 
-  // Enqueue one job immediately and any delayed jobs.
-  await _queue.add(jobName, serialized);
+  static add = async (data: EnqueueProcessTransactionReceiptsData) => {
+    const serialized = superjson.stringify(data);
+    // e.g. 8453:14423125-14423685
+    const jobName = `${data.chainId}:${data.fromBlock}-${data.toBlock}`;
+    const { contractSubscriptionsRequeryDelaySeconds } = await getConfig();
+    const requeryDelays = contractSubscriptionsRequeryDelaySeconds.split(",");
 
-  // The last attempt should attempt repeatedly to handle extended RPC issues.
-  // This backoff attempts at intervals:
-  // 30s, 1m, 2m, 4m, 8m, 16m, 32m, ~1h, ~2h, ~4h
-  for (let i = 0; i < requeryDelays.length; i++) {
-    const delay = parseInt(requeryDelays[i]) * 1000;
-    const attempts = i === requeryDelays.length - 1 ? 10 : 0;
-    await _queue.add(jobName, serialized, {
-      delay,
-      attempts,
-      backoff: { type: "exponential", delay: 30_000 },
-    });
-  }
-};
+    // Enqueue one job immediately and any delayed jobs.
+    await this.q.add(jobName, serialized);
+
+    // The last attempt should attempt repeatedly to handle extended RPC issues.
+    // This backoff attempts at intervals:
+    // 30s, 1m, 2m, 4m, 8m, 16m, 32m, ~1h, ~2h, ~4h
+    for (let i = 0; i < requeryDelays.length; i++) {
+      const delay = parseInt(requeryDelays[i]) * 1000;
+      const attempts = i === requeryDelays.length - 1 ? 10 : 0;
+      await this.q.add(jobName, serialized, {
+        delay,
+        attempts,
+        backoff: { type: "exponential", delay: 30_000 },
+      });
+    }
+  };
+}

src/worker/queues/pruneTransactionsQueue.ts

@@ -1,17 +1,25 @@
 import { Queue } from "bullmq";
+import { env } from "../../utils/env";
 import { redis } from "../../utils/redis/redis";
 import { defaultJobOptions } from "./queues";
 
-export const PRUNE_TRANSACTIONS_QUEUE_NAME = "prune-transactions";
+export class PruneTransactionsQueue {
+  private static name = "prune-transactions";
 
-const _queue = new Queue<string>(PRUNE_TRANSACTIONS_QUEUE_NAME, {
-  connection: redis,
-  defaultJobOptions,
-});
+  static q = new Queue<string>(this.name, {
+    connection: redis,
+    defaultJobOptions,
+  });
 
-_queue.setGlobalConcurrency(1);
-_queue.add("hourly-cron", "", {
-  repeat: { pattern: "*/10 * * * *" },
-  // Use a constant jobId to not insert multiple repeatable jobs.
-  jobId: "prune-transactions-cron",
-});
+  constructor() {
+    if (env.ENGINE_MODE === "server_only") {
+      return;
+    }
+
+    PruneTransactionsQueue.q.setGlobalConcurrency(1);
+    PruneTransactionsQueue.q.add("cron", "", {
+      repeat: { pattern: "*/10 * * * *" },
+      jobId: "cron",
+    });
+  }
+}

src/worker/queues/sendTransactionQueue.ts

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