feat(ingestion): add common ingestion pipeline builder prototype

[pl]

Problem

Every Kafka-fed ingestion pipeline in nodejs (analytics, AI, error tracking, heatmaps, client warnings, session replay) hand-assembles the same skeleton: beforeBatch hooks, messageAware, header parsing, pre-parse restrictions, body parse + team resolution, lifting the team into context, a teamAware processing block, ingestion warning handling, result handling, side effect handling, afterBatch flush hooks. Only the middle differs per product. The boilerplate is duplicated in every pipeline, the invariants (warnings before results, side effects last) rely on convention, and the deep callback nesting makes it hard to see at a glance what a pipeline actually does.

Changes

Adds a common ingestion pipeline builder (nodejs/src/ingestion/common/common-ingestion-pipeline.ts) that owns the skeleton and reads as a flat recipe. Phase markers carry the structure; there is no user-side nesting except batch hooks and grouping:

newCommonIngestionPipeline<TInput, TContext>({ teamManager, outputs, promiseScheduler, concurrentBatches: 1 })
    .beforeBatch((b) => b.pipe(createEventFiltersBatchAppMetricsBeforeBatchStep(outputs)))
    .parseHeaders()
    .pipe(createAllowEventsStep(['$$heatmap']))
    .pipe(createApplyBasicEventRestrictionsStep(eventIngestionRestrictionManager))
    .parseMessage()
    .resolveTeam()
    .pipe(createValidateHistoricalMigrationStep())
    .pipe(createValidateEventMetadataStep())
    .pipe(createApplyEventFiltersStep(eventFilterManager))
    .pipe(createDropOldEventsStep())
    .gather()
    .pipeChunk(createApplyCookielessProcessingStep(cookielessManager))
    .pipe(createCheckHeatmapOptInStep())
    // ...
    .afterBatch((b) => b.pipe(createFlushEventFiltersBatchAppMetricsStep()))
    .build()

Semantics:

