Move webhook update listener to worker (#262)

Author: adam-maj

server/controller/tx-update-listener.ts

@@ -6,7 +6,6 @@ import { URL } from "url";
 import { env } from "../src/utils/env";
 import { logger } from "../src/utils/logger";
 import { withRoutes } from "./api";
-import { startTxUpdatesNotificationListener } from "./controller/tx-update-listener";
 import { withAuth } from "./middleware/auth";
 import { withCors } from "./middleware/cors";
 import { withErrorHandler } from "./middleware/error";
@@ -65,8 +64,6 @@ const main = async () => {
       }
     },
   );
-
-  await startTxUpdatesNotificationListener();
 };
 
 main();

server/index.ts

@@ -2,11 +2,15 @@ import { deleteProcessedTx } from "./listeners/deleteProcessedTx";
 import { minedTxListener } from "./listeners/minedTxListener";
 import { queuedTxListener } from "./listeners/queuedTxListener";
 import { retryTxListener } from "./listeners/retryTxListener";
+import { updateTxListener } from "./listeners/updateTxListener";
 
 const worker = async () => {
   // Listen for queued transactions to process
   await queuedTxListener();
 
+  // Listen for transaction updates to send webhooks
+  await updateTxListener();
+
   // Poll for transactions stuck in mempool to retry
   await retryTxListener();
 

src/worker/index.ts

@@ -0,0 +1,60 @@
+import {
+  formatSocketMessage,
+  getStatusMessageAndConnectionStatus,
+} from "../../../server/helpers/websocket";
+import { subscriptionsData } from "../../../server/schemas/websocket";
+import { sendTxWebhook } from "../../../server/utilities/webhook";
+import { knex } from "../../db/client";
+import { getTxById } from "../../db/transactions/getTxById";
+import { logger } from "../../utils/logger";
+
+export const updateTxListener = async (): Promise<void> => {
+  logger.worker.info(`Listening for updated transactions`);
+
+  const connection = await knex.client.acquireConnection();
+  connection.query(`LISTEN updated_transaction_data`);
+
+  connection.on(
+    "notification",
+    async (msg: { channel: string; payload: string }) => {
+      const parsedPayload = JSON.parse(msg.payload);
+
+      // Send webhook
+      await sendTxWebhook(parsedPayload);
+
+      // Send websocket message
+      const index = subscriptionsData.findIndex(
+        (sub) => sub.requestId === parsedPayload.identifier,
+      );
+
+      if (index == -1) {
+        return;
+      }
+
+      const userSubscription = subscriptionsData[index];
+      const returnData = await getTxById({
+        queueId: parsedPayload.identifier,
+      });
+      const { message, closeConnection } =
+        await getStatusMessageAndConnectionStatus(returnData);
+      userSubscription.socket.send(
+        await formatSocketMessage(returnData, message),
+      );
+      closeConnection ? userSubscription.socket.close() : null;
+    },
+  );
+
+  connection.on("end", async () => {
+    logger.server.info(`Connection database ended`);
+    knex.client.releaseConnection(connection);
+    await knex.destroy();
+    logger.server.info(`Released sql connection : on end`);
+  });
+
+  connection.on("error", async (err: any) => {
+    logger.server.error(err);
+    knex.client.releaseConnection(connection);
+    await knex.destroy();
+    logger.server.info(`Released sql connection: on error`);
+  });
+};