fix(indexer): share one per-batch pipeline between the single and parallel ingest paths

[DevTobis]

closes #203

Summary

INGEST_WORKERS > 1 switched to pollParallel, which only fetched SAC contracts and only stored fungible transfers. It silently skipped NFT transfers and metadata, LP shares, host-fn logs, SAC tagging, token metadata and account summaries. Raising the worker count changed what was indexed, not just how fast.

Changes

• New src/indexer/batch.ts: processEventBatch(events, { network, knownLpPools }), the body of pollOnce moved verbatim (fungible + SAC tagging + token metadata + account summaries + WS emit, host-fn logs, LP shares, NFT + metadata). It returns per-type insert counts; batchTotal sums them.
• pollOnce now fetches and calls processEventBatch. pollParallel workers do the same: its last parameter is a required { fetchEvents, processBatch }, so a sharded run cannot exist without a batch handler.
• New exported ingestWindow(loop, from, to, workers = INGEST_WORKERS) in src/indexer.ts picks the path; the poll loop calls it. The sharded branch now shards allContractIds (SAC + NFT, previously SAC only), and fetches through the loop's source switcher (previously raw fetchEventsSafe, bypassing Horizon failover). createLoopState / LoopState are exported for the test.
• .env.example documents INGEST_WORKERS.

Parallel-unsafe work

None found. Workers process disjoint contracts, and every table the pipeline writes is keyed by contract (eventId unique; AccountSummary, NftMetadata, token metadata include contractId in their key). The only shared mutable state is the knownLpPools set, and a contract only ever adds itself to it. So nothing needed to be excluded from parallel mode.

Tests

src/__tests__/ingestParity.test.ts (in-memory db, fake source that honours the contract filter):

• the same events through ingestWindow(..., 1) and ingestWindow(..., 4) leave identical TokenTransfer, NftTransfer, NftMetadata, account summary, host-fn, LP-share, token-metadata and cursor state, and the same totalIndexed;
• both paths route every event through processEventBatch;
• a sensitivity test wires a sharded run whose handler skips steps (the old bug) and asserts the snapshots differ, i.e. a step added to one path only is caught.

Fail-before: with the parallel branch temporarily restored to the old behaviour (SAC ids only, fungible-only handler), 2 of the 5 tests fail (parity and same-pipeline). Restored, all 5 pass. All fixtures are built with StrKey so they are valid by construction.

Verification

• npx jest src/__tests__/ingestParity.test.ts src/__tests__/multiNetworkIndexer.test.ts src/__tests__/indexerSources.test.ts src/__tests__/lpShares.test.ts src/__tests__/tombstoneWiring.test.ts src/__tests__/nft.test.ts: 6 suites, 78 passed.
• npm run typecheck: clean.
• Full suite and vitest integration tests not run (no Docker; shared, heavily loaded machine).

Caveats

• pollParallel's signature changed (required io parameter); its only caller was src/indexer.ts.
• Merge note: PR fix(indexer): bound NFT metadata lookups with a worker pool and per-cycle budget #217 (bounded NFT metadata lookups) edits the NFT metadata block that this PR moves into src/indexer/batch.ts; whichever merges second needs a small mechanical rebase (the bounded call belongs in batch.ts). I did not touch src/api/offramp.ts, prisma/schema.prisma or the README, so no overlap with fix(offramp): require a bearer token to read an order #216.

[drips-wave]

@DevTobis Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits.

You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀

Learn more about application limits

[Miracle656]

Verified locally on a worktree merged with current main.

• npx tsc --noEmit -p tsconfig.test.json: clean
• Full jest suite: 503 passed / 42 suites, 0 failed

This is the right shape for #203. Moving the pipeline body into src/indexer/batch.ts verbatim and making ParallelIo.processBatch a required parameter means the old failure mode is now unrepresentable rather than merely tested against — a sharded run cannot exist without a batch handler. Two other real bugs got fixed along the way that the issue did not name: the sharded branch was sharding sacContractIds rather than allContractIds (so NFT contracts were never even requested), and it called fetchEventsSafe directly, bypassing the source switcher's Horizon failover.

ingestParity.test.ts does what was asked: the same fixture through ingestWindow(..., 1) and ingestWindow(..., 4) produces identical TokenTransfer, NftTransfer, AccountSummary, NftMetadata, host-fn, LP-share, token-metadata and cursor state, and the sensitivity case proves the snapshot comparison actually catches a path that skips steps. Building every fixture with StrKey.encodeContract / encodeEd25519PublicKey from raw bytes, plus the explicit isValidContract assertion, is the right habit — hand-typed strkeys in this repo have been wrong before.

Prometheus labels stay {network, type}, so nothing unbounded moved into batch.ts.

One small thing for a follow-up, not worth holding this: newLoop() sets SAC_CONTRACT_IDS_TESTNET / NFT_CONTRACT_IDS_TESTNET on process.env and never restores them. Jest reuses a worker process across test files, so that leaks into whatever runs next in the same worker. An afterAll restoring the previous values would close it.

https://claude.ai/code/session_01USgemLt4Rnz4SGB1Srf3GB