feat: Add subscription delay envvar (#513)

* feat: Add subscription delay envvar

* removed +1 of blockNumber

---------

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

Author: arcoraven

src/utils/env.ts

@@ -67,6 +67,10 @@ export const env = createEnv({
     SDK_BATCH_TIME_LIMIT: z.coerce.number().default(0),
     SDK_BATCH_SIZE_LIMIT: z.coerce.number().default(100),
     ENABLE_KEYPAIR_AUTH: boolSchema("false"),
+    CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS: z.coerce
+      .number()
+      .nonnegative()
+      .default(0),
   },
   clientPrefix: "NEVER_USED",
   client: {},
@@ -88,6 +92,8 @@ export const env = createEnv({
     SDK_BATCH_TIME_LIMIT: process.env.SDK_BATCH_TIME_LIMIT,
     SDK_BATCH_SIZE_LIMIT: process.env.SDK_BATCH_SIZE_LIMIT,
     ENABLE_KEYPAIR_AUTH: process.env.ENABLE_KEYPAIR_AUTH,
+    CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS:
+      process.env.CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS,
   },
   onValidationError: (error: ZodError) => {
     console.error(

src/worker/indexers/chainIndexerRegistry.ts

@@ -1,5 +1,6 @@
 import cron from "node-cron";
 import { getConfig } from "../../utils/cache/getConfig";
+import { env } from "../../utils/env";
 import { getBlockTimeSeconds } from "../../utils/indexer/getBlockTime";
 import { logger } from "../../utils/logger";
 import { createChainIndexerTask } from "../tasks/chainIndexer";
@@ -18,6 +19,8 @@ export const addChainIndexer = async (chainId: number) => {
 
   let processStarted = false;
   const config = await getConfig();
+
+  // Estimate block time.
   const blockTimeSeconds = await getBlockTimeSeconds(chainId);
 
   const blocksIn5Seconds = Math.round((1 / blockTimeSeconds) * 5);
@@ -26,7 +29,17 @@ export const addChainIndexer = async (chainId: number) => {
     blocksIn5Seconds * 4,
   );
 
-  const handler = await createChainIndexerTask(chainId, maxBlocksToIndex);
+  // Compute block offset based on delay.
+  // Example: 10s delay with a 3s block time = 4 blocks offset
+  const toBlockOffset = env.CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS
+    ? Math.ceil(env.CONTRACT_SUBSCRIPTIONS_DELAY_SECONDS / blockTimeSeconds)
+    : 0;
+
+  const handler = await createChainIndexerTask({
+    chainId,
+    maxBlocksToIndex,
+    toBlockOffset,
+  });
 
   const cronSchedule = createScheduleSeconds(
     Math.max(Math.round(blockTimeSeconds), 1),

src/worker/tasks/chainIndexer.ts

@@ -268,10 +268,13 @@ const indexTransactionReceipts = async ({
   }
 };
 
-export const createChainIndexerTask = async (
-  chainId: number,
-  maxBlocksToIndex: number,
-) => {
+export const createChainIndexerTask = async (args: {
+  chainId: number;
+  maxBlocksToIndex: number;
+  toBlockOffset: number;
+}) => {
+  const { chainId, maxBlocksToIndex, toBlockOffset } = args;
+
   const chainIndexerTask = async () => {
     try {
       await prisma.$transaction(
@@ -287,7 +290,8 @@ export const createChainIndexerTask = async (
           const sdk = await getSdk({ chainId });
 
           const provider = sdk.getProvider();
-          const currentBlockNumber = await provider.getBlockNumber();
+          const currentBlockNumber =
+            (await provider.getBlockNumber()) - toBlockOffset;
 
           // check if up-to-date
           if (lastIndexedBlock >= currentBlockNumber) {
@@ -296,8 +300,8 @@ export const createChainIndexerTask = async (
 
           // limit max block numbers
           let toBlockNumber = currentBlockNumber;
-          if (currentBlockNumber - (lastIndexedBlock + 1) > maxBlocksToIndex) {
-            toBlockNumber = lastIndexedBlock + 1 + maxBlocksToIndex;
+          if (currentBlockNumber - lastIndexedBlock > maxBlocksToIndex) {
+            toBlockNumber = lastIndexedBlock + maxBlocksToIndex;
           }
 
           const subscribedContracts = await getContractSubscriptionsByChainId(
@@ -315,14 +319,14 @@ export const createChainIndexerTask = async (
             indexContractEvents({
               pgtx,
               chainId,
-              fromBlockNumber: lastIndexedBlock + 1,
+              fromBlockNumber: lastIndexedBlock,
               toBlockNumber,
               subscribedContractAddresses,
             }),
             indexTransactionReceipts({
               pgtx,
               chainId,
-              fromBlockNumber: lastIndexedBlock + 1,
+              fromBlockNumber: lastIndexedBlock,
               toBlockNumber,
               subscribedContractAddresses,
             }),