convergence.test.ts

convergence.test.tstests/integration/realtime/convergence.test.tsraw ← realtime10,169 B← Back to the app
1/**
2 * Realtime integration (TASK-0092 slice 6c — the (b) CONVERGENCE loop,
3 * Docker-free): the full producer chain with NO fakes of the decision
4 * path. A room client edits a block → the room's REAL flush emits
5 * `page.updated` on the server's event sink → a loopback "Redis stream"
6 * (the in-memory stand-in for the API's `factory:events` XADD, recorded
7 * via `eventsTap`) → the REAL `createPageUpdatedSource` poll loop (fake
8 * `XREAD` honoring the raw keyword form + BLOCK pacing) → the REAL
9 * `blockChangedForPageUpdated` derivation → the REAL
10 * `publishBlockChanged` → the `block:changed` refetch-hint arrives on
11 * TWO independent roomless control sockets (the two-tab (b) story: every
12 * tab's overlay re-reads the source of truth at rest after a remote
13 * change).
14 *
15 * What is faked and why (recorded): the `XREAD` transport (Docker-gated —
16 * a real Redis) and the API's REST/flush EMIT side (the server's own
17 * `EventSink` is the EMIT seam the API mirrors). Everything from the event
18 * on (parsing, derivation, fan-out, socket delivery) is the production
19 * code path.
20 */
21import { afterAll, beforeAll, describe, expect, it } from "vitest";
22import * as Y from "yjs";
23import {
24  buildRealtimeFixture,
25  connectClient,
26  mintToken,
27  type RealtimeFixture,
28  type RtClient,
29} from "./rt-helpers";
30import {
31  createPageUpdatedSource,
32  type PageUpdatedStreamClient,
33  type PageUpdatedSource,
34} from "@factory/realtime/src/page-updated-source";
35import { FRAGMENT_NAME, type BlockLike } from "@factory/collaboration/codec";
36import type { DomainEvent } from "@factory/shared";
37 
38const PARA = (id: string, text: string, position: string): BlockLike => ({
39  id,
40  type: "paragraph",
41  content: { text },
42  parentId: null,
43  position,
44});
45 
46/** A loopback stream: the server's emitted events (via `eventsTap`) are the
47 *  XADD side; `xread` is the (fake) XREAD side honoring the raw keyword
48 *  form the real source sends + BLOCK pacing (sleep `blockMs` when there
49 *  is nothing new — the same hot-loop guard the unit-test fake uses). */
50function makeLoopbackStream(blockMs = 5) {
51  const entries: Array<[string, string[]]> = [];
52  let marker = -1; // the index "now" resolves to once the source starts ($ = tail)
53  let counter = 0;
54  const client: PageUpdatedStreamClient = {
55    async xread(...args: (string | number)[]) {
56      // XREAD COUNT n BLOCK ms STREAMS key lastId
57      const cursor = args[args.length - 1] as string;
58      let from: number;
59      if (cursor === "$") {
60        if (marker === -1) {
61          marker = entries.length; // first $ = start from the current tail
62          await new Promise((r) => setTimeout(r, blockMs));
63          return [];
64        }
65        from = marker;
66      } else {
67        const idx = entries.findIndex((e) => e[0] === cursor);
68        from = idx === -1 ? entries.length : idx + 1;
69      }
70      if (from >= entries.length) {
71        await new Promise((r) => setTimeout(r, blockMs)); // BLOCK — nothing new
72        return [];
73      }
74      const slice = entries.slice(from);
75      return [[String(args[5]!), slice]]; // the stream key + its entries
76    },
77  };
78  return {
79    client,
80    /** The XADD side (the API's EventBus mirrors this shape). */
81    xadd(event: DomainEvent): void {
82      const [event_, payload, ts] = [
83        event.name,
84        JSON.stringify(event.payload),
85        event.ts,
86      ];
87      entries.push([`${String(counter++).padStart(6, "0")}`, ["event", event_, "payload", payload, "ts", ts]]);
88    },
89  };
90}
91 
92/** Join + full-sync a page room into `localDoc`. */
93async function joinAndSync(fx: RealtimeFixture, room: string): Promise<{ client: RtClient; doc: Y.Doc }> {
94  const client = await connectClient(fx, { room });
95  await client.nextFrame("hello");
96  client.sendText({ type: "sync.request" });
97  await client.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
98  const doc = new Y.Doc();
99  for (const f of client.textFrames()) {
100    if (f.type === "sync.state" && typeof f.chunk === "string" && f.chunk.length > 0) {
101      Y.applyUpdate(doc, Buffer.from(f.chunk, "base64"), "it");
102    }
103  }
104  for (const b of client.binaryFrames()) {
105    if (b.kind === "update") Y.applyUpdate(doc, b.bytes, "it");
106  }
107  return { client, doc };
108}
109 
110/** The block's `text` slot (find the element by its `id` attr). */
111function textSlotOf(doc: Y.Doc, blockId: string): Y.XmlText {
112  const fragment = doc.getXmlFragment(FRAGMENT_NAME);
113  const walk = (els: Y.XmlElement[]): Y.XmlElement | null => {
114    for (const el of els) {
115      if (el.getAttribute("id") === blockId) return el;
116      const found = walk(el.toArray().filter((c): c is Y.XmlElement => c instanceof Y.XmlElement));
117      if (found !== null) return found;
118    }
119    return null;
120  };
121  const el = walk(fragment.toArray() as Y.XmlElement[]);
122  if (el === null) throw new Error(`block "${blockId}" not found in the doc`);
123  const text = el.toArray().find((c): c is Y.XmlText => c instanceof Y.XmlText && c.getAttribute("name") === "text");
124  if (text === undefined) throw new Error(`block "${blockId}" has no text slot`);
125  return text;
126}
127 
128/** Append to a block's text (a real Yjs edit) + send as a 0x01 frame + await the ack. */
129async function editBlock(client: RtClient, doc: Y.Doc, blockId: string, append: string): Promise<void> {
130  const before = Y.encodeStateAsUpdate(doc);
131  const slot = textSlotOf(doc, blockId);
132  slot.insert(slot.length, append);
133  client.sendBinary(1, Y.encodeStateAsUpdate(doc, before));
134  await client.nextFrame("ack");
135}
136 
137/** Connect a TOKEN-ONLY (roomless) control client + open its hello. */
138async function connectControl(fx: RealtimeFixture, sub: string): Promise<RtClient> {
139  const token = mintToken({ sub, workspaces: ["w1"] });
140  const url = `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}`;
141  const c = await connectClient(fx, { url });
142  await c.nextFrame("hello");
143  return c;
144}
145 
146describe("TASK-0092 6c — the (b) convergence loop (real flush → block:changed → 2 control clients)", () => {
147  let fx: RealtimeFixture;
148  let stream: ReturnType<typeof makeLoopbackStream>;
149 
150  beforeAll(async () => {
151    stream = makeLoopbackStream(5);
152    // The REAL production source, fed by the loopback stream (the in-memory
153    // factory:events). The fixture mirrors the server's EMIT side into it.
154    // A FACTORY (not an instance) because the server builds the sink first
155    // and starts/stops the source in start()/stop().
156    const factory = (on: (e: DomainEvent) => void): PageUpdatedSource =>
157      createPageUpdatedSource(on, { client: stream.client });
158    fx = await buildRealtimeFixture({
159      pages: new Set(["p-cv", "p-other"]),
160      workspaces: { "p-cv": "w1", "p-other": "w1" },
161      levels: { "page:p-cv": "edit", "page:p-other": "edit" },
162      eventsTap: (event) => stream.xadd(event),
163      pageUpdatedSource: factory,
164    });
165    // Let the source's FIRST XREAD (cursor "$" → set the tail marker) run
166    // before the test emits anything, so a flush event is never behind the
167    // marker (the real source starts from "$" = the current tail).
168    await fx.settle(30);
169  });
170 
171  afterAll(async () => {
172    await fx.stop();
173  });
174 
175  it("a room edit → the REAL flush → the REAL source → the hint reaches BOTH control clients of the page (a third client on another page gets nothing)", async () => {
176    fx.store.seed("p-cv", [PARA("a", "base a", "1"), PARA("b", "base b", "2")]);
177 
178    // Two control tabs on the page (different users/tokens — the two-tab
179    // story), plus a control tab on an UNRELATED page.
180    const tab1 = await connectControl(fx, "u-tab1");
181    const tab2 = await connectControl(fx, "u-tab2");
182    const otherTab = await connectControl(fx, "u-other");
183    try {
184      tab1.sendText({ type: "subscribe", channel: "page:p-cv" });
185      tab2.sendText({ type: "subscribe", channel: "page:p-cv" });
186      otherTab.sendText({ type: "subscribe", channel: "page:p-other" });
187      await fx.settle(50);
188 
189      // A real room client edits block "a" → the room is dirty → the 1 s
190      // debounce flush fires → `page.updated` on the sink → the loopback
191      // stream → the real source's XREAD → block:changed → fan-out.
192      const { client, doc } = await joinAndSync(fx, "page:p-cv");
193      try {
194        await editBlock(client, doc, "a", "+edit");
195        const persisted = await client.nextFrame("doc.persisted", 8_000);
196        expect(String(persisted.pageId)).toBe("p-cv");
197 
198        // BOTH control tabs get the refetch-hint (the (b) convergence
199        // signal — each tab's overlay clears + re-reads the blocks).
200        const h1 = await tab1.nextFrame("block:changed", 8_000);
201        const h2 = await tab2.nextFrame("block:changed", 8_000);
202        expect(h1.pageId).toBe("p-cv");
203        expect(h2.pageId).toBe("p-cv");
204        // The version is advisory (the flush's store version ≥ 2).
205        expect(Number(h1.version)).toBeGreaterThanOrEqual(2);
206        expect(h2).toEqual(h1);
207 
208        // The UNRELATED page's control tab gets NOTHING for this flush
209        // (channel isolation through the real path).
210        await expect(otherTab.nextFrame("block:changed", 400)).rejects.toThrow();
211 
212        // The loopback really carried the server's own emission (not a
213        // separately injected event): the sink recorded a page.updated with
214        // the minimal blockDelta (1 update) for THIS flush.
215        const evts = fx.events.pageUpdated("p-cv");
216        expect(evts.length).toBeGreaterThanOrEqual(1);
217        const delta = evts[evts.length - 1]!.payload.blockDelta;
218        expect(delta).toEqual({ inserted: 0, updated: 1, moved: 0, deleted: 0 });
219 
220        // The store converged (the refetch the hint triggers would read
221        // exactly this).
222        const final = (await fx.store.hydrate("p-cv")).blocks!;
223        expect(final.find((b) => b.id === "a")!.content).toEqual({ text: "base a+edit" });
224        expect(final.find((b) => b.id === "b")!.content).toEqual({ text: "base b" });
225      } finally {
226        client.close();
227        doc.destroy();
228      }
229    } finally {
230      tab1.close();
231      tab2.close();
232      otherTab.close();
233    }
234  });
235});
236