ingestion-pipeline-doctor-nodejs

パイプラインドクター

PostHog の取り込みパイプラインフレームワークと、その規約をチェックするエージェントのクイックリファレンスです。

アーキテクチャの概要

取り込みパイプラインでは、型付きで合成可能なステップチェーンを通じてイベントを処理します。

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
ドクターエージェント	.claude/agents/ingestion/
テストヘルパー	nodejs/src/ingestion/framework/docs/helpers.ts

使用するエージェント

対象	エージェント	使用する場面
ステップの構造	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() コンストラクターを使います。副作用は ok(value, [effects]) のように Promise として指定します。警告は第 3 引数で指定します。ステップは timeout() を返しません。これを生成するのは予算のチェックポイントだけです。

バッチの時間予算（feed(elements, batchContext, BatchBudget.softDeadline(softAt))）：バッチの予算が尽きると、フレームワークはステップの開始を停止します。打ち切られた要素は timeout() として完了し、ack されずに再配信されます。.timeoutBarrier() は最後のチェックポイントを示します。ここを過ぎた要素は、残り時間があれば必ず処理を完了します。書き込みを行うことで再配信よりも処理完了のほうが低コストになる最初のステップより前に配置してください。また、打ち切られた要素は再配信されてチェーンの先頭から再実行されるため、繰り返すと安全でない副作用より前にも配置してください。グループが一度に 1 つの要素を処理する場合、フィード順の先頭部分が完了しますが、concurrently とファンアウト／ファンインではこの順序は維持されません。詳細は nodejs/src/ingestion/framework/docs/18-batch-budgets.test.ts を参照してください。

合成：messageAware でパイプラインをラップします。messageAware の内側で handleResults を使います。その後に handleSideEffects を配置します。エンティティごとの処理には concurrentlyPerGroup を使います。チャンクのステップの前に gather を配置します。

バッチ処理のライフサイクルフック（BatchingPipeline の beforeBatch/afterBatch）：フックはエンリッチのみを行います。フックは要素とバッチコンテキストをエンリッチできますが、受け取った要素を正確にそのまま返す必要があります。件数が変わると不変条件が破られ、feed() は例外をスローします。フィルタリングは drop() を返すサブパイプラインのステップで行います。空の feed() は何もせず、フックも容量も消費しません。詳細は nodejs/src/ingestion/framework/docs/14-batching.test.ts を参照してください。

ファンアウト／ファンイン（fanOut(fn).via((sub) => …).fanIn(fn)）：要素ごとのサブ処理を行い、要素数を元に戻します。1 つの要素を 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() ヘルパーを使います。非同期処理にはフェイクタイマーを使います。結果のアサーションには型ガードを使います。any は使いません。

すべてのドクターを実行する

Claude に「最近の変更に対してすべてのパイプラインドクターを実行して」と依頼すると、4 つすべての懸念領域を対象とした包括的なレビューが行われます。