SSE Event Engine — Redis-Backed Real-Time Events

A four-layer Redis pub/sub and Streams engine that delivers typed, replayable Server-Sent Event envelopes from any backend service to any connected browser without requiring a direct socket between the browser and NestJS.

Business Purpose

Clinical workflows are inherently event-driven: a doctor saves a prescription and the patient portal must update within seconds; a TUCAN AI agent streams reasoning steps as it runs; a voice session transitions state and the UI must reflect it immediately. Polling would add 1–10 s of latency and impose unnecessary load. The SSE engine solves this by maintaining a single persistent connection from the browser to Next.js, which proxies through to a Redis-backed NestJS streaming endpoint.

The engine is the plumbing behind every real-time surface in Dudoxx HMS: AI assistant streaming, SSE-driven UI events, session management, voice state transitions, bulk-upload progress, and org-level announcements. It ships with built-in Last-Event-ID replay (reconnect without data loss) and a contract package (@ddx/sse-contract) that enforces type safety across the monorepo at build time.

Business outcome: A doctor navigating to a patient's active visit sees the AI assistant's reasoning appear word-by-word, with zero UI polling code and automatic reconnect after a network blip — because the event engine logs every envelope in Redis Streams before fan-out, allowing the browser to resume from its last known ID.

Audiences

• Investor: Real-time responsiveness is a clinical differentiator. The architecture handles sub-second latency for AI reasoning streams without WebSockets or long-polling — demonstrating production-grade infrastructure maturity.
• Clinical buyer (doctor/nurse/receptionist): Prescriptions, AI chat responses, appointment status changes, and incoming-message notifications appear instantly — the UI is always live without page refreshes.
• Developer/partner: Integrations can subscribe to typed SSE channels (session:*, user:*, org:*) using the @ddx/sse-contract package. The contract enforces channel construction and envelope shape at compile time.
• Internal (ops/support): Redis DB 1 (separate from cache/session on DB 0) isolates SSE traffic. Streams are trimmed at ~10 000 entries with a 1-hour TTL. The gateway log at sse:log:{channel} allows ops to replay missed events for debugging.

Architecture

The engine is organized into four numbered layers:

The architecture snapshot is anchored in coding_context/ddx-hms-context.md. Cross-cutting auth (guard pipeline) is documented in auth-jwt.md.

Two-transport delivery: every RedisSsePublisher.transportPublish() call first appends to Redis Streams (XADD sse:log:{channel}) then fans out via Redis Pub/Sub (PUBLISH). The browser receives events via Streams (durable, replayable) — the Pub/Sub fan-out serves live PSUBSCRIBE listeners on other NestJS replicas. A pubsubSubscribers=0 log line is NOT a delivery failure; it is the expected steady state when the browser is connected via SseController rather than directly via Pub/Sub.

Contract evolution: The canonical envelope contract is @ddx/sse-contract (v3, promoted 2026-05-18). DDX_SSE_MESSAGE_PUBLISHER_CONSUMER_CONTRACT.md (v2.0.0, 2026-04-23) is preserved for historical reference — all new code imports from packages/ddx-sse-contract/src/.

Tech Stack & Choices

Why @Sse() over WebSockets: SSE is HTTP/1.1-compatible, requires no extra protocol upgrade, and naturally fits the unidirectional (server → client) event shape. Browser reconnect with Last-Event-ID is built into the spec.

Why two Redis clients (SSE_REDIS_CLIENT publisher + a duplicated subscriber): ioredis enters subscriber mode on first SUBSCRIBE command — after that the connection cannot issue normal commands. SseRedisTransport.onModuleInit() duplicates the publisher client (this.publisher.duplicate()) to get a dedicated subscriber connection, sharing all connection parameters without re-reading env vars.

Why Redis DB 1 (SSE_REDIS_DB=1): isolates SSE traffic from the main cache (DB 0) so FLUSHDB in tests does not wipe SSE logs and so memory pressure from Streams does not affect Redis-backed sessions.

SSE_LEGACY_POST_BRIDGE: set to false in production since commit 9631dd7d7. The old POST-based bridge that forwarded events from the legacy TUCAN SSE path is disabled. The @Sse() stream is the only delivery path.

Data Flow

Domain service (e.g. TucanService)
  └── calls RedisSsePublisher.publish(SSEChannels.session(id), envelope)
        ├── Layer C: SseEventLogService.append(channel, envelope)
        │     └── XADD sse:log:{channel} MAXLEN ~ 10000 * envelope <json>
        │           (Redis DB 1, 1-hour TTL on key, ~10k entries max)
        └── Layer C: SseRedisTransport.publish(channel, envelope)
              └── PUBLISH {channel} <json>  (fan-out to subscribed replicas)

