feat: Add webhooks to contract subscriptions Redis workers

Author: arcoraven

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/getContractSubscriptions.ts

@@ -21,12 +21,18 @@ export const isContractSubscribed = async ({
   else return true;
 };
 
-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,
+    },
   });
 };
 

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/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/schema.prisma

@@ -162,14 +162,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 +190,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: Cascade)
+
   // optimize distinct lookups
   @@index([chainId])
   @@map("contract_subscriptions")

src/schema/webhooks.ts

@@ -7,16 +7,7 @@ export enum WebhooksEventTypes {
   ALL_TX = "all_transactions",
   BACKEND_WALLET_BALANCE = "backend_wallet_balance",
   AUTH = "auth",
-}
-
-export interface SanitizedWebHooksSchema {
-  url: string;
-  name: string | null;
-  eventType: string;
-  secret?: string;
-  createdAt: string;
-  active: boolean;
-  id: number;
+  CONTRACT_SUBSCRIPTION = "contract_subscription",
 }
 
 export interface WalletBalanceWebhookSchema {

src/server/middleware/auth.ts

@@ -400,17 +400,18 @@ const handleAuthWebhooks = async (
   const authWebhooks = await getWebhook(WebhooksEventTypes.AUTH);
   if (authWebhooks.length > 0) {
     const authResponses = await Promise.all(
-      authWebhooks.map((webhook) =>
-        sendWebhookRequest(webhook, {
+      authWebhooks.map(async (webhook) => {
+        const { ok } = await sendWebhookRequest(webhook, {
           url: req.url,
           method: req.method,
           headers: req.headers,
           params: req.params,
           query: req.query,
           cookies: req.cookies,
           body: req.body,
-        }),
-      ),
+        });
+        return ok;
+      }),
     );
 
     if (authResponses.every((ok) => !!ok)) {

src/server/routes/contract/events/getEventLogsByTimestamp.ts

@@ -2,10 +2,8 @@ import { Static, Type } from "@sinclair/typebox";
 import { FastifyInstance } from "fastify";
 import { StatusCodes } from "http-status-codes";
 import { getEventLogsByBlockTimestamp } from "../../../../db/contractEventLogs/getContractEventLogs";
-import {
-  eventLogsSchema,
-  standardResponseSchema,
-} from "../../../schemas/sharedApiSchemas";
+import { eventLogSchema } from "../../../schemas/eventLog";
+import { standardResponseSchema } from "../../../schemas/sharedApiSchemas";
 
 const requestQuerySchema = Type.Object({
   contractAddresses: Type.Optional(Type.Array(Type.String())),
@@ -16,7 +14,7 @@ const requestQuerySchema = Type.Object({
 
 const responseSchema = Type.Object({
   result: Type.Object({
-    logs: eventLogsSchema,
+    logs: Type.Array(eventLogSchema),
     status: Type.String(),
   }),
 });

src/server/routes/contract/events/paginateEventLogs.ts

@@ -3,10 +3,8 @@ import { FastifyInstance } from "fastify";
 import { StatusCodes } from "http-status-codes";
 import { getConfiguration } from "../../../../db/configuration/getConfiguration";
 import { getEventLogsByCursor } from "../../../../db/contractEventLogs/getContractEventLogs";
-import {
-  eventLogsSchema,
-  standardResponseSchema,
-} from "../../../schemas/sharedApiSchemas";
+import { eventLogSchema, toEventLogSchema } from "../../../schemas/eventLog";
+import { standardResponseSchema } from "../../../schemas/sharedApiSchemas";
 
 const requestQuerySchema = Type.Object({
   cursor: Type.Optional(Type.String()),
@@ -18,7 +16,7 @@ const requestQuerySchema = Type.Object({
 const responseSchema = Type.Object({
   result: Type.Object({
     cursor: Type.Optional(Type.String()),
-    logs: eventLogsSchema,
+    logs: Type.Array(eventLogSchema),
     status: Type.String(),
   }),
 });
@@ -89,42 +87,18 @@ export async function pageEventLogs(fastify: FastifyInstance) {
         Date.now() - config.cursorDelaySeconds * 1000,
       );
 
-      const { cursor: newCursor, logs: resultLogs } =
-        await getEventLogsByCursor({
-          cursor,
-          limit: pageSize,
-          topics,
-          contractAddresses: standardizedContractAddresses,
-          maxCreatedAt,
-        });
-
-      const logs = resultLogs.map((log) => {
-        const topics: string[] = [];
-        [log.topic0, log.topic1, log.topic2, log.topic3].forEach((val) => {
-          if (val) {
-            topics.push(val);
-          }
-        });
-
-        return {
-          chainId: log.chainId,
-          contractAddress: log.contractAddress,
-          blockNumber: log.blockNumber,
-          transactionHash: log.transactionHash,
-          topics,
-          data: log.data,
-          eventName: log.eventName ?? undefined,
-          decodedLog: log.decodedLog,
-          timestamp: log.timestamp.getTime(),
-          transactionIndex: log.transactionIndex,
-          logIndex: log.logIndex,
-        };
+      const { cursor: newCursor, logs } = await getEventLogsByCursor({
+        cursor,
+        limit: pageSize,
+        topics,
+        contractAddresses: standardizedContractAddresses,
+        maxCreatedAt,
       });
 
       reply.status(StatusCodes.OK).send({
         result: {
           cursor: newCursor,
-          logs,
+          logs: logs.map(toEventLogSchema),
           status: "success",
         },
       });

src/server/routes/contract/transactions/getTransactionReceipts.ts

@@ -7,8 +7,8 @@ import { createCustomError } from "../../../middleware/error";
 import {
   contractParamSchema,
   standardResponseSchema,
-  transactionReceiptsSchema,
 } from "../../../schemas/sharedApiSchemas";
+import { transactionReceiptSchema } from "../../../schemas/transactionReceipt";
 import { getChainIdFromChain } from "../../../utils/chain";
 
 const requestQuerySchema = Type.Object({
@@ -18,7 +18,7 @@ const requestQuerySchema = Type.Object({
 
 const responseSchema = Type.Object({
   result: Type.Object({
-    receipts: transactionReceiptsSchema,
+    receipts: Type.Array(transactionReceiptSchema),
     status: Type.String(),
   }),
 });