• Consecutive .pipe() calls coalesce into one sequential per-element block; any chunk-level call (.pipeChunk, .gather, .concurrentlyPerGroup, .compose) closes the block. Block boundaries fall out of where the chunk operations sit, which is how the hand-written pipelines were already shaped.
• .parseHeaders() / .parseMessage() / .resolveTeam() pipe the shared steps. .resolveTeam() also lifts the team into the element context and opens the team-aware scope: everything after it runs inside handleIngestionWarnings. It takes an optional wrap to decorate the resolution step (error tracking wraps it in topHog metrics).
• .build() wraps the whole thing in messageAware and bundles handleResults, handleSideEffects, and the batching hooks. The built pipeline owns all of its side effects: the beforeBatch/afterBatch hook pipelines are wrapped in side-effect handling too (via a new single-item SideEffectHandlingProcessor and PipelineBuilder.handleSideEffects, mirroring the chunk-level one), so BatchResult.sideEffects always comes out empty and drivers only drain results. An awaitSideEffects config flag (default false, i.e. schedule on the promise scheduler) applies to element results and hooks alike.
• Stage order is enforced by the type system (body steps don't typecheck before .parseMessage()), and redirect kinds are declared up front via the ROut type parameter so the outputs must cover every redirect a step can produce.
• .concurrentlyPerGroup mirrors the framework method (within-group ordering stays visible at the call site), and .compose(fn) is the escape hatch for existing subpipeline functions like createPostTeamPreprocessingSubpipeline, whose generics are widened to accept mid-chain builders.
• Internally each stage composes continuations over the chunk builder — no casts and no any; the coalescing block's start type stays existential inside closures.

Five pipelines are rewired onto the builder: client warnings, heatmaps, joined analytics (keeps its per-distinct-id concurrentlyPerGroup and post-team subpipeline), AI (keeps its topHog metric wrappers and per-step retries), and error tracking. Two mechanical enablers rode along: the schema-enforcement toggle moved into createValidateEventSchemaStep (a no-op step when disabled beats conditional chain composition), and the post-team subpipeline generics were widened.

Note

Error tracking previously built a bare chunk pipeline driven through ChunkPipelineUnwrapper; it is now a BatchingPipeline like the other consumers, with passthrough batch hooks and concurrentBatches: 1, and its custom team-lift filterMap became a plain messageBytes step. Its createErrorTrackingPipeline / runErrorTrackingPipeline signatures are unchanged, so the existing suites verify the conversion.

Note

Sequential-block boundaries shift slightly versus the hand-written versions (e.g. post-team-resolution validation now runs inside the team-aware scope, so its warnings route per-team even for dropped elements). Per-element step order is unchanged everywhere; the end-to-end suites pass unchanged, including all 22 ingestion-consumer snapshots.

Session replay is the one pipeline left (custom team resolution via TeamService rather than TeamManager); a doc-test chapter for the builder is also still pending.

How did you test this code?

Automated only, no manual testing:

• pnpm build (tsc across nodejs) and eslint on the touched files
• Existing suites pass unchanged, so the rewired pipelines are behaviorally identical: src/ingestion/ingestion-consumer.test.ts (57 tests, 22 snapshots, end-to-end joined pipeline), src/ingestion/pipelines/ai/ (806 tests), src/ingestion/pipelines/errortracking/ (146 tests, including the end-to-end pipeline and consumer suites, driven through the unchanged createErrorTrackingPipeline/runErrorTrackingPipeline API), src/ingestion/pipelines/heatmaps/ (42 tests), src/ingestion/pipelines/clientwarnings/
• One new test case in validate-event-schema.test.ts: a step built with enabled: false passes through an event that violates an enforced schema — guards the flag plumbing, which no existing test covered; the enabled-path cases already existed
• Otherwise no new tests: the builder is exercised end-to-end by the existing pipeline suites; a dedicated doc-test chapter is deferred until the API settles

Automatic notifications

• Publish to changelog?
• Alert Sales and Marketing teams?

Docs update

None yet, prototype. The framework's doc-test chapters (nodejs/src/ingestion/framework/docs/) get a builder chapter if this graduates.

🤖 Agent context

Autonomy: Human-driven (agent-assisted)

Built with Claude Code. Skills used: ingestion-pipeline-doctor-nodejs, writing-tests. The API went through three shapes on this branch: staged callbacks per phase (preParse/resolveTeam/perTeam), then a collapse of the post-resolution stage, then the current flat recipe — the flat form was chosen because the linear spine is the point of the template and the callback nesting added no information. Key implementation decisions: sequential-block coalescing with the block's start type kept existential inside closures (cast-free), ROut fixed at the factory so redirect coverage is checked where outputs are supplied, the post-team subpipeline generics widened instead of adding a special builder hook, and error tracking converted to a BatchingPipeline (rather than giving the builder a second, unwrapped build path) to keep all consumers on one pipeline shape.

[trunk-io]

     

Failed Test	Failure Summary	Logs
Event Pipeline E2E tests (prefetch=false) we do not alias users if distinct id changes but we are already identified		Logs ↗︎

View Full Report ↗︎ ⋅ Docs

[greptile-apps]

_{Reviews (1): Last reviewed commit: "chore(ingestion): handle batch hook side..." | Re-trigger Greptile}

[jose-sequeira]

Approved, overall change seems good to unify code. Think some improvements to testing across the pipelines can be made as well as other smallr call outs

[jose-sequeira]

No tests for the massive class we added. We should add a test file for it as most pipelines do

[jose-sequeira]

The two chain interfaces plus their four factory functions (pendingPreTeamChain/committedPreTeamChain and pendingTeamChain/committedTeamChain) implement the exact same pending/committed coalescing logic which leads to a good amount of duplicated code with only the type parameters changed (BatchContext<TContext> vs TeamAwareContext<TContext>, SubpipelineTransform vs TeamSubpipelineTransform).

Could we generalize and then provide type aliases?
Chain<TBuilderContext, TTransform> and then alias based on parameters?

[jose-sequeira]

consumer.test.ts in clientwarnings mocks this entire file, so none of this gets tested

[deployment-status-posthog]

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-07-16 13:33 UTC	Run
prod-us	✅ Deployed	2026-07-16 13:56 UTC	Run
prod-eu	✅ Deployed	2026-07-16 13:58 UTC	Run