page-updated-source.test.ts

page-updated-source.test.tsapps/realtime/test/page-updated-source.test.tsraw ← test7,506 B← Back to the app
1/**
2 * Unit tests for the `page.updated` control source (TASK-0092 6b).
3 *
4 * The stream read is faked (a hand-rolled {@link PageUpdatedStreamClient});
5 * every decision the server makes from an event — envelope parsing, the
6 * `page.updated` filter, the `block:changed` frame derivation — is pure and
7 * asserted here without Redis.
8 */
9import { describe, expect, it } from "vitest";
10 
11import { DomainEventName } from "@factory/shared";
12import { createPageUpdatedSource, blockChangedForPageUpdated, type PageUpdatedStreamClient } from "../src/page-updated-source.js";
13import type { DomainEvent } from "@factory/shared";
14 
15/** A controllable fake Redis stream client (records every XREAD). Honors
16 *  `BLOCK`: when nothing is queued it waits `blockMs` (a real
17 *  `XREAD BLOCK` blocks up to blockMs; a queued entry is returned
18 *  immediately). This paces the poll loop exactly like production. */
19function makeFakeClient(blockMs = 3) {
20  const calls: (string | number)[][] = [];
21  const queue: (Array<[string, Array<[string, string[]]>]> | null)[] = [];
22  const client: PageUpdatedStreamClient = {
23    // Raw keyword form (XREAD COUNT n BLOCK ms STREAMS key lastId) — the
24    // last arg is the cursor, so the cursor-advance assertions below hold.
25    async xread(...args: (string | number)[]) {
26      calls.push(args);
27      if (queue.length === 0) {
28        await new Promise((r) => setTimeout(r, blockMs));
29        return []; // BLOCK timeout — no entries (ioredis returns [])
30      }
31      return queue.shift() ?? [];
32    },
33  };
34  return {
35    client,
36    calls,
37    /** Queue one read result (consumed by the next read). */
38    prime(result: Array<[string, Array<[string, string[]]>]> | null): void {
39      queue.push(result);
40    },
41  };
42}
43 
44const sleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));
45/** Let the (paced) poll loop consume a primed read + settle. */
46const settle = (): Promise<void> => sleep(40);
47 
48const entry = (id: string, event: string, payload: Record<string, unknown>): [string, string[]] => [
49  id,
50  ["event", event, "payload", JSON.stringify(payload), "ts", "2026-09-26T00:00:00.000Z", "id", `evt_${id}`],
51];
52 
53describe("createPageUpdatedSource", () => {
54  it("delivers a page.updated entry as a parsed DomainEvent (pageId kept in the payload)", async () => {
55    const fake = makeFakeClient();
56    const seen: DomainEvent[] = [];
57    const source = createPageUpdatedSource((e) => seen.push(e), {
58      client: fake.client,
59    });
60    source.start();
61    fake.prime([["factory:events", [entry("1", "page.updated", { pageId: "pg_1", version: 7 })]]]);
62    await settle();
63    await source.stop();
64    expect(seen).toHaveLength(1);
65    expect(seen[0]?.name).toBe(DomainEventName.PAGE_UPDATED);
66    expect(seen[0]?.payload).toEqual({ pageId: "pg_1", version: 7 });
67    expect(seen[0]?.ts).toBe("2026-09-26T00:00:00.000Z");
68  });
69 
70  it("ignores non-page.updated events (the stream carries the whole domain event bus)", async () => {
71    const fake = makeFakeClient();
72    const seen: DomainEvent[] = [];
73    const source = createPageUpdatedSource((e) => seen.push(e), { client: fake.client });
74    source.start();
75    fake.prime([
76      [
77        "factory:events",
78        [
79          entry("2", "page.created", { pageId: "pg_1" }),
80          entry("3", "comment.created", { commentId: "c1" }),
81          entry("4", "page.updated", { pageId: "pg_2", version: 1 }),
82        ],
83      ],
84    ]);
85    await settle();
86    await source.stop();
87    expect(seen).toHaveLength(1);
88    expect((seen[0]?.payload as { pageId: string }).pageId).toBe("pg_2");
89  });
90 
91  it("skips a page.updated WITHOUT a pageId (defensive — no crash, no event)", async () => {
92    const fake = makeFakeClient();
93    const seen: DomainEvent[] = [];
94    const source = createPageUpdatedSource((e) => seen.push(e), { client: fake.client });
95    source.start();
96    fake.prime([["factory:events", [entry("5", "page.updated", { version: 9 })]]]);
97    await settle();
98    await source.stop();
99    expect(seen).toHaveLength(0);
100  });
101 
102  it("skips a MALFORMED payload (logged via onError, never crashes the loop)", async () => {
103    const fake = makeFakeClient();
104    const seen: DomainEvent[] = [];
105    const errors: unknown[] = [];
106    const source = createPageUpdatedSource((e) => seen.push(e), {
107      client: fake.client,
108      onError: (err) => errors.push(err),
109    });
110    source.start();
111    fake.prime([["factory:events", [["6", ["event", "page.updated", "payload", "{not-json", "ts", "t"]]]]]);
112    await settle();
113    fake.prime([["factory:events", [entry("7", "page.updated", { pageId: "pg_ok", version: 1 })]]]);
114    await settle();
115    await source.stop();
116    expect(errors).toHaveLength(1);
117    // The loop survived the bad entry and still delivered the next one.
118    expect(seen).toHaveLength(1);
119    expect((seen[0]?.payload as { pageId: string }).pageId).toBe("pg_ok");
120  });
121 
122  it("advances the cursor past delivered entries (XREAD starts at the last id)", async () => {
123    const fake = makeFakeClient();
124    const source = createPageUpdatedSource(() => {}, { client: fake.client });
125    source.start();
126    fake.prime([["factory:events", [entry("42", "page.updated", { pageId: "pg_1" })]]]);
127    await settle();
128    await source.stop();
129    // ioredis form: [count, block, key, lastId] — the last entry is the cursor.
130    const reads = fake.calls;
131    expect(reads.length).toBeGreaterThanOrEqual(2);
132    // The read AFTER the entry must start at "42" (the cursor advanced).
133    const after = reads.find((c) => c[c.length - 1] === "42");
134    expect(after).toBeDefined();
135    expect(reads[0]![reads[0]!.length - 1]).toBe("$");
136  });
137 
138  it("does nothing after stop() (the loop exits)", async () => {
139    const fake = makeFakeClient();
140    const seen: DomainEvent[] = [];
141    const source = createPageUpdatedSource((e) => seen.push(e), { client: fake.client });
142    source.start();
143    await source.stop();
144    fake.prime([["factory:events", [entry("9", "page.updated", { pageId: "pg_1" })]]]);
145    await settle();
146    expect(seen).toHaveLength(0);
147  });
148});
149 
150describe("blockChangedForPageUpdated (the (b) fan-out derivation)", () => {
151  const event = (payload: Record<string, unknown>, name: string = DomainEventName.PAGE_UPDATED): DomainEvent => ({
152    id: "evt_1",
153    name: name as DomainEvent["name"],
154    payload,
155    ts: "2026-09-26T00:00:00.000Z",
156  });
157 
158  it("derives a block:changed refetch-hint with the advisory version", () => {
159    expect(blockChangedForPageUpdated(event({ pageId: "pg_1", version: 7 }))).toEqual({
160      type: "block:changed",
161      pageId: "pg_1",
162      version: 7,
163    });
164  });
165 
166  it("derives the hint WITHOUT a version when the event has none (a REST write)", () => {
167    expect(blockChangedForPageUpdated(event({ pageId: "pg_1" }))).toEqual({ type: "block:changed", pageId: "pg_1" });
168  });
169 
170  it("drops a non-integer version (advisory only — the hint still goes out)", () => {
171    expect(blockChangedForPageUpdated(event({ pageId: "pg_1", version: 1.5 }))).toEqual({
172      type: "block:changed",
173      pageId: "pg_1",
174    });
175  });
176 
177  it("returns null for a non-page.updated event", () => {
178    expect(blockChangedForPageUpdated(event({ pageId: "pg_1" }, DomainEventName.PAGE_CREATED))).toBeNull();
179  });
180 
181  it("returns null when the payload has no pageId", () => {
182    expect(blockChangedForPageUpdated(event({ version: 3 }))).toBeNull();
183    expect(blockChangedForPageUpdated(event({ pageId: "" }))).toBeNull();
184  });
185});
186