/** * watch-shape.test.ts — behavioural tests for `watchShape` (src/watch-shape.ts), * against a STATEFUL fake `ng` with a CONTROLLABLE `doc_subscribe`. * * The fake emulates just enough of the broker: * - `doc_create` mints monotonic doc NURIs. * - `sparql_update` parses the shim account writes + the per-entity index * `contains` append + arbitrary anchored triple writes into an in-memory quad * store (same tolerant parser shape as store-registry.test / read-model.test). * - `sparql_query` answers the shim account SELECT, the scope-index `contains` * SELECT, and the anchored per-doc `?s ?p ?o` read (readUnion) — each scoped to * the anchor graph. * - `doc_subscribe` models the platform push order TabInfo→State: on subscribe it * records the callback and fires a `TabInfo` immediately, but the sync BARRIER * `State` is fired only when the TEST releases it (`fireState`) — so we can * assert isPending BEFORE the barrier and isSuccess AFTER. A later write to a * subscribed doc fires a `Patch` push (reactivity). * * These prove the four distinctions the surface exists for: * (a) isPending at first, isSuccess after the first State (barrier); * (b) isSuccess + data:[] on a synced-but-EMPTY scope (the key distinction); * (c) a write then push → data updates (reactivity, no polling); * (d) timed-out → isSuccess (best-effort), NOT isError. */ import { describe, it, expect, mock, beforeEach, afterEach, afterAll } from "bun:test"; import { watchShape } from "../src/watch-shape"; import { configure, configureStoreRegistry, resetStoreRegistry, resetConfig, setCurrentUser, } from "../src/polyfill"; import { resetRegistryCache, createEntityDoc } from "../src/store-registry"; import { resetOpenedRepos, setOpenTimeoutForTests, getSyncState } from "../src/open-repo"; const TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"; const FP = "http://festipod.org/"; const SESSION = { sessionId: "sid-ws", privateStoreId: "PRIV-WS" }; interface Quad { g: string; s: string; p: string; o: string } /** Reverse of escapeLiteral: single left-to-right pass over `\x`. */ function unescapeLiteral(s: string): string { let out = ""; for (let i = 0; i < s.length; i++) { if (s[i] === "\\" && i + 1 < s.length) { const next = s[++i]; out += next === "n" ? "\n" : next === "r" ? "\r" : next === "t" ? "\t" : next!; } else out += s[i]; } return out; } interface SubRec { nuri: string; cb: (r: unknown) => void } /** * The stateful fake with a controllable doc_subscribe. `holdState: true` means a * fresh subscription does NOT auto-fire its `State` — the test fires it via * `fireState(nuri)`. `holdState: false` (default) auto-fires `State` on subscribe * (synced immediately), which is the convenient mode for the reactivity/empty cases. */ function makeFake(opts?: { holdState?: boolean }) { const quads: Quad[] = []; let docCounter = 0; const subs: SubRec[] = []; const hold = opts?.holdState ?? false; // Nuris whose barrier `State` has been released (auto-fire on future subscribe). const released = new Set(); // The shim ANCHOR (private-store-root) is ALWAYS loaded/synced on the real broker // (the store repo is bootstrapped at connect), so its barrier `State` is always // available. `resolveAccount`/`ensureAccount` now open it (the cold-start heal) // before touching the shim — pre-release it here so `holdState` (which gates the // per-ENTITY docs the tests control) never blocks the anchor open. This mirrors the // real invariant the production heal relies on. released.add(`did:ng:${SESSION.privateStoreId}`); let releaseEverything = false; const doc_create = mock(async () => `did:ng:o:doc${++docCounter}`); const sparql_update = mock(async (...a: unknown[]) => { const query = a[1] as string; const anchor = a[2] as string | undefined; const gm = query.match(/GRAPH <([^>]+)>\s*\{([\s\S]*)\}/); let g: string; let body: string; if (gm) { g = gm[1]!; body = gm[2]!; } else { if (!anchor) return undefined; g = anchor; body = query.replace(/^\s*INSERT DATA\s*\{/, "").replace(/\}\s*$/, ""); } const sm = body.match(/<([^>]+)>/); if (!sm) return undefined; const s = sm[1]!; const pairRe = /(?:a|<([^>]+)>)\s+(?:"((?:[^"\\]|\\.)*)"|<([^>]+)>)/g; let m: RegExpExecArray | null; const after = body.slice(body.indexOf(sm[0]) + sm[0].length); while ((m = pairRe.exec(after)) !== null) { const p = m[1] ?? "urn:ng-eventually:shim:Account"; const o = m[2] !== undefined ? unescapeLiteral(m[2]) : (m[3] ?? ""); quads.push({ g, s, p, o }); } // A write to a subscribed doc fires a Patch push (reactivity signal). for (const sub of subs) { if (sub.nuri === g) sub.cb({ V0: { Patch: {} } }); } return undefined; }); const sparql_query = mock(async (...a: unknown[]) => { const query = a[1] as string; const anchor = a[3] as string | undefined; if (query.includes("")) { const subjM = query.match( /GRAPH <[^>]+>\s*\{\s*<([^>]+)>\s+a\s+/, ); const onlySubject = subjM ? subjM[1]! : null; const bySubject = new Map>(); for (const q of quads) { if (q.g !== anchor) continue; if (onlySubject !== null && q.s !== onlySubject) continue; const rec = bySubject.get(q.s) ?? {}; if (q.p === "urn:ng-eventually:shim:id") rec.id = q.o; if (q.p === "urn:ng-eventually:shim:docPublic") rec.docPublic = q.o; if (q.p === "urn:ng-eventually:shim:docProtected") rec.docProtected = q.o; if (q.p === "urn:ng-eventually:shim:docPrivate") rec.docPrivate = q.o; bySubject.set(q.s, rec); } const bindings = [...bySubject.values()] .filter((r) => r.id) .map((r) => ({ id: { value: r.id! }, docPublic: { value: r.docPublic ?? "" }, docProtected: { value: r.docProtected ?? "" }, docPrivate: { value: r.docPrivate ?? "" }, })); return { results: { bindings } }; } if (query.includes("")) { const bindings = quads .filter((q) => q.g === anchor && q.p === "urn:ng-eventually:shim:contains") .map((q) => ({ e: { value: q.o } })); return { results: { bindings } }; } // Anchored per-doc read (readUnion `SELECT ?s ?p ?o`). const bindings = quads .filter((q) => q.g === anchor) .map((q) => ({ s: { value: q.s }, p: { value: q.p }, o: { value: q.o } })); return { results: { bindings } }; }); const doc_subscribe = mock( async (nuri: string, _sid: string, cb: (r: unknown) => void) => { subs.push({ nuri, cb }); // Platform pushes TabInfo FIRST (never the barrier). setTimeout(() => cb({ V0: { TabInfo: {} } }), 0); // Fire the barrier State if this fake auto-syncs, or if this nuri was already // released (so a doc subscribed AFTER a release still crosses the barrier). if (!hold || releaseEverything || released.has(nuri)) { setTimeout(() => cb({ V0: { State: {} } }), 0); } return () => {}; }, ); /** Release the barrier for `nuri` (fire State now + auto-fire for future subs). */ function fireState(nuri: string): void { released.add(nuri); for (const sub of subs) if (sub.nuri === nuri) sub.cb({ V0: { State: {} } }); } /** Release the barrier for EVERY doc, present and future. */ function releaseAll(): void { releaseEverything = true; for (const sub of subs) sub.cb({ V0: { State: {} } }); } return { doc_create, sparql_update, sparql_query, doc_subscribe, _quads: quads, fireState, releaseAll, subs, }; } function inject(ng: ReturnType) { configure({ ng: ng as any, useShape: (() => {}) as any }); configureStoreRegistry({ getSession: async () => SESSION, normalizeId: (u: string) => u.trim().replace(/^@+/, "").toLowerCase(), }); resetRegistryCache(); resetOpenedRepos(); } // Insert a triple straight into a doc's graph in the fake store (no push). function seed(ng: ReturnType, doc: string, p: string, o: string): void { ng._quads.push({ g: doc, s: doc, p, o }); } const tick = () => new Promise((r) => setTimeout(r, 5)); // A minimal SHEX ShapeType pinning rdf:type to `${FP}Event`. const EventShape = { shape: `${FP}EventShape`, schema: { [`${FP}EventShape`]: { iri: `${FP}EventShape`, predicates: [{ iri: TYPE, dataTypes: [{ literals: [`${FP}Event`], valType: "iri" }] }], }, }, }; afterEach(() => { setCurrentUser(null); }); afterAll(() => { resetConfig(); resetStoreRegistry(); resetRegistryCache(); resetOpenedRepos(); }); describe("watchShape", () => { it("(a) isPending at first, then isSuccess after the first State (barrier)", async () => { const ng = makeFake({ holdState: true }); inject(ng); setCurrentUser("alice"); // One protected entity doc for alice, carrying an Event triple. const doc = await createEntityDoc("alice", "protected"); seed(ng, doc, TYPE, `${FP}Event`); seed(ng, doc, `${FP}title`, "Alpha"); const obs = watchShape(EventShape, "protected"); let notes = 0; const unsub = obs.subscribe(() => { notes += 1; }); // Before the barrier: pending, no data. await tick(); expect(obs.getSnapshot().isPending).toBe(true); expect(obs.getSnapshot().isSuccess).toBe(false); expect(obs.getSnapshot().data).toEqual([]); // Release the barrier for every opened doc (present + future) → synced. ng.releaseAll(); await tick(); await tick(); await tick(); const snap = obs.getSnapshot(); expect(snap.isPending).toBe(false); expect(snap.isSuccess).toBe(true); expect(snap.isError).toBe(false); expect(snap.data.length).toBe(1); expect(snap.data[0]!.props[`${FP}title`]).toEqual(["Alpha"]); expect(notes).toBeGreaterThan(0); unsub(); }); it("(b) isSuccess + data:[] on a synced-but-EMPTY scope (the key distinction)", async () => { const ng = makeFake(); // auto-fires State → synced immediately inject(ng); setCurrentUser("bob"); // bob has NO entity docs in this scope — the scope is genuinely empty. const obs = watchShape(EventShape, "protected"); const unsub = obs.subscribe(() => {}); await tick(); await tick(); const snap = obs.getSnapshot(); expect(snap.isPending).toBe(false); expect(snap.isSuccess).toBe(true); // synced, NOT stuck pending expect(snap.isError).toBe(false); expect(snap.data).toEqual([]); // empty — distinguishable from "still syncing" unsub(); }); it("(c) a write then push updates data (reactivity, no polling)", async () => { const ng = makeFake(); // synced immediately inject(ng); setCurrentUser("carol"); const doc = await createEntityDoc("carol", "protected"); seed(ng, doc, TYPE, `${FP}Event`); seed(ng, doc, `${FP}title`, "One"); const obs = watchShape(EventShape, "protected"); const unsub = obs.subscribe(() => {}); await tick(); await tick(); expect(obs.getSnapshot().data.length).toBe(1); // Write a SECOND event doc + fire the push via a write to the ALREADY-subscribed // doc. Because a new doc must appear in the set, write into the scope-INDEX // (createEntityDoc appends to it, and the index is subscribed → re-resolve). const doc2 = await createEntityDoc("carol", "protected"); seed(ng, doc2, TYPE, `${FP}Event`); seed(ng, doc2, `${FP}title`, "Two"); // createEntityDoc's index append fired a Patch on the index doc → re-resolve. await tick(); await tick(); const titles = obs .getSnapshot() .data.flatMap((s) => s.props[`${FP}title`] ?? []) .sort(); expect(titles).toEqual(["One", "Two"]); // No setInterval anywhere — reactivity was push-driven. unsub(); }); it("(d) timed-out → isSuccess (best-effort), NOT isError", async () => { // A doc whose subscription NEVER pushes a `State`: open-repo's bounded fallback // fires and marks the nuri "timed-out" (NOT "synced"). We shrink the fallback to // a few ms so this is fast, and assert the barrier is genuinely reached via // timed-out (getSyncState === "timed-out") and that the snapshot maps that to // isSuccess, never isError. const ng = makeFake({ holdState: true }); // State is never released inject(ng); setOpenTimeoutForTests(20); // fallback fires quickly instead of after 8s setCurrentUser("dave"); const doc = await createEntityDoc("dave", "protected"); seed(ng, doc, TYPE, `${FP}Event`); seed(ng, doc, `${FP}title`, "Timed"); const obs = watchShape(EventShape, "protected"); const unsub = obs.subscribe(() => {}); await tick(); // Before the fallback fires: still pending (subscribed, no State). expect(obs.getSnapshot().isPending).toBe(true); // Let the bounded fallback elapse → open-repo marks each opened doc timed-out. await new Promise((r) => setTimeout(r, 60)); await tick(); await tick(); // The entity doc's barrier resolved via timed-out (never a State). expect(getSyncState(doc)).toBe("timed-out"); const snap = obs.getSnapshot(); expect(snap.isError).toBe(false); expect(snap.isSuccess).toBe(true); // timed-out is best-effort success expect(snap.isPending).toBe(false); // The data still read (best-effort): the doc's triples resolved. expect(snap.data.length).toBe(1); unsub(); }); });