Add NATS JetStream ingest path; deprecate Kafka

.gitignore

@@ -28,5 +28,12 @@ settings.yaml
 __debug_bin*
 *.code-workspace
 .history/
-vehicle-triggers-api
+/vehicle-triggers-api
+/nats-testpublisher
+/triggers-bench
+/triggers-state
+/triggers-dlq
 
+CLAUDE.md
+
+triggers-kvbench

NATS_CONTRACT.md

@@ -0,0 +1,152 @@
+# NATS JetStream Ingest Contract
+
+This document is the contract between **producers** (DIS / device-ingest) and
+**vehicle-triggers-api** for the JetStream-based ingest path. It is what a
+producer team needs to publish signals and events that this service will
+evaluate against user webhooks.
+
+It applies when vehicle-triggers-api runs with `NATS_MODE=primary` (Kafka
+bridges into NATS) or `NATS_MODE=exclusive` (NATS only, no Kafka).
+
+---
+
+## ⚠️ Webhook receivers MUST dedup on `data.webhookId`
+
+This is the single most important thing to know if you build something that
+*receives* DIMO webhooks. The pipeline guarantees **at-least-once**
+delivery: the same logical fire can be delivered to your endpoint more than
+once due to JetStream redelivery, replica races during cluster events, or
+in-dispatcher retries on transient network failures.
+
+The CloudEvent ID (`data.webhookId` field, also surfaced as the CloudEvent
+`id` header) is **deterministic**: `sha256(triggerID || sourceID)[:16]`
+where `sourceID` is the inbound signal/event's CloudEvent ID. The same
+logical fire always produces the same `data.webhookId` regardless of how
+many times we deliver it.
+
+**Required receiver behavior:**
+
+```
+seen = persistent_set()                      // any KV / Redis / DB unique key
+on receive(webhook):
+  id = webhook.data.webhookId
+  if id in seen:
+    return 200                              // already processed; ack and move on
+  process(webhook)
+  seen.add(id, ttl=24h)                     // TTL > our redelivery window
+  return 200
+```
+
+`vehicle_triggers_state_cas_conflicts_total{outcome="fallback"}` ticking on
+our side is the public symptom that duplicates are reaching your endpoint.
+If your endpoint isn't dedup'ing on `webhookId`, your downstream side-effects
+WILL execute twice on any meaningful cluster event.
+
+---
+
+## Streams
+
+The service provisions these on startup (idempotent `CreateOrUpdateStream`).
+Producers only need to publish to the signal/event subjects; the service owns
+the audit and DLQ streams.
+
+| Stream | Subjects | Written by | Read by |
+|---|---|---|---|
+| `DIMO_SIGNALS` | `dimo.signals.>` | **producer** | triggers-api |
+| `DIMO_EVENTS` | `dimo.events.>` | **producer** | triggers-api |
+| `DIMO_TRIGGER_AUDIT` | `dimo.trigger.fired.>` | triggers-api | billing/usage |
+| `DIMO_TRIGGER_DLQ` | `dimo.dlq.>` | triggers-api | ops (`cmd/triggers-dlq`) |
+
+Stream names and subjects are configurable via `NATS_*` env (see `sample.env`),
+but the defaults above are the contract.
+
+## Subjects
+
+Cardinality scales with the number of signal/event **names**, not the number
+of vehicles. The vehicle identity travels in the payload, never the subject.
+
+```
+dimo.signals.<signalName>     e.g. dimo.signals.speed
+dimo.events.<eventName>       e.g. dimo.events.harshBraking
+```
+
+`<signalName>` / `<eventName>` are the raw VSS names (no `vss.` prefix on the
+subject). Characters illegal in NATS tokens (space, `.`, `*`, `>`, control
+chars) must be replaced with `_`. The service applies the same sanitization on
+its filter side, so a name containing a `.` (e.g. `powertrain.fuelLevel`)
+becomes the subject token `powertrain_fuelLevel`.
+
+Publishing one message **per signal** (single-element CloudEvent) is preferred
+so consumers can filter precisely. A multi-signal envelope on a single subject
+also works — the service unpacks it — but then the subject name is whichever
+signal you keyed on, which muddies filtering. Prefer one publish per signal.
+
+## Payload
+
+Standard DIMO CloudEvent, JSON-encoded, exactly as on the Kafka topics today.
+
+- Signals: `vss.SignalCloudEvent` (produced by `vss.PackSignals`)
+- Events: `vss.EventCloudEvent` (produced by `vss.PackEvents`)
+
+### Required CloudEvent header fields
+
+| Field | Requirement | Notes |
+|---|---|---|
+| `subject` | **required** | Full ERC721 DID of the vehicle: `did:erc721:<chainId>:<contract>:<tokenId>`. The service decodes this to route to subscriptions. A bad DID fails the message (→ retried → DLQ). |
+| `specversion` | `"1.0"` | |
+| `type` | `dimo.signal` / `dimo.event` | |
+| `time` | RFC3339 | Used for latency metrics. |
+| `producer` | recommended | Surfaced in the webhook payload. |
+| `source` | recommended | Surfaced in the webhook payload. |
+
+### Signal data fields (`vss.SignalData`)
+
+| Field | Notes |
+|---|---|
+| `name` | VSS signal name (e.g. `speed`). Must match the subject token (pre-sanitization). |
+| `timestamp` | Signal observation time. |
+| `valueNumber` / `valueString` / `valueLocation` | One set per the signal's value type. |
+
+## Delivery semantics
+
+- **At-least-once.** The service uses durable pull consumers with explicit
+  ack. A handler failure naks the message; JetStream redelivers per the
+  backoff ladder up to `NATS_MAX_DELIVER` (default 5).
+- **After max deliveries**, the message is republished to
+  `dimo.dlq.<original-subject>` with headers `X-Original-Subject`,
+  `X-Failure-Reason`, `X-Delivered-Count`, then terminated so it stops
+  redelivering. Inspect/replay with `cmd/triggers-dlq`.
+- **Duplicates are possible** (at-least-once + redelivery). Webhook evaluation
+  is idempotent w.r.t. cooldown: the `trigger_state` KV records the last fire
+  per (trigger, vehicle), so a redelivered message inside the cooldown window
+  does not re-fire.
+
+## Audit stream (consumers: billing/usage)
+
+Every successful webhook delivery publishes the full webhook CloudEvent
+payload to:
+
+```
+dimo.trigger.fired.<developerLicense>
+```
+
+`<developerLicense>` is the developer's license address (hex, `0x...`). The
+payload `data` carries `webhookId` (== the trigger ID), `assetDid`,
+`metricName`, `service`, and the signal/event detail. Aggregate per developer
+license for usage accounting. Publishes are async/best-effort and must never
+be assumed to block delivery — treat the stream as eventually-consistent.
+
+## ID semantics
+
+- **Trigger ID == webhook ID.** The API returns `id` on webhook registration;
+  that same value is `triggers.id` in Postgres, the `webhookId` in webhook and
+  audit payloads, and the trigger-half of the `trigger_state` KV key
+  (`<triggerID>.<assetDID>`).
+
+## Versioning
+
+Payload schema follows `vss` from `github.com/DIMO-Network/model-garage`. A
+breaking change to the CloudEvent shape requires coordinating the
+model-garage version across producer and this service. The subject layout
+(`dimo.signals.<name>`) is independent of payload version and is not expected
+to change.

