Cross-Service Event Delivery (Redis Streams)

Origem

services/foundation/id/engine/services/identity/internal/service/event_bus.go auto-documenta: "intentionally simple — a KDB-backed durable queue can replace it later without changing the publisher interface." Esse "later" chegou: tickets #091, #092, #093 precisam que auth/sso/keys services consumam identity.user.erased cross-process.

Ticket #097 levantou 3 opções (gRPC RPC, durable queue, webhook reuse). Esta spec ratifica Opção B — Durable queue via Redis Streams.

Decisão (2026-05-14)

Redis Streams é o substrate escolhido para event delivery cross-service até que kdb-stream layer da RFC-001 shipe, momento em que a migração é drop-in (mesma semântica pub/sub).

Razões:

1. Reuse-first — pkg/redisclient/redisclient.go já existe no engine; sem nova infra
2. Self-hosted-first — Redis é OSS, self-hostable, deployment já estabelecido em ops
3. Consumer groups — retry + offset + DLQ built-in; cada serviço tem seu próprio progresso
4. Migration path — kdb-stream futuro usa XADD/XREADGROUP semantics paralelas
5. Drop simplicity vs NATS JetStream — Redis já está in-stack; NATS exigiria nova infra
6. Webhook_deliveries reuse rejeitada — semantic stretch (intended for external webhooks, polling latency ruim pra cascade)

Contrato (R-rules)

R1 — Stream key naming

Único stream global por serviço-origem:

Prefixo koder:events: reservado.

R2 — Event envelope

Cada XADD carrega campos chave-valor (Redis Streams native format):

XADD koder:events:id * \
  type identity.user.erased \
  event_id evt_<ulid> \
  occurred_at 2026-05-14T12:34:56Z \
  tenant_id tenant-A \
  subject_id user-u123 \
  actor system \
  data '{"erasure_id":"ERS-2026-05-14-abc12345"}'

Campos obrigatórios:

• type — event type (e.g. identity.user.erased)
• event_id — globally unique (ULID); idempotency key for consumers
• occurred_at — ISO 8601 UTC
• tenant_id — tenant scope
• subject_id — primary subject (user_id, workspace_id, etc.)
• actor — who/what triggered (system, user:<id>, admin:<id>)
• data — JSON-encoded event-specific payload

R3 — Consumer group naming

Cada serviço consumidor cria seu próprio consumer group:

<service-name>-<purpose>
e.g.:
  auth-erasure-cascade
  sso-erasure-cascade
  keys-erasure-cascade

Pattern: <service>-<purpose>. Group names são globally unique per stream.

R4 — Consumer behavior

1. Boot-time: XGROUP CREATE koder:events:id <group> $ MKSTREAM (start from now; future events only)
2. Loop: XREADGROUP GROUP <group> <consumer-id> COUNT 100 BLOCK 5000 STREAMS koder:events:id >
3. Per message: deserialize → handle (idempotent — same event_id may be re-delivered)
4. Success: XACK koder:events:id <group> <message_id>
5. Failure: log + leave un-ack'd; XPENDING tracks; retry via XCLAIM after timeout
6. Hard failure (>3 retries): XCLAIM + move to DLQ stream koder:events:id:dlq for ops

R5 — Idempotency contract

Consumers MUST be idempotent per event_id. Recommended:

• Maintain processed_events table with event_id PK + TTL of 24h
• Skip if event_id already processed within window
• For erasure cascade: anonimization/deletion are naturally idempotent (re-applying = no-op)

R6 — Publisher behavior

Publisher (e.g. ErasureService.ExecuteErasure):

1. Generate event_id via ULID
2. XADD koder:events:id MAXLEN ~ 100000 * type ... event_id ... data ...
3. MAXLEN ~ 100000 (~ approx) trims stream to last 100k events; older purge automatically
4. On Redis failure: log + emit fallback in-process EventBus (existing) for SAME-process consumers; cross-service consumers retry on next poll if Redis recovers

R7 — TLS + auth

• Production: Redis TLS + AUTH password (env KODER_REDIS_URL=rediss://user:pass@host:6379/0)
• Development: plain Redis OK
• Connection pooling via pkg/redisclient (existing)

R8 — Observability

Publishers + consumers emit slog:

• Publish: event_emitted {type, event_id, subject_id, stream}
• Consume start: event_received {type, event_id, group, consumer}
• Consume ack: event_acked {type, event_id, elapsed_ms}
• Consume fail: event_failed {type, event_id, error, retry_count}
• DLQ: event_dlq {type, event_id, final_error} (warn level)

Metrics (Prometheus, future): event_count_total, event_lag_seconds, event_dlq_total.

R9 — Migration to kdb-stream

When kdb-stream layer ships per RFC-001:

1. Add KODER_EVENT_BUS=kdb-stream env flag
2. New pkg/eventbus/kdb_stream.go impl satisfying same publisher/consumer interfaces
3. Migration script copies in-flight stream state from Redis → kdb-stream
4. Cutover: env flip + restart services
5. Redis decommission

Drop-in because XADD/XREADGROUP have direct parallels in kdb-stream design.

Implementation

Wave 1 (now) — publisher abstraction in engine

• services/foundation/id/engine/pkg/eventbus/ package
• Publisher interface (Publish method) + Redis Streams impl
• Consumer interface (Subscribe + Process methods) + Redis Streams impl
• Wire publisher in ErasureService.ExecuteErasure (replace direct identity.PublishUserErased with bus.Publish; in-process EventBus keeps firing for SAME-process consumers via dual-emit)

Wave 2 — consumers in sibling services

• services/auth/internal/listener/erasure_listener.go (ticket #091)
• services/sso/internal/listener/erasure_listener.go (ticket #092)
• services/keys/internal/listener/erasure_listener.go (ticket #093)

Wave 3 — DLQ + observability + metrics

• DLQ stream + admin endpoint to inspect/replay
• Prometheus metrics

Test contract

T1 — Round-trip happy path

Publisher → XADD → Consumer XREADGROUP → handler invoked → XACK. Assert: message acked, handler received correct envelope.

T2 — Consumer idempotency

Same event delivered twice → handler invoked twice but business state changes once. Assert: anonimization/deletion idempotent.

T3 — Retry on failure

Handler returns error → message stays un-ack'd → XCLAIM after timeout → retry. Assert: 2nd attempt invokes handler.

T4 — DLQ on persistent failure

Handler fails 3× → message moved to DLQ stream. Assert: original stream cleared, DLQ has entry with error.

T5 — Concurrent consumers

2 consumer instances in same group → messages distributed (load balanced). Assert: each msg processed exactly once across instances.

N1 — Redis down on publish

Publisher Redis call fails → falls back to in-process EventBus + retry on next event. Assert: same-process consumers still work; cross-service consumers catch up on Redis recover.

N2 — Malformed event

Consumer receives event with missing required field → moves to DLQ immediately, doesn't crash.

Não-escopo

• Same-process pub/sub: existing in-process EventBus continues for badges, webhooks, etc. This spec only covers CROSS-process delivery.
• Synchronous request/reply (RPC pattern) — out-of-scope; events are fire-and-forget.
• Event sourcing / replay-from-zero — Wave 3 of erasure flow (#094 covers replay from erasure_requests table; not from stream).

Referências

• Redis Streams docs: https://redis.io/docs/data-types/streams/
• Consumer groups: https://redis.io/docs/data-types/streams/#consumer-groups
• RFC-001 kdb-as-unified-data-plane § Layers (kdb-stream future)
• services/foundation/id/engine#097 (decision ticket)