Wildcard partition

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

@@ -48,6 +48,19 @@ let shouldPreRegisterDynamicContracts = (chainConfig: chainConfig) => {
   })
 }
 
+let hasWildcard = (chainConfig: chainConfig) => {
+  chainConfig.contracts->Array.some(contract => {
+    contract.events->Array.some(event => {
+      let module(Event) = event
+
+      let {isWildcard} =
+        Event.handlerRegister->Types.HandlerTypes.Register.getEventOptions
+
+      isWildcard
+    })
+  })
+}
+
 type historyFlag = FullHistory | MinHistory
 type rollbackFlag = RollbackOnReorg | NoRollback
 type historyConfig = {rollbackFlag: rollbackFlag, historyFlag: historyFlag}

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

@@ -71,9 +71,9 @@ let makeAppState = (globalState: GlobalState.t): EnvioInkApp.appState => {
     globalState.chainManager.chainFetchers
     ->ChainMap.values
     ->Array.map(cf => {
-      let {numEventsProcessed, fetchState, numBatchesFetched} = cf
+      let {numEventsProcessed, partitionedFetchState, numBatchesFetched} = cf
       let latestFetchedBlockNumber = PartitionedFetchState.getLatestFullyFetchedBlock(
-        fetchState,
+        partitionedFetchState,
       ).blockNumber
       let hasProcessedToEndblock = cf->ChainFetcher.hasProcessedToEndblock
       let currentBlockHeight =

codegenerator/cli/templates/static/codegen/src/eventFetching/ChainFetcher.res

@@ -10,7 +10,7 @@ type processingFilter = {
 type addressToDynContractLookup = dict<TablesStatic.DynamicContractRegistry.t>
 type t = {
   logger: Pino.t,
-  fetchState: PartitionedFetchState.t,
+  partitionedFetchState: PartitionedFetchState.t,
   sourceManager: SourceManager.t,
   chainConfig: Config.chainConfig,
   //The latest known block of the chain
@@ -51,22 +51,25 @@ let make = (
   let module(ChainWorker) = chainConfig.chainWorker
   logger->Logging.childInfo("Initializing ChainFetcher with " ++ ChainWorker.name ++ " worker")
 
-  let fetchState = PartitionedFetchState.make(
+  let isPreRegisteringDynamicContracts = dynamicContractPreRegistration->Option.isSome
+  let partitionedFetchState = PartitionedFetchState.make(
     ~maxAddrInPartition,
     ~staticContracts,
     ~dynamicContractRegistrations,
     ~startBlock,
     ~endBlock,
+    ~isPreRegisteringDynamicContracts,
+    ~hasWildcard=Config.hasWildcard(chainConfig),
     ~logger,
   )
 
   {
     logger,
     chainConfig,
-    sourceManager: SourceManager.make(~maxPartitionConcurrency=Env.maxPartitionConcurrency, ~logger),
+    sourceManager: SourceManager.make(~maxPartitionConcurrency=Env.maxPartitionConcurrency, ~endBlock, ~logger),
     lastBlockScannedHashes,
     currentBlockHeight: 0,
-    fetchState,
+    partitionedFetchState,
     dbFirstEventBlockNumber,
     latestProcessedBlock,
     timestampCaughtUpToHeadOrEndblock,
@@ -303,7 +306,7 @@ let applyProcessingFilters = (
 //any that meet the cleanup condition
 let cleanUpProcessingFilters = (
   processingFilters: array<processingFilter>,
-  ~fetchState as {partitions}: PartitionedFetchState.t,
+  ~partitionedFetchState as {partitions}: PartitionedFetchState.t,
 ) => {
   switch processingFilters->Array.keep(processingFilter =>
     partitions->Array.reduce(false, (accum, partition) => {
@@ -333,7 +336,7 @@ let updateFetchState = (
   | Some(processingFilters) => fetchedEvents->applyProcessingFilters(~processingFilters)
   }
 
-  self.fetchState
+  self.partitionedFetchState
   ->PartitionedFetchState.update(
     ~id,
     ~latestFetchedBlock={
@@ -344,12 +347,12 @@ let updateFetchState = (
     ~currentBlockHeight,
     ~chain=self.chainConfig.chain,
   )
-  ->Result.map(fetchState => {
+  ->Result.map(partitionedFetchState => {
     {
       ...self,
-      fetchState,
+      partitionedFetchState,
       processingFilters: switch self.processingFilters {
-      | Some(processingFilters) => processingFilters->cleanUpProcessingFilters(~fetchState)
+      | Some(processingFilters) => processingFilters->cleanUpProcessingFilters(~partitionedFetchState)
       | None => None
       },
     }
@@ -368,7 +371,7 @@ let hasProcessedToEndblock = (self: t) => {
 }
 
 let hasNoMoreEventsToProcess = (self: t, ~hasArbQueueEvents) => {
-  !hasArbQueueEvents && self.fetchState->PartitionedFetchState.queueSize === 0
+  !hasArbQueueEvents && self.partitionedFetchState->PartitionedFetchState.queueSize === 0
 }
 
 /**
@@ -424,12 +427,12 @@ let getLastScannedBlockData = lastBlockData => {
 }
 
 let isFetchingAtHead = (chainFetcher: t) =>
-  chainFetcher.fetchState->PartitionedFetchState.isFetchingAtHead
+  chainFetcher.partitionedFetchState->PartitionedFetchState.isFetchingAtHead
 
 let getFirstEventBlockNumber = (chainFetcher: t) =>
   Utils.Math.minOptInt(
     chainFetcher.dbFirstEventBlockNumber,
-    chainFetcher.fetchState->PartitionedFetchState.getFirstEventBlockNumber,
+    chainFetcher.partitionedFetchState->PartitionedFetchState.getFirstEventBlockNumber,
   )
 
 let isPreRegisteringDynamicContracts = (chainFetcher: t) =>

codegenerator/cli/templates/static/codegen/src/eventFetching/ChainManager.res

@@ -326,8 +326,8 @@ let getFetchStateWithData = (self: t, ~shouldDeepCopy=false): ChainMap.t<fetchSt
   self.chainFetchers->ChainMap.map(cf => {
     {
       partitionedFetchState: shouldDeepCopy
-        ? cf.fetchState->PartitionedFetchState.copy
-        : cf.fetchState,
+        ? cf.partitionedFetchState->PartitionedFetchState.copy
+        : cf.partitionedFetchState,
       heighestBlockBelowThreshold: cf->ChainFetcher.getHeighestBlockBelowThreshold,
     }
   })
@@ -417,7 +417,7 @@ let createBatch = (self: t, ~maxBatchSize: int, ~onlyBelowReorgThreshold: bool)
       ->ChainMap.values
       ->Array.map(fetcher => (
         fetcher.chainConfig.chain->ChainMap.Chain.toString,
-        fetcher.fetchState->PartitionedFetchState.queueSize,
+        fetcher.partitionedFetchState->PartitionedFetchState.queueSize,
       ))
       ->Array.concat([("arbitrary", self.arbitraryEventQueue->Array.length)])
       ->Js.Dict.fromArray