OPERATIONS.md

@@ -0,0 +1,154 @@
+# vehicle-triggers-api Operations Runbook
+
+Operator-facing procedures. Read top to bottom in an incident.
+
+---
+
+## Health quick check
+
+```
+GET /health
+```
+
+Returns 200 when the connection to NATS is up and every configured stream
+(`DIMO_SIGNALS`, `DIMO_EVENTS`, `DIMO_TRIGGER_AUDIT`, `DIMO_TRIGGER_DLQ`)
+returns a valid `Info` with a leader. Returns 503 with a JSON body listing
+per-stream status when any of those probes fails.
+
+If 503: check `nats stream report` against the cluster; usually a node lost
+quorum or a stream lost a replica.
+
+---
+
+## Metric surfaces
+
+Prometheus namespace: `vehicle_triggers`.
+
+| Metric | Use |
+|---|---|
+| `vehicle_triggers_nats_publish_total{stream, outcome}` | Per-stream publish health (`ok` / `error`). |
+| `vehicle_triggers_nats_consume_total{stream, outcome}` | Consume outcomes (`ack`, `nak`, `dlq`). A non-zero `dlq` rate means poison messages are accumulating - inspect with `triggers-dlq list`. |
+| `vehicle_triggers_nats_eval_latency_seconds{stream, outcome}` | Histogram, JetStream timestamp to handler return. SLO surface. |
+| `vehicle_triggers_state_cas_conflicts_total{bucket, outcome}` | Concurrent writers on the same `(trigger, vehicle)`. `retry` = resolved on retry; `fallback` = persistent, ended in unconditional Put. A steady non-zero rate means duplicate webhook fires are happening; receivers must dedup via the deterministic CloudEvent ID. |
+| `vehicle_triggers_state_decode_errors_total{bucket}` | KV record JSON decode failures. Non-zero = schema mismatch or corruption. |
+| `vehicle_triggers_tokenexchange_cache_total{outcome}` | Permission cache outcomes. Steady-state hit rate should be > 90%. Spikes in `miss` at autoscale events are expected. |
+
+---
+
+## Inspection tools
+
+| Tool | Use |
+|---|---|
+| `triggers-state list / get / dump / watch` | Per-(trigger, vehicle) cooldown KV (`trigger_state` bucket). |
+| `triggers-dlq list / get / replay / purge` | Dead-letter stream. Use `replay --all` after fixing the upstream cause; messages re-enter the normal evaluation path. |
+| `triggers-bench` | Standalone load + latency probe. Useful for capacity tests on a fresh cluster. |
+
+---
+
+## Daily JetStream backup (Helm)
+
+Set in `values.yaml`:
+
+```yaml
+backup:
+  enabled: true
+  schedule: "0 6 * * *"
+  image: <internal-image-with-nats-and-aws-cli>
+  s3Bucket: my-backup-bucket
+  s3Prefix: vehicle-triggers-api/jetstream
+```
+
+The CronJob runs `nats stream backup` for each stream and KV bucket, tarballs
+the result, and uploads to `s3://<bucket>/<prefix>/vehicle-triggers-backup-<ts>.tar.gz`.
+
+`natsio/nats-box` does not include the AWS CLI. For S3 uploads in prod,
+bake an image that includes both and set `backup.image`.
+
+---
+
+## Restore from backup
+
+Restoring is destructive: it deletes the existing stream/KV and re-creates
+from the snapshot. Make sure traffic is paused first (scale the service to
+0 replicas) so the consumer isn't fighting the restore.
+
+```sh
+# 1. Pull the tarball
+aws s3 cp s3://my-backup-bucket/vehicle-triggers-api/jetstream/vehicle-triggers-backup-<ts>.tar.gz .
+tar -xzf vehicle-triggers-backup-<ts>.tar.gz
+cd backup-<ts>
+
+# 2. Restore each stream
+for d in stream-*; do
+  name="${d#stream-}"
+  nats stream restore --force "$name" "$d"
+done
+
+# 3. Restore each KV (stored as KV_<bucket>)
+for d in kv-*; do
+  name="${d#kv-}"
+  nats stream restore --force "KV_${name}" "$d"
+done
+
+# 4. Scale the service back up
+kubectl scale deployment vehicle-triggers-api --replicas=<N>
+```
+
+Verify with `/health` and `triggers-state dump` / `triggers-dlq list`.
+
+---
+
+## Common incidents
+
+### Webhook deliveries silently failing
+
+1. Check `vehicle_triggers_nats_consume_total{outcome="dlq"}` rate.
+2. `triggers-dlq list` to see what's poisoned, `triggers-dlq get <seq>` for
+   one record.
+3. If failures are receiver-side and resolved, `triggers-dlq replay --all`.
+4. If failures look like our parsing, log into a pod and inspect the body.
+
+### CAS conflict counter rising
+
+This means two replicas are racing on the same `(trigger, vehicle)`. The
+state store is still correct (last write wins) but webhooks are firing
+twice. Receivers should dedup on the CloudEvent ID. If they can't, see
+PROD_HARDENING.md P1 "decouple webhook delivery from eval" - that's the
+fix.
+
+### Token-exchange cache hit rate drops
+
+`vehicle_triggers_tokenexchange_cache_total{outcome=miss}` rate spikes
+typically happen at autoscale-up (cold caches on new pods). Sustained drop
+in hit rate means either:
+
+- token-exchange-api increased its TTL → ours is still 15m default, mismatch
+- subscriptions ballooned → cache is being evicted
+
+Tune `TOKEN_EXCHANGE_CACHE_EXPIRATION` if needed.
+
+### Latency histogram p99 climbs
+
+Buckets are 1ms-30s. p99 > 100ms typically means slow webhook receivers
+(HTTP dispatch is in-line). Mitigate: get receiver to fix; or build the P1
+async dispatcher.
+
+### NATS cluster lost a node
+
+`/health` returns 503 with the failed stream named. JetStream auto-recovers
+when the node returns. If permanent: rebalance replicas with `nats stream
+edit --replicas`.
+
+---
+
+## Cutover procedure
+
+Documented in PROD_HARDENING.md under "Cutover gates". Summary:
+
+- merge with `NATS_MODE=off`
+- enable in dev (`NATS_MODE=primary`)
+- soak 1 week
+- enable in staging
+- enable in prod
+- once DIS publishes natively, flip to `NATS_MODE=exclusive`
+- after exclusive is clean in prod, remove Kafka deps in a follow-up PR