Configuration + Webhooks : PG Notify/Trigger (#361)

* updated implementation to use configuration cache

* changed log level for cache to debug

* removed extra export fr cache

* updated config end-points to pull latest & return latest in response

* removed extra logging for webhooks

* cleaned webhooks cache debug logs

Author: farhanW3

src/db/transactions/getQueuedTxs.ts

@@ -2,8 +2,8 @@ import { Transactions } from "@prisma/client";
 import { Static } from "@sinclair/typebox";
 import { PrismaTransaction } from "../../schema/prisma";
 import { transactionResponseSchema } from "../../server/schemas/transaction";
+import { getConfig } from "../../utils/cache/getConfig";
 import { getPrismaWithPostgresTx } from "../client";
-import { getConfiguration } from "../configuration/getConfiguration";
 import { cleanTxs } from "./cleanTxs";
 
 interface GetQueuedTxsParams {
@@ -14,7 +14,7 @@ export const getQueuedTxs = async ({ pgtx }: GetQueuedTxsParams = {}): Promise<
   Static<typeof transactionResponseSchema>[]
 > => {
   const prisma = getPrismaWithPostgresTx(pgtx);
-  const config = await getConfiguration();
+  const config = await getConfig();
 
   // TODO: Don't use env var for transactions to batch
   const txs = await prisma.$queryRaw<Transactions[]>`

src/db/transactions/getSentTxs.ts

@@ -1,7 +1,7 @@
 import { Transactions } from "@prisma/client";
 import { PrismaTransaction } from "../../schema/prisma";
+import { getConfig } from "../../utils/cache/getConfig";
 import { getPrismaWithPostgresTx } from "../client";
-import { getConfiguration } from "../configuration/getConfiguration";
 
 interface GetSentTxsParams {
   pgtx?: PrismaTransaction;
@@ -11,7 +11,7 @@ export const getSentTxs = async ({ pgtx }: GetSentTxsParams = {}): Promise<
   Transactions[]
 > => {
   const prisma = getPrismaWithPostgresTx(pgtx);
-  const config = await getConfiguration();
+  const config = await getConfig();
 
   return prisma.$queryRaw<Transactions[]>`
     SELECT * FROM "transactions"

src/db/transactions/getSentUserOps.ts

@@ -1,7 +1,7 @@
 import { Transactions } from "@prisma/client";
 import { PrismaTransaction } from "../../schema/prisma";
+import { getConfig } from "../../utils/cache/getConfig";
 import { getPrismaWithPostgresTx } from "../client";
-import { getConfiguration } from "../configuration/getConfiguration";
 
 interface GetSentUserOpsParams {
   pgtx?: PrismaTransaction;
@@ -11,7 +11,7 @@ export const getSentUserOps = async ({
   pgtx,
 }: GetSentUserOpsParams = {}): Promise<Transactions[]> => {
   const prisma = getPrismaWithPostgresTx(pgtx);
-  const config = await getConfiguration();
+  const config = await getConfig();
 
   return prisma.$queryRaw<Transactions[]>`
     SELECT * FROM "transactions"

src/db/transactions/getTxToRetry.ts

@@ -1,7 +1,7 @@
 import { Transactions } from "@prisma/client";
 import type { PrismaTransaction } from "../../schema/prisma";
+import { getConfig } from "../../utils/cache/getConfig";
 import { getPrismaWithPostgresTx } from "../client";
-import { getConfiguration } from "../configuration/getConfiguration";
 
 interface GetTxToRetryParams {
   pgtx?: PrismaTransaction;
@@ -11,7 +11,7 @@ export const getTxToRetry = async ({ pgtx }: GetTxToRetryParams = {}): Promise<
   Transactions | undefined
 > => {
   const prisma = getPrismaWithPostgresTx(pgtx);
-  const config = await getConfiguration();
+  const config = await getConfig();
 
   // TODO: Remove transactionHash
   // TODO: For now, we're not retrying user ops

src/db/webhooks/createWebhook.ts

@@ -1,6 +1,5 @@
 import { createHash, randomBytes } from "crypto";
 import { WebhooksEventTypes } from "../../schema/webhooks";
-import { webhookCache } from "../../utils/cache/getWebhook";
 import { prisma } from "../client";
 
 interface CreateWebhooksParams {
@@ -19,9 +18,6 @@ export const insertWebhook = async ({
   // hash the bytes to create the secret (this will not be stored by itself)
   const secret = createHash("sha512").update(bytes).digest("base64url");
 
-  // Clear Cache
-  webhookCache.clear();
-
   return prisma.webhooks.create({
     data: {
       url,

src/db/webhooks/revokeWebhook.ts

@@ -1,15 +1,12 @@
 import { StatusCodes } from "http-status-codes";
 import { createCustomError } from "../../server/middleware/error";
-import { webhookCache } from "../../utils/cache/getWebhook";
 import { prisma } from "../client";
 
 interface RevokeWebhooksParams {
   id: number;
 }
 
 export const markWebhookAsRevoked = async ({ id }: RevokeWebhooksParams) => {
-  // Clear Cache
-  webhookCache.clear();
   const currentTimestamp = new Date();
 
   const exists = await prisma.webhooks.findUnique({

src/prisma/migrations/20231226100100_webhook_config_triggers/migration.sql

@@ -0,0 +1,65 @@
+-- Configuration Triggers
+CREATE OR REPLACE FUNCTION notify_configuration_insert()
+  RETURNS TRIGGER
+  LANGUAGE plpgsql
+AS $function$
+BEGIN
+  PERFORM pg_notify('new_configuration_data', row_to_json(NEW)::text);
+  RETURN NEW;
+END;
+$function$;
+
+CREATE OR REPLACE FUNCTION notify_configuration_update()
+  RETURNS TRIGGER
+  LANGUAGE plpgsql
+AS $function$
+BEGIN
+  PERFORM pg_notify('updated_configuration_data', json_build_object(
+      'id', NEW.id
+  )::text);
+  RETURN NEW;
+END;
+$function$;
+
+CREATE OR REPLACE TRIGGER configuration_insert_trigger
+  AFTER INSERT ON configuration
+  FOR EACH ROW
+  EXECUTE FUNCTION notify_configuration_insert();
+
+CREATE OR REPLACE TRIGGER configuration_update_trigger
+  AFTER UPDATE ON configuration
+  FOR EACH ROW
+  EXECUTE FUNCTION notify_configuration_update();
+
+-- Webhooks Triggers
+CREATE OR REPLACE FUNCTION notify_webhooks_insert()
+  RETURNS TRIGGER
+  LANGUAGE plpgsql
+AS $function$
+BEGIN
+  PERFORM pg_notify('new_webhook_data', row_to_json(NEW)::text);
+  RETURN NEW;
+END;
+$function$;
+
+CREATE OR REPLACE FUNCTION notify_webhooks_update()
+  RETURNS TRIGGER
+  LANGUAGE plpgsql
+AS $function$
+BEGIN
+  PERFORM pg_notify('updated_webhook_data', json_build_object(
+      'id', NEW.id
+  )::text);
+  RETURN NEW;
+END;
+$function$;
+
+CREATE OR REPLACE TRIGGER webhooks_insert_trigger
+  AFTER INSERT ON webhooks
+  FOR EACH ROW
+  EXECUTE FUNCTION notify_webhooks_insert();
+
+CREATE OR REPLACE TRIGGER webhooks_update_trigger
+  AFTER UPDATE ON webhooks
+  FOR EACH ROW
+  EXECUTE FUNCTION notify_webhooks_update();

src/server/middleware/auth.ts

@@ -8,14 +8,14 @@ import { GenericAuthWallet, LocalWallet } from "@thirdweb-dev/wallets";
 import { AsyncWallet } from "@thirdweb-dev/wallets/evm/wallets/async";
 import { utils } from "ethers";
 import { FastifyInstance } from "fastify";
-import { getConfiguration } from "../../db/configuration/getConfiguration";
 import { updateConfiguration } from "../../db/configuration/updateConfiguration";
 import { getPermissions } from "../../db/permissions/getPermissions";
 import { createToken } from "../../db/tokens/createToken";
 import { getToken } from "../../db/tokens/getToken";
 import { revokeToken } from "../../db/tokens/revokeToken";
 import { WebhooksEventTypes } from "../../schema/webhooks";
-import { getWebhookConfig } from "../../utils/cache/getWebhook";
+import { getConfig } from "../../utils/cache/getConfig";
+import { getWebhook } from "../../utils/cache/getWebhook";
 import { env } from "../../utils/env";
 import { logger } from "../../utils/logger";
 import { Permission } from "../schemas/auth";
@@ -73,7 +73,7 @@ const authWithApiServer = async (jwt: string, domain: string) => {
 };
 
 export const withAuth = async (server: FastifyInstance) => {
-  const config = await getConfiguration();
+  const config = await getConfig();
 
   // Configure the ThirdwebAuth fastify plugin
   const { authRouter, authMiddleware, getUser } = ThirdwebAuth<
@@ -86,7 +86,7 @@ export const withAuth = async (server: FastifyInstance) => {
     wallet: new AsyncWallet({
       // TODO: Use caching for the signer
       getSigner: async () => {
-        const config = await getConfiguration();
+        const config = await getConfig();
         const wallet = new LocalWallet();
 
         try {
@@ -209,7 +209,7 @@ export const withAuth = async (server: FastifyInstance) => {
       const thirdwebApiSecretKey = req.headers.authorization?.split(" ")[1];
       if (thirdwebApiSecretKey === env.THIRDWEB_API_SECRET_KEY) {
         // If the secret key is being used, treat the user as the auth wallet
-        const config = await getConfiguration();
+        const config = await getConfig();
         const wallet = new LocalWallet();
 
         try {
@@ -300,7 +300,7 @@ export const withAuth = async (server: FastifyInstance) => {
         // no-op
       }
 
-      const authWebhooks = await getWebhookConfig(WebhooksEventTypes.AUTH);
+      const authWebhooks = await getWebhook(WebhooksEventTypes.AUTH);
       if (authWebhooks) {
         const authResponses = await Promise.all(
           authWebhooks.map((webhook) =>

src/server/routes/auth/access-tokens/create.ts

@@ -3,9 +3,9 @@ import { buildJWT } from "@thirdweb-dev/auth";
 import { LocalWallet } from "@thirdweb-dev/wallets";
 import { FastifyInstance } from "fastify";
 import { StatusCodes } from "http-status-codes";
-import { getConfiguration } from "../../../../db/configuration/getConfiguration";
 import { updateConfiguration } from "../../../../db/configuration/updateConfiguration";
 import { createToken } from "../../../../db/tokens/createToken";
+import { getConfig } from "../../../../utils/cache/getConfig";
 import { env } from "../../../../utils/env";
 import { AccessTokenSchema } from "./getAll";
 
@@ -42,7 +42,7 @@ export async function createAccessToken(fastify: FastifyInstance) {
     handler: async (req, res) => {
       const { label } = req.body;
 
-      const config = await getConfiguration();
+      const config = await getConfig();
       const wallet = new LocalWallet();
 
       // TODO: Remove this with next breaking change

src/server/routes/backend-wallet/create.ts

@@ -1,8 +1,8 @@
 import { Static, Type } from "@sinclair/typebox";
 import { FastifyInstance } from "fastify";
 import { StatusCodes } from "http-status-codes";
-import { getConfiguration } from "../../../db/configuration/getConfiguration";
 import { WalletType } from "../../../schema/wallet";
+import { getConfig } from "../../../utils/cache/getConfig";
 import { standardResponseSchema } from "../../schemas/sharedApiSchemas";
 import { createAwsKmsWallet } from "../../utils/wallets/createAwsKmsWallet";
 import { createGcpKmsWallet } from "../../utils/wallets/createGcpKmsWallet";
@@ -48,7 +48,7 @@ export const createBackendWallet = async (fastify: FastifyInstance) => {
       const { label } = req.body;
 
       let walletAddress: string;
-      const config = await getConfiguration();
+      const config = await getConfig();
       switch (config.walletConfiguration.type) {
         case WalletType.local:
           walletAddress = await createLocalWallet({ label });