PD-1 Kafka Mode — Implementation Scope

PD-1 Kafka Mode — Implementation Scope

How to add Kafka-triggered DMs to the PD-1 executor: the firing mode that turns one Kafka message into 0..N FireEvents through the same 7-stage pipeline. This is the gating dependency for the Earnings / Post-Scan Coach (event-driven, not scheduled).

Status (2026-06-18). Cron mode is code-complete (review-ready PR stack #228–#237, #244). Kafka mode is not built — only the executor seams exist. One-off / future-fire mode is explicitly out of scope here (deferred, per the team call).

Decisions (2026-06-18):

• Receipt coverage = physical + Shop orders → BOTH topics. Unblocked research confirmed coverage is split and non-overlapping: physical receipts on pipeline-v1-processed-receipt-events (Status=FINISHED, non-digital), Shop/e-commerce orders on factoid-stream (Source=SHOP, Name=ORDER_FINALIZED), and digital receipts are not ingested today (DigitalReceiptData != nil is dropped). The Coach fires on physical + Shop, so the handler consumes both topics. Digital is excluded (relaxing it is separate upstream work).
• Multi-topic handler from day one. DMKafkaComponent registers N (topic, serde, extractor) bindings — the factoid topic uses Buf CSR protobuf, the receipt topic plain protobuf, so the serde is per-binding.
• The Points Optimization Engine call lives in the generic Enrich stage (ADR-0002 #241), not the extractor. The Coach’s extractor just shapes the raw receipt/order payload into a candidate; the POE enrichment is a registered enricher. This ties PR 3 below to ADR-0002.

Companion docs: dm-execution-e2e.md §3.2 (firing modes), decisions/0001-drop-pd1-sqs-handoff.md.

---

1. What already exists (the seams)

Kafka mode is deliberately easy to add because the pipeline was built mode-agnostic. Already present and tested:

• FireEvent.EventPayload []byte (internal/dm/executor/types.go) — raw Kafka bytes, nil for cron. TriggerKind: kafka is a defined constant.
• registry.ExtractorFunc + RegisterExtractor / LookupExtractor (internal/dm/registry/callable.go) — the symbol-table slot for an event-payload extractor.
• CandidateAdapter.Candidates (internal/dm/runner/runner.go) already dispatches candidate_query.kind: event_payload: it resolves the registration’s extractor ref and invokes fn(ctx, fe.EventPayload) → []any candidates. Stage 1 needs no change.
• The whole 7-stage Pipeline.Run is identical for both modes — once a FireEvent exists, enrich → eligibility → ranking → copy → flag → dispatch → audit are mode-blind.
• PD-3 schema already models trigger.kind: kafka (topic + optional filter) and candidate_query.kind: event_payload (extractor ref), with loader validation that an event_payload candidate query is only legal when trigger.kind == kafka.
• CGW’s Kafka consumer framework (internal/kafka/, cmd/unified-worker/kafka_consumer.go) — the yakl/franz/msk machinery, the KafkaConsumerComponent lifecycle, the feature-flag gate + Valkey idempotency patterns the factoid/receipt handlers use. We reuse this; we don’t build a new consumer stack.

So the missing work is the firing mode itself — the thing that produces FireEvents from a Kafka message — plus its wiring and the per-message concerns (idempotency, backpressure, flag-gating, user_id sourcing). The pipeline, the dispatch step, and the audit are done.

2. What’s missing (the work)

2.1 A DM Kafka handler — internal/dm/kafkamode/ (new package)

The analogue of internal/dm/executor/cron.go’s CronLoop, but message-driven instead of tick-driven. One handler instance per Kafka-triggered DM topic (a topic may serve more than one DM type — dispatch by type_id / filter). Per message:

message arrives (yakl.Message)
  │
  ├─ idempotency: dedup on the message's natural key (e.g. receipt_id) via Valkey,
  │  24h TTL — reuse the factoid/receipt handler pattern. Already-seen → ack, skip.
  │
  ├─ kill-switch gate (process-level + per-type flag.kill_switch) — mirror the cron-tick gate;
  │  off → ack, skip (no FireEvent, no audit, same rule as cron §4 / audit-boundary nuance).
  │
  ├─ for each registered kafka type whose trigger.topic matches (+ trigger.filter, if set):
  │     mint dm_run_id (uuid4) for this (message, type)
  │     build a run-level FireEvent{TriggerKind: kafka, EventPayload: msg.Value, TypeID, ...}
  │     pipeline stage 1 runs the type's extractor → 0..N candidates, each carrying its user_id
  │     for each candidate: mint session_id, build per-user FireEvent, pipeline.Run(...)
  │
  └─ ack the message only after all fan-out FireEvents have run (at-least-once; see 2.4)

Key differences from cron mode, each a real design point:

• No RunClaimer. Kafka gives each message to exactly one consumer in the group already, so the multi-replica claim cron needs is unnecessary. (Idempotency replaces it — see 2.4.)
• dm_run_id is per message, not per tick. One Kafka message = one dm_run_id grouping its fan-out (PD-1 §4.3). For most event types the extractor yields one candidate (the user on the receipt), but the contract allows N (e.g. one candidate per receipt item).
• user_id comes from the event payload, via the extractor — never from anything external. Same identity invariant as cron; the extractor is responsible for surfacing it.
• Backpressure (PD-1 §4.3): the fan-out must not block the Kafka commit loop unboundedly. A bounded intake buffer / worker pool in front of pipeline.Run, sized like cron’s pool, with the message ack gated on completion. This is the one genuinely new concurrency concern.

2.2 The extractor for the first Kafka type (Earnings Coach)

The DM-type package (e.g. internal/dm/types/earningscoach/) registers an ExtractorFunc at init:

func init() { registry.RegisterExtractor("earningscoach.extract:from_receipt", fromReceipt) }
func fromReceipt(ctx context.Context, message any) ([]any, error) { /* parse payload → candidate(s) */ }

The extractor deserializes the topic’s payload (protobuf, via the topic’s serde — receipt topic is plain protobuf per CLAUDE.md) into the candidate type the Coach’s eligibility/copy callables expect. This is where the Points Optimization Engine call lands if ADR-0002 (generic Enrich stage) is accepted — otherwise it’s a candidate-side concern. Flag here as the ADR-0002 intersection.

{receipt_id} becomes resolvable in the idempotency-key template for Kafka types (today renderIdempotencyKey rejects {receipt_id} as cron-only — see pipeline.go): the Kafka handler must thread the event’s receipt_id into the FireEvent so the renderer can fill it. Small executor change: add a ReceiptID/event-key field to FireEvent (or a generic EventKeys map[string]string) and teach renderIdempotencyKey to resolve it in Kafka mode.

2.3 Wiring — cmd/unified-worker/

• A DMKafkaComponent (sibling to DMExecutorComponent) constructed in the composition root, registering the handler on the matching topic via the existing yakl.NewConsumerHandler + TopicHandlerMap pattern, under the LifecycleManager. Gated by config (dark by default, same posture as DMExecutorEnabled).
• Config: the DM-trigger topic name(s), consumer group, worker-pool size. The registry already tells us which type_ids are trigger.kind: kafka and their topics — the component can derive its topic subscriptions from the loaded registry rather than hardcoding.
• FSD: consumer-side IAM/MSK config for the new consumer group (no new infra beyond the subscription, same as the factoid/receipt consumers).

2.4 Idempotency & delivery semantics (the careful part)

• At-least-once + idempotent. Kafka redelivers on rebalance/crash. The two dedup layers: (a) Valkey message-key dedup in the handler (don’t re-run the pipeline for a message already processed), and (b) the gateway’s 24h request_id dedup (the rendered idempotency key), which makes a re-fired send safe even if (a) misses. Same belt-and-suspenders the cron path leans on.
• Ack ordering. Ack after the fan-out completes (or after a durable audit row exists per fire), so a crash mid-fan-out redelivers rather than silently dropping. Mirror the factoid handler’s commit discipline.
• Audit. Unchanged — each per-user FireEvent writes its ScheduleEvent row exactly as cron does (trigger_kind: kafka distinguishes them). producer=pd2 dispatch update (#244) applies identically.

3. Explicit non-goals

• One-off / future-fire mode — deferred (team call). The queue code exists (#234) but stays unwired; not part of this work.
• No pipeline changes beyond the small FireEvent event-key addition (2.2) for {receipt_id} idempotency. Stages 2–7 are untouched.
• No new delivery path — Kafka-fired DMs dispatch through the same stage-7 gateway client.

4. Work breakdown (suggested PRs, stackable)

#	PR	Scope	Depends on
1	executor: event-key on FireEvent	Add EventKeys/ReceiptID to FireEvent; teach renderIdempotencyKey to resolve {receipt_id} in Kafka mode; tests	merged cron stack
2	internal/dm/kafkamode/ handler	The message→FireEvents fan-out, bounded worker pool, kill-switch gate, Valkey dedup; unit-tested with a fake pipeline + fake consumer	PR 1
3	Earnings Coach type pkg + extractor	internal/dm/types/earningscoach/ extractor + eligibility + copy; dm/types/earnings-coach.yaml (trigger.kind: kafka, candidate_query.kind: event_payload); registry resolves it at load	PR 1 (+ ADR-0002 if the POE call goes in Enrich)
4	wiring: DMKafkaComponent	composition root + config + FSD consumer IAM; derive topic subscriptions from the registry	PRs 2, 3
5	integration test	message → fan-out → dispatch → 2 audit rows, for the Coach; redelivery/idempotency test	PRs 2–4

PR 1 is small and unblocks the rest; PRs 2 and 3 are parallel; PR 4 stacks on both; PR 5 last.

5. Open questions

1. One topic, many types? Likely yes (a single DM-trigger topic, dispatch by type_id + trigger.filter). Confirm whether Earnings Coach reuses the existing receipt topic or gets a dedicated DM-trigger topic — affects serde + consumer-group config.
2. FireEvent event-key shape — a typed ReceiptID string (simple, Coach-specific) vs. a generic EventKeys map[string]string (extensible to future Kafka types’ idempotency placeholders). Recommend the generic map; it’s the same cost and avoids a second migration.
3. Extractor ↔ ADR-0002 boundary — does the POE call live in the extractor (candidate stage) or in a registered enricher (stage 2, per ADR-0002)? If ADR-0002 lands first, the enricher is the cleaner home and the extractor just shapes the raw candidate.
4. Backpressure sizing — receipt volume is far higher than cron audiences; confirm the worker-pool/buffer sizing and whether the kill-switch/flag gate should run before the per-type fan-out to shed load cheaply when a type is dark.