rebase.test.ts · Source

rebase.test.tstests/integration/realtime/rebase.test.tsraw ← realtime8,657 B← Back to the app
1/**
2 * Realtime integration (TASK-0092, acceptance criteria 2 + 3): the
3 * flush-rebase (a concurrent REST write vs a dirty room → both survive, no
4 * clobber, the next flush is a no-op) and the persistence diff minimality
5 * (the flush's `page.updated` `blockDelta` is the MINIMAL op set; unchanged
6 * blocks are untouched). Reuses TASK-0091's harness (the real server + the
7 * in-memory store, driven by a real `ws` client over real block updates).
8 */
9import { afterAll, beforeAll, describe, expect, it } from "vitest";
10import * as Y from "yjs";
11import {
12  buildRealtimeFixture,
13  connectClient,
14  type RealtimeFixture,
15  type RtClient,
16} from "./rt-helpers";
17import {
18  blocksToYDoc,
19  yDocToBlockList,
20  FRAGMENT_NAME,
21  type BlockLike,
22} from "@factory/collaboration/codec";
23 
24const PARA = (id: string, text: string, position: string): BlockLike => ({
25  id,
26  type: "paragraph",
27  content: { text },
28  parentId: null,
29  position,
30});
31 
32/** Apply every Yjs update (sync.state chunks + binary) to `doc`. */
33function applyAll(doc: Y.Doc, c: RtClient): void {
34  for (const f of c.textFrames()) {
35    if (f.type === "sync.state" && typeof f.chunk === "string" && f.chunk.length > 0) {
36      Y.applyUpdate(doc, Buffer.from(f.chunk, "base64"), "it");
37    }
38  }
39  for (const b of c.binaryFrames()) {
40    if (b.kind === "update") Y.applyUpdate(doc, b.bytes, "it");
41  }
42}
43 
44/**
45 * Join + full-sync a page room into `localDoc` (the client now shares the
46 * server's Yjs item ids). Returns the client + the local doc.
47 */
48async function joinAndSync(fx: RealtimeFixture, room: string): Promise<{ client: RtClient; doc: Y.Doc }> {
49  const client = await connectClient(fx, { room });
50  await client.nextFrame("hello");
51  client.sendText({ type: "sync.request" });
52  await client.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
53  const doc = new Y.Doc();
54  applyAll(doc, client);
55  return { client, doc };
56}
57 
58/** Find a block's `text` slot in a doc by its `id` attr. */
59function textSlotOf(doc: Y.Doc, blockId: string): Y.XmlText {
60  const fragment = doc.getXmlFragment(FRAGMENT_NAME);
61  const walk = (els: Y.XmlElement[]): Y.XmlElement | null => {
62    for (const el of els) {
63      if (el.getAttribute("id") === blockId) return el;
64      const nested = el.toArray().filter((c): c is Y.XmlElement => c instanceof Y.XmlElement);
65      const found = walk(nested);
66      if (found !== null) return found;
67    }
68    return null;
69  };
70  const el = walk(fragment.toArray() as Y.XmlElement[]);
71  if (el === null) throw new Error(`block "${blockId}" not found in the doc`);
72  const text = el.toArray().find((c): c is Y.XmlText => c instanceof Y.XmlText && c.getAttribute("name") === "text");
73  if (text === undefined) throw new Error(`block "${blockId}" has no text slot`);
74  return text;
75}
76 
77/** Edit a block's text (a real Yjs text edit on the shared item) + return the
78 *  delta bytes to send as a 0x01 frame. */
79function blockTextUpdate(doc: Y.Doc, blockId: string, append: string): Uint8Array {
80  const before = Y.encodeStateAsUpdate(doc);
81  const slot = textSlotOf(doc, blockId);
82  slot.insert(slot.length, append);
83  return Y.encodeStateAsUpdate(doc, before);
84}
85 
86/** Edit a block + await the server's ack (the update is applied + logged). */
87async function editBlock(fx: RealtimeFixture, client: RtClient, doc: Y.Doc, blockId: string, append: string): Promise<void> {
88  void fx;
89  client.sendBinary(1, blockTextUpdate(doc, blockId, append));
90  await client.nextFrame("ack");
91}
92 
93describe("flush-rebase (TASK-0092 R2, acceptance criterion 2)", () => {
94  let fx: RealtimeFixture;
95  beforeAll(async () => {
96    fx = await buildRealtimeFixture({
97      pages: new Set(["p-rb"]),
98      workspaces: { "p-rb": "w1" },
99      levels: { "page:p-rb": "edit" },
100    });
101  });
102  afterAll(async () => {
103    await fx.stop();
104  });
105 
106  it("a concurrent REST write + a room edit on DIFFERENT blocks → both survive (no clobber), and the next flush is a no-op", async () => {
107    fx.store.seed("p-rb", [PARA("a", "base a", "1"), PARA("b", "base b", "2")]);
108    const { client, doc } = await joinAndSync(fx, "page:p-rb");
109    try {
110      // The room's client edits block "a".
111      await editBlock(fx, client, doc, "a", "+room edit");
112      // A concurrent REST write (bypassing the room) edits block "b" + bumps
113      // pages.version — exactly the race the flush's CAS detects.
114      fx.store.writeExternal("p-rb", [PARA("a", "base a", "1"), PARA("b", "base b+rest edit", "2")]);
115 
116      // Wait for the flush (1 s debounce → the rebase + the retry persist).
117      const persisted = await client.nextFrame("doc.persisted", 8_000);
118      expect(String(persisted.pageId)).toBe("p-rb");
119 
120      // The FINAL store state has BOTH edits (the acceptance criterion).
121      const final = (await fx.store.hydrate("p-rb")).blocks!;
122      const a = final.find((b) => b.id === "a")!;
123      const b = final.find((b) => b.id === "b")!;
124      expect(a.content).toEqual({ text: "base a+room edit" }); // the room's edit
125      expect(b.content).toEqual({ text: "base b+rest edit" }); // the REST edit
126 
127      // The persist retried at the external version and won (v3).
128      expect(Number(persisted.version)).toBe(3);
129      // A rejoiner's doc == the store (the rebase converged the CRDT).
130      const ref = await joinAndSync(fx, "page:p-rb");
131      ref.doc.destroy();
132      expect(ref.client.textFrames().some((f) => f.type === "error")).toBe(false);
133 
134      // The flush emitted `page.updated` with the additive `blockDelta`
135      // (both real changes vs the base → 2 updates).
136      const evts = fx.events.pageUpdated("p-rb");
137      expect(evts.length).toBeGreaterThanOrEqual(1);
138      const delta = evts[evts.length - 1]!.payload.blockDelta;
139      expect(delta).toMatchObject({ inserted: 0, updated: 2, moved: 0, deleted: 0 });
140 
141      // The room is now CLEAN + in sync — the NEXT flush is a no-op (the
142      // idempotency assert): a clean persist emits a ZERO blockDelta + the
143      // store content is unchanged.
144      const room = fx.server.manager.get("page:p-rb")!;
145      expect(room.isDirty).toBe(false);
146      await room.flush();
147      const noop = fx.events.pageUpdated("p-rb")[fx.events.pageUpdated("p-rb").length - 1]!;
148      expect(noop.payload.blockDelta).toEqual({ inserted: 0, updated: 0, moved: 0, deleted: 0 });
149      const unchanged = (await fx.store.hydrate("p-rb")).blocks!;
150      expect(unchanged.find((b) => b.id === "a")!.content).toEqual({ text: "base a+room edit" });
151      expect(unchanged.find((b) => b.id === "b")!.content).toEqual({ text: "base b+rest edit" });
152    } finally {
153      client.close();
154      doc.destroy();
155    }
156  });
157});
158 
159describe("persistence diff minimality (TASK-0092 R3, acceptance criterion 3)", () => {
160  let fx: RealtimeFixture;
161  beforeAll(async () => {
162    fx = await buildRealtimeFixture({
163      pages: new Set(["p-dm"]),
164      workspaces: { "p-dm": "w1" },
165      levels: { "page:p-dm": "edit" },
166    });
167  });
168  afterAll(async () => {
169    await fx.stop();
170  });
171 
172  it("editing ONE block → the blockDelta is exactly that 1 update; the other blocks are untouched", async () => {
173    fx.store.seed("p-dm", [PARA("a", "alpha", "1"), PARA("b", "beta", "2"), PARA("c", "gamma", "3")]);
174    const { client, doc } = await joinAndSync(fx, "page:p-dm");
175    try {
176      // Edit ONLY block "a".
177      await editBlock(fx, client, doc, "a", " edited");
178      await client.nextFrame("doc.persisted", 8_000);
179 
180      // The blockDelta is MINIMAL: exactly 1 update (block "a"), nothing else.
181      const evts = fx.events.pageUpdated("p-dm");
182      const delta = evts[evts.length - 1]!.payload.blockDelta;
183      expect(delta).toEqual({ inserted: 0, updated: 1, moved: 0, deleted: 0 });
184      // The emitted ops are exactly the one changed block.
185      const ops = evts[evts.length - 1]!.payload.ops as Array<{ op: string; id?: string }>;
186      expect(ops).toHaveLength(1);
187      expect(ops[0]).toMatchObject({ op: "update", id: "a" });
188 
189      // The store: block "a" changed; "b" and "c" UNCHANGED (same content).
190      const final = (await fx.store.hydrate("p-dm")).blocks!;
191      expect(final.find((b) => b.id === "a")!.content).toEqual({ text: "alpha edited" });
192      expect(final.find((b) => b.id === "b")!.content).toEqual({ text: "beta" });
193      expect(final.find((b) => b.id === "c")!.content).toEqual({ text: "gamma" });
194      // The room doc agrees (the CRDT state == the persisted state).
195      expect(yDocToBlockList(doc).map((b) => b.id).sort()).toEqual(["a", "b", "c"]);
196      void blocksToYDoc; // (imported for the round-trip type check)
197    } finally {
198      client.close();
199      doc.destroy();
200    }
201  });
202});
203