feat: Add filters to Contract Subscriptions

src/db/contractSubscriptions/createContractSubscription.ts

@@ -4,18 +4,30 @@ interface CreateContractSubscriptionParams {
   chainId: number;
   contractAddress: string;
   webhookId?: number;
+  processEventLogs: boolean;
+  filterEvents: string[];
+  processTransactionReceipts: boolean;
+  filterFunctions: string[];
 }
 
 export const createContractSubscription = async ({
   chainId,
   contractAddress,
   webhookId,
+  processEventLogs,
+  filterEvents,
+  processTransactionReceipts,
+  filterFunctions,
 }: CreateContractSubscriptionParams) => {
   return prisma.contractSubscriptions.create({
     data: {
       chainId,
       contractAddress,
       webhookId,
+      processEventLogs,
+      filterEvents,
+      processTransactionReceipts,
+      filterFunctions,
     },
     include: {
       webhook: true,

src/prisma/migrations/20240603232427_contract_subscription_filters/migration.sql

@@ -0,0 +1,4 @@
+-- AlterTable
+ALTER TABLE "contract_subscriptions" ADD COLUMN     "filterEventLogs" TEXT[] DEFAULT ARRAY[]::TEXT[],
+ADD COLUMN     "parseEventLogs" BOOLEAN NOT NULL DEFAULT true,
+ADD COLUMN     "parseTransactionReceipts" BOOLEAN NOT NULL DEFAULT true;

src/prisma/migrations/20240604034241_rename_contract_subscription_vars/migration.sql

@@ -0,0 +1,15 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `filterEventLogs` on the `contract_subscriptions` table. All the data in the column will be lost.
+  - You are about to drop the column `parseEventLogs` on the `contract_subscriptions` table. All the data in the column will be lost.
+  - You are about to drop the column `parseTransactionReceipts` on the `contract_subscriptions` table. All the data in the column will be lost.
+
+*/
+-- AlterTable
+ALTER TABLE "contract_subscriptions" DROP COLUMN "filterEventLogs",
+DROP COLUMN "parseEventLogs",
+DROP COLUMN "parseTransactionReceipts",
+ADD COLUMN     "filterEvents" TEXT[] DEFAULT ARRAY[]::TEXT[],
+ADD COLUMN     "processEventLogs" BOOLEAN NOT NULL DEFAULT true,
+ADD COLUMN     "processTransactionReceipts" BOOLEAN NOT NULL DEFAULT true;

src/prisma/migrations/20240611023057_add_contract_subscriptions_filter_functions/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "contract_subscriptions" ADD COLUMN     "filterFunctions" TEXT[] DEFAULT ARRAY[]::TEXT[];

src/prisma/migrations/20240612015710_add_function_name/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "contract_transaction_receipts" ADD COLUMN     "functionName" TEXT;

src/prisma/schema.prisma

@@ -8,26 +8,26 @@ generator client {
 }
 
 model Configuration {
-  id                             String  @id @default("default") @map("id")
+  id                                     String  @id @default("default") @map("id")
   // Chains
-  chainOverrides                 String? @map("chainOverrides")
+  chainOverrides                         String? @map("chainOverrides")
   // Tx Processing
-  minTxsToProcess                Int     @map("minTxsToProcess")
-  maxTxsToProcess                Int     @map("maxTxsToProcess")
+  minTxsToProcess                        Int     @map("minTxsToProcess")
+  maxTxsToProcess                        Int     @map("maxTxsToProcess")
   // Tx Updates
-  minedTxListenerCronSchedule    String? @map("minedTxsCronSchedule")
-  maxTxsToUpdate                 Int     @map("maxTxsToUpdate")
+  minedTxListenerCronSchedule            String? @map("minedTxsCronSchedule")
+  maxTxsToUpdate                         Int     @map("maxTxsToUpdate")
   // Tx Retries
-  retryTxListenerCronSchedule    String? @map("retryTxsCronSchedule")
-  minEllapsedBlocksBeforeRetry   Int     @map("minEllapsedBlocksBeforeRetry")
-  maxFeePerGasForRetries         String  @map("maxFeePerGasForRetries")
-  maxPriorityFeePerGasForRetries String  @map("maxPriorityFeePerGasForRetries")
-  maxRetriesPerTx                Int     @map("maxRetriesPerTx")
+  retryTxListenerCronSchedule            String? @map("retryTxsCronSchedule")
+  minEllapsedBlocksBeforeRetry           Int     @map("minEllapsedBlocksBeforeRetry")
+  maxFeePerGasForRetries                 String  @map("maxFeePerGasForRetries")
+  maxPriorityFeePerGasForRetries         String  @map("maxPriorityFeePerGasForRetries")
+  maxRetriesPerTx                        Int     @map("maxRetriesPerTx")
   // Contract Indexer Updates
-  indexerListenerCronSchedule    String? @map("indexerListenerCronSchedule")
-  maxBlocksToIndex               Int     @default(25) @map("maxBlocksToIndex")
-  cursorDelaySeconds             Int     @default(2) @map("cursorDelaySeconds")
-  contractSubscriptionsRetryDelaySeconds String @default("10") @map("contractSubscriptionsRetryDelaySeconds")
+  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")
@@ -188,10 +188,14 @@ model Relayers {
 }
 
 model ContractSubscriptions {
-  id              String @id @default(uuid()) @map("id")
-  chainId         Int
-  contractAddress String
-  webhookId       Int?
+  id                         String   @id @default(uuid()) @map("id")
+  chainId                    Int
+  contractAddress            String
+  webhookId                  Int?
+  processEventLogs           Boolean  @default(true)
+  filterEvents               String[] @default([]) // empty array = no filter
+  processTransactionReceipts Boolean  @default(true)
+  filterFunctions            String[] @default([]) // empty array = no filter
 
   createdAt DateTime  @default(now())
   updatedAt DateTime  @updatedAt
@@ -242,6 +246,7 @@ model ContractTransactionReceipts {
   blockHash       String
   timestamp       DateTime
   data            String
+  functionName    String?
 
   to               String
   from             String

src/server/routes/contract/subscriptions/addContractSubscription.ts

@@ -29,6 +29,26 @@ const bodySchema = Type.Object({
       examples: ["https://example.com/webhook"],
     }),
   ),
+  processEventLogs: Type.Boolean({
+    description: "If true, parse event logs for this contract.",
+  }),
+  filterEvents: Type.Optional(
+    Type.Array(Type.String(), {
+      description:
+        "A case-sensitive list of event names to filter event logs. Parses all event logs by default.",
+      examples: ["Transfer"],
+    }),
+  ),
+  processTransactionReceipts: Type.Boolean({
+    description: "If true, parse transaction receipts for this contract.",
+  }),
+  filterFunctions: Type.Optional(
+    Type.Array(Type.String(), {
+      description:
+        "A case-sensitive list of function names to filter transaction receipts. Parses all transaction receipts by default.",
+      examples: ["mintTo"],
+    }),
+  ),
 });
 
 const responseSchema = Type.Object({
@@ -63,10 +83,26 @@ export async function addContractSubscription(fastify: FastifyInstance) {
       },
     },
     handler: async (request, reply) => {
-      const { chain, contractAddress, webhookUrl } = request.body;
+      const {
+        chain,
+        contractAddress,
+        webhookUrl,
+        processEventLogs,
+        filterEvents = [],
+        processTransactionReceipts,
+        filterFunctions = [],
+      } = request.body;
 
       const chainId = await getChainIdFromChain(chain);
-      const standardizedContractAddress = contractAddress.toLowerCase();
+
+      // Must parse logs or receipts.
+      if (!processEventLogs && !processTransactionReceipts) {
+        throw createCustomError(
+          "Contract Subscriptions must parse event logs and/or receipts.",
+          StatusCodes.BAD_REQUEST,
+          "BAD_REQUEST",
+        );
+      }
 
       // If not currently indexed, upsert the latest block number.
       const subscribedChainIds = await getContractSubscriptionsUniqueChainIds();
@@ -103,8 +139,12 @@ export async function addContractSubscription(fastify: FastifyInstance) {
       // Create the contract subscription.
       const contractSubscription = await createContractSubscription({
         chainId,
-        contractAddress: standardizedContractAddress,
+        contractAddress: contractAddress.toLowerCase(),
         webhookId,
+        processEventLogs,
+        filterEvents,
+        processTransactionReceipts,
+        filterFunctions,
       });
 
       reply.status(StatusCodes.OK).send({

src/server/routes/system/health.ts

@@ -64,7 +64,7 @@ const getFeatures = (): EngineFeature[] => {
   const features: EngineFeature[] = [];
 
   if (env.ENABLE_KEYPAIR_AUTH) features.push("KEYPAIR_AUTH");
-  // Contract subscriptions requires Redis.
+  // Contract Subscriptions requires Redis.
   if (redis) features.push("CONTRACT_SUBSCRIPTIONS");
 
   return features;

src/server/schemas/contractSubscription.ts

@@ -7,6 +7,10 @@ export const contractSubscriptionSchema = Type.Object({
   chainId: Type.Number(),
   contractAddress: Type.String(),
   webhook: Type.Optional(WebhookSchema),
+  processEventLogs: Type.Boolean(),
+  filterEvents: Type.Array(Type.String()),
+  processTransactionReceipts: Type.Boolean(),
+  filterFunctions: Type.Array(Type.String()),
   createdAt: Type.Unsafe<Date>({
     type: "string",
     format: "date",
@@ -22,5 +26,9 @@ export const toContractSubscriptionSchema = (
   webhook: contractSubscription.webhook
     ? toWebhookSchema(contractSubscription.webhook)
     : undefined,
+  processEventLogs: contractSubscription.processEventLogs,
+  filterEvents: contractSubscription.filterEvents,
+  processTransactionReceipts: contractSubscription.processTransactionReceipts,
+  filterFunctions: contractSubscription.filterFunctions,
   createdAt: contractSubscription.createdAt,
 });

src/worker/queues/processEventLogsQueue.ts

@@ -1,5 +1,6 @@
 import { Queue } from "bullmq";
 import SuperJSON from "superjson";
+import { Address } from "thirdweb";
 import { getConfig } from "../../utils/cache/getConfig";
 import { redis } from "../../utils/redis/redis";
 import { defaultJobOptions } from "./queues";
@@ -17,9 +18,13 @@ const _queue = redis
     })
   : null;
 
+// Each job handles a block range for a given chain, filtered by addresses + events.
 export type EnqueueProcessEventLogsData = {
   chainId: number;
-  contractAddresses: string[];
+  filters: {
+    address: Address;
+    events: string[];
+  }[];
   fromBlock: number; // inclusive
   toBlock: number; // inclusive
 };

src/worker/queues/processTransactionReceiptsQueue.ts

@@ -1,5 +1,6 @@
 import { Queue } from "bullmq";
 import SuperJSON from "superjson";
+import { Address } from "thirdweb";
 import { getConfig } from "../../utils/cache/getConfig";
 import { redis } from "../../utils/redis/redis";
 import { defaultJobOptions } from "./queues";
@@ -18,9 +19,13 @@ const _queue = redis
     })
   : null;
 
+// Each job handles a block range for a given chain, filtered by addresses + events.
 export type EnqueueProcessTransactionReceiptsData = {
   chainId: number;
-  contractAddresses: string[];
+  filters: {
+    address: Address;
+    functions: string[];
+  }[];
   fromBlock: number; // inclusive
   toBlock: number; // inclusive
 };

src/worker/tasks/chainIndexer.ts

@@ -1,4 +1,5 @@
 import { StaticJsonRpcBatchProvider } from "@thirdweb-dev/sdk";
+import { Address } from "thirdweb";
 import { getBlockForIndexing } from "../../db/chainIndexers/getChainIndexer";
 import { upsertChainIndexer } from "../../db/chainIndexers/upsertChainIndexer";
 import { prisma } from "../../db/client";
@@ -32,7 +33,7 @@ export const createChainIndexerTask = async (args: {
           const currentBlockNumber =
             (await provider.getBlockNumber()) - toBlockOffset;
 
-          // Limit toBlock to avoid hitting rate or file size limits when querying logs.
+          // Limit toBlock to avoid hitting rate or block range limits when querying logs.
           const toBlock = Math.min(
             currentBlockNumber,
             fromBlock + maxBlocksToIndex,
@@ -43,11 +44,9 @@ export const createChainIndexerTask = async (args: {
             return;
           }
 
-          // Ensuring that the block data exists.
-          // Sometimes the RPC providers nodes are aware of the latest block
-          // but the block data is not available yet.
+          // Ensure that the block data exists.
+          // Sometimes the RPC nodes do not yet return data for the latest block.
           const block = await provider.getBlockWithTransactions(toBlock);
-
           if (!block) {
             logger({
               service: "worker",
@@ -59,32 +58,48 @@ export const createChainIndexerTask = async (args: {
             return;
           }
 
-          // Get contract addresses to filter event logs and transaction receipts by.
           const contractSubscriptions = await getContractSubscriptionsByChainId(
             chainId,
             true,
           );
-          const contractAddresses = [
-            ...new Set<string>(
-              contractSubscriptions.map(
-                (subscription) => subscription.contractAddress,
-              ),
-            ),
-          ];
 
-          await enqueueProcessEventLogs({
-            chainId,
-            fromBlock,
-            toBlock,
-            contractAddresses,
-          });
+          // Identify contract addresses + event names to parse event logs, if any.
+          const eventLogFilters: {
+            address: Address;
+            events: string[];
+          }[] = contractSubscriptions
+            .filter((c) => c.processEventLogs)
+            .map((c) => ({
+              address: c.contractAddress as Address,
+              events: c.filterEvents,
+            }));
+          if (eventLogFilters.length > 0) {
+            await enqueueProcessEventLogs({
+              chainId,
+              fromBlock,
+              toBlock,
+              filters: eventLogFilters,
+            });
+          }
 
-          await enqueueProcessTransactionReceipts({
-            chainId,
-            fromBlock,
-            toBlock,
-            contractAddresses,
-          });
+          // Identify addresses + function names to parse transaction receipts, if any.
+          const transactionReceiptFilters: {
+            address: Address;
+            functions: string[];
+          }[] = contractSubscriptions
+            .filter((c) => c.processTransactionReceipts)
+            .map((c) => ({
+              address: c.contractAddress as Address,
+              functions: c.filterFunctions,
+            }));
+          if (transactionReceiptFilters.length > 0) {
+            await enqueueProcessTransactionReceipts({
+              chainId,
+              fromBlock,
+              toBlock,
+              filters: transactionReceiptFilters,
+            });
+          }
 
           // Update the latest block number.
           try {
@@ -103,7 +118,7 @@ export const createChainIndexerTask = async (args: {
           }
         },
         {
-          timeout: 5 * 60000, // 3 minutes timeout
+          timeout: 60 * 1000, // 1 minute timeout
         },
       );
     } catch (err: any) {