Feature: Contract Events (#431)

* added table schema

* add trigger for new indexed contracts

* delete migrations

* update schema

* add migration for indexer tables

* add migration for indexer triggers

* adding db util functions

* adding indexer routes

* remove indexer trigger migration

* update db contract log getter

* add getLogs route

* save contract indexer listener

* add indexer listener cron schedule

* adding indexer config

* updating the db layer

* updating routes

* updating prisma schema

* updating indexer worker

* add failure message

* estimate blocktime, add route

* cleanup

* remove relations and fk constraints

* remove fk constraints migration

* removing get logs endpoint

* update indexer

* remove test routes

* cleaning up migration history

* add contract indexer migration

* fix for edge case

* redo the migration and update schema

* table rename

* updating naming to events and subscriptions

* fix import

* automatically delete logs on subscribe

* add logging

* use lowercase contract address

* adding error handling

* all log values toString

* adding new getContractLogs route

* adding event name to get logs response

* remove export, this feels dirty imo

* lint fix

* add deletedAt to contract subscriptions

* add update endpoints

* add contract log indexes

* add default max blocks to index

* remove logs

* added query limits

* remove debug log

* remove performance logging

* remove debug log

* fix lint

* move chain indexer

* clean migration history

* minor fix

* use unon insted of prevalidation handler

* added default indexer schedule

* fix block time error

* minor fix

* fixing subscribe / unsubscribe bug with deletedAt

* minor fix: update contractSubscrition query with id

* adding base64 package

* updating schema

* adding pagination endpoint

* clean up migrations

* add comment

* minor fix: duplicate indexed transactions

* fixing chainIndexer

* add last block indexed

* Feature/contract tx indexer (#446)

* update docs

* add ContractTransactionReceipts data model

* helper for contractId

* update ContractTransactionReceipts with create / update time

* crud db apis for ContractTransactionReceipts

* adding examples for docs

* lowercase all str fields

* reorg contract subscriptions

* remove console.logs

* add transactionReceiptSchema

* update db apis

* add routes for getting tx receipts

* update imports for routes

* add Contract Transaction routes

* update prisma model with transaction value and receipt

* update transaction value and receipt

* delete transaction receipts on unsubscribe

* add migration for transaction receipts

* update endpoint

* update endpoints

* add value / data to transaction receipt
response

* transaction status bug fix

* cleanup

* parallelize transaction receipts and event logs

* remove log

* bug fix: transaction receipt timestamp

* variable max blocks to index

* parallelize block receipts + perf logging

* update blocks to estimate 100 and parallelize

* Added chainIndexerListener to restart on config DB changes

* remove perf logs

* adding decode failed warning back in

---------

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

Author: IDubuque

package.json

@@ -46,6 +46,8 @@
     "@thirdweb-dev/sdk": "^4.0.46-nightly-3e7f9cb94-20240314012159",
     "@thirdweb-dev/service-utils": "0.4.17",
     "@thirdweb-dev/wallets": "^2.1.5",
+    "@types/base-64": "^1.0.2",
+    "base-64": "^1.0.0",
     "body-parser": "^1.20.2",
     "cookie": "^0.5.0",
     "cookie-parser": "^1.4.6",

src/db/chainIndexers/getChainIndexer.ts

@@ -0,0 +1,49 @@
+import { PrismaTransaction } from "../../schema/prisma";
+import { getPrismaWithPostgresTx } from "../client";
+
+interface GetLastIndexedBlockParams {
+  chainId: number;
+  pgtx?: PrismaTransaction;
+}
+
+export const getLastIndexedBlock = async ({
+  chainId,
+  pgtx,
+}: GetLastIndexedBlockParams) => {
+  const prisma = getPrismaWithPostgresTx(pgtx);
+
+  const indexedChain = await prisma.chainIndexers.findUnique({
+    where: {
+      chainId,
+    },
+  });
+
+  if (indexedChain) {
+    return indexedChain.lastIndexedBlock;
+  }
+};
+
+interface GetBlockForIndexingParams {
+  chainId: number;
+  pgtx?: PrismaTransaction;
+}
+
+export const getBlockForIndexing = async ({
+  chainId,
+  pgtx,
+}: GetBlockForIndexingParams) => {
+  const prisma = getPrismaWithPostgresTx(pgtx);
+
+  const lastIndexedBlock = await prisma.$queryRaw<
+    { lastIndexedBlock: number }[]
+  >`
+    SELECT
+      "lastIndexedBlock"
+    FROM
+      "chain_indexers"
+    WHERE
+      "chainId"=${chainId}
+    FOR UPDATE NOWAIT
+  `;
+  return lastIndexedBlock[0]["lastIndexedBlock"];
+};

src/db/chainIndexers/upsertChainIndexer.ts

@@ -0,0 +1,29 @@
+import { PrismaTransaction } from "../../schema/prisma";
+import { getPrismaWithPostgresTx } from "../client";
+
+interface UpsertChainIndexerParams {
+  chainId: number;
+  currentBlockNumber: number;
+  pgtx?: PrismaTransaction;
+}
+
+export const upsertChainIndexer = async ({
+  chainId,
+  currentBlockNumber,
+  pgtx,
+}: UpsertChainIndexerParams) => {
+  const prisma = getPrismaWithPostgresTx(pgtx);
+  return prisma.chainIndexers.upsert({
+    where: {
+      chainId,
+    },
+    update: {
+      chainId,
+      lastIndexedBlock: currentBlockNumber,
+    },
+    create: {
+      chainId,
+      lastIndexedBlock: currentBlockNumber,
+    },
+  });
+};

src/db/configuration/getConfiguration.ts

@@ -179,6 +179,9 @@ export const getConfiguration = async (): Promise<Config> => {
         minedTxListenerCronSchedule: "*/5 * * * * *",
         maxTxsToUpdate: 50,
         retryTxListenerCronSchedule: "*/15 * * * * *",
+        indexerListenerCronSchedule: "*/5 * * * * *",
+        maxBlocksToIndex: 25,
+        cursorDelaySeconds: 2,
         minEllapsedBlocksBeforeRetry: 12,
         maxFeePerGasForRetries: ethers.utils
           .parseUnits("10000", "gwei")
@@ -205,6 +208,10 @@ export const getConfiguration = async (): Promise<Config> => {
     config = await updateConfiguration({
       accessControlAllowOrigin: mandatoryAllowedCorsUrls.join(","),
     });
+  } else if (!config.indexerListenerCronSchedule) {
+    config = await updateConfiguration({
+      indexerListenerCronSchedule: "*/5 * * * * *",
+    });
   }
 
   return withWalletConfig(config);

src/db/contractEventLogs/createContractEventLogs.ts

@@ -0,0 +1,35 @@
+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[];
+}
+
+export const bulkInsertContractEventLogs = async ({
+  pgtx,
+  logs,
+}: BulkInsertContractLogsParams) => {
+  const prisma = getPrismaWithPostgresTx(pgtx);
+  return await prisma.contractEventLogs.createMany({
+    data: logs,
+    skipDuplicates: true,
+  });
+};

src/db/contractEventLogs/deleteContractEventLogs.ts

@@ -0,0 +1,18 @@
+import { prisma } from "../client";
+
+interface DeleteContractEventLogsParams {
+  chainId: number;
+  contractAddress: string;
+}
+
+export const deleteContractEventLogs = async ({
+  chainId,
+  contractAddress,
+}: DeleteContractEventLogsParams) => {
+  return prisma.contractEventLogs.deleteMany({
+    where: {
+      chainId,
+      contractAddress,
+    },
+  });
+};