diff --git a/packages/client/e2e/run.ts b/packages/client/e2e/run.ts index 260747e..9af6b3c 100644 --- a/packages/client/e2e/run.ts +++ b/packages/client/e2e/run.ts @@ -294,6 +294,47 @@ async function main(): Promise { check("scope resolvers return NURIs (private distinct from protected/public)", !!r.priv && !!r.prot && !!r.pub && r.priv !== r.prot, `priv=${r.priv?.slice(0,16)}… prot=${r.prot?.slice(0,16)}…`); }); + // ── watchShape (reactive useQuery-shaped read) ────────────────────────── + // The real cycle: first subscription reads isPending (barrier not yet crossed), + // then after the broker sync it reaches isSuccess with the seeded datum present. + // A separate empty scope reaches isSuccess with data:[] (synced-but-empty — the + // distinction useShape's upgrade will make native, surfaced here from getSyncState). + console.log("\n── watchShape (reactive useQuery-shaped read) ──"); + await step("watchShape: first subscribe isPending → isSuccess with data present", async () => { + const h = "cyc" + Date.now(); + const CLS = "urn:e2e:ws:Event"; + const seed = await sdk(frame, "watchShapeSeedAndSubscribe", h, CLS); + check("first snapshot after subscribe is isPending (barrier not crossed)", seed.initial.isPending === true && seed.initial.isSuccess === false, `initial=${JSON.stringify(seed.initial)}`); + // Event-driven: wait for the barrier to cross + the datum to land. + await frame.waitForFunction( + (hh) => { + const s = (window as any).__sdk.watchShapeSnapshot(hh as string); + return s && s.isSuccess && s.dataLen >= 1; + }, + h, + { timeout: 30000 }, + ); + const snap = await sdkGet(frame, "watchShapeSnapshot", h); + check("reaches isSuccess with the seeded datum (titles include 'seeded')", snap.isSuccess && !snap.isPending && !snap.isError && snap.dataLen >= 1 && snap.titles.includes("seeded"), `snap=${JSON.stringify(snap)}`); + await sdk(frame, "watchShapeStop", h); + }); + await step("watchShape: empty scope reaches isSuccess with data:[]", async () => { + const h = "empty" + Date.now(); + const CLS = "urn:e2e:ws:Event"; + await sdk(frame, "watchShapeEmptyStart", h, CLS); + await frame.waitForFunction( + (hh) => { + const s = (window as any).__sdk.watchShapeSnapshot(hh as string); + return s && s.isSuccess; + }, + h, + { timeout: 30000 }, + ); + const snap = await sdkGet(frame, "watchShapeSnapshot", h); + check("empty scope: isSuccess, data:[] (synced-but-empty, not stuck pending)", snap.isSuccess && !snap.isPending && !snap.isError && snap.dataLen === 0, `snap=${JSON.stringify(snap)}`); + await sdk(frame, "watchShapeStop", h); + }); + // ── caps / read-filter (in-memory cap model) ──────────────────────────── console.log("\n── caps / read-filter (in-memory cap model) ──"); await step("read-filter: protected hidden from stranger", async () => { diff --git a/packages/client/e2e/sdk-entry.ts b/packages/client/e2e/sdk-entry.ts index 8c0e00d..925f6f3 100644 --- a/packages/client/e2e/sdk-entry.ts +++ b/packages/client/e2e/sdk-entry.ts @@ -25,8 +25,10 @@ import { discovery, storeRegistry, useShape as libUseShape, + watchShape, accounts, } from "@ng-eventually/client"; +import type { ShapeObservable, ShapeQuery } from "@ng-eventually/client"; const { IdentityStore } = accounts; @@ -563,6 +565,123 @@ const identity = new IdentityStore( return { priv, prot, pub }; }, + // ── watchShape (reactive useQuery-shaped read) ─────────────────────────── + // A minimal SHEX-ish ShapeType pinning rdf:type to an e2e class IRI. watchShape + // reads only `.shape`/`.schema[...].predicates[rdf:type].dataTypes[].literals` + // to know which @type to keep — generic, no application domain. + _wsProbes: {} as Record< + string, + { obs: ShapeObservable; initial: ShapeQuery; unsub: () => void } + >, + /** + * REAL-BROKER cycle proof. Under a fresh identity, create ONE protected entity + * document carrying an rdf:type= triple, then open a `watchShape` over + * (that class shape, "protected"). Capture the FIRST snapshot right after subscribe + * (must be isPending) so the caller can then wait event-driven for isSuccess with + * the seeded datum present. Returns the handle + the initial snapshot + the seed + * doc/type so the runner can assert the data landed. + */ + async watchShapeSeedAndSubscribe(handle: string, cls: string) { + storeRegistry.resetRegistryCache(); + const id = "@ws-" + handle; + setCurrentUser(id); + const doc = await storeRegistry.createEntityDoc(id, "protected"); + const s = await sessionReady; + // Seed the entity doc with the shape's type + a title (anchored default graph). + await docs.sparqlUpdate( + s.session_id, + `INSERT DATA { <${doc}> <${cls}> ; "seeded" }`, + doc, + ); + // Wait until this session sees the index append (data persisted on the broker). + for (let i = 0; i < 15; i++) { + storeRegistry.resetRegistryCache(); + const listed = await storeRegistry.listMyEntityDocs(id, "protected"); + if (listed.includes(doc)) break; + await new Promise((r) => setTimeout(r, 1000)); + } + const shape = { + shape: "urn:e2e:ws:Shape", + schema: { + "urn:e2e:ws:Shape": { + iri: "urn:e2e:ws:Shape", + predicates: [ + { + iri: "http://www.w3.org/1999/02/22-rdf-syntax-ns#type", + readablePredicate: "type", + maxCardinality: 1, + minCardinality: 1, + dataTypes: [{ literals: [cls], valType: "iri" }], + }, + ], + }, + }, + }; + const obs = watchShape(shape, "protected") as ShapeObservable; + const unsub = obs.subscribe(() => {}); + // First snapshot immediately after subscribe — the barrier is not yet crossed + // (docs opening via doc_subscribe), so this MUST be pending. + const initial = obs.getSnapshot(); + (window as any).__sdk._wsProbes[handle] = { obs, initial, unsub }; + // NB: leave the current user SET for the probe's lifetime — watchShape resolves + // the scope from getCurrentUser() on every (reactive) refresh, exactly as the app + // keeps a stable identity. watchShapeStop clears it. + return { doc, cls, initial: { isPending: initial.isPending, isSuccess: initial.isSuccess, dataLen: initial.data.length } }; + }, + /** Current snapshot of a watchShape probe (event-driven poll target). */ + watchShapeSnapshot(handle: string) { + const rec = (window as any).__sdk._wsProbes[handle]; + if (!rec) return null; + const snap: ShapeQuery = rec.obs.getSnapshot(); + return { + isPending: snap.isPending, + isSuccess: snap.isSuccess, + isError: snap.isError, + dataLen: snap.data.length, + // The seeded title, if the datum is present (proves the real data landed). + titles: snap.data.flatMap((d: any) => d.props?.["urn:e2e:ws:title"] ?? []), + }; + }, + watchShapeStop(handle: string) { + const rec = (window as any).__sdk._wsProbes[handle]; + if (rec) rec.unsub(); + setCurrentUser(null); + }, + /** + * EMPTY-scope proof: a brand-new identity with NO entity docs of `scope`. watchShape + * must reach isSuccess with data:[] (synced-but-empty), NOT stay pending. Returns + * the handle; poll watchShapeSnapshot for the transition. + */ + watchShapeEmptyStart(handle: string, cls: string) { + storeRegistry.resetRegistryCache(); + const id = "@ws-empty-" + handle; + setCurrentUser(id); + const shape = { + shape: "urn:e2e:ws:Shape", + schema: { + "urn:e2e:ws:Shape": { + iri: "urn:e2e:ws:Shape", + predicates: [ + { + iri: "http://www.w3.org/1999/02/22-rdf-syntax-ns#type", + readablePredicate: "type", + maxCardinality: 1, + minCardinality: 1, + dataTypes: [{ literals: [cls], valType: "iri" }], + }, + ], + }, + }, + }; + const obs = watchShape(shape, "protected") as ShapeObservable; + const unsub = obs.subscribe(() => {}); + const initial = obs.getSnapshot(); + (window as any).__sdk._wsProbes[handle] = { obs, initial, unsub }; + // Keep the identity set for the probe's lifetime (watchShapeStop clears it): a + // faithful "empty scope for a real identity", not "no identity at all". + return { initial: { isPending: initial.isPending, isSuccess: initial.isSuccess } }; + }, + // ── caps / read-filter (in-memory cap model) ───────────────────────────── // The read-filter over the injected useShape Set-like. Boundary note: the // caps/read-filter are EMULATED in-memory (CapRegistry) — the real broker does diff --git a/packages/client/src/discovery.ts b/packages/client/src/discovery.ts index 1f20439..7def0ed 100644 --- a/packages/client/src/discovery.ts +++ b/packages/client/src/discovery.ts @@ -101,6 +101,17 @@ async function indexInboxNuri(): Promise { return record.docPublic; } +/** + * The NURI of the global discovery-index document (the inbox where submissions + * land). Exposed so a reactive reader ({@link watchShape}) that folds discovery + * into the public read-set can SUBSCRIBE to this document and re-resolve when a + * new public entity is announced. This is exactly {@link watchIndex}'s subscribe + * anchor. Removed against real NextGraph along with the special account. + */ +export async function indexDocNuri(): Promise { + return indexInboxNuri(); +} + /** * Submit a reference to the global discovery index — the SDK act "make this * discoverable". Deposits `ref` into the index document's inbox via diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index fa1dec0..fd0478c 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -13,6 +13,8 @@ export * from "./types"; export { useShape } from "./use-shape"; +export { watchShape } from "./watch-shape"; +export type { ShapeQuery, ShapeObservable } from "./watch-shape"; export { init, initNg } from "./lifecycle"; export * as inbox from "./inbox"; export * as discovery from "./discovery"; diff --git a/packages/client/src/open-repo.ts b/packages/client/src/open-repo.ts index f2dd85c..0ff2f0e 100644 --- a/packages/client/src/open-repo.ts +++ b/packages/client/src/open-repo.ts @@ -91,7 +91,17 @@ let boundSessionId: string | number | null = null; * pathological case (a doc that never pushes), so a read is never blocked forever — * it just proceeds (and yields 0 rows, exactly as before, for a genuinely-absent doc). */ -const OPEN_TIMEOUT_MS = 8000; +let OPEN_TIMEOUT_MS = 8000; + +/** + * Override the bootstrap-open fallback timeout (ms). TEST-ONLY: the timed-out + * branch (a doc that never pushes a `State`) is otherwise only reachable after the + * 8s production wait, too slow for a unit test. Production never calls this — the + * default stands. `resetOpenedRepos` restores the default. + */ +export function setOpenTimeoutForTests(ms: number): void { + OPEN_TIMEOUT_MS = ms; +} /** Reset the open registry (mainly for tests / a switched wallet). Tears down the * held bootstrap subscriptions so a subsequent open re-subscribes cleanly. */ @@ -108,6 +118,7 @@ export function resetOpenedRepos(): void { held.clear(); syncState.clear(); boundSessionId = null; + OPEN_TIMEOUT_MS = 8000; } /** diff --git a/packages/client/src/store-registry.ts b/packages/client/src/store-registry.ts index 40b8286..de442a2 100644 --- a/packages/client/src/store-registry.ts +++ b/packages/client/src/store-registry.ts @@ -495,6 +495,20 @@ export async function listEntityDocs(scope: Scope): Promise { return out; } +/** + * The scope-INDEX document NURI of ONE account (`id`) for `scope` — the + * store-container document that LISTS the account's per-entity document NURIs + * (what {@link listMyEntityDocs} reads). Exposed so a reactive reader + * ({@link watchShape}) can SUBSCRIBE to this index document and re-resolve the + * entity-doc set when the index changes (a new entity created appends a NURI + * here). Idempotent via `ensureAccount`'s cache. At migration this becomes the + * user's real per-scope store NURI (the container the store itself provides). + */ +export async function scopeIndexDoc(id: string, scope: Scope): Promise { + const record = await ensureAccount(id); + return indexDocOf(record, scope); +} + /** * The entity-document NURIs of `scope` belonging to ONE account (`id`) — * the read-by-need path for one account's own entities. Bounded to a SINGLE diff --git a/packages/client/src/watch-shape.ts b/packages/client/src/watch-shape.ts new file mode 100644 index 0000000..168ebfd --- /dev/null +++ b/packages/client/src/watch-shape.ts @@ -0,0 +1,384 @@ +/** + * watch-shape — a REACTIVE, TanStack-`useQuery`-shaped read over one SHEX shape in + * one logical scope. This is the surface the consuming app will bind (phase B) with + * `useSyncExternalStore` — the polyfill deliberately exposes an OBSERVABLE, never a + * React hook (the lib has NO React dependency, same constraint as `subscribe.ts`). + * + * ── Why an observable, and why this exact shape ──────────────────────────── + * It anticipates NextGraph's planned `useShape(shape, scope)` upgrade, which will + * natively distinguish "sync in progress" from "synced but empty". That distinction + * ALREADY exists lib-internally (`open-repo.ts` `getSyncState`: syncing / synced / + * timed-out); `watchShape` merely SURFACES it as a `useQuery`-minimal snapshot: + * `ShapeQuery = { data: T[]; isPending; isSuccess; isError; error }`. + * `data` is ALWAYS an array (never `undefined`), so a synced-but-empty scope reads + * `{ data: [], isPending: false, isSuccess: true }` — the key distinction — while a + * scope still syncing reads `{ data: [], isPending: true, isSuccess: false }`. + * + * ── What the observable OWNS (the whole read pipeline) ───────────────────── + * 1. Resolve the logical scope → the doc set: the current identity's per-entity + * docs for that scope (`storeRegistry.listMyEntityDocs`), PLUS — for `public` + * only — the discovery index folded in (`discovery.readIndex`), so the app + * never orchestrates discovery to read. Faithful to the future + * `useShape(shape, 'public')`. + * 2. Open the docs (`ensureReposOpen`) — this AWAITS the sync BARRIER (first + * `State` per doc, `getSyncState` → `synced`, or `timed-out` on the bounded + * fallback). `isPending` holds until the barrier is reached for the current + * doc set AND the first `readUnion` has rendered. + * 3. `readUnion(docs)` — the read-model (cap filter already applied inside; we do + * NOT double-filter), then FILTER the union by the requested shape's `@type` + * (a `readUnion` union spans multiple types; each `watchShape` yields only the + * subjects of its shape). Non-domain: the type IRI is read from the SHEX + * ShapeType, not from any application concept. + * + * ── Reactivity WITHOUT polling (no `setInterval`) ────────────────────────── + * Reactivity is push-only (rule no-broker-polling): `subscribeDoc` on every doc in + * the current set re-runs `readUnion` on any push. The set is DYNAMIC (creating an + * entity appends a NURI to the scope-index doc; announcing a public entity appends + * to the discovery index), so we ALSO subscribe to the scope-index document (and, + * for `public`, the discovery-index document): a push there re-RESOLVES the scope + * and re-keys the subscribed set. Subscriptions are idempotent — an already-followed + * doc is not re-subscribed. Everything reuses `subscribe.ts` / `open-repo.ts`; no + * parallel channel. + * + * ── timed-out → isSuccess (best-effort), NOT isError ─────────────────────── + * A doc whose barrier fell back to `timed-out` still counts as "barrier reached" + * (`isSuccess`): a slow-but-empty wallet must read as empty-success, not error. + * `isError` fires ONLY on a real thrown exception in the pipeline. + */ + +import { getCurrentUser } from "./polyfill"; +import { ensureReposOpen, getSyncState } from "./open-repo"; +import { readUnion, type UnionSubject } from "./read-model"; +import { subscribeDoc, type Unsubscribe } from "./subscribe"; +import { listMyEntityDocs, scopeIndexDoc } from "./store-registry"; +import { readIndex, indexDocNuri } from "./discovery"; +import type { Nuri, Scope } from "./types"; + +/** + * The RDF `type` predicate IRI. A SHEX shape pins its class via a triple + * constraint on this predicate; we filter the read union by it. + */ +const RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"; + +/** + * A minimal TanStack-`useQuery`-shaped read snapshot. `data` is ALWAYS an array + * (never `undefined`). Defaults `T` to {@link UnionSubject} — `watchShape` yields + * the generic per-subject property bags of the read-model (NO application domain); + * the app maps them to its own entity types in phase B. + */ +export interface ShapeQuery { + /** The subjects of the requested shape/scope. Empty array when none (never undefined). */ + data: T[]; + /** True while the sync barrier for the current doc set is not yet reached OR the + * first `readUnion` has not rendered. Mutually exclusive with `isSuccess`. */ + isPending: boolean; + /** True once the barrier is reached (all docs `synced` OR `timed-out`) AND the + * first `readUnion` has rendered. A synced-but-EMPTY scope is `isSuccess` with + * `data: []` — the distinction this surface exists for. */ + isSuccess: boolean; + /** True ONLY on a real thrown exception in the read pipeline (never for `timed-out`). */ + isError: boolean; + /** The caught error when `isError`, else `undefined`. */ + error: unknown; +} + +/** The observable a caller binds with `useSyncExternalStore` (phase B). */ +export interface ShapeObservable { + /** The current snapshot. STABLE across calls until it actually changes (so + * `useSyncExternalStore` does not loop): the same reference is returned until a + * state transition produces a new one. */ + getSnapshot(): ShapeQuery; + /** Register a change listener; returns an unsubscribe. The last listener's + * unsubscribe tears down the underlying doc subscriptions. */ + subscribe(onChange: () => void): () => void; + /** Force a re-resolve + re-read now (e.g. an imperative refresh). Idempotent + * w.r.t. subscriptions; never polls. */ + refetch(): void; +} + +/** + * Derive the class IRI(s) a SHEX {@link ShapeType} pins on `rdf:type`, if any. + * A generated shape constrains its subject's type via a triple constraint on the + * `rdf:type` predicate whose `literals` carry the class IRI(s). Returns the set of + * those IRIs, or `null` when the shape pins NO type (then no type-filter is applied + * and every subject in the doc set flows through). Purely structural — reads only + * the SHEX schema, no application domain. + */ +function shapeTypeIris(shapeType: unknown): Set | null { + try { + const st = shapeType as { + shape?: string; + schema?: Record }> }>; + }; + const shape = st?.shape && st.schema ? st.schema[st.shape] : undefined; + const preds = shape?.predicates ?? []; + const iris = new Set(); + for (const p of preds) { + if (p?.iri !== RDF_TYPE) continue; + for (const dt of p.dataTypes ?? []) { + for (const lit of dt.literals ?? []) { + if (typeof lit === "string") iris.add(lit); + } + } + } + return iris.size > 0 ? iris : null; + } catch { + return null; + } +} + +/** Whether a subject satisfies the shape's `@type` constraint (or the shape pins none). */ +function matchesShape(subject: UnionSubject, typeIris: Set | null): boolean { + if (!typeIris) return true; // shape pins no rdf:type → accept every subject + const types = subject.props[RDF_TYPE] ?? []; + return types.some((t) => typeIris.has(t)); +} + +/** + * Whether the sync BARRIER is reached for the whole doc set. Called only AFTER + * `ensureReposOpen(docs)` has resolved, so each doc has been requested; the only + * state that still holds the barrier open is `syncing` (subscribed, first `State` + * not yet received). `synced` and `timed-out` both count as reached (`timed-out` is + * best-effort, not an error). `unknown` means the injected `ng` has no + * `doc_subscribe` (the fake/no-op open path, which has NO barrier semantics) — after + * a completed open it can only mean that path, so it counts as reached (opened, + * nothing to wait on). An EMPTY doc set is trivially past the barrier. + */ +function barrierReached(docs: Nuri[]): boolean { + for (const d of docs) { + if (getSyncState(d) === "syncing") return false; + } + return true; +} + +/** + * Create a reactive, `useQuery`-shaped observable over one SHEX `shapeType` in one + * logical `scope` (`'public' | 'protected' | 'private'`). See the module header for + * the full pipeline. The returned observable is inert until its first + * {@link ShapeObservable.subscribe} (or {@link ShapeObservable.refetch}) — that is + * what kicks off resolution, opening and the first read; before then `getSnapshot` + * reports the initial pending snapshot. + */ +export function watchShape( + shapeType: unknown, + scope: Scope, +): ShapeObservable { + const typeIris = shapeTypeIris(shapeType); + + // The current, STABLE snapshot (same reference until a transition rebuilds it). + let snapshot: ShapeQuery = { + data: [], + isPending: true, + isSuccess: false, + isError: false, + error: undefined, + }; + + const listeners = new Set<() => void>(); + let started = false; + // The docs currently subscribed for change signals, keyed by NURI → unsubscribe. + // Idempotent: a doc already here is not re-subscribed. Excludes the container + // (scope-index / discovery-index) subscriptions, held separately. + const docSubs = new Map(); + // Container subscriptions (scope-index doc; discovery-index doc for `public`) — + // a push here means the doc SET may have changed → re-resolve. + const containerSubs = new Map(); + // Monotonic token so a slow in-flight refresh cannot clobber a newer one. + let refreshToken = 0; + + function emit(): void { + for (const l of listeners) { + try { + l(); + } catch (error) { + console.error("[watch-shape] listener threw", error); + } + } + } + + function setSnapshot(next: ShapeQuery): void { + snapshot = next; + emit(); + } + + /** Extract candidate document NURIs from an opaque discovery `ref` — every + * string, recursively, that looks like a NextGraph doc NURI (`did:ng:`). Generic: + * the app puts the entity doc NURI inside the ref it submits; we fold those docs + * into the read-set so the app need not orchestrate discovery. Non-NURI refs + * contribute nothing (and readUnion+shape-filter drop anything irrelevant). */ + function nurisFromRef(ref: unknown, out: Set): void { + if (typeof ref === "string") { + if (ref.startsWith("did:ng:")) out.add(ref); + return; + } + if (Array.isArray(ref)) { + for (const v of ref) nurisFromRef(v, out); + return; + } + if (ref && typeof ref === "object") { + for (const v of Object.values(ref)) nurisFromRef(v, out); + } + } + + /** Resolve the logical scope → the current doc set (my entity docs + discovery + * fold for `public`). Tolerant: a resolution failure yields whatever resolved. */ + async function resolveDocs(): Promise { + const user = getCurrentUser(); + const set = new Set(); + if (user) { + try { + for (const d of await listMyEntityDocs(user, scope)) set.add(d); + } catch (error) { + console.error("[watch-shape] listMyEntityDocs failed", error); + } + } + if (scope === "public") { + try { + for (const e of await readIndex()) nurisFromRef(e.ref, set); + } catch (error) { + console.error("[watch-shape] discovery readIndex failed", error); + } + } + return [...set]; + } + + /** Subscribe to the CONTAINER documents (scope-index; discovery-index for public) + * so a change to the doc SET re-resolves. Idempotent per NURI. */ + async function ensureContainerSubs(): Promise { + const containers: Nuri[] = []; + const user = getCurrentUser(); + if (user) { + try { + containers.push(await scopeIndexDoc(user, scope)); + } catch (error) { + console.error("[watch-shape] scopeIndexDoc failed", error); + } + } + if (scope === "public") { + try { + containers.push(await indexDocNuri()); + } catch (error) { + console.error("[watch-shape] indexDocNuri failed", error); + } + } + for (const c of containers) { + if (!c || containerSubs.has(c)) continue; + // A push on a container doc means the set may have changed → full re-resolve. + containerSubs.set(c, subscribeDoc(c, () => void refresh())); + } + } + + /** Re-key the per-doc change subscriptions to exactly `docs` (idempotent adds, + * prune removed). A push on any of these re-reads (data-only, no re-resolve). */ + function syncDocSubs(docs: Nuri[]): void { + const wanted = new Set(docs.filter(Boolean)); + for (const [nuri, unsub] of docSubs) { + if (!wanted.has(nuri)) { + try { + unsub(); + } catch { + /* ignore */ + } + docSubs.delete(nuri); + } + } + for (const nuri of wanted) { + if (docSubs.has(nuri)) continue; + docSubs.set(nuri, subscribeDoc(nuri, () => void reread())); + } + } + + /** Read (union + shape filter) the CURRENT doc set and publish a snapshot. + * Derives isPending/isSuccess from the barrier + whether the read rendered. */ + async function readAndPublish(docs: Nuri[], token: number): Promise { + let subjects: UnionSubject[]; + try { + subjects = await readUnion(docs); + } catch (error) { + if (token !== refreshToken) return; + setSnapshot({ data: [], isPending: false, isSuccess: false, isError: true, error }); + return; + } + if (token !== refreshToken) return; // superseded by a newer refresh/reread + const data = subjects.filter((s) => matchesShape(s, typeIris)); + const past = barrierReached(docs); + setSnapshot({ + data, + isPending: !past, + isSuccess: past, + isError: false, + error: undefined, + }); + } + + /** Full cycle: resolve the scope, (re)establish container subs, open the docs + * (await the barrier), sync per-doc subs, then read + publish. */ + async function refresh(): Promise { + const token = ++refreshToken; + try { + await ensureContainerSubs(); + const docs = await resolveDocs(); + if (token !== refreshToken) return; + syncDocSubs(docs); + // Open/await the barrier (first State per doc, or timed-out). No-op once open. + await ensureReposOpen(docs); + if (token !== refreshToken) return; + await readAndPublish(docs, token); + } catch (error) { + if (token !== refreshToken) return; + setSnapshot({ data: [], isPending: false, isSuccess: false, isError: true, error }); + } + } + + /** A push on an already-open doc: re-read the CURRENT set only (no re-resolve, + * the set is unchanged). Reuses the docs we are subscribed to. */ + async function reread(): Promise { + const token = ++refreshToken; + const docs = [...docSubs.keys()]; + await readAndPublish(docs, token); + } + + function start(): void { + if (started) return; + started = true; + void refresh(); + } + + return { + getSnapshot(): ShapeQuery { + return snapshot as unknown as ShapeQuery; + }, + subscribe(onChange: () => void): () => void { + listeners.add(onChange); + start(); + return () => { + listeners.delete(onChange); + if (listeners.size === 0) { + // Last listener gone → tear down the underlying subscriptions. A later + // subscribe restarts a fresh cycle. + for (const u of docSubs.values()) { + try { + u(); + } catch { + /* ignore */ + } + } + for (const u of containerSubs.values()) { + try { + u(); + } catch { + /* ignore */ + } + } + docSubs.clear(); + containerSubs.clear(); + started = false; + } + }; + }, + refetch(): void { + start(); + void refresh(); + }, + }; +} diff --git a/packages/client/test/watch-shape.test.ts b/packages/client/test/watch-shape.test.ts new file mode 100644 index 0000000..f83ebf0 --- /dev/null +++ b/packages/client/test/watch-shape.test.ts @@ -0,0 +1,349 @@ +/** + * 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(); + 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(); + }); +});