fix(core): deliver domain events at least once

.rulesync/rules/messaging-and-microservices.md

@@ -34,7 +34,8 @@ Use the seams and contracts in code; this rule owns channel selection and the sa
 
 - Handlers and jobs are at-least-once: they must be idempotent. Money-adjacent work needs a durable database guard, not only an event ID or idempotency key.
 - Money and any needed-now answer never travel over events. Use a synchronous, transactional command port.
-- `emit()` is best-effort. Use `emitInTransaction()` only when the transactional outbox is explicitly enabled by `OUTBOX_ENABLED`, `AMQP_URL`, or `RABBITMQ_URL`; those AMQP variables enable the outbox but do not bind an AMQP broker.
+- An event reporting a committed change on a money, compliance, KYC, RG or account-status path uses `emitInTransaction(tx, ...)` inside the same transaction (the outbox is always on). `emit()` is best-effort and only for informational, catalog and cache-invalidation events. Both throw on an invalid payload.
+- A throwing handler is redelivered and dead-lettered to `oss:dlq:<topic>` after 5 deliveries (ADR-0045); plugin handlers are deduped per consumer via `processed_events`, a bare `bus.on` without a `consumer` is not.
 - A Redis Streams deployment uses `SERVICE_NAME` as its durable consumer-group identity. Every independently deployed service sharing Redis needs a distinct name.
 - The shipped Redis Streams and BullMQ drivers do not honor `orderingKey`; use an overlay when strict ordering is required.
 - `REALTIME_TRANSPORT` is Redis Pub/Sub (`RedisPubSubRealtimeTransport`, auto-bound on `REDIS_URL`, no in-process fallback - ADR-0031) - fire-and-forget UI push, not a durable event: a message published while nobody is subscribed is simply lost, which is why a realtime channel carries a change _signal_ the client refetches, never the state itself. Channels are prefixed with `SERVICE_NAME` for the same cross-deployment isolation reason as the streams broker's consumer group. `RealtimePresence`/`getOnlineUserIds` are shared across replicas via `RedisPresenceStore`, a per-channel sorted set scored by each member's last heartbeat and read with a TTL cutoff, so a crashed replica's members self-expire instead of lingering forever.

AGENTS.md

@@ -130,7 +130,8 @@ Use the seams and contracts in code; this rule owns channel selection and the sa
 
 - Handlers and jobs are at-least-once: they must be idempotent. Money-adjacent work needs a durable database guard, not only an event ID or idempotency key.
 - Money and any needed-now answer never travel over events. Use a synchronous, transactional command port.
-- `emit()` is best-effort. Use `emitInTransaction()` only when the transactional outbox is explicitly enabled by `OUTBOX_ENABLED`, `AMQP_URL`, or `RABBITMQ_URL`; those AMQP variables enable the outbox but do not bind an AMQP broker.
+- An event reporting a committed change on a money, compliance, KYC, RG or account-status path uses `emitInTransaction(tx, ...)` inside the same transaction (the outbox is always on). `emit()` is best-effort and only for informational, catalog and cache-invalidation events. Both throw on an invalid payload.
+- A throwing handler is redelivered and dead-lettered to `oss:dlq:<topic>` after 5 deliveries (ADR-0045); plugin handlers are deduped per consumer via `processed_events`, a bare `bus.on` without a `consumer` is not.
 - A Redis Streams deployment uses `SERVICE_NAME` as its durable consumer-group identity. Every independently deployed service sharing Redis needs a distinct name.
 - The shipped Redis Streams and BullMQ drivers do not honor `orderingKey`; use an overlay when strict ordering is required.
 - `REALTIME_TRANSPORT` is Redis Pub/Sub (`RedisPubSubRealtimeTransport`, auto-bound on `REDIS_URL`, no in-process fallback - ADR-0031) - fire-and-forget UI push, not a durable event: a message published while nobody is subscribed is simply lost, which is why a realtime channel carries a change _signal_ the client refetches, never the state itself. Channels are prefixed with `SERVICE_NAME` for the same cross-deployment isolation reason as the streams broker's consumer group. `RealtimePresence`/`getOnlineUserIds` are shared across replicas via `RedisPresenceStore`, a per-channel sorted set scored by each member's last heartbeat and read with a TTL cutoff, so a crashed replica's members self-expire instead of lingering forever.

docs/adr/0017-extraction-readiness-manifest-outbox-command-ports.md

@@ -4,6 +4,8 @@
 **Status**: Accepted; implemented (manifest boot + outbox seam + `WALLET_COMMANDS` reference; outbox off by default).
 **Relates to**: ADR-0010 (event-driven broker direction), ADR-0014 (job-queue/realtime seams), ADR-0016 (event envelope).
 
+> **Update (2026-10-13)**: [ADR-0045](./0045-at-least-once-event-delivery.md) made the outbox always on - `OUTBOX_ENABLED` is gone, `createApp` always binds `OUTBOX` and starts the `OutboxRelay`, and money and compliance events are emitted with `emitInTransaction`. The "Opt-in" paragraph below no longer holds. The relay now claims rows atomically with a lease, and consumers dedup per handler through `processed_events` instead of on `eventId` alone.
+
 ## Context
 
 ADR-0010/0016 made the inter-module _event_ transport swappable so a module can be

