feat: Add webhooks to contract subscriptions Redis workers (#521)

* feat: Add webhooks to contract subscriptions Redis workers

* updates for logs & worker

* CRUD endpoints

* make Redis backward compatible

* add processEventLogs worker

* Added contract-subscriptions config to DB

* Contract Subs Event Missing updates

* updated endpoint check

* assert valid numbers

* move webhooks query above db call

---------

Co-authored-by: farhanW3 <farhan@thirdweb.com>

Author: arcoraven

docker-compose-infra.yml

@@ -20,5 +20,29 @@ services:
           cpus: "2"
           memory: 2G
 
+  redis:
+    container_name: redis
+    image: redis:latest
+    restart: always
+    ports:
+      - 6379:6379
+    volumes:
+      - redis_data:/data
+
+  bullboard:
+    container_name: bullboard
+    image: deadly0/bull-board:latest
+    restart: always
+    ports:
+      - 3000:3000
+    environment:
+      REDIS_HOST: redis
+      REDIS_PORT: 6379
+      REDIS_USE_TLS: "false"
+      BULL_PREFIX: bull
+    depends_on:
+      - redis
+
 volumes:
   psql_db:
+  redis_data:

package.json

@@ -36,7 +36,7 @@
     "@fastify/type-provider-typebox": "^3.2.0",
     "@fastify/websocket": "^8.2.0",
     "@google-cloud/kms": "^4.0.0",
-    "@prisma/client": "5.2.0",
+    "@prisma/client": "^5.14.0-dev.65",
     "@sinclair/typebox": "^0.31.28",
     "@t3-oss/env-core": "^0.6.0",
     "@thirdweb-dev/auth": "^4.1.55",
@@ -47,6 +47,7 @@
     "@types/base-64": "^1.0.2",
     "base-64": "^1.0.0",
     "body-parser": "^1.20.2",
+    "bullmq": "^5.7.8",
     "cookie": "^0.5.0",
     "cookie-parser": "^1.4.6",
     "copyfiles": "^2.4.1",
@@ -59,14 +60,16 @@
     "fastify": "^4.15.0",
     "fastify-plugin": "^4.5.0",
     "http-status-codes": "^2.2.0",
+    "ioredis": "^5.4.1",
     "jsonwebtoken": "^9.0.2",
     "knex": "^3.1.0",
     "mnemonist": "^0.39.8",
     "node-cron": "^3.0.2",
     "pg": "^8.11.3",
     "pino": "^8.15.1",
     "pino-pretty": "^10.0.0",
-    "prisma": "^5.2.0",
+    "prisma": "5.14.0-dev.61",
+    "superjson": "^2.2.1",
     "thirdweb": "^5.1.0",
     "uuid": "^9.0.1",
     "viem": "^1.14.0",

src/db/contractEventLogs/createContractEventLogs.ts

@@ -1,34 +1,18 @@
+import { ContractEventLogs, Prisma } from "@prisma/client";
 import { PrismaTransaction } from "../../schema/prisma";
 import { getPrismaWithPostgresTx } from "../client";
 
-interface ContractEventLogEntry {
-  chainId: number;
-  blockNumber: number;
-  contractAddress: string;
-  transactionHash: string;
-  topic0?: string;
-  topic1?: string;
-  topic2?: string;
-  topic3?: string;
-  data: string;
-  decodedLog?: any; // Assuming JSON object for decodedLog
-  eventName?: string;
-  timestamp: Date;
-  transactionIndex: number;
-  logIndex: number;
-}
-
 export interface BulkInsertContractLogsParams {
   pgtx?: PrismaTransaction;
-  logs: ContractEventLogEntry[];
+  logs: Prisma.ContractEventLogsCreateInput[];
 }
 
 export const bulkInsertContractEventLogs = async ({
   pgtx,
   logs,
-}: BulkInsertContractLogsParams) => {
+}: BulkInsertContractLogsParams): Promise<ContractEventLogs[]> => {
   const prisma = getPrismaWithPostgresTx(pgtx);
-  return await prisma.contractEventLogs.createMany({
+  return await prisma.contractEventLogs.createManyAndReturn({
     data: logs,
     skipDuplicates: true,
   });

src/db/contractSubscriptions/createContractSubscription.ts

@@ -3,13 +3,22 @@ import { prisma } from "../client";
 interface CreateContractSubscriptionParams {
   chainId: number;
   contractAddress: string;
+  webhookId?: number;
 }
 
 export const createContractSubscription = async ({
   chainId,
   contractAddress,
+  webhookId,
 }: CreateContractSubscriptionParams) => {
   return prisma.contractSubscriptions.create({
-    data: { chainId, contractAddress },
+    data: {
+      chainId,
+      contractAddress,
+      webhookId,
+    },
+    include: {
+      webhook: true,
+    },
   });
 };

src/db/contractSubscriptions/deleteContractSubscription.ts

@@ -1,20 +1,8 @@
 import { prisma } from "../client";
 
-interface RemoveContractSubscriptionParams {
-  chainId: number;
-  contractAddress: string;
-}
-
-export const deleteContractSubscription = async ({
-  chainId,
-  contractAddress,
-}: RemoveContractSubscriptionParams) => {
-  return prisma.contractSubscriptions.updateMany({
-    where: {
-      chainId,
-      contractAddress,
-      deletedAt: null,
-    },
+export const deleteContractSubscription = async (id: string) => {
+  return prisma.contractSubscriptions.update({
+    where: { id },
     data: {
       deletedAt: new Date(),
     },

src/db/contractSubscriptions/getContractSubscriptions.ts

@@ -9,24 +9,28 @@ export const isContractSubscribed = async ({
   chainId,
   contractAddress,
 }: GetContractSubscriptionsParams) => {
-  const subscribedContract = await prisma.contractSubscriptions.findMany({
+  const contractSubscription = await prisma.contractSubscriptions.findFirst({
     where: {
       chainId,
       contractAddress,
       deletedAt: null,
     },
   });
-
-  if (subscribedContract.length === 0) return false;
-  else return true;
+  return contractSubscription !== null;
 };
 
-export const getContractSubscriptionsByChainId = async (chainId: number) => {
+export const getContractSubscriptionsByChainId = async (
+  chainId: number,
+  includeWebhook = false,
+) => {
   return await prisma.contractSubscriptions.findMany({
     where: {
       chainId,
       deletedAt: null,
     },
+    include: {
+      webhook: includeWebhook,
+    },
   });
 };
 
@@ -35,6 +39,9 @@ export const getAllContractSubscriptions = async () => {
     where: {
       deletedAt: null,
     },
+    include: {
+      webhook: true,
+    },
   });
 };
 

src/db/contractTransactionReceipts/createContractTransactionReceipts.ts

@@ -1,36 +1,19 @@
+import { ContractTransactionReceipts, Prisma } from "@prisma/client";
 import { PrismaTransaction } from "../../schema/prisma";
 import { getPrismaWithPostgresTx } from "../client";
 
-interface ContractTransactionReceiptEntry {
-  chainId: number;
-  blockNumber: number;
-  contractId: string;
-  contractAddress: string;
-  transactionHash: string;
-  blockHash: string;
-  timestamp: Date;
-  to: string;
-  from: string;
-  transactionIndex: number;
-  value: string;
-  data: string;
-  gasUsed: string;
-  effectiveGasPrice: string;
-  status: number;
-}
-
 export interface BulkInsertContractLogsParams {
   pgtx?: PrismaTransaction;
-  txReceipts: ContractTransactionReceiptEntry[];
+  receipts: Prisma.ContractTransactionReceiptsCreateInput[];
 }
 
 export const bulkInsertContractTransactionReceipts = async ({
   pgtx,
-  txReceipts,
-}: BulkInsertContractLogsParams) => {
+  receipts,
+}: BulkInsertContractLogsParams): Promise<ContractTransactionReceipts[]> => {
   const prisma = getPrismaWithPostgresTx(pgtx);
-  return await prisma.contractTransactionReceipts.createMany({
-    data: txReceipts,
+  return await prisma.contractTransactionReceipts.createManyAndReturn({
+    data: receipts,
     skipDuplicates: true,
   });
 };

src/db/webhooks/createWebhook.ts

@@ -1,3 +1,4 @@
+import { Webhooks } from "@prisma/client";
 import { createHash, randomBytes } from "crypto";
 import { WebhooksEventTypes } from "../../schema/webhooks";
 import { prisma } from "../client";
@@ -12,7 +13,7 @@ export const insertWebhook = async ({
   url,
   name,
   eventType,
-}: CreateWebhooksParams) => {
+}: CreateWebhooksParams): Promise<Webhooks> => {
   // generate random bytes
   const bytes = randomBytes(4096);
   // hash the bytes to create the secret (this will not be stored by itself)

src/db/webhooks/getAllWebhooks.ts

@@ -1,27 +1,10 @@
 import { Webhooks } from "@prisma/client";
-import { SanitizedWebHooksSchema } from "../../schema/webhooks";
 import { prisma } from "../client";
 
-export const getAllWebhooks = async (): Promise<SanitizedWebHooksSchema[]> => {
-  let webhooks = await prisma.webhooks.findMany({
+export const getAllWebhooks = async (): Promise<Webhooks[]> => {
+  return await prisma.webhooks.findMany({
     orderBy: {
       id: "asc",
     },
   });
-
-  return sanitizeData(webhooks);
-};
-
-const sanitizeData = (data: Webhooks[]): SanitizedWebHooksSchema[] => {
-  return data.map((webhook) => {
-    return {
-      url: webhook.url,
-      name: webhook.name,
-      eventType: webhook.eventType,
-      secret: webhook.secret ? webhook.secret : undefined,
-      createdAt: webhook.createdAt.toISOString(),
-      active: webhook.revokedAt ? false : true,
-      id: webhook.id,
-    };
-  });
 };

src/db/webhooks/getWebhook.ts

@@ -0,0 +1,7 @@
+import { prisma } from "../client";
+
+export const getWebhook = async (id: number) => {
+  return await prisma.webhooks.findUnique({
+    where: { id },
+  });
+};

src/db/webhooks/revokeWebhook.ts

@@ -1,34 +1,13 @@
-import { StatusCodes } from "http-status-codes";
-import { createCustomError } from "../../server/middleware/error";
 import { prisma } from "../client";
 
-interface RevokeWebhooksParams {
-  id: number;
-}
-
-export const markWebhookAsRevoked = async ({ id }: RevokeWebhooksParams) => {
-  const currentTimestamp = new Date();
-
-  const exists = await prisma.webhooks.findUnique({
-    where: {
-      id,
-    },
-  });
-
-  if (!exists)
-    throw createCustomError(
-      `Webhook with id ${id} does not exist`,
-      StatusCodes.BAD_REQUEST,
-      "BAD_REQUEST",
-    );
+export const deleteWebhook = async (id: number) => {
+  const now = new Date();
 
   return prisma.webhooks.update({
-    where: {
-      id,
-    },
+    where: { id },
     data: {
-      revokedAt: currentTimestamp,
-      updatedAt: currentTimestamp,
+      revokedAt: now,
+      updatedAt: now,
     },
   });
 };

src/prisma/migrations/20240510023239_add_webhook_id/migration.sql

@@ -0,0 +1,5 @@
+-- AlterTable
+ALTER TABLE "contract_subscriptions" ADD COLUMN     "webhookId" INTEGER;
+
+-- AddForeignKey
+ALTER TABLE "contract_subscriptions" ADD CONSTRAINT "contract_subscriptions_webhookId_fkey" FOREIGN KEY ("webhookId") REFERENCES "webhooks"("id") ON DELETE CASCADE ON UPDATE CASCADE;

src/prisma/migrations/20240511011737_set_on_delete_set_null/migration.sql

@@ -0,0 +1,5 @@
+-- DropForeignKey
+ALTER TABLE "contract_subscriptions" DROP CONSTRAINT "contract_subscriptions_webhookId_fkey";
+
+-- AddForeignKey
+ALTER TABLE "contract_subscriptions" ADD CONSTRAINT "contract_subscriptions_webhookId_fkey" FOREIGN KEY ("webhookId") REFERENCES "webhooks"("id") ON DELETE SET NULL ON UPDATE CASCADE;

src/prisma/migrations/20240513204722_add_contract_sub_retry_delay_to_config/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "configuration" ADD COLUMN     "contractSubscriptionsRetryDelaySeconds" TEXT NOT NULL DEFAULT '10';

src/prisma/schema.prisma

@@ -27,6 +27,7 @@ model Configuration {
   indexerListenerCronSchedule    String? @map("indexerListenerCronSchedule")
   maxBlocksToIndex               Int     @default(25) @map("maxBlocksToIndex")
   cursorDelaySeconds             Int     @default(2) @map("cursorDelaySeconds")
+  contractSubscriptionsRetryDelaySeconds String @default("10") @map("contractSubscriptionsRetryDelaySeconds")
 
   // AWS
   awsAccessKeyId                     String? @map("awsAccessKeyId")
@@ -162,14 +163,15 @@ model Transactions {
 }
 
 model Webhooks {
-  id        Int       @id @default(autoincrement()) @map("id")
-  name      String?   @map("name")
-  url       String    @map("url")
-  secret    String    @map("secret")
-  eventType String    @map("evenType")
-  createdAt DateTime  @default(now()) @map("createdAt")
-  updatedAt DateTime  @updatedAt @map("updatedAt")
-  revokedAt DateTime? @map("revokedAt")
+  id                    Int                     @id @default(autoincrement()) @map("id")
+  name                  String?                 @map("name")
+  url                   String                  @map("url")
+  secret                String                  @map("secret")
+  eventType             String                  @map("evenType")
+  createdAt             DateTime                @default(now()) @map("createdAt")
+  updatedAt             DateTime                @updatedAt @map("updatedAt")
+  revokedAt             DateTime?               @map("revokedAt")
+  ContractSubscriptions ContractSubscriptions[]
 
   @@map("webhooks")
 }
@@ -189,11 +191,14 @@ model ContractSubscriptions {
   id              String @id @default(uuid()) @map("id")
   chainId         Int
   contractAddress String
+  webhookId       Int?
 
   createdAt DateTime  @default(now())
   updatedAt DateTime  @updatedAt
   deletedAt DateTime?
 
+  webhook Webhooks? @relation(fields: [webhookId], references: [id], onDelete: SetNull)
+
   // optimize distinct lookups
   @@index([chainId])
   @@map("contract_subscriptions")