Implementation plan — drop PD-1 SQS handoff (ADR-0001)

Companion to ADR-0001. Sequenced so each PR is independently green and the executor is never left calling a non-existent interface.

End state

PD-1 pipeline (per FireEvent)
  stage 1 candidate → 2 CCS enrich → 3 eligibility → 4 ranking → 5 copy
  → 6 cohort flag (fire-time) → 7 DISPATCH
                                   │
                                   ▼
              internal/dm/dispatch.Dispatcher  (was: SQS Handoff)
                 - map HandoffMessage → notifclient.NotifyRequest (PD-2 §4.1)
                 - send_time translation → scheduled_time (PD-2 §4.3)
                 - per-type weekly cap (Valkey Lua, PD-2 §4.5)
                 - POST /v1/gateway/notify via notification-service SDK
                   (github.com/fetch-rewards/notification-service/pkg/client)
                 - inline bounded retry on 5xx/timeout
                 - producer="pd2" audit UpdateItem (PD-2 §5.2)
                                   │
                                   ▼
                       notification-service gateway
                 (dedup / quota / quiet-hours / scheduling — all gateway-owned)

No SQS. No separate consumer process. PD-2’s logic runs inline as the dispatch adapter.

What changes vs. what stays

Stays as-is (already correct):

The 7-stage pipeline shape and stages 1–6.
HandoffMessage struct — it’s a fine internal carrier of the assembled fields; we keep it as the input to the dispatcher (rename optional, see PR2).
The synchronous ScheduleEvent audit (internal/dm/audit, internal/dm/scheduleevent).
The gateway’s ownership of dedup/quota/quiet-hours/scheduling.

Changes:

executor.Handoff interface: Publish(ctx, HandoffMessage) (sqsMessageID string, err) → executor.Dispatcher: Dispatch(ctx, HandoffMessage) (DispatchResult, err) where DispatchResult carries the gateway notification_id + the PD-2 outcome.
Pipeline.Handoff Handoff field → Pipeline.Dispatcher Dispatcher. Stage 7 comment + the Publish call site in pipeline.go.
FireResult.SQSMessageID → FireResult.NotificationID (audit row column follows).
The scheduleevent row: drop sqs_message_id, add notification_id (gateway response). PD-2 §5.2 already expected notification_id; this just makes PD-1 write it directly.

Deleted:

The (never-built) SQS handoff adapter — no code to remove, just don’t build it.
{env}-dm-delivery-queue + {env}-dm-delivery-dlq FSD deps in consumer-graph-worker.yml.

PR sequence

PR-A — spec edits (this PR, doc-only). ADR-0001 + the PD-1 / PD-2 spec diffs below. No code. Merges first so the design is the source of truth before implementation.

PR-B — executor interface rename (Handoff → Dispatcher).

deps.go: rename interface + method; HandoffMessage kept; add DispatchResult.
pipeline.go: stage-7 rename; call Dispatcher.Dispatch; map result → FireResult.
types.go: FireResult.SQSMessageID → NotificationID.
Update pipeline_test.go fakes. Pure rename + signature change; no behavior yet.
Green: build/vet/test.

PR-C — scheduleevent codec: sqs_message_id → notification_id, add scheduled_at.

internal/dm/scheduleevent: swap sqs_message_id → notification_id (field + tag); add scheduled_at (RFC3339 UTC, omitempty) — the gateway-returned delivery instant, so the funnel can answer “fired 00:00 UTC → scheduled 08:00 user-local.” Update NewRow.
internal/dm/audit + fake follow (PD-1’s fire-time write leaves notification_id / scheduled_at empty; the dispatch step fills them on its producer="pd2" UpdateItem).
Snowflake column note: sqs_message_id → notification_id is a straight rename (nothing dispatched in prod); scheduled_at is additive. Confirm both with the PLT-629 owner.

PR-D — internal/dm/dispatch adapter (the real work).

New package implementing executor.Dispatcher.
Holds a *notifclient.Client (the notification-service/pkg/client SDK already vendored and used by internal/notification/weeklyforecast/adapter.go).
Dispatch does: field map (PD-2 §4.1) → send_time translation (PD-2 §4.3) → per-type cap check (Valkey Lua, PD-2 §4.5) → client.Send → outcome mapping (PD-2 §4.6 table) → inline bounded retry on 5xx/408/network → producer="pd2" audit UpdateItem.
Capture scheduled_at from the gateway’s 202 response (NotifyResponse.scheduled_time, per the SDK) into DispatchResult and onto the audit row alongside notification_id, so the funnel records the resolved delivery instant, not just fire_time.
Reuse the cap Lua + outcome enum already specified in PD-2 §4.5/§5.1.
Tests: stub the SDK transport (same http.RoundTripper pattern as the CCS enricher tests); cover 202/202-idempotent/429/dropped/5xx-retry/cap-exceeded.

PR-E — FSD: remove the queue + DLQ deps.

Delete the {env}-dm-delivery-queue (+ DLQ) aws_sqs_queue block from consumer-graph-worker.yml. Confirm nothing else references it.
This is the only infra change; the DDB audit table + Snowflake projection are unchanged.

PR-F — wire it up in cmd/unified-worker.

Construct the dispatch.Dispatcher with the gateway client + cap + auditor; inject into the Pipeline. (This rides with the broader worker-registration PR already on the roadmap.)

Open coordination items

PLT-629 column: sqs_message_id → notification_id in the Snowflake projection. Nothing dispatched in prod yet, so a straight rename is safe — confirm with the funnel owner before PR-C merges.
Growth sign-off: this shifts retry + scheduling reliance fully onto the gateway. Share ADR-0001 with Steve/Growth; the retry-durability tradeoff is the one item they may want to weigh in on (does any DM type need at-least-once delivery stronger than in-process retry + visible failed audit + next-tick re-fire?).
Per-type cap host: the Valkey Lua cap now runs inside the unified-worker dispatch path. Same Valkey the gateway uses? If so, mind the PR-#148 lesson — keep cap ops cheap (one EVAL per dispatch, no churn).