Websocket event updates fixed (#376)

* moved updateTxListener to server side to help with websockets realtime update

* updated logs for websocket

Author: farhanW3

src/server/index.ts

@@ -6,6 +6,7 @@ import { URL } from "url";
 import { deleteAllWalletNonces } from "../db/wallets/deleteAllWalletNonces";
 import { env } from "../utils/env";
 import { logger } from "../utils/logger";
+import { updateTxListener } from "./listerners/updateTxListener";
 import { withAuth } from "./middleware/auth";
 import { withCors } from "./middleware/cors";
 import { withErrorHandler } from "./middleware/error";
@@ -89,6 +90,7 @@ const main = async () => {
   });
 
   writeOpenApiToFile(server);
+  await updateTxListener();
 };
 
 main();

src/worker/listeners/updateTxListener.ts renamed to src/server/listerners/updateTxListener.ts

@@ -9,7 +9,7 @@ import { logger } from "../../utils/logger";
 
 export const updateTxListener = async (): Promise<void> => {
   logger({
-    service: "worker",
+    service: "server",
     level: "info",
     message: `Listening for updated transactions`,
   });
@@ -24,7 +24,7 @@ export const updateTxListener = async (): Promise<void> => {
 
       // Send websocket message
       const index = subscriptionsData.findIndex(
-        (sub) => sub.requestId === parsedPayload.identifier,
+        (sub) => sub.requestId === parsedPayload.id,
       );
 
       if (index == -1) {
@@ -33,7 +33,13 @@ export const updateTxListener = async (): Promise<void> => {
 
       const userSubscription = subscriptionsData[index];
       const returnData = await getTxById({
-        queueId: parsedPayload.identifier,
+        queueId: parsedPayload.id,
+      });
+
+      logger({
+        service: "server",
+        level: "info",
+        message: `[updateTxListener] Sending websocket update for queueId: ${parsedPayload.id}, status ${returnData?.status}.`,
       });
 
       const { message, closeConnection } =
@@ -47,26 +53,26 @@ export const updateTxListener = async (): Promise<void> => {
 
   connection.on("end", async () => {
     logger({
-      service: "worker",
+      service: "server",
       level: "info",
-      message: `Connection database ended`,
+      message: `[updateTxListener] Connection database ended`,
     });
 
     knex.client.releaseConnection(connection);
     await knex.destroy();
 
     logger({
-      service: "worker",
+      service: "server",
       level: "info",
-      message: `Released database connection on end`,
+      message: `[updateTxListener] Released database connection on end`,
     });
   });
 
   connection.on("error", async (err: any) => {
     logger({
-      service: "worker",
+      service: "server",
       level: "error",
-      message: `Database connection error`,
+      message: `[updateTxListener] Database connection error`,
       error: err,
     });
 
@@ -76,7 +82,7 @@ export const updateTxListener = async (): Promise<void> => {
     logger({
       service: "worker",
       level: "info",
-      message: `Released database connection on error`,
+      message: `[updateTxListener] Released database connection on error`,
     });
   });
 };

src/worker/index.ts

@@ -8,7 +8,6 @@ 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";
 import {
   newWebhooksListener,
   updatedWebhooksListener,
@@ -18,9 +17,6 @@ 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();