Execute processTx on a cron schedule (#349)

Author: adam-maj

src/worker/listeners/queuedTxListener.ts

@@ -1,73 +1,22 @@
-import PQueue from "p-queue";
-import { knex } from "../../db/client";
+import cron from "node-cron";
+import { getConfiguration } from "../../db/configuration/getConfiguration";
 import { logger } from "../../utils/logger";
 import { processTx } from "../tasks/processTx";
 
-const queue = new PQueue({
-  concurrency: 1,
-  autoStart: true,
-});
-
-queue.on("error", (err) => {
-  logger({
-    service: "worker",
-    level: "error",
-    message: `[Queue Error] Size: ${queue.size}, Pending: ${queue.pending}`,
-    error: err,
-  });
-
-  throw err;
-});
-
 export const queuedTxListener = async (): Promise<void> => {
   logger({
     service: "worker",
     level: "info",
     message: `Listening for queued transactions`,
   });
 
-  // TODO: This doesn't even need to be a listener
-  const connection = await knex.client.acquireConnection();
-  connection.query(`LISTEN new_transaction_data`);
-
-  // Queue transaction processing immediately on startup
-  queue.add(processTx);
-
-  // Whenever we receive a new transaction, process it
-  connection.on(
-    "notification",
-    async (msg: { channel: string; payload: string }) => {
-      queue.add(processTx);
-    },
-  );
-
-  connection.on("end", async () => {
-    await knex.destroy();
-    knex.client.releaseConnection(connection);
-
-    logger({
-      service: "worker",
-      level: "info",
-      message: `Released database connection on end`,
-    });
-  });
-
-  connection.on("error", async (err: any) => {
-    logger({
-      service: "worker",
-      level: "error",
-      message: `Database connection error`,
-      error: err,
-    });
+  const config = await getConfiguration();
 
-    await knex.destroy();
-    knex.client.releaseConnection(connection);
+  if (!config.minedTxListenerCronSchedule) {
+    return;
+  }
 
-    logger({
-      service: "worker",
-      level: "info",
-      message: `Released database connection on error`,
-      error: err,
-    });
+  cron.schedule(config.minedTxListenerCronSchedule, async () => {
+    await processTx();
   });
 };

src/worker/tasks/processTx.ts

@@ -49,6 +49,12 @@ export const processTx = async () => {
         // 1. Select a batch of transactions and lock the rows so no other workers pick them up
         const txs = await getQueuedTxs({ pgtx });
 
+        logger({
+          service: "worker",
+          level: "info",
+          message: `Received ${txs.length} transactions to process`,
+        });
+
         const config = await getConfiguration();
         if (txs.length < config.minTxsToProcess) {
           return;