feat(client): watchShape — lecture réactive à la forme TanStack useQuery

Phase A du refactor des lectures. Surface la barrière de sync interne (open-repo `getSyncState`) dans une API useQuery-shaped, en anticipation de la mise à jour prévue de useShape par NextGraph — distingue nativement « sync en cours » de « synchronisé mais vide ». `watchShape<T>(shapeType, scope): { getSnapshot(): ShapeQuery<T>, subscribe(cb), refetch() }` avec `ShapeQuery = { data, isPending, isSuccess, isError, error }`. OBSERVABLE (pas de dépendance React — l'app câblera useSyncExternalStore en phase B) ; getSnapshot rend une référence stable. - Scope LOGIQUE (public/protected/private) résolu au wallet virtuel : listMyEntityDocs(getCurrentUser, scope) + découverte foldée pour public. - isPending tant que la barrière n'est pas atteinte / 1er readUnion non rendu ; isSuccess après ; timed-out → isSuccess (best-effort, pas isError). - Réactif SANS polling : subscribeDoc sur les docs + le doc d'index de scope (+ index découverte) → re-read/re-résolution au push ; souscriptions idempotentes. - Générique : aucune logique domaine Festipod dans le lib ; filtre par la shape SHEX (rdf:type). Pas de double filtre cap. gate : tsc 0 ; bun test 120 (+4 : pending→success, synced-vide, réactif, timed-out) ; test:e2e 42 passed (+scénario broker réel : isPending au 1er snapshot → isSuccess avec données, scope vide → isSuccess data:[]). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-09 23:17:18 +02:00
parent 38b152136b
commit 7c233df5c0
8 changed files with 932 additions and 1 deletions
@@ -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<T> = { 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<T = UnionSubject> {
+  /** 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<T = UnionSubject> {
+  /** 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<T>;
+  /** 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<string> | null {
+  try {
+    const st = shapeType as {
+      shape?: string;
+      schema?: Record<string, { predicates?: Array<{ iri?: string; dataTypes?: Array<{ literals?: unknown[] }> }> }>;
+    };
+    const shape = st?.shape && st.schema ? st.schema[st.shape] : undefined;
+    const preds = shape?.predicates ?? [];
+    const iris = new Set<string>();
+    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<string> | 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<T = UnionSubject>(
+  shapeType: unknown,
+  scope: Scope,
+): ShapeObservable<T> {
+  const typeIris = shapeTypeIris(shapeType);
+
+  // The current, STABLE snapshot (same reference until a transition rebuilds it).
+  let snapshot: ShapeQuery<UnionSubject> = {
+    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<Nuri, Unsubscribe>();
+  // 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<Nuri, Unsubscribe>();
+  // 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<UnionSubject>): 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<Nuri>): 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<Nuri[]> {
+    const user = getCurrentUser();
+    const set = new Set<Nuri>();
+    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<void> {
+    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<void> {
+    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<void> {
+    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<void> {
+    const token = ++refreshToken;
+    const docs = [...docSubs.keys()];
+    await readAndPublish(docs, token);
+  }
+
+  function start(): void {
+    if (started) return;
+    started = true;
+    void refresh();
+  }
+
+  return {
+    getSnapshot(): ShapeQuery<T> {
+      return snapshot as unknown as ShapeQuery<T>;
+    },
+    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();
+    },
+  };
+}