docs/adr/0030-distributed-only-production-seams.md

@@ -7,6 +7,8 @@
 > **Update (2026-07-25)**: [ADR-0039](./0039-tests-run-the-production-seams.md) took the next step and deleted the four in-process seam impls entirely - the "survive as test-only doubles" and "zero infra" clauses below no longer hold, and Postgres + Redis are now prerequisites for the suite. `InProcessRealtimeTransport`/`SseClientAuthorizer` remain as `createApp`'s production default. The distributed-only production decision itself is unchanged and still in force.
 >
 > **Update (2026-08-31)**: [ADR-0031](./0031-realtime-transport-sse-and-http-over-web-sockets.md) closed the last exception: `InProcessRealtimeTransport` is deleted and `REALTIME_TRANSPORT` joins `assertDurableSeamsBound`, auto-binding `RedisPubSubRealtimeTransport` on `REDIS_URL` with no in-process fallback. Production is now distributed-only for all five durable seams, not four; `SseClientAuthorizer` (a stateless grant-issuer, not a seam with delivery semantics) is unaffected and stays as-is.
+>
+> **Update (2026-10-13)**: [ADR-0045](./0045-at-least-once-event-delivery.md) changed the Redis Streams broker's delivery semantics: an entry is XACKed only when every local handler succeeded, failed entries are redelivered via `XAUTOCLAIM` and dead-lettered to `oss:dlq:<topic>` after 5 deliveries, and retention is `MINID ~ now - 7d` instead of `MAXLEN ~ 10000`. The distributed-only decision is unchanged.
 
 ## Context
 

docs/adr/0045-at-least-once-event-delivery.md

