page-updated-source.ts

page-updated-source.tsapps/realtime/src/page-updated-source.tsraw ← src7,278 B← Back to the app
1/**
2 * The `page.updated` event source (TASK-0092 slice 6b — the (b) producer
3 * seam).
4 *
5 * The API publishes every block mutation (REST write AND the CRDT flush)
6 * as a `page.updated` domain event onto the Redis stream `factory:events`
7 * (apps/api `EventBus`, the same stream the worker consumes). The realtime
8 * server consumes that stream and fans a `block:changed` REFETCH-HINT out
9 * to the page's control sockets — the web overlay then re-reads the DB
10 * (the source of truth at rest, correct timestamps).
11 *
12 * The source is an INJECTABLE SEAM (the server tests pass a fake; the
13 * production wiring in index.ts lazily loads ioredis, mirroring the
14 * `RedisUpdateLog` pattern in update-log.ts — the ioredis client
15 * satisfies {@link PageUpdatedStreamClient} structurally). The stream
16 * read itself is the Redis-gated part; every decision made from an event
17 * (which sockets get the hint) is pure and tested without Redis.
18 */
19import {
20  DomainEventName,
21  type DomainEvent,
22} from "@factory/shared";
23import { BlockChangedFrameSchema, type BlockChangedFrame } from "@factory/shared/realtime/frames";
24 
25/** The Redis stream the API's event bus appends to (single source of the
26 *  name — keep in lockstep with `EVENTS_STREAM` in apps/api constants). */
27export const PAGE_UPDATED_STREAM = "factory:events";
28 
29/** The minimal ioredis surface the source needs (a real `Redis` satisfies
30 *  it structurally; the integration tests inject a fake). The signature is
31 *  the RAW keyword form ioredis forwards to Redis (`XREAD COUNT n BLOCK ms
32 *  STREAMS key lastId`) — ioredis's typed overloads accept the literal
33 *  `"STREAMS"`/`"BLOCK"` tokens as the leading args, and that is exactly
34 *  what `readOnce` below sends. */
35export interface PageUpdatedStreamClient {
36  /** `XREAD COUNT {count} BLOCK {block} STREAMS {key} {lastId}` →
37   *  `[[key, [[id, [f1,v1,...]]]], ...]` (an empty array on BLOCK timeout). */
38  xread(
39    ...args: (string | number)[]
40  ): Promise<Array<[string, Array<[string, string[]]>]> | null>;
41}
42 
43export interface PageUpdatedSourceOptions {
44  readonly client: PageUpdatedStreamClient;
45  /** `XREAD BLOCK` wait (ms). Default 5000. */
46  readonly blockMs?: number;
47  /** Max entries per read. Default 100. */
48  readonly count?: number;
49  /** Log line for errors (a throw would kill the loop). */
50  readonly onError?: (err: unknown) => void;
51  /** The last stream entry id consumed (`$` = start from now). */
52  readonly lastId?: () => string;
53}
54 
55export interface PageUpdatedSource {
56  /** Begin the poll loop (no-op if already started). Returns when the
57   *  source is disposed. */
58  start(): void;
59  /** Stop the loop (resolves when the in-flight read settles). */
60  stop(): Promise<void>;
61}
62 
63/**
64 * A stream-polling `page.updated` source: `XREAD` in a loop, parse each
65 * envelope (`{ event, payload(JSON), ts, id }`), keep the `page.updated`
66 * ones (a `pageId` in the payload), and hand the parsed
67 * {@link DomainEvent} to `onPageUpdated`.
68 *
69 * Envelope parsing mirrors the API's `EventBus.publish` field names and
70 * the worker's `eventEnvelopeSchema` (event / payload / ts). `payload` is
71 * re-parsed here and each consumer validates its own shape (defense in
72 * depth) — a malformed entry is logged + SKIPPED, never a crash.
73 */
74export function createPageUpdatedSource(
75  onPageUpdated: (event: DomainEvent) => void,
76  opts: PageUpdatedSourceOptions,
77): PageUpdatedSource {
78  const blockMs = opts.blockMs ?? 5000;
79  const count = opts.count ?? 100;
80  const onError = opts.onError ?? ((err: unknown) => console.error("page-updated-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      PAGE_UPDATED_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        if (body.event !== DomainEventName.PAGE_UPDATED) continue;
110        let payload: Record<string, unknown>;
111        try {
112          payload = JSON.parse(body.payload ?? "{}") as Record<string, unknown>;
113        } catch (err) {
114          onError(err);
115          continue; // malformed payload — skip, never crash the loop
116        }
117        const pageId = payload.pageId;
118        if (typeof pageId !== "string" || pageId.length === 0) continue;
119        onPageUpdated({
120          id: body.id ?? id,
121          name: DomainEventName.PAGE_UPDATED,
122          payload,
123          ts: body.ts ?? new Date().toISOString(),
124        });
125      }
126    }
127  };
128 
129  const tick = async (): Promise<void> => {
130    while (running && !stopped) {
131      try {
132        await readOnce();
133      } catch (err) {
134        onError(err);
135        // Back off so a dead Redis doesn't hot-loop the error path.
136        await new Promise((r) => setTimeout(r, 1000));
137      }
138    }
139  };
140 
141  return {
142    start(): void {
143      if (running || stopped) return;
144      running = true;
145      inFlight = tick();
146      inFlight.finally(() => {
147        inFlight = null;
148      });
149    },
150    async stop(): Promise<void> {
151      if (stopped) return;
152      stopped = true;
153      running = false;
154      if (inFlight !== null) await inFlight;
155    },
156  };
157}
158 
159/**
160 * The (b) fan-out: the `block:changed` REFETCH-HINT for a `page.updated`
161 * event. Pure — no sockets, no Redis (unit-testable):
162 *
163 *  - a `page.updated` WITHOUT a `pageId` is a no-op (defensive; every
164 *    producer carries one);
165 *  - `version` is ADVISORY (int ≥ 1 only) — the client MUST NOT gate on
166 *    it (the refetch is authoritative); a missing/non-integer version is
167 *    dropped from the frame, never a reason to skip the hint;
168 *  - the returned frame is the SHARED `BlockChangedFrame` (the single
169 *    source of truth the server emits and the web consumes — slice 6a).
170 */
171export function blockChangedForPageUpdated(event: DomainEvent): BlockChangedFrame | null {
172  if (event.name !== DomainEventName.PAGE_UPDATED) return null;
173  const payload = event.payload as { pageId?: unknown; version?: unknown } | null;
174  const pageId = payload?.pageId;
175  if (typeof pageId !== "string" || pageId.length === 0) return null;
176  const version = payload?.version;
177  const frame = BlockChangedFrameSchema.safeParse(
178    version !== undefined ? { type: "block:changed", pageId, version } : { type: "block:changed", pageId },
179  );
180  if (frame.success) return frame.data;
181  // A non-integer version is advisory noise — DROP it, still emit the
182  // hint (a bad version is never a reason to skip the refetch).
183  const fallback = BlockChangedFrameSchema.safeParse({ type: "block:changed", pageId });
184  return fallback.success ? fallback.data : null;
185}
186