flush-rebase-http.test.ts

flush-rebase-http.test.tstests/integration/realtime/flush-rebase-http.test.tsraw ← realtime12,087 B← Back to the app
1/**
2 * Realtime integration (TASK-0092 slice 4): the flush-rebase exercised
3 * over the PRODUCTION persistence path — the room's `flush()` persists
4 * through a real `HttpDocStore` (a `node:http` fake standing in for the
5 * API's `/internal/*` surface, not an in-memory seam). This spans the
6 * seams no unit test covers: room `flush()` → `HttpDocStore.persist` →
7 * the HTTP flush's `pages.version` CAS → the conflict response
8 * (`externalBlocks`) → the room's slice-1 rebase (absorb the REST write
9 * for blocks it didn't touch) → the retry persist at the fresh version.
10 *
11 * Acceptance criterion 2 (flush-rebase): "a concurrent REST write … while
12 * the room is dirty → the flush's CAS fails → the rebase path runs → the
13 * final state has BOTH the REST write and the CRDT edits (no clobber), and
14 * the next flush is a no-op." Same contract, but the concurrent writer now
15 * arrives through the real flush-CAS over HTTP (the slice-3 REST batch and
16 * the flush serialize on the one `pages.version` column).
17 *
18 * Docker-free: the API is faked (no Postgres); the store, server, and
19 * client are the real production classes. The Postgres-backed
20 * `PrismaDocStore` variant of this scenario is the Docker-gated DoD item
21 * (Testcontainers), recorded in the task file.
22 */
23import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";
24import { AddressInfo } from "node:net";
25import { afterAll, beforeAll, describe, expect, it } from "vitest";
26import * as Y from "yjs";
27import { HttpDocStore } from "@factory/realtime/src/http-store";
28import {
29  buildRealtimeFixture,
30  connectClient,
31  type RealtimeFixture,
32  type RtClient,
33} from "./rt-helpers";
34import {
35  yDocToBlockList,
36  FRAGMENT_NAME,
37  type BlockLike,
38} from "@factory/collaboration/codec";
39 
40const TOKEN = "svc-token";
41 
42const PARA = (id: string, text: string, position: string): BlockLike => ({
43  id,
44  type: "paragraph",
45  content: { text },
46  parentId: null,
47  position,
48});
49 
50interface WireBlock {
51  id: string;
52  parentId: string | null;
53  type: string;
54  position: string;
55  content: Record<string, unknown>;
56  properties?: Record<string, unknown>;
57}
58 
59interface FakeApi {
60  state: {
61    pageId: string;
62    blocks: WireBlock[];
63    version: number;
64    casBase: number;
65  };
66  /** Simulate a REST batch (slice 3): the write lands + the CAS advanced
67   *  the version (a concurrent non-Yjs writer). */
68  restWrite: (blocks: WireBlock[]) => void;
69  server: Server;
70  url: string;
71}
72 
73/** The API's `/internal/blocks/*` surface, faked over node:http (the same
74 *  wire contract the API implements: cursor read + flush CAS). */
75async function startFakeApi(pageId: string): Promise<FakeApi> {
76  const state = {
77    pageId,
78    blocks: [] as WireBlock[],
79    version: 1,
80    casBase: 1,
81  };
82  const server = createServer((req: IncomingMessage, res: ServerResponse) => {
83    const send = (status: number, body: unknown) => {
84      res.writeHead(status, { "content-type": "application/json" });
85      res.end(JSON.stringify(body));
86    };
87    if (req.headers.authorization !== `Bearer ${TOKEN}`) {
88      return send(401, { error: { code: "UNAUTHORIZED", message: "bad token" } });
89    }
90    if (req.method === "GET" && typeof req.url === "string") {
91      const u = new URL(req.url, "http://x");
92      if (u.pathname === `/internal/blocks/${pageId}`) {
93        const pageSize = Number(u.searchParams.get("pageSize") ?? "500");
94        const cursor = u.searchParams.get("cursor");
95        let rows = state.blocks;
96        if (cursor !== null) rows = rows.filter((b) => b.position > (cursor as string));
97        const slice = rows.slice(0, pageSize);
98        const hasMore = rows.length > pageSize;
99        const last = slice[slice.length - 1];
100        return send(200, {
101          pageId,
102          version: state.version,
103          blocks: slice,
104          nextCursor: hasMore && last ? last.position : null,
105          hasMore,
106        });
107      }
108      return send(404, { error: { code: "NOT_FOUND", message: "no page" } });
109    }
110    if (req.method === "POST" && req.url === "/internal/blocks/flush") {
111      let raw = "";
112      req.on("data", (c) => (raw += c));
113      req.on("end", () => {
114        const body = JSON.parse(raw) as { pageId: string; baseVersion: number; blocks: WireBlock[] };
115        if (body.pageId !== state.pageId) return send(404, { error: { code: "NOT_FOUND", message: "no page" } });
116        if (body.baseVersion !== state.casBase) {
117          return send(200, {
118            version: state.version,
119            conflict: { externalBlocks: state.blocks },
120          });
121        }
122        state.blocks = body.blocks;
123        state.version = state.casBase + 1;
124        state.casBase = state.version;
125        return send(200, { version: state.version });
126      });
127      return;
128    }
129    send(404, { error: { code: "NOT_FOUND", message: "no route" } });
130  });
131  await new Promise<void>((r) => server.listen(0, "127.0.0.1", () => r()));
132  const addr = server.address() as AddressInfo;
133  return {
134    state,
135    restWrite: (blocks) => {
136      // A REST batch (slice 3): the write lands + the version advances
137      // (the flush's CAS now sees a stale base → the conflict path).
138      state.blocks = blocks;
139      state.version += 1;
140      state.casBase = state.version;
141    },
142    server,
143    url: `http://127.0.0.1:${addr.port}`,
144  };
145}
146 
147function applyAll(doc: Y.Doc, c: RtClient): void {
148  for (const f of c.textFrames()) {
149    if (f.type === "sync.state" && typeof f.chunk === "string" && f.chunk.length > 0) {
150      Y.applyUpdate(doc, Buffer.from(f.chunk, "base64"), "it");
151    }
152  }
153  for (const b of c.binaryFrames()) {
154    if (b.kind === "update") Y.applyUpdate(doc, b.bytes, "it");
155  }
156}
157 
158async function joinAndSync(fx: RealtimeFixture, room: string): Promise<{ client: RtClient; doc: Y.Doc }> {
159  const client = await connectClient(fx, { room });
160  await client.nextFrame("hello");
161  client.sendText({ type: "sync.request" });
162  await client.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
163  const doc = new Y.Doc();
164  applyAll(doc, client);
165  return { client, doc };
166}
167 
168function textSlotOf(doc: Y.Doc, blockId: string): Y.XmlText {
169  const fragment = doc.getXmlFragment(FRAGMENT_NAME);
170  const walk = (els: Y.XmlElement[]): Y.XmlElement | null => {
171    for (const el of els) {
172      if (el.getAttribute("id") === blockId) return el;
173      const nested = el.toArray().filter((c): c is Y.XmlElement => c instanceof Y.XmlElement);
174      const found = walk(nested);
175      if (found !== null) return found;
176    }
177    return null;
178  };
179  const el = walk(fragment.toArray() as Y.XmlElement[]);
180  if (el === null) throw new Error(`block "${blockId}" not found in the doc`);
181  const text = el.toArray().find((c): c is Y.XmlText => c instanceof Y.XmlText && c.getAttribute("name") === "text");
182  if (text === undefined) throw new Error(`block "${blockId}" has no text slot`);
183  return text;
184}
185 
186function blockTextUpdate(doc: Y.Doc, blockId: string, append: string): Uint8Array {
187  const before = Y.encodeStateAsUpdate(doc);
188  const slot = textSlotOf(doc, blockId);
189  slot.insert(slot.length, append);
190  return Y.encodeStateAsUpdate(doc, before);
191}
192 
193async function editBlock(fx: RealtimeFixture, client: RtClient, doc: Y.Doc, blockId: string, append: string): Promise<void> {
194  void fx;
195  client.sendBinary(1, blockTextUpdate(doc, blockId, append));
196  await client.nextFrame("ack");
197}
198 
199describe("flush-rebase over the HTTP store (TASK-0092 slice 4, acceptance criterion 2)", () => {
200  const pageId = "p-http";
201  let fx: RealtimeFixture;
202  let api: FakeApi;
203  let store: HttpDocStore;
204 
205  beforeAll(async () => {
206    api = await startFakeApi(pageId);
207    store = new HttpDocStore({ apiUrl: api.url, serviceToken: TOKEN });
208    // Seed through the (faked) API, NOT the in-memory store — the server
209    // persists to the HttpDocStore, so the fake IS the database.
210    api.state.blocks = [
211      { id: "a", parentId: null, type: "paragraph", position: "1", content: { text: "base a" } },
212      { id: "b", parentId: null, type: "paragraph", position: "2", content: { text: "base b" } },
213    ];
214    fx = await buildRealtimeFixture({
215      pages: new Set([pageId]),
216      workspaces: { [pageId]: "w1" },
217      levels: { [`page:${pageId}`]: "edit" },
218      store,
219    });
220  });
221  afterAll(async () => {
222    await fx.stop();
223    await new Promise<void>((r) => api.server.close(() => r()));
224  });
225 
226  it("hydration over HTTP: the joiner's doc == the API's blocks (the production hydrate path)", async () => {
227    const { client, doc } = await joinAndSync(fx, `page:${pageId}`);
228    try {
229      const blocks = yDocToBlockList(doc);
230      expect(blocks.map((b) => b.id).sort()).toEqual(["a", "b"]);
231      const a = blocks.find((b) => b.id === "a")!;
232      expect(a.content).toEqual({ text: "base a" });
233      // The room's CAS base == the API's version (1).
234      expect(api.state.version).toBe(1);
235    } finally {
236      client.close();
237      doc.destroy();
238    }
239  });
240 
241  it("a concurrent REST write + a room edit on DIFFERENT blocks → both survive (no clobber), next flush is a no-op", async () => {
242    const { client, doc } = await joinAndSync(fx, `page:${pageId}`);
243    try {
244      // The room's client edits block "a" (the CRDT path).
245      await editBlock(fx, client, doc, "a", "+room edit");
246      // A concurrent REST write (bypassing the room) edits block "b" +
247      // advances pages.version — exactly the race the flush's HTTP CAS
248      // detects (the slice-3 REST batch and the flush on one column).
249      api.restWrite([
250        { id: "a", parentId: null, type: "paragraph", position: "1", content: { text: "base a" } },
251        { id: "b", parentId: null, type: "paragraph", position: "2", content: { text: "base b+rest edit" } },
252      ]);
253      const versionAfterRest = api.state.version;
254 
255      // Wait for the flush (1 s debounce → the HTTP CAS fails → rebase →
256      // the retry persist at the fresh version).
257      const persisted = await client.nextFrame("doc.persisted", 15_000);
258      expect(String(persisted.pageId)).toBe(pageId);
259 
260      // The FINAL store state (the fake API) has BOTH edits (the
261      // acceptance criterion) — the rebase absorbed the REST write for
262      // block "b" (the room didn't touch it) and kept the room's "a".
263      const a = api.state.blocks.find((b) => b.id === "a")!;
264      const b = api.state.blocks.find((b) => b.id === "b")!;
265      expect(a.content).toEqual({ text: "base a+room edit" });
266      expect(b.content).toEqual({ text: "base b+rest edit" });
267 
268      // The retry persist won at the external version + 1 (the rebase
269      // re-based onto the REST write, then flushed over it).
270      expect(Number(persisted.version)).toBe(versionAfterRest + 1);
271 
272      // The flush emitted `page.updated` with the additive blockDelta
273      // (both real changes vs the base → 2 updates).
274      const evts = fx.events.pageUpdated(pageId);
275      expect(evts.length).toBeGreaterThanOrEqual(1);
276      const delta = evts[evts.length - 1]!.payload.blockDelta;
277      expect(delta).toMatchObject({ inserted: 0, updated: 2, moved: 0, deleted: 0 });
278 
279      // The room is now CLEAN + in sync — the NEXT flush is a no-op (the
280      // idempotency assert): a clean persist emits a ZERO blockDelta + the
281      // store content is unchanged (the HTTP CAS now matches, so it writes
282      // the identical set at version + 1, no external conflict).
283      const room = fx.server.manager.get(`page:${pageId}`)!;
284      expect(room.isDirty).toBe(false);
285      await room.flush();
286      const noop = fx.events.pageUpdated(pageId)[fx.events.pageUpdated(pageId).length - 1]!;
287      expect(noop.payload.blockDelta).toEqual({ inserted: 0, updated: 0, moved: 0, deleted: 0 });
288      const unchanged = api.state.blocks;
289      expect(unchanged.find((b) => b.id === "a")!.content).toEqual({ text: "base a+room edit" });
290      expect(unchanged.find((b) => b.id === "b")!.content).toEqual({ text: "base b+rest edit" });
291    } finally {
292      client.close();
293      doc.destroy();
294    }
295  });
296});
297