ingestion-pipeline-doctor-nodejs

Pipeline Doctor

PostHog 摄取流水线框架及其约定检查 Agent 的快速参考。

架构概览

摄取流水线通过类型化、可组合的步骤链处理事件：

Kafka message
  → messageAware()
    → parse headers/body
    → sequentially() for preprocessing
    → filterMap() to enrich context (e.g., team lookup)
    → teamAware()
      → concurrentlyPerGroup(token:distinctId) for per-entity processing
      → gather()
      → pipeChunk() for chunk operations
      → handleIngestionWarnings()
    → handleResults()
  → handleSideEffects()
  → build()

请参阅 nodejs/src/ingestion/pipelines/analytics/joined-ingestion-pipeline.ts，了解实际实现。

关键文件位置

内容	位置
步骤类型	nodejs/src/ingestion/framework/steps.ts
结果类型	nodejs/src/ingestion/framework/results.ts
文档测试章节	nodejs/src/ingestion/framework/docs/*.test.ts
联合流水线	nodejs/src/ingestion/pipelines/analytics/joined-ingestion-pipeline.ts
Doctor Agent	.claude/agents/ingestion/
测试辅助工具	nodejs/src/ingestion/framework/docs/helpers.ts

应使用哪个 Agent

关注点	Agent	使用场景
步骤结构	pipeline-step-doctor	工厂模式、类型扩展、配置注入、命名
结果处理	pipeline-result-doctor	ok/dlq/drop/redirect/timeout、副作用、警告
组合	pipeline-composition-doctor	构建器链、并发、分组、分支、重试
测试	pipeline-testing-doctor	测试辅助工具、断言、模拟计时器、文档测试风格

快速约定参考

步骤：返回具名内部函数的工厂函数。使用泛型 <T extends Input> 扩展类型。不要使用 any。通过闭包传递配置。

结果：使用 ok()、dlq()、drop()、redirect() 构造函数。副作用以 Promise 的形式传入 ok(value, [effects])。警告作为第三个参数传入。步骤绝不返回 timeout()：只有预算检查点会产生它。

批次时间预算（feed(elements, batchContext, BatchBudget.softDeadline(softAt))）：批次预算耗尽后，框架会停止启动步骤；被截断的元素会以 timeout() 完成，这些元素不会被确认，并会重新投递。.timeoutBarrier() 用于标记最后一个检查点：经过该检查点且仍有剩余时间的元素一定会完成。将它放在第一个满足以下条件的步骤之前：该步骤的写入操作使完成处理比重新投递更省成本；也要放在任何不适合重复执行的副作用之前，因为被截断的元素会重新投递，其处理链会从头再次运行。每次处理一个元素的分组会按 feed 顺序完成一个前缀，但 concurrently 和扇出/扇入不会保留此前缀。详情：nodejs/src/ingestion/framework/docs/18-batch-budgets.test.ts。

组合：messageAware 包装管道。handleResults 放在 messageAware 内部。handleSideEffects 放在其后。按实体处理时使用 concurrentlyPerGroup。分块步骤之前使用 gather。

批处理生命周期钩子（BatchingPipeline 的 beforeBatch/afterBatch）：仅用于增强数据。钩子可以增强元素和批次上下文，但必须原样返回收到的元素——元素数量变化会破坏不变量，feed() 会抛出异常。过滤应放在返回 drop() 的子管道步骤中。空的 feed() 不执行任何操作（不调用钩子，也不占用容量）。详情：nodejs/src/ingestion/framework/docs/14-batching.test.ts。

扇出/扇入（fanOut(fn).via((sub) => …).fanIn(fn)）：执行逐元素子任务，并恢复基数——一个元素扇出为 N 个子元素（例如逐 Blob 上传），常规子管道会处理这些元素（在子级 concurrently 块上设置 maxConcurrency，在逐子元素步骤上设置 retry），扇入则将 OK 结果折叠回父元素。与 concurrently/concurrentlyPerGroup 相比，当并发处理单位小于元素时，应优先使用此方式；在步骤内手动编写 p-limit/Promise.all 就是这种情况的信号。阶段顺序在编译时强制检查（未闭合的阶段无法构建）。子结果约定：收集 OK；静默排除 DROP；DLQ 会以聚合原因使父元素失败；REDIRECT 会被排除并发出警告——子重定向不会逸出该阶段；TIMEOUT 会使父元素超时，且优先于 DLQ。子管道与上下文无关：团队/消息数据应放在子元素值中，受上下文约束的接口（teamAware、handleIngestionWarnings，等等）不可调用。扇出/扇入函数应轻量、同步且具名。父元素会在完成时以无序方式发出。详情：nodejs/src/ingestion/framework/docs/17-fan-out-fan-in.test.ts。

测试：步骤测试直接调用工厂函数。使用 consumeAll()/collectChunks() 辅助函数。异步操作使用 Fake timers。断言结果时使用类型守卫。不要使用 any。

运行所有 doctor

请 Claude“对我最近的更改运行所有管道 doctor”，以便全面审查全部 4 个关注领域。