diff --git a/packages/cli/templates/static/shared/.claude/skills/indexer-testing/SKILL.md b/packages/cli/templates/static/shared/.claude/skills/indexer-testing/SKILL.md index ee738dc4bc..8b0393f891 100644 --- a/packages/cli/templates/static/shared/.claude/skills/indexer-testing/SKILL.md +++ b/packages/cli/templates/static/shared/.claude/skills/indexer-testing/SKILL.md @@ -99,6 +99,17 @@ await indexer.EntityName.getOrThrow("id"); // throws if not found await indexer.EntityName.getAll(); // returns all entities of this type ``` +## Test isolation + +Each `createTestIndexer()` has its own entity store; within one indexer, entity +state and block progress persist across `process()` calls (each continues where +the last stopped). + +Tests run in-process, so **module-level** state in your handler code (top-level +`let`/`const`, memoized clients, effect caches) is shared across every indexer +and test in the file and persists between them. Reset it yourself (e.g. in a +`beforeEach`) if a test depends on it. + ## result.changes `result.changes` is an array of per-block change objects. Each entry has `block`, `chainId`, `eventsProcessed`, plus entity names as keys with `sets` arrays of created/updated entities. Dynamic contract registrations appear under `addresses.sets`. diff --git a/packages/envio/package.json b/packages/envio/package.json index 0c6edb1e9c..ff572db4bb 100644 --- a/packages/envio/package.json +++ b/packages/envio/package.json @@ -74,6 +74,7 @@ "tsx": "4.21.0" }, "devDependencies": { - "rescript": "12.2.0" + "rescript": "12.2.0", + "vitest": "4.1.0" } } diff --git a/packages/envio/src/Api.res b/packages/envio/src/Api.res index 7ff60f545e..11b02006e8 100644 --- a/packages/envio/src/Api.res +++ b/packages/envio/src/Api.res @@ -4,12 +4,5 @@ let indexer: unknown = Main.getGlobalIndexer() let createTestIndexer: unit => unknown = () => { - let workerPath = - NodeJs.Path.join( - NodeJs.Path.getDirname(NodeJs.ImportMeta.importMeta), - "TestIndexerWorker.res.mjs", - )->NodeJs.Path.toString - TestIndexer.makeCreateTestIndexer(~config=Config.load(), ~workerPath)()->( - Utils.magic: TestIndexer.t<'a> => unknown - ) + TestIndexer.createTestIndexer()->(Utils.magic: TestIndexer.t<'a> => unknown) } diff --git a/packages/envio/src/ExitOnCaughtUp.res b/packages/envio/src/ExitOnCaughtUp.res index 1c0399b338..5d989dc504 100644 --- a/packages/envio/src/ExitOnCaughtUp.res +++ b/packages/envio/src/ExitOnCaughtUp.res @@ -10,8 +10,12 @@ let run = async (state: IndexerState.t) => { ->IndexerState.simulateDeadInputTracker ->Option.flatMap(SimulateDeadInputTracker.failureMessage) { | None => - Logging.info("Exiting with success") - NodeJs.process->NodeJs.exitWithCode(Success) + switch state->IndexerState.onExit { + | Some(onExit) => onExit() + | None => + Logging.info("Exiting with success") + NodeJs.process->NodeJs.exitWithCode(Success) + } | Some(message) => state->IndexerState.errorExit(ErrorHandling.make(Utils.Error.make(message))) } } diff --git a/packages/envio/src/IndexerState.res b/packages/envio/src/IndexerState.res index a4ef3bbea7..a838714d5c 100644 --- a/packages/envio/src/IndexerState.res +++ b/packages/envio/src/IndexerState.res @@ -115,6 +115,11 @@ type t = { exitAfterFirstEventBlock: bool, // The single fatal-error handler. onError: ErrorHandling.t => unit, + // Invoked once when the indexer catches up and would otherwise exit the + // process. `None` keeps the production behavior (exit the process); the + // in-process test runner injects a callback that resolves its run promise + // instead, so a caught-up run doesn't kill the test process. + onExit: option unit>, // Set once on any fatal error. Every loop checks it to stop iterating and // every launch skips when it's set, so a single failure quiesces the indexer. mutable isStopped: bool, @@ -149,6 +154,7 @@ let make = ( ~shouldUseTui=false, ~exitAfterFirstEventBlock=false, ~onError: ErrorHandling.t => unit, + ~onExit=?, ) => { let chainMetaThrottler = { let intervalMillis = Env.ThrottleWrites.chainMetadataIntervalMillis @@ -194,6 +200,7 @@ let make = ( keepProcessAlive: isDevelopmentMode || shouldUseTui, exitAfterFirstEventBlock, onError, + onExit, isStopped: false, epoch: 0, simulateDeadInputTracker: SimulateDeadInputTracker.makeFromConfig(config), @@ -229,6 +236,7 @@ let makeFromDbState = ( ~reducedPollingInterval=?, ~targetBufferSize=CrossChainState.calculateTargetBufferSize(), ~onError, + ~onExit=?, ) => { let isInReorgThreshold = if initialState.cleanRun { false @@ -280,6 +288,7 @@ let makeFromDbState = ( ~shouldUseTui, ~exitAfterFirstEventBlock, ~onError, + ~onExit?, ) initialState.cache->Utils.Dict.forEach(({effectName, count, scope}) => { state.effectState->EffectState.setUnregisteredCacheCount(~effectName, ~scope, ~count) @@ -415,6 +424,7 @@ let indexerStartTime = (state: t) => state.indexerStartTime let loadManager = (state: t) => state.loadManager let keepProcessAlive = (state: t) => state.keepProcessAlive let exitAfterFirstEventBlock = (state: t) => state.exitAfterFirstEventBlock +let onExit = (state: t) => state.onExit let isStopped = (state: t) => state.isStopped let epoch = (state: t) => state.epoch let lastPrunedAtMillis = (state: t) => state.lastPrunedAtMillis diff --git a/packages/envio/src/IndexerState.resi b/packages/envio/src/IndexerState.resi index 25401de6da..2c864d7f20 100644 --- a/packages/envio/src/IndexerState.resi +++ b/packages/envio/src/IndexerState.resi @@ -30,6 +30,7 @@ let make: ( ~shouldUseTui: bool=?, ~exitAfterFirstEventBlock: bool=?, ~onError: ErrorHandling.t => unit, + ~onExit: unit => unit=?, ) => t let makeFromDbState: ( @@ -43,6 +44,7 @@ let makeFromDbState: ( ~reducedPollingInterval: int=?, ~targetBufferSize: int=?, ~onError: ErrorHandling.t => unit, + ~onExit: unit => unit=?, ) => t let unexpectedErrorMsg: string @@ -96,6 +98,7 @@ let indexerStartTime: t => Date.t let loadManager: t => LoadManager.t let keepProcessAlive: t => bool let exitAfterFirstEventBlock: t => bool +let onExit: t => option unit> let isStopped: t => bool let epoch: t => int let lastPrunedAtMillis: t => dict diff --git a/packages/envio/src/TestIndexer.res b/packages/envio/src/TestIndexer.res index babfdb9cc1..042a440e7a 100644 --- a/packages/envio/src/TestIndexer.res +++ b/packages/envio/src/TestIndexer.res @@ -70,48 +70,31 @@ let getIndexingAddressesByChain = (state: testIndexerState): dict< byChain } -let handleLoad = (state: testIndexerState, ~tableName: string, ~filter: EntityFilter.t): JSON.t => { +let handleLoad = (state: testIndexerState, ~tableName: string, ~filter: EntityFilter.t): array< + Internal.entity, +> => { // Loads for non-entity tables (e.g. effect caches `envio_effect_`) reach // here too. TestIndexer never persists those, so there's nothing to return — // an empty result makes the effect recompute instead of crashing on a missing // entityConfig. switch state.entityConfigs->Dict.get(tableName) { - | None => []->JSON.Encode.array - | Some(entityConfig) => + | None => [] + | Some(_) => let entityDict = state.entities->Dict.get(tableName)->Option.getOr(Dict.make()) - let results = [] - - // Field values arrive as JSON from the worker boundary, so parse them - // with the field's schema before comparing. This properly handles - // bigint and BigDecimal comparisons - let parseLeaf = (~fieldName, ~fieldValue: unknown, ~isArray): unknown => { - let queryField = switch entityConfig.table->Table.queryFields->Dict.get(fieldName) { - | Some(queryField) => queryField - | None => JsError.throwWithMessage(`Field ${fieldName} not found in entity ${tableName}`) - } - fieldValue->S.convertOrThrow(isArray ? queryField.arrayFieldSchema : queryField.fieldSchema) - } - let filter = filter->EntityFilter.mapValues(~mapValue=parseLeaf) - entityDict ->Dict.valuesToArray - ->Array.forEach(entity => { - // Cast entity to dict of field values (same approach as InMemoryTable) + ->Array.filter(entity => { + // The store holds decoded entities and the filter carries decoded values, + // so compare directly (same approach as InMemoryTable) — no JSON round-trip. let entityAsDict = entity->(Utils.magic: Internal.entity => dict) - if filter->EntityFilter.matches(~entity=entityAsDict) { - // Serialize entity back to JSON for worker thread - let jsonEntity = entity->S.reverseConvertToJsonOrThrow(entityConfig.schema) - results->Array.push(jsonEntity)->ignore - } + filter->EntityFilter.matches(~entity=entityAsDict) }) - - results->JSON.Encode.array } } let handleWriteBatch = ( state: testIndexerState, - ~updatedEntities: array, + ~updatedEntities: array, ~checkpointIds: array, ~checkpointChainIds: array, ~checkpointBlockNumbers: array, @@ -121,7 +104,8 @@ let handleWriteBatch = ( // checkpointId -> entityName -> entityChange let changesByCheckpoint: dict> = Dict.make() - updatedEntities->Array.forEach(({entityName, changes}) => { + updatedEntities->Array.forEach(({entityConfig, changes}: Persistence.updatedEntity) => { + let entityName = entityConfig.name let entityDict = switch state.entities->Dict.get(entityName) { | Some(dict) => dict | None => @@ -129,57 +113,35 @@ let handleWriteBatch = ( state.entities->Dict.set(entityName, dict) dict } - let entityConfig = state.entityConfigs->Dict.getUnsafe(entityName) - let processChange = (change: TestIndexerProxyStorage.serializableChange) => { + let entityChangeFor = checkpointId => { + let checkpointKey = checkpointId->BigInt.toString + let entityChanges = switch changesByCheckpoint->Dict.get(checkpointKey) { + | Some(changes) => changes + | None => + let changes = Dict.make() + changesByCheckpoint->Dict.set(checkpointKey, changes) + changes + } + switch entityChanges->Dict.get(entityName) { + | Some(change) => change + | None => + let change = {sets: [], deleted: []} + entityChanges->Dict.set(entityName, change) + change + } + } + + let processChange = (change: Change.t) => { switch change { | Set({entityId, entity, checkpointId}) => - // Parse entity immediately to store decoded values for proper comparisons - // (bigint/BigDecimal need actual values, not JSON strings) - let parsedEntity = entity->S.parseOrThrow(entityConfig.schema) - - // Update entities dict with parsed entity for load operations - entityDict->Dict.set(entityId, parsedEntity) - - // Track change by checkpoint - let checkpointKey = checkpointId->BigInt.toString - let entityChanges = switch changesByCheckpoint->Dict.get(checkpointKey) { - | Some(changes) => changes - | None => - let changes = Dict.make() - changesByCheckpoint->Dict.set(checkpointKey, changes) - changes - } - let entityChange = switch entityChanges->Dict.get(entityName) { - | Some(change) => change - | None => - let change = {sets: [], deleted: []} - entityChanges->Dict.set(entityName, change) - change - } - entityChange.sets->Array.push(parsedEntity->Utils.magic)->ignore - + // The store keeps decoded entities so load comparisons (bigint / + // BigDecimal) work on real values. + entityDict->Dict.set(entityId, entity) + entityChangeFor(checkpointId).sets->Array.push(entity->Utils.magic)->ignore | Delete({entityId, checkpointId}) => - // Update entities dict for load operations Dict.delete(entityDict->Obj.magic, entityId) - - // Track change by checkpoint - let checkpointKey = checkpointId->BigInt.toString - let entityChanges = switch changesByCheckpoint->Dict.get(checkpointKey) { - | Some(changes) => changes - | None => - let changes = Dict.make() - changesByCheckpoint->Dict.set(checkpointKey, changes) - changes - } - let entityChange = switch entityChanges->Dict.get(entityName) { - | Some(change) => change - | None => - let change = {sets: [], deleted: []} - entityChanges->Dict.set(entityName, change) - change - } - entityChange.deleted->Array.push(entityId)->ignore + entityChangeFor(checkpointId).deleted->Array.push(entityId)->ignore } } @@ -285,8 +247,6 @@ let makeInitialState = ( chains, checkpointId: InternalTable.Checkpoints.initialCheckpointId, reorgCheckpoints: [], - // TestIndexer fakes the resume path; mirror what Main.start passes as - // ~envioInfo so the compat check always sees an empty diff. envioInfo: Some(Config.getPublicConfigJson()->Config.stripSensitiveData), } } @@ -406,6 +366,18 @@ let parseBlockRange = ( {startBlock, endBlock} } +// The store owns its entities. Copy on the boundary with user code — both when +// handing one out (get/getAll/getOrThrow) and when taking one in (set) — so a +// user mutating a returned entity, or an object they passed to `set`, can't +// corrupt the in-memory store. The copy is shallow (matching InMemoryTable): +// scalar fields (string/bigint/BigDecimal) are immutable, but array-valued +// fields still share the backing array, so in-place mutation of those leaks. +let copyEntity = (entity: Internal.entity): Internal.entity => + entity + ->(Utils.magic: Internal.entity => dict) + ->Utils.Dict.shallowCopy + ->(Utils.magic: dict => Internal.entity) + // Entity operations for direct manipulation outside of handlers let getEntityFromState = ( ~state: testIndexerState, @@ -419,7 +391,7 @@ let getEntityFromState = ( ) } let entityDict = state.entities->Dict.get(entityConfig.name)->Option.getOr(Dict.make()) - entityDict->Dict.get(entityId) + entityDict->Dict.get(entityId)->Option.map(copyEntity) } let makeEntityGet = (~state: testIndexerState, ~entityConfig: Internal.entityConfig): ( @@ -462,7 +434,7 @@ let makeEntitySet = (~state: testIndexerState, ~entityConfig: Internal.entityCon state.entities->Dict.set(entityConfig.name, dict) dict } - entityDict->Dict.set(entity.id, entity) + entityDict->Dict.set(entity.id, copyEntity(entity)) } } @@ -476,7 +448,7 @@ let makeEntityGetAll = (~state: testIndexerState, ~entityConfig: Internal.entity ) } let entityDict = state.entities->Dict.get(entityConfig.name)->Option.getOr(Dict.make()) - Promise.resolve(entityDict->Dict.valuesToArray) + Promise.resolve(entityDict->Dict.valuesToArray->Array.map(copyEntity)) } } @@ -487,411 +459,432 @@ type entityOperations = { set: Internal.entity => unit, } -type workerData = { - chainId: int, - startBlock: int, - endBlock: option, - simulate: option>, - initialState: Persistence.initialState, +// Adapt the real storage interface to the in-memory entity store. In-process +// there's no worker boundary, so entities are stored and loaded decoded — no +// JSON serialization round-trip. +let makeInMemoryStorage = (~state: testIndexerState): Persistence.storage => { + name: "test-inmemory", + isInitialized: async () => true, + // The runner injects the config-derived initial state by setting + // `persistence.storageStatus = Ready(...)` directly, bypassing `Persistence.init`, + // so neither of these is reached. + initialize: async (~chainConfigs as _=?, ~entities as _=?, ~enums as _=?, ~envioInfo as _) => + JsError.throwWithMessage( + "TestIndexer: initialize should not be called; the initial state is derived from config.", + ), + resumeInitialState: async () => + JsError.throwWithMessage( + "TestIndexer: resumeInitialState should not be called; the initial state is derived from config.", + ), + loadOrThrow: async (~filter, ~table: Table.table) => + state + ->handleLoad(~tableName=table.tableName, ~filter) + ->(Utils.magic: array => array), + writeBatch: async ( + ~batch, + ~rollback as _, + ~isInReorgThreshold as _, + ~config as _, + ~allEntities as _, + ~updatedEffectsCache as _, + ~updatedEntities, + ~chainMetaData as _, + ~onWrite as _, + ) => + state->handleWriteBatch( + ~updatedEntities, + ~checkpointIds=batch.checkpointIds, + ~checkpointChainIds=batch.checkpointChainIds, + ~checkpointBlockNumbers=batch.checkpointBlockNumbers, + ~checkpointEventsProcessed=batch.checkpointEventsProcessed, + ), + dumpEffectCache: async () => (), + reset: async () => (), + setChainMeta: async _ => Obj.magic(), + pruneStaleCheckpoints: async (~safeCheckpointId as _) => (), + pruneStaleEntityHistory: async (~entityName as _, ~entityIndex as _, ~safeCheckpointId as _) => + (), + getRollbackTargetCheckpoint: async (~reorgChainId as _, ~lastKnownValidBlockNumber as _) => + JsError.throwWithMessage( + "TestIndexer: Rollback is not supported. The runner forces rollbackOnReorg off, so this should be unreachable.", + ), + getRollbackProgressDiff: async (~rollbackTargetCheckpointId as _) => + JsError.throwWithMessage( + "TestIndexer: Rollback is not supported. The runner forces rollbackOnReorg off, so this should be unreachable.", + ), + getRollbackData: async (~entityConfig as _, ~rollbackTargetCheckpointId as _) => + JsError.throwWithMessage( + "TestIndexer: Rollback is not supported. The runner forces rollbackOnReorg off, so this should be unreachable.", + ), + close: async () => (), } -let makeCreateTestIndexer = (~config: Config.t, ~workerPath: string): ( - unit => t<'processConfig> -) => { - () => { - let allEntities = config.allEntities - let entities = Dict.make() - let entityConfigs = Dict.make() - allEntities->Array.forEach(entityConfig => { - entities->Dict.set(entityConfig.name, Dict.make()) - entityConfigs->Dict.set(entityConfig.name, entityConfig) - }) +// Copy the per-chain registration arrays so a process() run's simulate-source +// additions (SimulateItems.patchConfig pushes onEventRegistrations) never +// mutate the shared base registration — lets independent createTestIndexer runs +// proceed in parallel without clobbering each other's registrations. +let cloneRegistrations = ( + base: HandlerRegister.registrationsByChainId, +): HandlerRegister.registrationsByChainId => { + let clone = Dict.make() + base + ->Dict.toArray + ->Array.forEach(((chainIdStr, chainRegistrations: HandlerRegister.chainRegistrations)) => + clone->Dict.set( + chainIdStr, + { + HandlerRegister.onEventRegistrations: chainRegistrations.onEventRegistrations->Array.copy, + onBlockRegistrations: chainRegistrations.onBlockRegistrations->Array.copy, + }, + ) + ) + clone +} - // Populate config addresses into the entity dict, mirroring PgStorage.initialize - let envioAddressesDict = entities->Dict.getUnsafe(InternalTable.EnvioAddresses.name) - config.chainMap - ->ChainMap.values - ->Array.forEach(chainConfig => { - chainConfig.contracts->Array.forEach(contract => { - contract.addresses->Array.forEach( - address => { - let entity: InternalTable.EnvioAddresses.t = { - id: Config.EnvioAddresses.makeId(~chainId=chainConfig.id, ~address), - chainId: chainConfig.id, - contractName: contract.name, - registrationBlock: -1, - registrationLogIndex: -1, - } - envioAddressesDict->Dict.set(entity.id, entity->Config.EnvioAddresses.castToInternal) - }, - ) - }) - }) +// User handlers register into the process-global HandlerRegister as an import +// side effect. Capture the resolved registrations once per process (imports are +// module-cached anyway) and reuse them across every createTestIndexer run, so +// the global registration cycle runs a single time and never races. +let registrationsRef: ref>> = ref(None) +let getRegistrations = (~config) => + switch registrationsRef.contents { + | Some(promise) => promise + | None => + let promise = HandlerLoader.registerAllHandlers(~config) + registrationsRef := Some(promise) + promise + } - let state = { - processInProgress: false, - progressBlockByChain: Dict.make(), - entities, - entityConfigs, - processChanges: [], - } +let createTestIndexer = (): t<'processConfig> => { + let config = Config.load() + let allEntities = config.allEntities + let entities = Dict.make() + let entityConfigs = Dict.make() + allEntities->Array.forEach(entityConfig => { + entities->Dict.set(entityConfig.name, Dict.make()) + entityConfigs->Dict.set(entityConfig.name, entityConfig) + }) - // Build entity operations for each user entity - let entityOpsDict: dict = Dict.make() - allEntities->Array.forEach(entityConfig => { - // Only create ops for user entities (not internal tables like envio_addresses) - if entityConfig.name !== InternalTable.EnvioAddresses.name { - entityOpsDict->Dict.set( - entityConfig.name, - { - get: makeEntityGet(~state, ~entityConfig), - getAll: makeEntityGetAll(~state, ~entityConfig), - getOrThrow: makeEntityGetOrThrow(~state, ~entityConfig), - set: makeEntitySet(~state, ~entityConfig), - }, - ) - } + // Populate config addresses into the entity dict, mirroring PgStorage.initialize + let envioAddressesDict = entities->Dict.getUnsafe(InternalTable.EnvioAddresses.name) + config.chainMap + ->ChainMap.values + ->Array.forEach(chainConfig => { + chainConfig.contracts->Array.forEach(contract => { + contract.addresses->Array.forEach( + address => { + let entity: InternalTable.EnvioAddresses.t = { + id: Config.EnvioAddresses.makeId(~chainId=chainConfig.id, ~address), + chainId: chainConfig.id, + contractName: contract.name, + registrationBlock: -1, + registrationLogIndex: -1, + } + envioAddressesDict->Dict.set(entity.id, entity->Config.EnvioAddresses.castToInternal) + }, + ) }) + }) - // Build chain info from config (similar to Main.getGlobalIndexer but static) - let chainIds = [] - let chains = Utils.Object.createNullObject() - config.chainMap - ->ChainMap.values - ->Array.forEach(chainConfig => { - let chainIdStr = chainConfig.id->Int.toString - chainIds->Array.push(chainConfig.id)->ignore + let state = { + processInProgress: false, + progressBlockByChain: Dict.make(), + entities, + entityConfigs, + processChanges: [], + } - let chainObj = Utils.Object.createNullObject() - chainObj - ->Utils.Object.definePropertyWithValue("id", {enumerable: true, value: chainConfig.id}) - ->Utils.Object.definePropertyWithValue( - "startBlock", - {enumerable: true, value: chainConfig.startBlock}, + // Per-instance in-memory storage over `state.entities`. Separate indexers + // get separate storages, so independent indexers run in parallel without + // shared mutable state. + let storage = makeInMemoryStorage(~state) + let persistence = Persistence.make( + ~userEntities=config.userEntities, + ~allEnums=config.allEnums, + ~storage, + ) + + // Silence logs by default in test mode unless LOG_LEVEL is explicitly set. + switch Env.userLogLevel { + | None => Logging.setLogLevel(#silent) + | Some(_) => () + } + + // Build entity operations for each user entity + let entityOpsDict: dict = Dict.make() + allEntities->Array.forEach(entityConfig => { + // Only create ops for user entities (not internal tables like envio_addresses) + if entityConfig.name !== InternalTable.EnvioAddresses.name { + entityOpsDict->Dict.set( + entityConfig.name, + { + get: makeEntityGet(~state, ~entityConfig), + getAll: makeEntityGetAll(~state, ~entityConfig), + getOrThrow: makeEntityGetOrThrow(~state, ~entityConfig), + set: makeEntitySet(~state, ~entityConfig), + }, ) - ->Utils.Object.definePropertyWithValue( - "endBlock", - {enumerable: true, value: chainConfig.endBlock}, + } + }) + + // Build chain info from config (similar to Main.getGlobalIndexer but static) + let chainIds = [] + let chains = Utils.Object.createNullObject() + config.chainMap + ->ChainMap.values + ->Array.forEach(chainConfig => { + let chainIdStr = chainConfig.id->Int.toString + chainIds->Array.push(chainConfig.id)->ignore + + let chainObj = Utils.Object.createNullObject() + chainObj + ->Utils.Object.definePropertyWithValue("id", {enumerable: true, value: chainConfig.id}) + ->Utils.Object.definePropertyWithValue( + "startBlock", + {enumerable: true, value: chainConfig.startBlock}, + ) + ->Utils.Object.definePropertyWithValue( + "endBlock", + {enumerable: true, value: chainConfig.endBlock}, + ) + ->Utils.Object.definePropertyWithValue("name", {enumerable: true, value: chainConfig.name}) + ->Utils.Object.definePropertyWithValue("isRealtime", {enumerable: true, value: false}) + ->ignore + + // Add contracts to chain object + chainConfig.contracts->Array.forEach(contract => { + let contractObj = Utils.Object.createNullObject() + contractObj + ->Utils.Object.definePropertyWithValue("name", {enumerable: true, value: contract.name}) + ->Utils.Object.definePropertyWithValue("abi", {enumerable: true, value: contract.abi}) + ->Utils.Object.defineProperty( + "addresses", + { + enumerable: true, + get: () => { + if state.processInProgress { + JsError.throwWithMessage( + `Cannot access ${contract.name}.addresses while indexer.process() is running. ` ++ "Wait for process() to complete before reading contract addresses.", + ) + } + getIndexingAddressesByChain(state) + ->Dict.get(chainConfig.id->Int.toString) + ->Option.getOr([]) + ->Array.filterMap(ia => ia.contractName === contract.name ? Some(ia.address) : None) + }, + }, ) - ->Utils.Object.definePropertyWithValue("name", {enumerable: true, value: chainConfig.name}) - ->Utils.Object.definePropertyWithValue("isRealtime", {enumerable: true, value: false}) ->ignore - // Add contracts to chain object - chainConfig.contracts->Array.forEach(contract => { - let contractObj = Utils.Object.createNullObject() - contractObj - ->Utils.Object.definePropertyWithValue("name", {enumerable: true, value: contract.name}) - ->Utils.Object.definePropertyWithValue("abi", {enumerable: true, value: contract.abi}) - ->Utils.Object.defineProperty( - "addresses", - { - enumerable: true, - get: () => { - if state.processInProgress { - JsError.throwWithMessage( - `Cannot access ${contract.name}.addresses while indexer.process() is running. ` ++ "Wait for process() to complete before reading contract addresses.", - ) - } - getIndexingAddressesByChain(state) - ->Dict.get(chainConfig.id->Int.toString) - ->Option.getOr([]) - ->Array.filterMap(ia => ia.contractName === contract.name ? Some(ia.address) : None) - }, - }, - ) - ->ignore - - chainObj - ->Utils.Object.definePropertyWithValue( - contract.name, - {enumerable: true, value: contractObj}, - ) - ->ignore - }) + chainObj + ->Utils.Object.definePropertyWithValue(contract.name, {enumerable: true, value: contractObj}) + ->ignore + }) + chains + ->Utils.Object.definePropertyWithValue(chainIdStr, {enumerable: true, value: chainObj}) + ->ignore + + if chainConfig.name !== chainIdStr { chains - ->Utils.Object.definePropertyWithValue(chainIdStr, {enumerable: true, value: chainObj}) + ->Utils.Object.definePropertyWithValue(chainConfig.name, {enumerable: false, value: chainObj}) ->ignore + } + }) - if chainConfig.name !== chainIdStr { - chains - ->Utils.Object.definePropertyWithValue( - chainConfig.name, - {enumerable: false, value: chainObj}, - ) - ->ignore - } - }) + // Build the result object with process + entity operations + chain info + let result: dict = Dict.make() + result->Dict.set("chainIds", chainIds->(Utils.magic: array => unknown)) + result->Dict.set("chains", chains->(Utils.magic: {..} => unknown)) + entityOpsDict + ->Dict.toArray + ->Array.forEach(((name, ops)) => { + result->Dict.set(name, ops->(Utils.magic: entityOperations => unknown)) + }) - // Build the result object with process + entity operations + chain info - let result: dict = Dict.make() - result->Dict.set("chainIds", chainIds->(Utils.magic: array => unknown)) - result->Dict.set("chains", chains->(Utils.magic: {..} => unknown)) - entityOpsDict - ->Dict.toArray - ->Array.forEach(((name, ops)) => { - result->Dict.set(name, ops->(Utils.magic: entityOperations => unknown)) - }) + result->Dict.set( + "process", + ( + processConfig => { + // Check if already processing + if state.processInProgress { + JsError.throwWithMessage( + "createTestIndexer process is already running. Only one process call is allowed at a time", + ) + } - result->Dict.set( - "process", - ( - processConfig => { - // Check if already processing - if state.processInProgress { - JsError.throwWithMessage( - "createTestIndexer process is already running. Only one process call is allowed at a time", - ) - } + // Parse and validate processConfig + let parsedConfig = try processConfig->S.parseOrThrow(processConfigSchema) catch { + | S.Raised(exn) => + JsError.throwWithMessage( + `Invalid processConfig: ${exn->Utils.prettifyExn->(Utils.magic: exn => string)}`, + ) + } + let rawChains = parsedConfig["chains"] + let chainKeys = rawChains->Dict.keysToArray - // Parse and validate processConfig - let parsedConfig = try processConfig->S.parseOrThrow(processConfigSchema) catch { - | S.Raised(exn) => + if chainKeys->Array.length === 0 { + JsError.throwWithMessage("createTestIndexer requires at least one chain to be defined") + } + + // Sort chain keys by chain ID for deterministic ordering + let sortedChainKeys = chainKeys->Array.copy + sortedChainKeys->Array.sort((a, b) => { + let aId = a->Int.fromString->Option.getOr(0) + let bId = b->Int.fromString->Option.getOr(0) + Int.compare(aId, bId) + }) + + // Parse and validate the block ranges upfront before running any chain. + let chainEntries = sortedChainKeys->Array.map(chainIdStr => { + let rawChainConfig = rawChains->Dict.getUnsafe(chainIdStr) + if chainIdStr->Int.fromString->Option.isNone { JsError.throwWithMessage( - `Invalid processConfig: ${exn->Utils.prettifyExn->(Utils.magic: exn => string)}`, + `Invalid chain ID "${chainIdStr}": expected a numeric chain ID`, ) } - let rawChains = parsedConfig["chains"] - let chainKeys = rawChains->Dict.keysToArray - - if chainKeys->Array.length === 0 { - JsError.throwWithMessage("createTestIndexer requires at least one chain to be defined") + let processChainConfig = parseBlockRange( + ~chainIdStr, + ~config, + ~rawChainConfig, + ~progressBlock=state.progressBlockByChain->Dict.get(chainIdStr), + ) + (chainIdStr, rawChainConfig, processChainConfig) + }) + + // Reset processChanges for this run + state.processChanges = [] + + let runChainInProcess = async (( + chainIdStr, + rawChainConfig: rawChainConfig, + processChainConfig, + )) => { + // Build initialState from resolved block range. Rebuilt per chain so + // later chains in the same process() call see contracts registered + // by earlier ones. + let chains: dict = Dict.make() + chains->Dict.set(chainIdStr, processChainConfig) + let indexingAddressesByChain = getIndexingAddressesByChain(state) + let initialState = makeInitialState( + ~config, + ~processConfigChains=chains, + ~indexingAddressesByChain, + ) + + // No endBlock means auto-exit mode: process one block checkpoint at a + // time and stop after the first block with events. + let exitAfterFirstEventBlock = processChainConfig.endBlock->Option.isNone + + // Rebuild the processConfig JSON that SimulateItems.patchConfig reads + // to turn `simulate` items into a SimulateSource for the chain. + let resolvedChainDict: dict = Dict.make() + resolvedChainDict->Dict.set( + "startBlock", + processChainConfig.startBlock->(Utils.magic: int => unknown), + ) + switch processChainConfig.endBlock { + | Some(eb) => resolvedChainDict->Dict.set("endBlock", eb->(Utils.magic: int => unknown)) + | None => () } - - // Sort chain keys by chain ID for deterministic ordering - let sortedChainKeys = chainKeys->Array.copy - sortedChainKeys->Array.sort((a, b) => { - let aId = a->Int.fromString->Option.getOr(0) - let bId = b->Int.fromString->Option.getOr(0) - Int.compare(aId, bId) - }) - - // Parse and validate the block ranges upfront before starting any workers. - let chainEntries = sortedChainKeys->Array.map(chainIdStr => { - let rawChainConfig = rawChains->Dict.getUnsafe(chainIdStr) - let chainId = switch chainIdStr->Int.fromString { - | Some(id) => id - | None => - JsError.throwWithMessage( - `Invalid chain ID "${chainIdStr}": expected a numeric chain ID`, - ) - } - let processChainConfig = parseBlockRange( - ~chainIdStr, - ~config, - ~rawChainConfig, - ~progressBlock=state.progressBlockByChain->Dict.get(chainIdStr), - ) - (chainIdStr, chainId, rawChainConfig, processChainConfig) - }) - - // Reset processChanges for this run - state.processChanges = [] - - let runChainWorker = (( + switch rawChainConfig.simulate { + | Some(s) => + resolvedChainDict->Dict.set("simulate", s->(Utils.magic: array => unknown)) + | None => () + } + let resolvedChainsDict: dict = Dict.make() + resolvedChainsDict->Dict.set( chainIdStr, - chainId, - rawChainConfig: rawChainConfig, - processChainConfig, - )) => { - // Build initialState from resolved block range - let chains: dict = Dict.make() - chains->Dict.set(chainIdStr, processChainConfig) - - // Rebuilt per chain so workers see contracts registered by earlier - // chains in the same process() call. - let indexingAddressesByChain = getIndexingAddressesByChain(state) - - let initialState = makeInitialState( - ~config, - ~processConfigChains=chains, - ~indexingAddressesByChain, - ) - - Promise.make((resolve, reject) => { - let workerData: workerData = { - chainId, - startBlock: processChainConfig.startBlock, - endBlock: processChainConfig.endBlock, - simulate: rawChainConfig.simulate, - initialState, - } - let worker = try { - NodeJs.WorkerThreads.makeWorker( - workerPath, - { - workerData: workerData->(Utils.magic: workerData => JSON.t), - // Explicitly forward parent env so handlers running in - // the worker observe the same environment as the test - // process (e.g. E2E_EXPECTED_END_BLOCK). - env: %raw(`process.env`), - }, - ) - } catch { - | exn => - reject(exn->Utils.magic) - throw(exn) - } - - // Handle messages from worker - worker->NodeJs.WorkerThreads.onMessage(( - msg: TestIndexerProxyStorage.workerMessage, - ) => { - let respond = data => - worker->NodeJs.WorkerThreads.workerPostMessage( - { - TestIndexerProxyStorage.id: msg.id, - payload: TestIndexerProxyStorage.Response({data: data}), - }->Utils.magic, - ) - - switch msg.payload { - | Load({tableName, filter}) => state->handleLoad(~tableName, ~filter)->respond - - | WriteBatch({ - updatedEntities, - checkpointIds, - checkpointChainIds, - checkpointBlockNumbers, - checkpointBlockHashes: _, - checkpointEventsProcessed, - }) => - state->handleWriteBatch( - ~updatedEntities, - ~checkpointIds, - ~checkpointChainIds, - ~checkpointBlockNumbers, - ~checkpointEventsProcessed, - ) - JSON.Encode.null->respond - } - }) - - worker->NodeJs.WorkerThreads.onError(err => { - worker->NodeJs.WorkerThreads.terminate->ignore - reject(err) - }) - - worker->NodeJs.WorkerThreads.onExit(code => { - if code !== 0 { - reject(Utils.Error.make(`Worker exited with code ${code->Int.toString}`)) - } else { - resolve() + resolvedChainDict->(Utils.magic: dict => unknown), + ) + let processConfigJson = + {"chains": resolvedChainsDict}->(Utils.magic: {"chains": dict} => JSON.t) + + // Each run gets its own copy of the shared base registration so the + // simulate-source registration it appends stays isolated. + let registrationsByChainId = cloneRegistrations(await getRegistrations(~config)) + let patchedConfig: Config.t = SimulateItems.patchConfig( + ~config, + ~processConfig=processConfigJson, + ~registrationsByChainId, + ) + let runConfig = {...patchedConfig, shouldRollbackOnReorg: false} + let runConfig = exitAfterFirstEventBlock ? {...runConfig, batchSize: 1} : runConfig + + // Bypass Persistence.init: hand the loop the config-derived initial + // state directly (never a real DB) and mark the storage Ready so + // writes go through. + persistence.storageStatus = Ready(initialState) + + let indexerStateRef = ref(None) + // Stop the loop and let any in-flight processing/write settle, so a + // finished run leaves nothing driving the shared state into the next. + let cleanup = async () => { + switch indexerStateRef.contents { + | Some(indexerState) => + indexerState->IndexerState.stop + while ( + indexerState->IndexerState.isProcessing || + indexerState->IndexerState.writeFiber->Option.isSome + ) { + switch indexerState->IndexerState.writeFiber { + // Await the in-flight write directly; only fall back to a tick + // yield while processing hasn't yet spawned a write fiber. + | Some(fiber) => await fiber + | None => await Utils.delay(0) } - }) - }) + } + | None => () + } } - - // Set flag before starting workers - state.processInProgress = true - - // Run worker threads sequentially, one chain at a time - let rec runChains = idx => { - if idx >= chainEntries->Array.length { - state.processInProgress = false - Promise.resolve({changes: state.processChanges}) - } else { - runChainWorker(chainEntries->Array.getUnsafe(idx))->Promise.then(_ => - runChains(idx + 1) + try { + await Promise.make((resolve, reject) => { + let indexerState = IndexerState.makeFromDbState( + ~config=runConfig, + ~persistence, + ~initialState, + ~registrationsByChainId, + ~exitAfterFirstEventBlock, + ~onError=errHandler => { + errHandler->ErrorHandling.log + reject(errHandler.exn->Utils.prettifyExn) + }, + // Caught up: resolve the run instead of exiting the process. + ~onExit=() => resolve(), ) - } + indexerStateRef := Some(indexerState) + indexerState->IndexerLoop.start + }) + await cleanup() + } catch { + | exn => + await cleanup() + throw(exn) } - - runChains(0)->Promise.catch(err => { - state.processInProgress = false - Promise.reject(err->Utils.prettifyExn) - }) } - )->(Utils.magic: ('a => promise) => unknown), - ) - - result->(Utils.magic: dict => t<'processConfig>) - } -} - -let initTestWorker = () => { - if NodeJs.WorkerThreads.isMainThread { - JsError.throwWithMessage("initTestWorker must be called from a worker thread") - } - - let parentPort = switch NodeJs.WorkerThreads.parentPort->Nullable.toOption { - | Some(port) => port - | None => JsError.throwWithMessage("initTestWorker: No parent port available") - } - - let workerData: option = NodeJs.WorkerThreads.workerData->Nullable.toOption - switch workerData { - | Some({chainId, startBlock, endBlock, simulate, initialState}) => - let chainIdStr = chainId->Int.toString - - // auto-exit mode: no endBlock means fetch first block with events and exit - let exitAfterFirstEventBlock = endBlock->Option.isNone - - // Build processConfig JSON for SimulateItems.patchConfig - let resolvedChainDict: dict = Dict.make() - resolvedChainDict->Dict.set("startBlock", startBlock->(Utils.magic: int => unknown)) - switch endBlock { - | Some(eb) => resolvedChainDict->Dict.set("endBlock", eb->(Utils.magic: int => unknown)) - | None => () - } - switch simulate { - | Some(s) => resolvedChainDict->Dict.set("simulate", s->(Utils.magic: array => unknown)) - | None => () - } - let resolvedChainsDict: dict = Dict.make() - resolvedChainsDict->Dict.set( - chainIdStr, - resolvedChainDict->(Utils.magic: dict => unknown), - ) - let processConfig = - {"chains": resolvedChainsDict}->(Utils.magic: {"chains": dict} => JSON.t) - - // Create proxy storage that communicates with main thread - let proxy = TestIndexerProxyStorage.make(~parentPort, ~initialState) - let storage = TestIndexerProxyStorage.makeStorage(proxy) - let config = Config.load() - let persistence = Persistence.make( - ~userEntities=config.userEntities, - ~allEnums=config.allEnums, - ~storage, - ) - // Silence logs by default in test mode unless LOG_LEVEL is explicitly set - switch Env.userLogLevel { - | None => Logging.setLogLevel(#silent) - | Some(_) => () - } + // Set flag before starting the run + state.processInProgress = true - let patchConfig = (config: Config.t, registrationsByChainId) => { - let config = SimulateItems.patchConfig(~config, ~processConfig, ~registrationsByChainId) + // Run chains sequentially, one at a time + let rec runChains = idx => { + if idx >= chainEntries->Array.length { + state.processInProgress = false + Promise.resolve({changes: state.processChanges}) + } else { + runChainInProcess(chainEntries->Array.getUnsafe(idx))->Promise.then(_ => + runChains(idx + 1) + ) + } + } - // In auto-exit mode, set batchSize=1 to process one block checkpoint at a time - if exitAfterFirstEventBlock { - {...config, batchSize: 1} - } else { - config - } - } - Main.start(~persistence, ~isTest=true, ~patchConfig, ~exitAfterFirstEventBlock) - ->Promise.catch(exn => { - // `Main.start` rejects on any fatal error: a runtime failure arrives wrapped - // in `Main.FatalError` (already logged), a setup throw (e.g. an invalid - // simulate item) arrives raw. The parent only learns of failures - // through the worker `error` event, which fires on an *uncaught* exception. - // Throwing synchronously in this catch would just reject the catch's own - // promise (swallowed by `ignore`); `setImmediate` re-throws outside the - // promise chain so it becomes uncaught and reaches the parent. - let toThrow = switch exn { - | Main.FatalError(inner) => inner - | _ => exn->Utils.prettifyExn + runChains(0)->Promise.catch(err => { + state.processInProgress = false + Promise.reject(err->Utils.prettifyExn) + }) } - NodeJs.setImmediate(() => throw(toThrow)) - Promise.resolve() - }) - ->ignore - | None => - Logging.error("TestIndexerWorker: No worker data provided") - NodeJs.process->NodeJs.exitWithCode(Failure) - } + )->(Utils.magic: ('a => promise) => unknown), + ) + + result->(Utils.magic: dict => t<'processConfig>) } diff --git a/packages/envio/src/TestIndexerProxyStorage.res b/packages/envio/src/TestIndexerProxyStorage.res deleted file mode 100644 index 6440e0648a..0000000000 --- a/packages/envio/src/TestIndexerProxyStorage.res +++ /dev/null @@ -1,196 +0,0 @@ -// Message types for communication between worker and main thread -type requestId = int - -// Serializable change with entity as JSON (for worker thread messaging) -@tag("type") -type serializableChange = - | @as("SET") Set({entityId: string, entity: JSON.t, checkpointId: bigint}) - | @as("DELETE") Delete({entityId: string, checkpointId: bigint}) - -type serializableUpdatedEntity = { - entityName: string, - changes: array, -} - -// Worker -> Main thread payloads -@tag("type") -type workerPayload = - | @as("load") - Load({ - tableName: string, - // Leaf field values are JSON-serialized with the table's field schemas - filter: EntityFilter.t, - }) - | @as("writeBatch") - WriteBatch({ - updatedEntities: array, - checkpointIds: array, - checkpointChainIds: array, - checkpointBlockNumbers: array, - checkpointBlockHashes: array>, - checkpointEventsProcessed: array, - }) - -// Main thread -> Worker payloads -@tag("type") -type mainPayload = - | @as("response") Response({data: JSON.t}) - | @as("error") Error({message: string}) - -// Message wrapper with id -type message<'payload> = {id: requestId, payload: 'payload} -type workerMessage = message -type mainMessage = message - -// Pending request tracker -type pendingRequest = { - resolve: JSON.t => unit, - reject: exn => unit, -} - -type t = { - parentPort: NodeJs.WorkerThreads.messagePort, - initialState: Persistence.initialState, - pendingRequests: dict, - mutable requestCounter: int, -} - -let make = (~parentPort, ~initialState): t => { - let proxy = { - parentPort, - initialState, - pendingRequests: Dict.make(), - requestCounter: 0, - } - - // Set up message listener for responses from main thread - parentPort->NodeJs.WorkerThreads.onPortMessage((msg: mainMessage) => { - let idStr = msg.id->Int.toString - let {resolve, reject} = switch proxy.pendingRequests->Utils.Dict.dangerouslyGetNonOption( - idStr, - ) { - | Some(pending) => pending - | None => JsError.throwWithMessage(`TestIndexer: No pending request found for id ${idStr}`) - } - Dict.delete(proxy.pendingRequests->Obj.magic, idStr) - - switch msg.payload { - | Response({data}) => resolve(data) - | Error({message}) => reject(Utils.Error.make(message)) - } - }) - - proxy -} - -let nextRequestId = (proxy: t): requestId => { - proxy.requestCounter = proxy.requestCounter + 1 - proxy.requestCounter -} - -let sendRequest = (proxy: t, ~payload: workerPayload): promise => { - Promise.make((resolve, reject) => { - let id = proxy->nextRequestId - proxy.pendingRequests->Dict.set(id->Int.toString, {resolve, reject}) - proxy.parentPort->NodeJs.WorkerThreads.postMessage({id, payload}) - }) -} - -let makeStorage = (proxy: t): Persistence.storage => { - name: "test-proxy", - isInitialized: async () => true, - initialize: async (~chainConfigs as _=?, ~entities as _=?, ~enums as _=?, ~envioInfo as _) => { - JsError.throwWithMessage( - "TestIndexer: initialize should not be called. Use resumeInitialState instead.", - ) - }, - resumeInitialState: async () => proxy.initialState, - loadOrThrow: async (~filter, ~table: Table.table) => { - let serializeLeafOrThrow = (~fieldName, ~fieldValue: unknown, ~isArray) => { - let queryField = switch table->Table.queryFields->Dict.get(fieldName) { - | Some(queryField) => queryField - | None => - JsError.throwWithMessage( - `TestIndexer: The table "${table.tableName}" doesn't have the field "${fieldName}"`, - ) - } - fieldValue - ->S.reverseConvertToJsonOrThrow( - isArray ? queryField.arrayFieldSchema : queryField.fieldSchema, - ) - ->(Utils.magic: JSON.t => unknown) - } - let response = await proxy->sendRequest( - ~payload=Load({ - tableName: table.tableName, - // Field values must be JSON-safe to survive the worker thread boundary - filter: filter->EntityFilter.mapValues(~mapValue=serializeLeafOrThrow), - }), - ) - response->S.parseOrThrow(table->Table.rowsSchema) - }, - writeBatch: async ( - ~batch, - ~rollback as _, - ~isInReorgThreshold as _, - ~config as _, - ~allEntities as _, - ~updatedEffectsCache as _, - ~updatedEntities, - ~chainMetaData as _, - ~onWrite as _, - ) => { - // Encode entities to JSON for serialization across worker boundary - let serializableEntities = updatedEntities->Array.map(( - {entityConfig, changes}: Persistence.updatedEntity, - ) => { - let encodeChange = (change: Change.t): serializableChange => { - switch change { - | Set({entityId, entity, checkpointId}) => - Set({ - entityId, - entity: entity->S.reverseConvertToJsonOrThrow(entityConfig.schema), - checkpointId, - }) - | Delete({entityId, checkpointId}) => Delete({entityId, checkpointId}) - } - } - { - entityName: entityConfig.name, - changes: changes->Array.map(encodeChange), - } - }) - let _ = await proxy->sendRequest( - ~payload=WriteBatch({ - updatedEntities: serializableEntities, - checkpointIds: batch.checkpointIds, - checkpointChainIds: batch.checkpointChainIds, - checkpointBlockNumbers: batch.checkpointBlockNumbers, - checkpointBlockHashes: batch.checkpointBlockHashes, - checkpointEventsProcessed: batch.checkpointEventsProcessed, - }), - ) - }, - dumpEffectCache: async () => (), - reset: async () => (), - setChainMeta: async _ => Obj.magic(), - pruneStaleCheckpoints: async (~safeCheckpointId as _) => (), - pruneStaleEntityHistory: async (~entityName as _, ~entityIndex as _, ~safeCheckpointId as _) => - (), - getRollbackTargetCheckpoint: async (~reorgChainId as _, ~lastKnownValidBlockNumber as _) => { - JsError.throwWithMessage( - "TestIndexer: Rollback is not supported. Set rollbackOnReorg to false in config.", - ) - }, - getRollbackProgressDiff: async (~rollbackTargetCheckpointId as _) => { - JsError.throwWithMessage( - "TestIndexer: Rollback is not supported. Set rollbackOnReorg to false in config.", - ) - }, - getRollbackData: async (~entityConfig as _, ~rollbackTargetCheckpointId as _) => { - JsError.throwWithMessage( - "TestIndexer: Rollback is not supported. Set rollbackOnReorg to false in config.", - ) - }, - close: async () => (), -} diff --git a/packages/envio/src/TestIndexerWorker.res b/packages/envio/src/TestIndexerWorker.res deleted file mode 100644 index fb1dee867f..0000000000 --- a/packages/envio/src/TestIndexerWorker.res +++ /dev/null @@ -1,4 +0,0 @@ -// Spawned by `TestIndexer.makeCreateTestIndexer` as a sibling of `Api.res.mjs` -// so `import.meta.url` resolves inside the envio package. - -TestIndexer.initTestWorker() diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 82f61f2319..0e6ae17310 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -124,6 +124,9 @@ importers: rescript: specifier: 12.2.0 version: 12.2.0 + vitest: + specifier: 4.1.0 + version: 4.1.0(@opentelemetry/api@1.9.0)(@types/node@24.12.2)(jsdom@16.7.0)(vite@7.3.1(@types/node@24.12.2)(tsx@4.21.0)) packages/envio-tests: dependencies: diff --git a/scenarios/test_codegen/src/handlers/EventHandlers.ts b/scenarios/test_codegen/src/handlers/EventHandlers.ts index 209008c1d2..e22b040031 100644 --- a/scenarios/test_codegen/src/handlers/EventHandlers.ts +++ b/scenarios/test_codegen/src/handlers/EventHandlers.ts @@ -129,6 +129,40 @@ expectType< > >(true); +// Type-only surface checks for the registration API. Guarded by `if (0)` so +// tsc validates them but they never execute: invalid contract/event combos must +// stay compile errors, and these must not register real handlers. +if (0) { + indexer.onEvent( + // @ts-expect-error - "BadContract" is not a configured contract + { contract: "BadContract", event: "X" }, + async () => {}, + ); + indexer.onEvent( + // @ts-expect-error - "BadEvent" is not an event of Gravatar + { contract: "Gravatar", event: "BadEvent" }, + async () => {}, + ); + indexer.onEvent( + { contract: "Gravatar", event: "NewGravatar" }, + async ({ event }) => { + expectType>(true); + }, + ); + indexer.contractRegister( + { contract: "NftFactory", event: "SimpleNftCreated" }, + async ({ event, context }) => { + expectType< + TypeEqual + >(true); + context.chain.SimpleNft.add(event.params.contractAddress); + context.chain.NftFactory.add(event.params.contractAddress); + // @ts-expect-error - UnknownContract is not configured + context.chain.UnknownContract.add(event.params.contractAddress); + }, + ); +} + const zeroAddress = "0x0000000000000000000000000000000000000000"; indexer.onEvent({ contract: "Gravatar", event: "CustomSelection" }, async ({ event, context }) => { @@ -834,6 +868,10 @@ indexer.onEvent({ contract: "Gravatar", event: "FactoryEvent" }, async ({ event, fail("Should have thrown"); } + case "throwInHandler": { + throw new Error("Error from handler"); + } + // Reproduction for https://github.com/enviodev/hyperindex/issues/1199: // getWhere on a linkedEntity field (db column name `_id`) must // resolve the field schema in the in-memory TestIndexer. Production diff --git a/scenarios/test_codegen/test/EventHandler.test.ts b/scenarios/test_codegen/test/EventHandler.test.ts index 8f2b858cc9..dbd5e23925 100644 --- a/scenarios/test_codegen/test/EventHandler.test.ts +++ b/scenarios/test_codegen/test/EventHandler.test.ts @@ -460,6 +460,36 @@ describe("Use Envio test framework to test event handlers", () => { assert.deepEqual(users, [existingUser]); }); + it("mutating entities across the set/get boundary doesn't corrupt the store", async () => { + const indexer = createTestIndexer(); + + const original: User = { + id: "0", + address: "existing", + updatesCountOnUserForTesting: 0, + gravatar_id: undefined, + accountType: "USER", + }; + indexer.User.set(original); + + // Mutating the object passed to set, or any entity handed back, must not + // leak into the in-memory store. + Object.assign(original, { address: "mutated-after-set" }); + Object.assign(await indexer.User.getOrThrow("0"), { address: "mutated-after-getOrThrow" }); + Object.assign((await indexer.User.get("0"))!, { address: "mutated-after-get" }); + (await indexer.User.getAll()).forEach((u) => Object.assign(u, { address: "mutated-after-getAll" })); + + assert.deepEqual(await indexer.User.getAll(), [ + { + id: "0", + address: "existing", + updatesCountOnUserForTesting: 0, + gravatar_id: undefined, + accountType: "USER", + }, + ]); + }); + it("entity.getOrThrow throws if entity doesn't exist", async () => { const indexer = createTestIndexer(); const dcAddress = "0x1234567890123456789012345678901234567890"; @@ -935,6 +965,30 @@ describe("Use Envio test framework to test event handlers", () => { ); }); + it("propagates a handler throw to the process() call site with its message", async () => { + const indexer = createTestIndexer(); + const dcAddress = "0x1234567890123456789012345678901234567890"; + + await assert.rejects( + indexer.process({ + chains: { + 1337: { + startBlock: 1, + endBlock: 100, + simulate: [ + { + contract: "Gravatar", + event: "FactoryEvent", + params: { contract: dcAddress, testCase: "throwInHandler" }, + }, + ], + }, + }, + }), + /Error from handler/, + ); + }); + it("Should throw when registering a handler after the indexer has finished initializing", async () => { const indexer = createTestIndexer(); const dcAddress = "0x1234567890123456789012345678901234567890"; @@ -1757,44 +1811,10 @@ describe("onEvent / contractRegister types", () => { type _userOnCr = EvmContractRegisterContext["User"]; }); - it("indexer.onEvent rejects invalid contract/event combinations", () => { - indexer.onEvent( - // @ts-expect-error - "BadContract" is not a configured contract - { contract: "BadContract", event: "X" }, - async () => {}, - ); - - indexer.onEvent( - // @ts-expect-error - "BadEvent" is not an event of Gravatar - { contract: "Gravatar", event: "BadEvent" }, - async () => {}, - ); - - // Valid combination should compile (no @ts-expect-error) - indexer.onEvent( - { contract: "Gravatar", event: "NewGravatar" }, - async ({ event }) => { - expectType>(true); - }, - ); - }); - - it("indexer.contractRegister exposes context.chain.ContractName.add()", () => { - indexer.contractRegister( - { contract: "NftFactory", event: "SimpleNftCreated" }, - async ({ event, context }) => { - // event is typed - expectType< - TypeEqual - >(true); - // chain.ContractName.add(address) is available - context.chain.SimpleNft.add(event.params.contractAddress); - context.chain.NftFactory.add(event.params.contractAddress); - // @ts-expect-error - UnknownContract is not configured - context.chain.UnknownContract.add(event.params.contractAddress); - }, - ); - }); + // `indexer.onEvent` / `indexer.contractRegister` type-surface checks live in + // `src/handlers/EventHandlers.ts` (compile-time only): the registration API + // throws once handlers are registered, so they can't run as post-registration + // test cases here. it("EvmOnEventHandler defaults to union of all events", () => { // Without args, the handler accepts the union of all EVM events