Browser (EventSource via ddx-web/src/app/api/sse/route.ts)
  └── GET /api/v1/sse?channels=session:abc,user:xyz [Last-Event-ID: <id>]
        └── SseController.stream() [Layer D]
              ├── validateApiKey(X-API-Key: GATEWAY_API_KEY_NEXTJS)
              ├── if Last-Event-ID: SseReplayService.replay(channels, lastId)
              │     └── XRANGE sse:log:{ch} (<lastId> +   [per channel]
              │           yields buffered envelopes in ts order
              └── SseRedisTransport.subscribe(channel, handler)  [per channel]
                    └── browser receives live MessageEvents via RxJS Observable

Resume semantics: SseReplayService.replay() runs XRANGE per channel with an exclusive start (lastEventId, then merges across channels by envelope.ts ascending. If the requested ID has been MAXLEN-evicted, it emits a synthetic replay:gap envelope so the consumer can trigger a full resync rather than silently losing events.

Heartbeat: SseController emits a typed sse:heartbeat envelope every 25 seconds (HEARTBEAT_INTERVAL_MS = 25_000) to prevent intermediary proxies from closing idle connections. This is documented workaround T018 — NestJS @Sse() does not expose raw : comment keepalive lines.

Channel naming (enforced at build time by packages/ddx-sse-contract/src/channels.ts):

Business outcome → Technical mechanism: A doctor opens an AI session; the browser opens one EventSource to GET /api/v1/sse?channels=session:abc. As the AI reasons, TucanService calls RedisSsePublisher which appends each text:chunk envelope to sse:log:session:abc (Redis Streams) then publishes to the channel. SseController fans the live Observable into the open HTTP response. If the doctor's tab loses network for 3 seconds, the browser automatically reconnects with Last-Event-ID: <last-seen-stream-id> and SseReplayService delivers the missed chunks before re-attaching live.

Implicated Code

• packages/ddx-sse-contract/src/channels.ts:57 — SSEChannels branded channel builder; 7 channel constructors; CI audit blocks raw template literals
• packages/ddx-sse-contract/src/envelope.ts — SseEnvelope<R, N> generic type + wrap() / parseFrame() helpers
• ddx-api/src/platform/sse/sse.module.ts:105 — @Global() module declaration; SseRedisClientProvider factory on line 68 reads REDIS_HOST/REDIS_PORT/SSE_REDIS_DB
• ddx-api/src/platform/sse/sse.controller.ts:82 — SseController; @Public() + manual X-API-Key validation at line 281; HEARTBEAT_INTERVAL_MS = 25_000 at line 47
• ddx-api/src/platform/sse/sse.controller.ts:205 — runReplayThenSubscribe(): replay-before-subscribe ordering enforces monotonic envelope delivery
• ddx-api/src/platform/sse/redis-sse-publisher.ts:27 — RedisSsePublisher extends SsePublisherBase; Streams-first then Pub/Sub at lines 55–56; HIGH_VOLUME_EVENTS set suppresses text:chunk debug noise at line 69
• ddx-api/src/platform/sse/sse-redis.transport.ts:42 — SseRedisTransport; onModuleInit() duplicates the publisher Redis client for the subscriber connection at line 54
• ddx-api/src/platform/sse/sse-event-log.service.ts:41 — SseEventLogService; MAXLEN = 10_000, TTL_SECONDS = 3600, key prefix sse:log:; XADD call at line 77
• ddx-api/src/platform/sse/sse-replay.service.ts:60 — SseReplayService.replay(); replay:gap synthetic envelope at line 151; cross-channel merge by ts at line 79
• ddx-api/src/platform/sse/sse.tokens.ts — DI tokens SSE_PUBLISHER, SSE_REDIS_CLIENT, SSE_TRANSPORT

Operational Notes

Critical pitfalls (from context/REMINDERS.md):

1. SSE_LEGACY_POST_BRIDGE=false — verify this is set in production before every deploy. The POST bridge has been disabled since 9631dd7d7; enabling it on a live cluster would create duplicate delivery paths.
2. Raw template literals forbidden — never write `session:${id}` directly. Always use SSEChannels.session(id). The CI audit script at scripts/audit/sse-event-inventory.ts fails the build on violations. Enforced by packages/ddx-sse-contract/src/channels.ts:57.
3. organizationId vs. org slug drift — SSEChannels.org() must receive the clinic slug, not the UUID. Passing a UUID passes TypeScript's string check but violates the channel naming contract (§9.3 of v2.0.0 contract) and breaks org-wide delivery. Use request.organizationSlug (set by TenantIsolationInterceptor) not request.organizationId.
4. pubsubSubscribers=0 is not a bug — this is the expected steady state. Browsers receive events via Redis Streams (XRANGE), not Pub/Sub. Log messages referencing pubsubSubscribers=0 can be safely ignored.
5. SseController is @Public() — the global GatewayAuthGuard is bypassed. Instead, SseController.validateApiKey() manually checks X-API-Key against GATEWAY_API_KEY_NEXTJS. This is the established pattern — the SSE endpoint is reached only from the trusted Next.js gateway route ddx-web/src/app/api/sse/route.ts.

Env vars:

Validation commands:

# Verify SSE stack health
redis-cli -n 1 keys 'sse:log:*' | head -20          # list active stream keys
redis-cli -n 1 xlen sse:log:session:<id>              # event count for a session
redis-cli -n 1 xrange sse:log:session:<id> - + COUNT 5  # last 5 envelopes

# Check for raw template literal violations (must return 0 hits)
rg --type ts 'agent:session-\$\{' ddx-api/src ddx-web/src

Known defects / future work: SseEventLogService EXPIRE is process-local (expiredKeys: Set<string>). On multi-replica deploys each pod issues EXPIRE once per stream key, which is idempotent but slightly wasteful. A Redis SET-based distributed flag is the clean fix (tracked but not yet scheduled).

Related Topics

• auth-jwt.md — GatewayAuthGuard pipeline runs before SSE channel connect; the SSE endpoint opts out with @Public() and validates the API key inline
• rbac-roles.md — SseController carries @RbacExempt() because role enforcement runs after GatewayAuthGuard; the endpoint is trusted-gateway-only, not user-role-gated
• tenant-isolation.md — org: channels must use the clinic slug resolved by TenantIsolationInterceptor (request.organizationSlug), not the raw X-Clinic-ID UUID