refactor(layout): ranger les modules par destin à la migration
Les 25 modules étaient à plat, nommés d'après ce qu'ils font mécaniquement (`store-registry`, `read-model`, `reach`, `caps`). Rien dans l'arborescence ne disait lesquels DEVIENDRONT le vrai SDK, lesquels tiennent lieu du travail que le verifier fera nativement, et lesquels n'existent que parce qu'un wallet est partagé — trois destins sans rapport. Quatre dossiers, les deux fichiers d'entrée restant à la racine pour que l'`exports` du paquet et le code du consommateur ne bougent pas : - `model/` — le modèle d'adressage de la cible, transcrit : vocabulaire pur, pas d'I/O. Survit comme connaissance. - `surface/` — ce que l'app touche, chaque symbole ayant un pendant cible documenté. Supprimé quand l'alias bascule ; le code de l'app est inchangé. - `emulated-verifier/` — les doublures de ce que le verifier fait nativement : possession, dépôt des caps, frontière, non-livraison, traitement des inbox, registres de branche, ouverture de repo. **C'est le dossier où diverger du modèle est possible.** Le préfixe `emulated-` porte le sens : tient lieu de, jamais est — cette bibliothèque ne réside dans aucune couche de la cible, elle les référence. - `shared-wallet/` — n'existe que parce qu'un wallet héberge toutes les identités. Aucun pendant, rien sur quoi s'aligner ; sa seule loi est de rester invisible depuis `surface/`. S'évapore, remplacé par rien. `store-registry-api.ts` devient `surface/placement.ts` : il faisait déjà à la main ce que la frontière de dossier fait structurellement — c'est la meilleure preuve interne du bien-fondé de ce rangement. Ce commit ne fait que déplacer et recâbler les imports (src, test, e2e). Les scissions des modules à cheval suivent. 157 tests unitaires, typecheck src/test/e2e vert.
This commit is contained in:
Sylvain Duchesne
@@ -0,0 +1,386 @@
|
||||
/**
|
||||
* 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 wallet's own per-entity
|
||||
* documents for that scope (`storeRegistry.listMyEntityDocs`), and nothing
|
||||
* else. There is no "everything public" to fold in — **you cannot discover,
|
||||
* you can only follow links** (see docs/readcap-and-nuri-model.md §4ter-bis),
|
||||
* and a link reaches you through an inbox or through a document you already
|
||||
* hold. A document whose cap you were given is read by NAMING it
|
||||
* (`readModel.readUnion`), not by turning up in a scope you never put it in.
|
||||
* 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). It has TWO sources: document
|
||||
* pushes, and the KEYRING — a cap that arrives asynchronously (an inbox delivery
|
||||
* absorbed by the consumer's `inbox.watch`) makes documents readable that were not,
|
||||
* so `CapRegistry.onChange` re-reads. Without that, a view stays stale until an
|
||||
* unrelated push happens to fire. On the document side, `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), so we ALSO subscribe to the
|
||||
* scope-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 { getCaps, getCurrentUser } from "../polyfill";
|
||||
import { ensureReposOpen, getSyncState } from "../emulated-verifier/open-repo";
|
||||
import { readUnion, type UnionSubject } from "./read-model";
|
||||
import { subscribeDoc, type Unsubscribe } from "./subscribe";
|
||||
import { listMyEntityDocs, userStoreDoc } from "../shared-wallet/account-registry";
|
||||
import type { Nuri, Scope } from "../model/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>();
|
||||
// Unsubscribe from the held-caps change signal (see the subscription in `start`).
|
||||
let capsUnsub: (() => void) | null = null;
|
||||
// True while `resolveDocs` runs. Folding a repo link files a cap, which fires the
|
||||
// held-caps signal; the resolution in progress already accounts for it, so the
|
||||
// signal is ignored during that window instead of restarting the cycle.
|
||||
let resolving = false;
|
||||
// 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();
|
||||
}
|
||||
|
||||
/** Resolve the logical scope → the current doc set: the CURRENT wallet's own
|
||||
* entity documents for that scope, and nothing else. Tolerant: a resolution
|
||||
* failure yields whatever resolved.
|
||||
*
|
||||
* There is no "everything public" to fold in. You cannot discover; you can only
|
||||
* follow links, and a link reaches you through an inbox or through a document
|
||||
* you already hold — never through a shared index. A document someone gave you
|
||||
* the cap for is read by naming it (`readModel.readUnion`), not by appearing in
|
||||
* a scope you did not put it in. */
|
||||
async function resolveDocs(): Promise<Nuri[]> {
|
||||
const user = getCurrentUser();
|
||||
const set = new Set<Nuri>();
|
||||
resolving = true;
|
||||
try {
|
||||
if (user) {
|
||||
try {
|
||||
for (const d of await listMyEntityDocs(user, scope)) set.add(d);
|
||||
} catch (error) {
|
||||
console.error("[watch-shape] listMyEntityDocs failed", error);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
resolving = false;
|
||||
}
|
||||
return [...set];
|
||||
}
|
||||
|
||||
/** Subscribe to the CONTAINER document (the scope index) 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 userStoreDoc(user, scope));
|
||||
} catch (error) {
|
||||
console.error("[watch-shape] userStoreDoc 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;
|
||||
// A cap that arrives ASYNCHRONOUSLY (an inbox deposit absorbed by the
|
||||
// consumer's `inbox.watch`) makes documents readable that were not. Without
|
||||
// this the view would stay stale until some unrelated push happened to fire —
|
||||
// so re-read whenever they change. This is the delivery channel key
|
||||
// ROTATION uses too, which is why keeping an access needs no subscription
|
||||
// obligation on the consumer's side.
|
||||
capsUnsub = getCaps().onChange(() => {
|
||||
if (!resolving) void refresh();
|
||||
});
|
||||
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();
|
||||
if (capsUnsub) {
|
||||
capsUnsub();
|
||||
capsUnsub = null;
|
||||
}
|
||||
started = false;
|
||||
}
|
||||
};
|
||||
},
|
||||
refetch(): void {
|
||||
start();
|
||||
void refresh();
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user