1/** 2 * @factory/realtime — the generic `factory:events` stream source 3 * (TASK-0094; generalizes the TASK-0092 6b `page-updated-source`). 4 * 5 * The API publishes EVERY domain event onto the Redis stream 6 * `factory:events` (apps/api `EventBus`). The realtime app polls that ONE 7 * stream and routes events to fan-outs: 8 * - `page.updated` → `block:changed` refetch-hint (TASK-0092 6b) 9 * - `comment.*` → `comment.event` room frames (TASK-0094 R1) 10 * - `notification.created` → `notification.push` control-room push (R7) 11 * 12 * This module is the shared STREAM POLLING core (the `XREAD` loop, the 13 * envelope parse, the malformed-entry skip). It is an INJECTABLE SEAM (the 14 * server tests pass a fake client; the production wiring in index.ts lazily 15 * loads ioredis). The stream read is the Redis-gated part; every decision 16 * made from an event is pure and tested without Redis (the per-event 17 * builders live in `comments.ts` / `notifications.ts` / 18 * `page-updated-source.ts`). 19 * 20 * The client interface is the SAME as the old `PageUpdatedStreamClient` 21 * (kept exported from `page-updated-source.ts` for lockstep — a real 22 * `ioredis` `Redis` satisfies it structurally). 23 */ 24 25import type { DomainEvent } from "@factory/shared"; 26 27/** The Redis stream the API's event bus appends to (single source of the 28 * name — keep in lockstep with `EVENTS_STREAM` in apps/api constants). */ 29export const DOMAIN_EVENTS_STREAM = "factory:events"; 30 31/** The minimal ioredis surface the source needs (a real `Redis` satisfies 32 * it structurally; the integration tests inject a fake). The signature is 33 * the RAW keyword form ioredis forwards to Redis (`XREAD COUNT n BLOCK ms 34 * STREAMS key lastId`). */ 35export interface DomainEventStreamClient { 36 /** `XREAD COUNT {count} BLOCK {block} STREAMS {key} {lastId}` → 37 * `[[key, [[id, [f1,v1,...]]]], ...]` (an empty array on BLOCK timeout). */ 38 xread(...args: (string | number)[]): Promise<Array<[string, Array<[string, string[]]>]> | null>; 39} 40 41export interface DomainEventSourceOptions { 42 readonly client: DomainEventStreamClient; 43 /** `XREAD BLOCK` wait (ms). Default 5000. */ 44 readonly blockMs?: number; 45 /** Max entries per read. Default 100. */ 46 readonly count?: number; 47 /** Log line for errors (a throw would kill the loop). */ 48 readonly onError?: (err: unknown) => void; 49 /** The last stream entry id consumed (`$` = start from now). */ 50 readonly lastId?: () => string; 51} 52 53export interface DomainEventSource { 54 /** Begin the poll loop (no-op if already started). Returns nothing. */ 55 start(): void; 56 /** Stop the loop (resolves when the in-flight read settles). */ 57 stop(): Promise<void>; 58} 59 60/** 61 * A stream-polling domain-event source: `XREAD` in a loop, parse each 62 * envelope (`{ event, payload(JSON), ts, id }`), and hand the parsed 63 * {@link DomainEvent} to `onEvent` for EVERY event name (the caller 64 * routes by `event.name`). 65 * 66 * Envelope parsing mirrors the API's `EventBus.publish` field names and the 67 * worker's `eventEnvelopeSchema` (event / payload / ts). `payload` is 68 * re-parsed here and each consumer validates its own shape (defense in 69 * depth) — a malformed entry is logged + SKIPPED, never a crash. The 70 * envelope carries no top-level `actorId`/`workspaceId` (the API's bus 71 * does not emit them); the per-event payload carries whatever the emitter 72 * included (TASK-0094 emitters include `actorId` — recorded). 73 */ 74export function createDomainEventSource( 75 onEvent: (event: DomainEvent) => void, 76 opts: DomainEventSourceOptions, 77): DomainEventSource { 78 const blockMs = opts.blockMs ?? 5000; 79 const count = opts.count ?? 100; 80 const onError = opts.onError ?? ((err: unknown) => console.error("domain-event-source:", err)); 81 const lastId = opts.lastId ?? (() => "$"); 82 83 let running = false; 84 let stopped = false; 85 let cursor = lastId(); 86 let inFlight: Promise<void> | null = null; 87 88 const readOnce = async (): Promise<void> => { 89 // Raw keyword form — ioredis forwards the args to Redis verbatim 90 // (XREAD COUNT n BLOCK ms STREAMS key lastId); the fake and a real 91 // client both see the same shape. Empty array / null = BLOCK timeout. 92 const res = await opts.client.xread( 93 "COUNT", 94 count, 95 "BLOCK", 96 blockMs, 97 "STREAMS", 98 DOMAIN_EVENTS_STREAM, 99 cursor, 100 ); 101 if (res === null || res.length === 0) return; 102 for (const [, entries] of res) { 103 for (const [id, fields] of entries) { 104 cursor = id; 105 const body: Record<string, string> = {}; 106 for (let i = 0; i + 1 < fields.length; i += 2) { 107 body[fields[i]!] = fields[i + 1]!; 108 } 109 const name = body.event; 110 if (typeof name !== "string" || name.length === 0) continue; 111 let payload: Record<string, unknown>; 112 try { 113 payload = JSON.parse(body.payload ?? "{}") as Record<string, unknown>; 114 } catch (err) { 115 onError(err); 116 continue; // malformed payload — skip, never crash the loop 117 } 118 onEvent({ 119 id: body.id ?? id, 120 name: name as DomainEvent["name"], 121 payload, 122 ts: body.ts ?? new Date().toISOString(), 123 }); 124 } 125 } 126 }; 127 128 const tick = async (): Promise<void> => { 129 while (running && !stopped) { 130 try { 131 await readOnce(); 132 } catch (err) { 133 onError(err); 134 // Back off so a dead Redis doesn't hot-loop the error path. 135 await new Promise((r) => setTimeout(r, 1000)); 136 } 137 } 138 }; 139 140 return { 141 start(): void { 142 if (running || stopped) return; 143 running = true; 144 inFlight = tick(); 145 inFlight.finally(() => { 146 inFlight = null; 147 }); 148 }, 149 async stop(): Promise<void> { 150 if (stopped) return; 151 stopped = true; 152 running = false; 153 if (inFlight !== null) await inFlight; 154 }, 155 }; 156} 157