@nodii/outbox-dispatcher

Producer-side Pattern-3 outbox drainer for the Nodii microservice stack — the symmetric peer of @nodii/replica-consumer (the consumer side). Drains a service's <service>_outbox table to Redis Streams via XADD, with single-active-worker leasing, a retry-backoff cap, a 24h-unpublished ops-alert sweep (D401 — no publish-side DLQ), HWM exposure for snapshot live-stitch, lag/depth metrics, and graceful drain.

Spec: planning hub feature_doc serviceId=nodii-libs docKey=outbox-dispatcher. Implements 01-communication-doctrine § 9.2 (drainer, softened by D209: SET-NX-EX lease is canonical, BullMQ optional) + 09-replication-doctrine § 2.4 (XADD envelope) + § 7 (snapshot HWM cursor).

Quick start

import { defineDispatcher, startDispatcher } from "@nodii/outbox-dispatcher";

const dispatcher = defineDispatcher({
  serviceName: "nodii-tenant-service",
  outboxTable: "outbox",                 // 09-replication § 2.2
  streamFor: (row) => `tenant.${row.topic.split(".")[1]}`,
  leaseKey: "lock:outbox:tenant-service", // D209 SET-NX-EX
  leaseTtlMs: 30_000,                      // renew at ttl/2
  pollMs: 500,                             // § 9.2
  batchSize: 50,                           // § 9.2
  publishMaxAttempts: 8,                   // retry-backoff cap (D401: never deletes/DLQs)
  drainDeadlineSeconds: 30,                // graceful drain
});

const handle = await startDispatcher([dispatcher], { pgPool, redis, telemetry });

// snapshot RPC handler at freeze:
const hwm = handle.currentHwm("tenant.user_membership"); // → "1748100000000-0"

Drain sequence (§ 5.2)

acquire-lease (D209) → SELECT … FOR UPDATE SKIP LOCKED → XADD canonical envelope → UPDATE … SET dispatched_at = now(), stream_id = … → release. At-least-once: the consumer dedupes on event_id via consumed_events. Rows with dispatched_at IS NOT NULL are never re-claimed (idempotent re-XADD on restart).

XADD envelope — consumer contract

The XADD output is byte-identical to what @nodii/replica-consumer reads: a single Redis-Streams field named event carrying JSON.stringify({ event_id, topic, aggregate_kind, aggregate_id, entity_version, payload, stream_id, committed_at }). The outbox column source_entity_version maps to the envelope key entity_version; the producer telemetry_envelope is folded into payload._telemetry. The contract test (tests/contract.test.ts) drives this output back through the consumer's read path to prove round-trip. (The feature_doc § 5.2 lists individual XADD fields + source_entity_version; the shipped consumer reads a single JSON event field keyed entity_version — the consumer's read path is the binding contract per the feature_doc's own statement; drift filed.)

Publish failures + the 24h-unpublished sweep (D401)

There is no publish-side DLQ. Per locked decision D401 (01-comms § 9.2 / 09-replication § 2.5), a failed XADD only increments publish_attempts + records last_error on the outbox row — the row is never deleted and never moved to a <service>_outbox_dlq table, so the at-least-once / replay-from-outbox invariant holds for financial + business events. publishMaxAttempts caps retry backoff only (the claim query stops re-claiming a row once publish_attempts reaches it); it never affects row lifecycle.

Liveness is surfaced by a per-service 24h-unpublished sweep. Schedule sweepUnpublishedOutbox (e.g. hourly) — it selects rows un-dispatched past a threshold (default 24h) and raises an M17 page-tier ops alert (outbox.unpublished_alert, severity=page) carrying outbox_id, topic, event_id, publish_attempts, last_error, age:

import { sweepUnpublishedOutbox } from "@nodii/outbox-dispatcher";

// in your scheduler (BullMQ per 01 § 9.3, or your existing cron):
await sweepUnpublishedOutbox({ pool: pgPool, dispatcher, telemetry });

This is distinct from the subscriber-side replication_dlq (09 § 8.1), which handles consume-side poison applies and is unaffected.

Metrics (§ 5.6)

outbox.depth{topic} · outbox.lag_seconds{topic} · outbox.dispatch_rate{topic} · outbox.publish_failures{topic,reason} · outbox.hwm{stream} · outbox.lease_renewals{result}. D401 sweep: outbox.unpublished_over_threshold{service} (gauge) · outbox.unpublished_alert{topic,service,severity=page} (the M17 alert).

migrate-gen (§ 5.8)

bunx @nodii/outbox-dispatcher migrate-gen --service nodii-tenant-service [--out-dir .]

Emits the <service>_outbox table DDL (D401: no publish-side DLQ table) + a src/outbox/boot.ts skeleton.

Optional @nodii/db-rls (08-rls § 4)

Pass tenantId on the dispatcher + a dbRls handle at startDispatcher and the drain SELECT runs inside withSystemContext({ tenantId }). Without it, the raw pool is used (platform-tenant outbox only).