Remove eventOrigin, introduce context chains

Author: Segfaultd

codegenerator/cli/npm/envio/src/Internal.gen.ts

@@ -42,7 +42,7 @@ export type entityHandlerContext<entity> = {
   readonly deleteUnsafe: (_1:string) => void
 };
 
-export type eventOrigin = "Historical" | "Live";
+export type chainInfo = { readonly isReady: boolean };
 
 export type genericHandlerWithLoader<loader,handler,eventFilters> = {
   readonly loader: loader; 

codegenerator/cli/npm/envio/src/Internal.res

@@ -55,12 +55,16 @@ type entityHandlerContext<'entity> = {
 }
 
 @genType
-type eventOrigin = Historical | Live
+type chainInfo = {
+  isReady: bool,
+}
+
+type chains = dict<chainInfo>
 
 type loaderReturn
 type handlerContext = private {
   isPreload: bool,
-  eventOrigin: eventOrigin,
+  chains: chains,
 }
 type handlerArgs = {
   event: event,

codegenerator/cli/templates/dynamic/codegen/src/TestHelpers_MockDb.res.hbs

@@ -405,20 +405,24 @@ and makeProcessEvents = (mockDb: t, ~chainId=?) => async (
     }
   }
 
+  // Create a mock chains state where the processing chain is ready (simulating "Live" mode)
+  let chains = Js.Dict.empty()
+  chains->Js.Dict.set(processingChainId->Int.toString, {Internal.isReady: true})
+
   try {
     await items->EventProcessing.preloadBatchOrThrow(
       ~loadManager,
       ~persistence,
       ~inMemoryStore,
-      ~eventOrigin=Internal.Live,
+      ~chains,
     )
     await items->EventProcessing.runBatchHandlersOrThrow(
       ~inMemoryStore,
       ~loadManager,
       ~config,
       ~shouldSaveHistory=false,
       ~shouldBenchmark=false,
-      ~eventOrigin=Internal.Live,
+      ~chains,
     )
   } catch {
   | EventProcessing.ProcessingError({message, exn, item}) =>

codegenerator/cli/templates/dynamic/codegen/src/Types.res.hbs

@@ -29,7 +29,7 @@ type handlerContext = {
   log: Envio.logger,
   effect: 'input 'output. (Envio.effect<'input, 'output>, 'input) => promise<'output>,
   isPreload: bool,
-  eventOrigin: Internal.eventOrigin,
+  chains: Internal.chains,
   {{#each entities as | entity |}}
   @as("{{entity.name.original}}") {{entity.name.uncapitalized}}: entityHandlerContext<Entities.{{entity.name.capitalized}}.t, Entities.{{entity.name.capitalized}}.indexedFieldOperations>,
   {{/each}}
@@ -50,7 +50,7 @@ type loaderContext = {
   log: Envio.logger,
   effect: 'input 'output. (Envio.effect<'input, 'output>, 'input) => promise<'output>,
   isPreload: bool,
-  eventOrigin: Internal.eventOrigin,
+  chains: Internal.chains,
   {{#each entities as | entity |}}
   @as("{{entity.name.original}}") {{entity.name.uncapitalized}}: entityLoaderContext<Entities.{{entity.name.capitalized}}.t, Entities.{{entity.name.capitalized}}.indexedFieldOperations>,
   {{/each}}
@@ -63,7 +63,7 @@ type entityHandlerContext<'entity> = Internal.entityHandlerContext<'entity>
 type handlerContext = {
   log: Envio.logger,
   effect: 'input 'output. (Envio.effect<'input, 'output>, 'input) => promise<'output>,
-  eventOrigin: Internal.eventOrigin,
+  chains: Internal.chains,
   {{#each entities as | entity |}}
   @as("{{entity.name.original}}") {{entity.name.uncapitalized}}: entityHandlerContext<Entities.{{entity.name.capitalized}}.t>,
   {{/each}}

codegenerator/cli/templates/static/codegen/src/EventProcessing.res

@@ -1,9 +1,20 @@
 open Belt
 
-let allChainsEventsProcessedToEndblock = (chainFetchers: ChainMap.t<ChainFetcher.t>) => {
+let computeChainsState = (chainFetchers: ChainMap.t<ChainFetcher.t>): Internal.chains => {
+  let chains = Js.Dict.empty()
+
   chainFetchers
-  ->ChainMap.values
-  ->Array.every(cf => cf->ChainFetcher.hasProcessedToEndblock)
+  ->ChainMap.entries
+  ->Array.forEach(((chain, chainFetcher)) => {
+    let chainId = chain->ChainMap.Chain.toChainId->Int.toString
+    let isReady = chainFetcher->ChainFetcher.hasProcessedToEndblock
+
+    chains->Js.Dict.set(chainId, {
+      Internal.isReady: isReady,
+    })
+  })
+
+  chains
 }
 
 let convertFieldsToJson = (fields: option<dict<unknown>>) => {
@@ -87,7 +98,7 @@ let runEventHandlerOrThrow = async (
   ~persistence,
   ~shouldSaveHistory,
   ~shouldBenchmark,
-  ~eventOrigin: Internal.eventOrigin,
+  ~chains: Internal.chains,
 ) => {
   let eventItem = item->Internal.castUnsafeEventItem
 
@@ -106,7 +117,7 @@ let runEventHandlerOrThrow = async (
             persistence,
             shouldSaveHistory,
             isPreload: false,
-            eventOrigin,
+            chains,
           }),
         }: Internal.handlerArgs
       ),
@@ -139,7 +150,7 @@ let runHandlerOrThrow = async (
   ~config: Config.t,
   ~shouldSaveHistory,
   ~shouldBenchmark,
-  ~eventOrigin: Internal.eventOrigin,
+  ~chains: Internal.chains,
 ) => {
   switch item {
   | Block({onBlockConfig: {handler, chainId}, blockNumber}) =>
@@ -158,7 +169,7 @@ let runHandlerOrThrow = async (
               persistence: config.persistence,
               shouldSaveHistory,
               isPreload: false,
-              eventOrigin,
+              chains,
             }),
           }: Internal.onBlockArgs
         ),
@@ -183,7 +194,7 @@ let runHandlerOrThrow = async (
           ~persistence=config.persistence,
           ~shouldSaveHistory,
           ~shouldBenchmark,
-          ~eventOrigin,
+          ~chains,
         )
       | None => ()
       }
@@ -200,7 +211,7 @@ let preloadBatchOrThrow = async (
   ~loadManager,
   ~persistence,
   ~inMemoryStore,
-  ~eventOrigin: Internal.eventOrigin,
+  ~chains: Internal.chains,
 ) => {
   // On the first run of loaders, we don't care about the result,
   // whether it's an error or a return type.
@@ -224,7 +235,7 @@ let preloadBatchOrThrow = async (
                   persistence,
                   isPreload: true,
                   shouldSaveHistory: false,
-                  eventOrigin,
+                  chains,
                 }),
               })->Promise.silentCatch,
               // Must have Promise.catch as well as normal catch,
@@ -250,7 +261,7 @@ let preloadBatchOrThrow = async (
                 persistence,
                 isPreload: true,
                 shouldSaveHistory: false,
-                eventOrigin,
+                chains,
               }),
             })->Promise.silentCatch,
           )
@@ -269,7 +280,7 @@ let runBatchHandlersOrThrow = async (
   ~config,
   ~shouldSaveHistory,
   ~shouldBenchmark,
-  ~eventOrigin: Internal.eventOrigin,
+  ~chains: Internal.chains,
 ) => {
   for i in 0 to eventBatch->Array.length - 1 {
     let item = eventBatch->Js.Array2.unsafe_get(i)
@@ -280,7 +291,7 @@ let runBatchHandlersOrThrow = async (
       ~config,
       ~shouldSaveHistory,
       ~shouldBenchmark,
-      ~eventOrigin,
+      ~chains,
     )
   }
 }
@@ -319,12 +330,8 @@ let processEventBatch = async (
   ~config: Config.t,
   ~chainFetchers: ChainMap.t<ChainFetcher.t>,
 ) => {
-  // Determine event origin based on whether all chains have caught up
-  let eventOrigin: Internal.eventOrigin = if chainFetchers->allChainsEventsProcessedToEndblock {
-    Live
-  } else {
-    Historical
-  }
+  // Compute chains state for this batch
+  let chains: Internal.chains = chainFetchers->computeChainsState
   let batchSize = items->Array.length
   let byChain = Js.Dict.empty()
   progressedChains->Js.Array2.forEach(data => {
@@ -350,7 +357,7 @@ let processEventBatch = async (
   try {
     let timeRef = Hrtime.makeTimer()
 
-    await items->preloadBatchOrThrow(~loadManager, ~persistence=config.persistence, ~inMemoryStore, ~eventOrigin)
+    await items->preloadBatchOrThrow(~loadManager, ~persistence=config.persistence, ~inMemoryStore, ~chains)
 
     let elapsedTimeAfterLoaders = timeRef->Hrtime.timeSince->Hrtime.toMillis->Hrtime.intFromMillis
 
@@ -360,7 +367,7 @@ let processEventBatch = async (
       ~config,
       ~shouldSaveHistory=config->Config.shouldSaveHistory(~isInReorgThreshold),
       ~shouldBenchmark=Env.Benchmark.shouldSaveData,
-      ~eventOrigin,
+      ~chains,
     )
 
     let elapsedTimeAfterProcessing =

codegenerator/cli/templates/static/codegen/src/UserContext.res

@@ -24,7 +24,7 @@ type contextParams = {
   persistence: Persistence.t,
   isPreload: bool,
   shouldSaveHistory: bool,
-  eventOrigin: Internal.eventOrigin,
+  chains: Internal.chains,
 }
 
 let rec initEffect = (params: contextParams) => (
@@ -237,7 +237,7 @@ let handlerTraps: Utils.Proxy.traps<contextParams> = {
       )
 
     | "isPreload" => params.isPreload->Utils.magic
-    | "eventOrigin" => params.eventOrigin->Utils.magic
+    | "chains" => params.chains->Utils.magic
     | _ =>
       switch Entities.byName->Utils.Dict.dangerouslyGetNonOption(prop) {
       | Some(entityConfig) =>
@@ -248,7 +248,7 @@ let handlerTraps: Utils.Proxy.traps<contextParams> = {
           loadManager: params.loadManager,
           persistence: params.persistence,
           shouldSaveHistory: params.shouldSaveHistory,
-          eventOrigin: params.eventOrigin,
+          chains: params.chains,
           entityConfig,
         }
         ->Utils.Proxy.make(entityTraps)

codegenerator/cli/templates/static/codegen/src/globalState/GlobalState.res

@@ -629,9 +629,11 @@ let actionReducer = (state: t, action: action) => {
       processedBatches: state.processedBatches + 1,
     }
 
-    let shouldExit = EventProcessing.allChainsEventsProcessedToEndblock(
-      state.chainManager.chainFetchers,
-    )
+    // Check if all chains have reached their end blocks
+    let chains = EventProcessing.computeChainsState(state.chainManager.chainFetchers)
+    let allChainsReady = chains->Js.Dict.values->Array.every(chainInfo => chainInfo.isReady)
+
+    let shouldExit = allChainsReady
       ? {
           // state.config.persistence.storage
           Logging.info("All chains are caught up to end blocks.")

scenarios/test_codegen/pnpm-lock.yaml

@@ -141,16 +141,20 @@ Handlers.Gravatar.TestEvent.handler(async _ => {
   ()
 })
 
-// Test eventOrigin accessibility - exposed for testing
-let lastEmptyEventOrigin: ref<option<Internal.eventOrigin>> = ref(None)
+// Test chains accessibility - exposed for testing
+// Instead of a single eventOrigin enum, we store the entire chains dict
+let lastEmptyEventChains: ref<option<Internal.chains>> = ref(None)
 
 Handlers.Gravatar.EmptyEvent.handler(async ({context}) => {
-  // This handler tests that eventOrigin is accessible in the context
-  // It will be Historical during sync and Live during live indexing
-  lastEmptyEventOrigin := Some(context.eventOrigin)
-  // Log it so we can verify it's working
-  switch context.eventOrigin {
-  | Historical => context.log.debug("Processing historical event")
-  | Live => context.log.debug("Processing live event")
-  }
+  // This handler tests that chains state is accessible in the context
+  // Chains will have isReady: false during sync and isReady: true during live indexing
+  lastEmptyEventChains := Some(context.chains)
+
+  // Log chain states for verification
+  context.chains
+  ->Js.Dict.entries
+  ->Belt.Array.forEach(((chainId, chainInfo)) => {
+    let status = chainInfo.isReady ? "ready (live)" : "syncing (historical)"
+    context.log.debug(`Chain ${chainId} status: ${status}`)
+  })
 })