Added Webhook Config to DB & updated logic to cache data & reset when…

… config updated (#233)

server/api/configuration/webhooks/get.ts

@@ -0,0 +1,38 @@
+import { Static, Type } from "@sinclair/typebox";
+import { FastifyInstance } from "fastify";
+import { StatusCodes } from "http-status-codes";
+import { getConfiguration } from "../../../../src/db/configuration/getConfiguration";
+
+export const ReplySchema = Type.Object({
+  result: Type.Object({
+    webhookUrl: Type.String(),
+    webhookAuthBearerToken: Type.String(),
+  }),
+});
+
+export async function getWebhookConfiguration(fastify: FastifyInstance) {
+  fastify.route<{
+    Reply: Static<typeof ReplySchema>;
+  }>({
+    method: "GET",
+    url: "/configuration/webhook",
+    schema: {
+      summary: "Get webhook configuration",
+      description: "Get the engine configuration for webhook",
+      tags: ["Configuration"],
+      operationId: "getWebhookConfiguration",
+      response: {
+        [StatusCodes.OK]: ReplySchema,
+      },
+    },
+    handler: async (req, res) => {
+      const config = await getConfiguration();
+      res.status(200).send({
+        result: {
+          webhookAuthBearerToken: config.webhookAuthBearerToken || "",
+          webhookUrl: config.webhookUrl || "",
+        },
+      });
+    },
+  });
+}

server/api/configuration/webhooks/update.ts

@@ -0,0 +1,40 @@
+import { Static, Type } from "@sinclair/typebox";
+import { FastifyInstance } from "fastify";
+import { StatusCodes } from "http-status-codes";
+import { updateConfiguration } from "../../../../src/db/configuration/updateConfiguration";
+import { ReplySchema } from "./get";
+
+const BodySchema = Type.Partial(
+  Type.Object({
+    webhookUrl: Type.String(),
+    webhookAuthBearerToken: Type.String(),
+  }),
+);
+
+export async function updateWebhookConfiguration(fastify: FastifyInstance) {
+  fastify.route<{
+    Body: Static<typeof BodySchema>;
+  }>({
+    method: "POST",
+    url: "/configuration/webhook",
+    schema: {
+      summary: "Update webhook configuration",
+      description: "Update the engine configuration for webhook",
+      tags: ["Configuration"],
+      operationId: "updateWebhookConfiguration",
+      body: BodySchema,
+      response: {
+        [StatusCodes.OK]: ReplySchema,
+      },
+    },
+    handler: async (req, res) => {
+      const config = await updateConfiguration({ ...req.body });
+      res.status(200).send({
+        result: {
+          webhookUrl: config.webhookUrl,
+          webhookAuthBearerToken: config.webhookAuthBearerToken,
+        },
+      });
+    },
+  });
+}

server/api/index.ts

@@ -49,12 +49,18 @@ import { getBalance } from "./backend-wallet/getBalance";
 import { importWallet } from "./backend-wallet/import";
 import { sendTransaction } from "./backend-wallet/send";
 import { transfer } from "./backend-wallet/transfer";
+
+// Configuration
 import { getChainsConfiguration } from "./configuration/chains/get";
 import { updateChainsConfiguration } from "./configuration/chains/update";
 import { getTransactionConfiguration } from "./configuration/transactions/get";
 import { updateTransactionConfiguration } from "./configuration/transactions/update";
 import { getWalletsConfiguration } from "./configuration/wallets/get";
 import { updateWalletsConfiguration } from "./configuration/wallets/update";
+import { getWebhookConfiguration } from "./configuration/webhooks/get";
+import { updateWebhookConfiguration } from "./configuration/webhooks/update";
+
+// Accounts
 import { accountRoutes } from "./contract/extensions/account";
 import { accountFactoryRoutes } from "./contract/extensions/accountFactory";
 
@@ -74,6 +80,8 @@ export const apiRoutes = async (fastify: FastifyInstance) => {
   await fastify.register(updateChainsConfiguration);
   await fastify.register(getTransactionConfiguration);
   await fastify.register(updateTransactionConfiguration);
+  await fastify.register(getWebhookConfiguration);
+  await fastify.register(updateWebhookConfiguration);
 
   // Chains
   await fastify.register(getChainData);

server/controller/tx-update-listener.ts

@@ -1,6 +1,5 @@
 import { knex } from "../../src/db/client";
 import { getTxById } from "../../src/db/transactions/getTxById";
-import { env } from "../../src/utils/env";
 import { logger } from "../../src/utils/logger";
 import {
   formatSocketMessage,
@@ -23,13 +22,7 @@ export const startTxUpdatesNotificationListener = async (): Promise<void> => {
         const parsedPayload = JSON.parse(msg.payload);
 
         // Send webhook
-        if (env.WEBHOOK_URL.length > 0) {
-          await sendWebhook(parsedPayload);
-        } else {
-          logger.server.debug(
-            `Webhooks are disabled or no URL is provided. Skipping webhook update`,
-          );
-        }
+        await sendWebhook(parsedPayload);
 
         // Send websocket message
         const index = subscriptionsData.findIndex(

server/utilities/webhook.ts

@@ -1,9 +1,16 @@
 import { getTxById } from "../../src/db/transactions/getTxById";
-import { env } from "../../src/utils/env";
 import { logger } from "../../src/utils/logger";
+import { getWebhookConfig } from "../utils/cache/getWebhookConfig";
 
 export const sendWebhook = async (data: any): Promise<void> => {
   try {
+    const webhookConfig = await getWebhookConfig();
+
+    if (!webhookConfig.webhookUrl) {
+      logger.server.debug("No WebhookURL set, skipping webhook send");
+      return;
+    }
+
     const txData = await getTxById({ queueId: data.id });
     const headers: {
       Accept: string;
@@ -14,13 +21,13 @@ export const sendWebhook = async (data: any): Promise<void> => {
       "Content-Type": "application/json",
     };
 
-    if (process.env.WEBHOOK_AUTH_BEARER_TOKEN) {
+    if (webhookConfig.webhookAuthBearerToken) {
       headers[
         "Authorization"
-      ] = `Bearer ${process.env.WEBHOOK_AUTH_BEARER_TOKEN}`;
+      ] = `Bearer ${webhookConfig.webhookAuthBearerToken}`;
     }
 
-    const response = await fetch(env.WEBHOOK_URL, {
+    const response = await fetch(webhookConfig.webhookUrl, {
       method: "POST",
       headers,
       body: JSON.stringify(txData),

server/utils/cache/getWebhookConfig.ts

@@ -0,0 +1,29 @@
+import { getConfiguration } from "../../../src/db/configuration/getConfiguration";
+
+interface WebhookConfig {
+  webhookUrl: string;
+  webhookAuthBearerToken: string | null;
+}
+
+export const webhookCache = new Map<string, WebhookConfig>();
+
+export const getWebhookConfig = async (): Promise<WebhookConfig> => {
+  const cacheKey = `webhookConfig`;
+  if (webhookCache.has(cacheKey)) {
+    return webhookCache.get(cacheKey)! as WebhookConfig;
+  }
+
+  const config = await getConfiguration();
+
+  if (config.webhookAuthBearerToken || config.webhookUrl) {
+    webhookCache.set(cacheKey, {
+      webhookUrl: config.webhookUrl!,
+      webhookAuthBearerToken: config.webhookAuthBearerToken,
+    });
+  }
+
+  return {
+    webhookUrl: config.webhookUrl!,
+    webhookAuthBearerToken: config.webhookAuthBearerToken,
+  };
+};

src/db/configuration/updateConfiguration.ts

@@ -1,4 +1,5 @@
 import { Prisma } from "@prisma/client";
+import { webhookCache } from "../../../server/utils/cache/getWebhookConfig";
 import { encrypt } from "../../utils/cypto";
 import { prisma } from "../client";
 import { getConfiguration } from "./getConfiguration";
@@ -7,6 +8,10 @@ export const updateConfiguration = async (
   data: Prisma.ConfigurationUpdateArgs["data"],
 ) => {
   const config = await getConfiguration();
+
+  // Clearing webhook cache on update
+  webhookCache.clear();
+
   return prisma.configuration.update({
     where: {
       id: config.id,

src/prisma/migrations/20231017214123_webhook_config/migration.sql

@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE "configuration" ADD COLUMN     "webhookAuthBearerToken" TEXT,
+ADD COLUMN     "webhookUrl" TEXT;

src/prisma/schema.prisma

@@ -33,6 +33,9 @@ model Configuration {
   gcpKmsKeyRingId                    String? @map("gcpKmsKeyRingId")
   gcpApplicationCredentialEmail      String? @map("gcpApplicationCredentialEmail")
   gcpApplicationCredentialPrivateKey String? @map("gcpApplicationCredentialPrivateKey")
+  // Webhook
+  webhookUrl                         String? @map("webhookUrl")
+  webhookAuthBearerToken             String? @map("webhookAuthBearerToken")
 
   @@map("configuration")
 }

src/utils/env.ts

@@ -55,16 +55,6 @@ export const env = createEnv({
     PORT: z.coerce.number().default(3005),
     HOST: z.string().default("0.0.0.0"),
     ACCESS_CONTROL_ALLOW_ORIGIN: z.string().default("*"),
-    WEBHOOK_URL: z
-      .string()
-      .default("")
-      .transform((url) => {
-        if (url.length > 0) {
-          return url;
-        }
-        return "";
-      }),
-    WEBHOOK_AUTH_BEARER_TOKEN: z.string().default(""),
   },
   clientPrefix: "NEVER_USED",
   client: {},
@@ -77,8 +67,6 @@ export const env = createEnv({
     HOST: process.env.HOST,
     OPENAPI_BASE_ORIGIN: process.env.OPENAPI_BASE_ORIGIN,
     ACCESS_CONTROL_ALLOW_ORIGIN: process.env.ACCESS_CONTROL_ALLOW_ORIGIN,
-    WEBHOOK_URL: process.env.WEBHOOK_URL,
-    WEBHOOK_AUTH_BEARER_TOKEN: process.env.WEBHOOK_AUTH_BEARER_TOKEN,
   },
   onValidationError: (error: ZodError) => {
     console.error(

src/worker/listeners/queuedTxListener.ts

@@ -1,7 +1,6 @@
 import PQueue from "p-queue";
 import { sendWebhook } from "../../../server/utilities/webhook";
 import { knex } from "../../db/client";
-import { env } from "../../utils/env";
 import { logger } from "../../utils/logger";
 import { processTx } from "../tasks/processTx";
 
@@ -31,14 +30,8 @@ export const queuedTxListener = async (): Promise<void> => {
   connection.on(
     "notification",
     async (msg: { channel: string; payload: string }) => {
-      if (env.WEBHOOK_URL.length > 0) {
-        const parsedPayload = JSON.parse(msg.payload);
-        await sendWebhook(parsedPayload);
-      } else {
-        logger.server.debug(
-          `Webhooks are disabled or no URL is provided. Skipping webhook update`,
-        );
-      }
+      const parsedPayload = JSON.parse(msg.payload);
+      await sendWebhook(parsedPayload);
       queue.add(processTx);
     },
   );