feat: Add filters to Contract Subscriptions

arcoraven · arcoraven · commit 1c9aead07847 · 2024-06-03T20:43:05.000-07:00
diff --git a/src/db/contractSubscriptions/createContractSubscription.ts b/src/db/contractSubscriptions/createContractSubscription.ts
@@ -4,18 +4,27 @@ interface CreateContractSubscriptionParams {
   chainId: number;
   contractAddress: string;
   webhookId?: number;
+  parseEventLogs: boolean;
+  filterEventLogs: string[];
+  parseTransactionReceipts: boolean;
 }
 
 export const createContractSubscription = async ({
   chainId,
   contractAddress,
   webhookId,
+  parseEventLogs,
+  filterEventLogs,
+  parseTransactionReceipts,
 }: CreateContractSubscriptionParams) => {
   return prisma.contractSubscriptions.create({
     data: {
       chainId,
       contractAddress,
       webhookId,
+      parseEventLogs,
+      filterEventLogs,
+      parseTransactionReceipts,
     },
     include: {
       webhook: true,
diff --git a/src/prisma/migrations/20240603232427_contract_subscription_filters/migration.sql b/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;
diff --git a/src/prisma/migrations/20240604034241_rename_contract_subscription_vars/migration.sql b/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;
diff --git a/src/prisma/schema.prisma b/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,13 @@ 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)
 
   createdAt DateTime  @default(now())
   updatedAt DateTime  @updatedAt
diff --git a/src/server/routes/contract/subscriptions/addContractSubscription.ts b/src/server/routes/contract/subscriptions/addContractSubscription.ts
@@ -29,6 +29,22 @@ const bodySchema = Type.Object({
       examples: ["https://example.com/webhook"],
     }),
   ),
+  parseEventLogs: Type.Optional(
+    Type.Boolean({
+      description: "If true, parse event logs for this contract.",
+    }),
+  ),
+  filterEventLogs: Type.Optional(
+    Type.Array(Type.String(), {
+      description:
+        "A case-sensitive list of event log names to parse. If empty, parse all event logs.",
+    }),
+  ),
+  parseTransactionReceipts: Type.Optional(
+    Type.Boolean({
+      description: "If true, parse transaction receipts for this contract.",
+    }),
+  ),
 });
 
 const responseSchema = Type.Object({
@@ -63,11 +79,27 @@ export async function addContractSubscription(fastify: FastifyInstance) {
       },
     },
     handler: async (request, reply) => {
-      const { chain, contractAddress, webhookUrl } = request.body;
+      const {
+        chain,
+        contractAddress,
+        webhookUrl,
+        parseEventLogs = true,
+        filterEventLogs = [],
+        parseTransactionReceipts = true,
+      } = request.body;
 
       const chainId = await getChainIdFromChain(chain);
       const standardizedContractAddress = contractAddress.toLowerCase();
 
+      // Must parse logs or receipts.
+      if (!parseEventLogs && !parseTransactionReceipts) {
+        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();
       if (!subscribedChainIds.includes(chainId)) {
@@ -105,6 +137,9 @@ export async function addContractSubscription(fastify: FastifyInstance) {
         chainId,
         contractAddress: standardizedContractAddress,
         webhookId,
+        parseEventLogs,
+        filterEventLogs,
+        parseTransactionReceipts,
       });
 
       reply.status(StatusCodes.OK).send({
diff --git a/src/server/routes/system/health.ts b/src/server/routes/system/health.ts
@@ -62,7 +62,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;
diff --git a/src/server/schemas/contractSubscription.ts b/src/server/schemas/contractSubscription.ts
@@ -7,6 +7,9 @@ export const contractSubscriptionSchema = Type.Object({
   chainId: Type.Number(),
   contractAddress: Type.String(),
   webhook: Type.Optional(WebhookSchema),
+  parseEventLogs: Type.Boolean(),
+  filterEventLogs: Type.Array(Type.String()),
+  parseTransactionReceipts: Type.Boolean(),
   createdAt: Type.Unsafe<Date>({
     type: "string",
     format: "date",
@@ -22,5 +25,8 @@ export const toContractSubscriptionSchema = (
   webhook: contractSubscription.webhook
     ? toWebhookSchema(contractSubscription.webhook)
     : undefined,
+  parseEventLogs: contractSubscription.parseEventLogs,
+  filterEventLogs: contractSubscription.filterEventLogs,
+  parseTransactionReceipts: contractSubscription.parseTransactionReceipts,
   createdAt: contractSubscription.createdAt,
 });
diff --git a/src/worker/queues/processEventLogsQueue.ts b/src/worker/queues/processEventLogsQueue.ts
@@ -19,7 +19,11 @@ const _queue = redis
 
 export type EnqueueProcessEventLogsData = {
   chainId: number;
-  contractAddresses: string[];
+  filters: {
+    address: string;
+    events: string[];
+    functions: string[];
+  }[];
   fromBlock: number; // inclusive
   toBlock: number; // inclusive
 };
diff --git a/src/worker/queues/processTransactionReceiptsQueue.ts b/src/worker/queues/processTransactionReceiptsQueue.ts
@@ -20,7 +20,7 @@ const _queue = redis
 
 export type EnqueueProcessTransactionReceiptsData = {
   chainId: number;
-  contractAddresses: string[];
+  filters: { address: string }[];
   fromBlock: number; // inclusive
   toBlock: number; // inclusive
 };
diff --git a/src/worker/tasks/chainIndexer.ts b/src/worker/tasks/chainIndexer.ts
@@ -32,7 +32,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 +43,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 +57,47 @@ 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: string;
+            events: string[];
+            functions: string[];
+          }[] = contractSubscriptions
+            .filter((c) => c.parseEventLogs)
+            .map((c) => ({
+              address: c.contractAddress,
+              events: c.filterEventNames,
+              functions: c.filterFunctionNames,
+            }));
+          if (eventLogFilters.length > 0) {
+            await enqueueProcessEventLogs({
+              chainId,
+              fromBlock,
+              toBlock,
+              filters: eventLogFilters,
+            });
+          }
 
-          await enqueueProcessTransactionReceipts({
-            chainId,
-            fromBlock,
-            toBlock,
-            contractAddresses,
-          });
+          // Identify contract addresses to parse transaction receipts, if any.
+          const transactionReceiptFilters: { address: string }[] =
+            contractSubscriptions
+              .filter((cs) => cs.parseTransactionReceipts)
+              .map((cs) => ({
+                address: cs.contractAddress.toLowerCase(),
+              }));
+          if (transactionReceiptFilters.length > 0) {
+            await enqueueProcessTransactionReceipts({
+              chainId,
+              fromBlock,
+              toBlock,
+              filters: transactionReceiptFilters,
+            });
+          }
 
           // Update the latest block number.
           try {
@@ -103,7 +116,7 @@ export const createChainIndexerTask = async (args: {
           }
         },
         {
-          timeout: 5 * 60000, // 3 minutes timeout
+          timeout: 60 * 1000, // 1 minute timeout
         },
       );
     } catch (err: any) {
diff --git a/src/worker/tasks/processEventLogsWorker.ts b/src/worker/tasks/processEventLogsWorker.ts
diff --git a/src/worker/tasks/processTransactionReceiptsWorker.ts b/src/worker/tasks/processTransactionReceiptsWorker.ts
