wip

Author: arcoraven

src/db/contractSubscriptions/createContractSubscription.ts

@@ -7,6 +7,7 @@ interface CreateContractSubscriptionParams {
   processEventLogs: boolean;
   filterEvents: string[];
   processTransactionReceipts: boolean;
+  filterFunctions: string[];
 }
 
 export const createContractSubscription = async ({
@@ -16,6 +17,7 @@ export const createContractSubscription = async ({
   processEventLogs,
   filterEvents,
   processTransactionReceipts,
+  filterFunctions,
 }: CreateContractSubscriptionParams) => {
   return prisma.contractSubscriptions.create({
     data: {
@@ -25,6 +27,7 @@ export const createContractSubscription = async ({
       processEventLogs,
       filterEvents,
       processTransactionReceipts,
+      filterFunctions,
     },
     include: {
       webhook: 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/schema.prisma

@@ -195,6 +195,7 @@ model ContractSubscriptions {
   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

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

@@ -35,12 +35,18 @@ const bodySchema = Type.Object({
   filterEvents: Type.Optional(
     Type.Array(Type.String(), {
       description:
-        "A case-sensitive list of event log names to parse. If empty, parse all event logs.",
+        "A case-sensitive list of event names to filter event logs. Parses all event logs by default.",
     }),
   ),
   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.",
+    }),
+  ),
 });
 
 const responseSchema = Type.Object({
@@ -82,6 +88,7 @@ export async function addContractSubscription(fastify: FastifyInstance) {
         processEventLogs,
         filterEvents = [],
         processTransactionReceipts,
+        filterFunctions = [],
       } = request.body;
 
       const chainId = await getChainIdFromChain(chain);
@@ -136,6 +143,7 @@ export async function addContractSubscription(fastify: FastifyInstance) {
         processEventLogs,
         filterEvents,
         processTransactionReceipts,
+        filterFunctions,
       });
 
       reply.status(StatusCodes.OK).send({

src/server/schemas/contractSubscription.ts

@@ -10,6 +10,7 @@ export const contractSubscriptionSchema = Type.Object({
   processEventLogs: Type.Boolean(),
   filterEvents: Type.Array(Type.String()),
   processTransactionReceipts: Type.Boolean(),
+  filterFunctions: Type.Array(Type.String()),
   createdAt: Type.Unsafe<Date>({
     type: "string",
     format: "date",
@@ -28,5 +29,6 @@ export const toContractSubscriptionSchema = (
   processEventLogs: contractSubscription.processEventLogs,
   filterEvents: contractSubscription.filterEvents,
   processTransactionReceipts: contractSubscription.processTransactionReceipts,
+  filterFunctions: contractSubscription.filterFunctions,
   createdAt: contractSubscription.createdAt,
 });

src/worker/queues/processTransactionReceiptsQueue.ts

@@ -20,7 +20,10 @@ const _queue = redis
 
 export type EnqueueProcessTransactionReceiptsData = {
   chainId: number;
-  filters: { address: string }[];
+  filters: {
+    address: string;
+    functions: string[];
+  }[];
   fromBlock: number; // inclusive
   toBlock: number; // inclusive
 };

src/worker/tasks/chainIndexer.ts

@@ -81,12 +81,13 @@ export const createChainIndexerTask = async (args: {
             });
           }
 
-          // Identify contract addresses to parse transaction receipts, if any.
+          // Identify addresses + function names to parse transaction receipts, if any.
           const transactionReceiptFilters: { address: string }[] =
             contractSubscriptions
-              .filter((cs) => cs.processTransactionReceipts)
-              .map((cs) => ({
-                address: cs.contractAddress.toLowerCase(),
+              .filter((c) => c.processTransactionReceipts)
+              .map((c) => ({
+                address: c.contractAddress.toLowerCase(),
+                functions: c.filterFunctions,
               }));
           if (transactionReceiptFilters.length > 0) {
             await enqueueProcessTransactionReceipts({

src/worker/tasks/processEventLogsWorker.ts

@@ -103,13 +103,20 @@ export const getWebhooksByContractAddresses = async (
 
 type GetLogsParams = EnqueueProcessEventLogsData;
 
-const getLogs = async (
-  params: GetLogsParams,
-): Promise<Prisma.ContractEventLogsCreateInput[]> => {
-  const chain = defineChain(params.chainId);
+const getLogs = async ({
+  chainId,
+  fromBlock,
+  toBlock,
+  filters,
+}: GetLogsParams): Promise<Prisma.ContractEventLogsCreateInput[]> => {
+  if (filters.length === 0) {
+    return [];
+  }
+
+  const chain = defineChain(chainId);
 
   // Get events for each contract address. Apply any filters.
-  const promises = params.filters.map(async (f) => {
+  const promises = filters.map(async (f) => {
     const contract = getContract({
       client: thirdwebClient,
       chain,
@@ -128,12 +135,10 @@ const getLogs = async (
       }
     }
 
-    // TODO: add filter for function name.
-
     return await getContractEvents({
       contract,
-      fromBlock: BigInt(params.fromBlock),
-      toBlock: BigInt(params.toBlock),
+      fromBlock: BigInt(fromBlock),
+      toBlock: BigInt(toBlock),
       events,
     });
   });
@@ -146,7 +151,7 @@ const getLogs = async (
   // Transform logs into the DB schema.
   return allEvents.map(
     (event): Prisma.ContractEventLogsCreateInput => ({
-      chainId: params.chainId,
+      chainId,
       blockNumber: Number(event.blockNumber),
       contractAddress: event.address.toLowerCase(),
       transactionHash: event.transactionHash,
@@ -156,15 +161,21 @@ const getLogs = async (
       topic3: event.topics[3],
       data: event.data,
       eventName: event.eventName,
-      decodedLog: event.args,
+      decodedLog: event.args ?? {},
       timestamp: blockTimestamps[event.blockHash],
       transactionIndex: event.transactionIndex,
       logIndex: event.logIndex,
     }),
   );
 };
 
-export const getBlockTimestamps = async (
+/**
+ * Gets the timestamps for a list of block hashes. Falls back to the current time.
+ * @param chain
+ * @param blockHashes
+ * @returns Record<Hash, Date>
+ */
+const getBlockTimestamps = async (
   chain: Chain,
   blockHashes: Hash[],
 ): Promise<Record<Hash, Date>> => {
@@ -180,7 +191,7 @@ export const getBlockTimestamps = async (
     dedupe.map(async (blockHash) => {
       try {
         const block = await eth_getBlockByHash(rpcRequest, { blockHash });
-        return new Date(Number(block.timestamp * 1000n));
+        return new Date(Number(block.timestamp) * 1000);
       } catch (e) {
         return now;
       }