@@ -0,0 +1,77 @@
+# ADR-0045: At-least-once event delivery - always-on outbox, redelivery, dead-letter, dedup
+
+**Date**: 2026-10-13
+**Status**: Accepted
+**Amends**: [ADR-0017](./0017-extraction-readiness-manifest-outbox-command-ports.md) (outbox no longer opt-in), [ADR-0030](./0030-distributed-only-production-seams.md) (the Redis Streams broker's ACK and retention semantics).
+
+## Context
+
+The event path claimed at-least-once delivery but lost or duplicated events in six places:
+
+1. **The outbox was opt-in.** Without `OUTBOX_ENABLED` every money and compliance event went
+   through `emit()`, a fire-and-forget publish after commit. A broker blip or a crash between
+   commit and publish lost the event; a rolled-back transaction could still publish one.
+2. **A failing handler lost its event.** The Redis Streams broker XACKed every entry after
+   dispatch, whether or not a handler threw. Nothing ever retried it.
+3. **Retention was a count.** `MAXLEN ~ 10000` per stream dropped unread entries on a busy topic
+   within minutes, long before a lagging group caught up.
+4. **No consumer-side dedup.** A redelivered event re-ran every handler, including the ones
+   that had already succeeded.
+5. **The relay was single-writer.** Every replica polled the same pending rows with no claim,
+   so two replicas published the same row twice, and one failing row stopped the batch behind it.
+6. **Invalid payloads were logged and delivered anyway**, on both the emit and the receive side.
+
+## Decision
+
+**Outbox always on.** `createApp` always binds `DrizzleOutboxWriter` and starts the
+`OutboxRelay`; `OUTBOX_ENABLED` is gone. `emitInTransaction(tx, ...)` is the rule for any event
+that reports a committed change on a money, compliance, KYC, RG or account-status path: the
+event exists if and only if the transaction commits. `emit()` stays for informational,
+catalog and cache-invalidation events, where losing one on a broker outage is acceptable.
+
+**ACK only on success, redeliver, then dead-letter.** The broker XACKs an entry only when every
+local handler for the topic succeeded. A failed entry stays pending; `XAUTOCLAIM` reclaims it
+after it has been idle for 60s (checked every 30s) and redelivers it. On the 5th delivery
+(`MAX_DELIVERIES`) a still-failing entry is copied to `oss:dlq:<topic>` with its group, stream
+id, delivery count and last error, then XACKed. An entry that is not valid JSON goes straight to
+the DLQ. The DLQ is never trimmed; an operator drains it.
+
+**Time-based retention.** Every `XADD` trims with `MINID ~ now - 7d` (`EVENT_RETENTION_MS`).
+A group more than 7 days behind loses entries; stream memory is traffic x 7 days. Both are
+accepted ceilings; alert on group lag.
+
+**Per-handler dedup in `processed_events`.** Table `processed_events (event_id, consumer)`, PK
+on both, owned by the engine `outbox` migration set. `createApp` subscribes each plugin handler
+with a stable consumer id `<serviceName>/<owner>/<topic>` (`#n` for the nth extra handler a
+plugin registers on the same topic). `EventBus.on` skips a handler whose (eventId, consumer) is
+already recorded, runs it, and records it only on success - so a redelivery re-runs only the
+handlers that failed.
+
+**Relay with an atomic claim and a lease.** A drain claims a batch in one
+`UPDATE ... WHERE event_id IN (SELECT ... FOR UPDATE SKIP LOCKED) RETURNING`, stamping
+`claimed_at`. Concurrent relays never claim the same row inside the 30s lease; a relay that
+dies mid-publish leaves its rows to be reclaimed once the lease expires. A row that fails to
+publish has its lease released and the rest of the batch carries on. The relay polls every
+200ms and, at most once an hour, deletes published outbox rows and `processed_events` rows
+older than 7 days.
+
+**Validate at both ends.** `emit`/`emitInTransaction` throw on a known topic whose payload fails
+its schema (`emitInTransaction` therefore rolls the caller's transaction back). An inbound
+payload that fails validation makes the handler throw, so it follows the redeliver-then-DLQ path.
+
+## Consequences
+
+- Money and compliance events are durable and transaction-atomic. They now publish up to
+  ~200ms after commit (relay poll) instead of immediately.
+- Handlers must still be idempotent. Dedup is check, run, record: two concurrent deliveries
+  of the same event (a reclaim racing a slow first attempt) can both run a handler. Money
+  handlers keep their database guards.
+- A handler registered through `bus.on` without a `consumer` option is not deduped and re-runs
+  on every redelivery of its topic.
+- A consumer id is positional for repeat handlers: reordering two handlers a plugin registers
+  on the same topic swaps their ids, and in-flight events may re-run them once.
+- A poison event retries 5 times over ~5 minutes before it is dead-lettered.
+- Every deployment needs the engine outbox migration applied (`0001_event_delivery`).
+- Events still on `emit()` (gamification payout notifications, geo access informational events,
+  catalog changes) remain best-effort. Moving gamification payouts to `emitInTransaction`
+  needs the event threaded into the payout services' transactions.

docs/platform/architecture.md

@@ -175,8 +175,10 @@ for the full direction; the shape:
   process boundary, a service emits it inside its transaction with
   `events.emitInTransaction(tx, ...)`: the envelope is written to `event_outbox`
   atomically with the state change, and the `OutboxRelay` publishes it after commit
-  (at-least-once; consumers dedup on `eventId`). Opt-in via `OUTBOX_ENABLED` / a
-  durable broker; off in the in-process monolith.
+  (at-least-once). Always on; required for money and compliance events, `emit()`
+  stays for informational ones. A failing handler is redelivered and dead-lettered
+  after 5 attempts; handlers are deduped per consumer via `processed_events`
+  (ADR-0045).
 - **Client push is separate.** WebSockets/SSE are client-facing only (chat,
   balance/bonus toasts), never the transport between modules.
 - **Modular monolith now, microservices later.** The no-cross-module-imports rule,

packages/core/src/audit/plugin.ts

@@ -1,4 +1,4 @@
-import { EVENT_BUS, DRIZZLE, ADMIN_GUARD, createLogger } from '@openora/core/server';
+import { EVENT_BUS, DRIZZLE, ADMIN_GUARD } from '@openora/core/server';
 import type { CoreTokenCatalog, Plugin } from '@openora/core/server';
 import { AUDIT_WRITER, IDENTITY_READER, type DomainEventName } from '@openora/core/contracts';
 import { AuditService, type RecordInput } from './service/audit.service.js';
@@ -1508,8 +1508,6 @@ const SUBSCRIBED_TOPICS: DomainEventName[] = [
 export default {
   id: 'audit',
   register(ctx) {
-    const logger = createLogger('audit');
-
     // Subscriptions are wired before router factories run (create-app.ts boot order),
     // so svcRef is null at registration but set before any real event arrives.
     let svcRef: AuditService | null = null;
@@ -1526,19 +1524,21 @@ export default {
     });
 
     for (const topic of SUBSCRIBED_TOPICS) {
-      ctx.events.on(topic, (payload) => {
-        if (!svcRef || !isRecord(payload)) {
+      // Awaited and thrown, never swallowed: a failed write leaves the event pending, so the
+      // broker redelivers it instead of the change going unaudited (ADR-0045).
+      ctx.events.on(topic, async (payload) => {
+        if (!isRecord(payload)) {
           return;
         }
         // KYC exemptions and game geo rule syncs append their audit record inside the state
         // transaction, then publish this event for realtime and other consumers after commit.
         if (payload['auditRecorded'] === true) {
           return;
         }
-        const svc = svcRef;
-        void mapEventToRecord(topic, payload)
-          .then((record) => svc.record(record))
-          .catch((err) => logger.error({ err, topic }, 'audit record failed'));
+        if (!svcRef) {
+          throw new Error(`audit record for '${topic}' skipped - service not constructed`);
+        }
+        await svcRef.record(await mapEventToRecord(topic, payload));
       });
     }