Reconnexion/ng-eventually
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat!: useShapeQuery lit par la requete de forme du SDK

Browse Source
This commit is contained in:
Sylvain Duchesne
2026-09-21 11:41:31 +02:00
parent a5e553e65b
commit 77e78ec81d
29 changed files with 1190 additions and 971 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/polyfill/src/surface/use-shape-query.ts
+221 -392
View File
@@ -1,280 +1,155 @@
/**
* use-shape-query — a REACTIVE, TanStack-`useQuery`-shaped read over one SHEX shape in
* use-shape-query - a REACTIVE, TanStack-`useQuery`-shaped read over one SHEX shape in
* one SDK `Scope`. This is the surface the consuming app binds with
* `useSyncExternalStore` — the polyfill deliberately exposes an OBSERVABLE, never a
* `useSyncExternalStore` - the polyfill deliberately exposes an OBSERVABLE, never a
* React hook (the lib has NO React dependency, same constraint as `subscribe.ts`).
*
* ── The name, and the one difference from the SDK that is deliberate ───────
* -- The name, and the one difference from the SDK that is deliberate -------
* It was `watchShape` until `1.0.0-dev.5`. That name has ZERO occurrences upstream, and
* it spent `watch`, a verb the ORM has already spent on signal observation
* (`watch as watchDeepSignal`). The target's own name is `useShapeQuery`, and what it
* returns is the snapshot below rather than the bare `DeepSignalSet` today's `useShape`
* hands back — announced by NextGraph's developers and on record in
* hands back - announced by NextGraph's developers and on record in
* `.project/rules/nextgraph/surface-mirrors-the-sdk.md`, which is what the annotation on
* the function cites. Deviating TOWARD an announced target is alignment; the rest of the
* surface is what will catch up:
* `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
* `{ data: [], isPending: false, isSuccess: true }` - the key distinction - while a
* scope still syncing reads `{ data: [], isPending: true, isSuccess: false }`.
*
* ── The scope is the SDK's, and translating it is EXPLICITLY TEMPORARY ─────
* `scope` is `@ng-org/orm`'s own `Scope` — `{ graphs: string[] | string; subjects?:
* string[] }` — not the three strings this module took before. A `graphs` entry names
* ONE document's graph, `did:ng:i` (or `""`) discards the rest and means the whole local
* dataset, and `[]` or `undefined` returns no objects at all: all three are upstream's
* rules, reproduced by `normalizeScope` below.
* -- What it reads THROUGH, since `1.0.0-dev.6` ----------------------------
* The SDK's own shape query - `RdfOrmSubscription.getOrCreate(shapeType, scope)` - and
* nothing of this library's own. Four steps went with the change, and they were one
* mechanism: a logical scope resolved to a document set (`listMyEntityDocs`), those
* documents opened through `ensureReposOpen` for a sync barrier, `readUnion` over them,
* and a filter of the union by the shape's `@type` IRI. All four existed because the
* parameter had to be TRANSLATED; once the parameter is upstream's own `Scope`, the
* question goes straight to the thing that answers it. `graphs`, `subjects` and the
* whole-dataset marker `did:ng:i` are now read by the engine, literally, and this module
* does not interpret them at all - so the three readings are upstream's, not ours.
*
* `graphs` is REQUIRED as of `@ng-org/orm@0.1.2-alpha.21`; it used to be optional. Only
* the spelling changed, never the meaning: a caller who wants "everything" still has to
* say `did:ng:i`, because on this axis an empty list is the EMPTY set and not the absent
* filter. `subjects` keeps the opposite convention - there `[]` means no filtering - so
* the two axes read `[]` in opposite directions and `normalizeScope` keeps them apart.
* It does NOT delegate to the SDK's `useShape`: that is React's hook, it returns a bare
* `DeepSignalSet`, and a set cannot carry the request state this surface exists to
* publish. What is taken from the SDK is the QUERY; the observable, the snapshot and the
* state tracking stay here.
*
* Only the whole-dataset reading still has to be TRANSLATED, and the translation is the
* tell that the mechanism under it is about to stop being ours: it is served by
* enumerating this identity's own documents across its three stores
* (`account-registry.listMyEntityDocs`, no longer published), because this pipeline has
* no other way to say "everything". When the read queries the store itself, the
* enumeration and this branch go together. A named `graphs` list needs no translation at
* all — it is already the doc set.
* -- What stays ours, and why ----------------------------------------------
* 1. **The three states.** `readyPromise` is the ORM's own answer to "has the initial
* payload arrived" - it resolves on that path and on no other - so it is what tells
* `isPending` from a genuinely empty `isSuccess`. The React adapter one floor up
* drops that distinction (an empty set "if still loading"); this does not.
* 2. **The capability filter.** `emulated-verifier/read-filter.ts` keeps only items
* whose document (`@graph`) the current holder holds a cap for - the emulation of
* what the broker will bound at delivery, applied per document on what comes back,
* exactly where `surface/use-shape.ts` applies it. Before the first cap is issued
* the registry is not enforcing and everything passes through.
* 3. **The keyring signal.** A cap that arrives asynchronously (an inbox delivery
* absorbed by `inbox.watch`) makes items readable that were not, so
* `CapRegistry.onChange` re-publishes. Without it a view stays stale until an
* unrelated change happens to fire.
*
* ── What the observable OWNS (the whole read pipeline) ─────────────────────
* 1. Resolve the scope → the doc set: the graphs named, or — for the whole-dataset
* reading — the CURRENT wallet's own per-entity documents, 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, 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 read 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` and by
* the scope's `subjects` axis (a union spans multiple types; each `useShapeQuery`
* 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`) --------------------------
* Two push sources, both of them signals: the subscription's own
* `addChangeListener` (root objects added, updated or removed - the engine's patches
* reaching the signal object), and the keyring above. No document subscription of this
* library's is opened here any more: the ORM holds the one that matters.
*
* ── 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. Under the whole-dataset reading the
* set is DYNAMIC (creating an entity appends a NURI to its store's index doc), so we ALSO
* subscribe to those index documents: a push there re-RESOLVES and re-keys the subscribed
* set. A named `graphs` list has no container to follow — the caller fixed the set.
* Subscriptions are idempotent — an already-followed doc is not re-subscribed. Everything
* reuses `subscribe.ts` / `open-repo.ts`; no parallel channel.
* -- `isError` is wired to what CAN fail, and says so ----------------------
* One thing on this path throws, and it throws at the call: opening the shape query -
* a shape the ORM cannot canonicalise, or a `configure()` that was never given the
* query to open. That is what `isError` carries.
*
* ── 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.
* What it does NOT carry, recorded rather than assumed: a start the ENGINE refuses.
* `RdfOrmSubscription` awaits its session and calls `orm_start_graph` inside a
* `try { ... } catch (e) { console.error(e) }`, and its readiness promise is built with
* `resolve` alone - nothing rejects it, and `resolveReady` runs on one path only, the
* arrival of initial data. So upstream's *"I could not find out"* IS its *"still
* pending"*, and a refused read reads here as `isPending` for ever, never as
* `isSuccess` with `data: []`. The distinction this surface exists for therefore holds
* - an application is never told "there is nothing" for "nobody answered" - but the
* third state is narrower than it was while the pipeline was ours. `isError` stays
* because this package's job is to ENRICH the SDK in its announced direction, not to be
* bounded by what it exposes today; the day the ORM surfaces a failed start, this is
* where it lands.
*
* ── A scope that did not answer is NOT an empty scope ──────────────────────
* The pipeline's first step ASKS a question — which documents are mine — and until
* 2026-08-17 a failure to answer it was logged and treated as "none": the empty set
* flowed on, the barrier over zero docs is trivially reached, and the surface published
* `{ data: [], isSuccess: true }`. That snapshot is BYTE-FOR-BYTE the synced-but-empty
* one this module exists to distinguish, so the one distinction it publishes was the one
* it destroyed — an application rendered "you have nothing" for "the store did not
* answer". `listMyEntityDocs` stopped handing out that reading one layer down
* (`90712e0`); catching it here put it straight back.
*
* A resolution failure now travels the LOAD-STATE channel, which is where "this is
* not an answer" already lives on this surface: `isError` with the thrown `error`,
* and never `isSuccess`. The channel is the honest place because it is the one an
* application already has to read to tell pending from empty — the same three-way
* question, and the third state costs it no new vocabulary.
*
* ── An observable cannot un-emit: `data` survives an error ─────────────────
* -- An observable cannot un-emit: `data` survives an error -----------------
* A one-shot call rejects and is done. An observable has already handed a list to a
* subscriber that rendered it, so its failure snapshot has to say something about
* that list — and collapsing it to `[]` would hand out, in `data`, exactly the empty
* that list - and collapsing it to `[]` would hand out, in `data`, exactly the empty
* answer this fix removes. So `data` KEEPS the last read that actually answered, and
* `isError` says the current answer is unknown. `isSuccess` is false throughout: the
* array is a memory, not a reply to the question just asked.
*/
import { getCaps, getCurrentUser } from "../shared-wallet/bootstrap";
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 { toNuri } from "../model/nuri";
import type { Nuri, StoreType } from "../model/types";
// The SDK's own scope type, imported as a TYPE (erased at build, so no runtime `@ng-org`
// import is added here). Its provenance is recorded where it is PUBLISHED, in `index.ts`.
import { getCaps, getConfig } from "../shared-wallet/bootstrap";
import { filterReadable } from "../emulated-verifier/read-filter";
import type { OrmSubscriptionHandle, OrmSubscriptionLike } from "../model/types";
// The SDK's own types, imported as TYPES (erased at build, so no runtime `@ng-org`
// import is added here). The provenance of what is PUBLISHED is recorded in `index.ts`.
import type { Scope } from "@ng-org/orm";
/** Upstream's whole-dataset marker. `normalizeScope` maps `""` onto it, and
* `orm_start_graph` lets an entry equal to it DISCARD every other entry. */
const ENTIRE_DATASET = "did:ng:i";
/** The three stores an identity's own documents can sit in. */
const STORE_TYPES: readonly StoreType[] = ["private", "protected", "public"];
/** What `normalizeScope` produces: two lists, `""` already mapped onto `did:ng:i`. */
type NormalizedScope = { graphs: string[]; subjects: string[] };
/**
* `normalizeScope`, reproduced rather than imported: this package never hard-imports the
* SDK at RUNTIME (see the module header of `use-shape.ts` and `configure`'s `NgLike`), and
* a type-only import is erased. Same three rules as upstream - a bare string is one graph,
* `""` becomes `did:ng:i`, an absent list is empty.
*/
// @provenance normalizeScope kind=aligned level=3 ref=sdk/js/orm/src/types.ts:normalizeScope — the same three rules; reproduced because the SDK is injected, never imported at runtime
function normalizeScope(scope: Scope | string | undefined): NormalizedScope {
if (typeof scope === "string") {
return { graphs: [scope === "" ? ENTIRE_DATASET : scope], subjects: [] };
}
const graphs = (!scope?.graphs ? [] : [scope.graphs]).flat().map((g) => (g === "" ? ENTIRE_DATASET : g));
return { graphs, subjects: scope?.subjects ?? [] };
}
/**
* 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";
import type { BaseType, ShapeType } from "@ng-org/shex-orm";
/**
* A minimal TanStack-`useQuery`-shaped read snapshot. `data` is ALWAYS an array
* (never `undefined`). Defaults `T` to {@link UnionSubject} — `useShapeQuery` 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.
* (never `undefined`), and its items are the SHAPE's own objects - what the ORM
* returns, carrying `@id`, `@graph` and the shape's readable predicates.
*/
// @provenance ShapeQuery kind=invention level=none ref=none — no upstream type carries load state; `OrmSubscription.readyPromise` expresses pending-vs-ready and nothing else (built with `resolve` alone, and `orm_start_graph` failing is only logged — a failed read is an eternal pending), so `isError` has no counterpart at any level
export interface ShapeQuery<T = UnionSubject> {
/** The subjects of the requested shape/scope. Empty array when none (never undefined).
* When `isError`, this is the last read that ANSWERED — kept, not cleared, because an
// @provenance ShapeQuery kind=invention level=none ref=none - no upstream type carries load state; `RdfOrmSubscription.readyPromise` expresses pending-vs-ready and nothing else (built with `resolve` alone, and a refused `orm_start_graph` is only `console.error`'d - a failed read is an eternal pending), so `isError` has no counterpart at any level
export interface ShapeQuery<T extends BaseType = BaseType> {
/** The objects of the requested shape/scope. Empty array when none (never undefined).
* When `isError`, this is the last read that ANSWERED - kept, not cleared, because an
* observable cannot un-emit and `[]` here would be the false "you have nothing" the
* error state exists to prevent. Read it as a memory then, not as a reply. */
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`. */
/** True until the shape query's initial payload has arrived (`readyPromise`).
* 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. */
/** True once the initial payload has arrived. A ready-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`):
* the scope could not be resolved, or the union could not be read. It means the
* current answer is UNKNOWN — never that the scope is empty. Mutually exclusive with
* both `isPending` and `isSuccess`. */
/** True ONLY when opening the shape query threw: the shape could not be subscribed, or
* `configure()` was never given the query to open it with. It means the current answer
* is UNKNOWN - never that the scope is empty. Mutually exclusive with both `isPending`
* and `isSuccess`. See the module header for the failure it does NOT cover. */
isError: boolean;
/** The caught error when `isError`, else `undefined`. */
error: unknown;
}
/** The observable a caller binds with `useSyncExternalStore` (phase B). */
// @provenance ShapeObservable kind=invention level=none ref=none — no upstream observable exists; the ORM exposes a hook returning a DeepSignalSet and nothing else
export interface ShapeObservable<T = UnionSubject> {
// @provenance ShapeObservable kind=invention level=none ref=none - no upstream observable exists; the ORM exposes a hook returning a DeepSignalSet and nothing else
export interface ShapeObservable<T extends BaseType = BaseType> {
/** 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. */
* unsubscribe closes the underlying shape query. */
subscribe(onChange: () => void): () => void;
/** Force a re-resolve + re-read now (e.g. an imperative refresh). Idempotent
* w.r.t. subscriptions; never polls. */
/** Ask again, now: close the shape query in force and open a fresh one. Idempotent
* w.r.t. subscribers, and it never polls - the ORM pools a subscription it has just
* closed, so asking again over an answering broker costs nothing and answers at
* once. The snapshot is NOT reset: whatever was rendered stays until the new
* subscription answers or fails. */
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 a subject passes the scope's `subjects` axis. An empty list is upstream's "no
* filtering" (`Scope.subjects`: *"Set to `[]` or leave it `undefined` for no filtering"*),
* and the filter is literal - a subject IRI is in the list or it is not, exactly as
* `FILTER(?s IN (…))` is emitted upstream.
*/
function matchesSubjects(subject: UnionSubject, wanted: string[]): boolean {
return wanted.length === 0 || wanted.includes(subject.subject);
}
/**
* 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
* `scope` - the SDK's own `{ graphs, subjects }`. See the module header for the full
* pipeline and for what the `graphs` axis costs while the read path is still this
* library's. 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.
* `scope` - the SDK's own `{ graphs, subjects }`, passed through untouched. See the
* module header for what is read underneath and what each state means. The returned
* observable is inert until its first {@link ShapeObservable.subscribe} (or
* {@link ShapeObservable.refetch}) - that is what opens the shape query; before then
* `getSnapshot` reports the initial pending snapshot.
*/
// @provenance useShapeQuery kind=announced-not-written level=3 ref=.project/rules/nextgraph/surface-mirrors-the-sdk.md:useShapeQuery announced-by=nextgraph-developers@2026-09-16 — the SDK ships `useShape` returning a bare `DeepSignalSet`; the announced direction is a `useShapeQuery` returning a TanStack-shaped snapshot
export function useShapeQuery<T = UnionSubject>(
shapeType: unknown,
// @provenance useShapeQuery kind=announced-not-written level=3 ref=.project/rules/nextgraph/surface-mirrors-the-sdk.md:useShapeQuery announced-by=nextgraph-developers@2026-09-16 - the SDK ships `useShape` returning a bare `DeepSignalSet`; the announced direction is a `useShapeQuery` returning a TanStack-shaped snapshot
export function useShapeQuery<T extends BaseType = BaseType>(
shapeType: ShapeType<T>,
scope: Scope,
): ShapeObservable<T> {
const typeIris = shapeTypeIris(shapeType);
const asked = normalizeScope(scope);
// `did:ng:i` (or `""`) discards every other entry upstream and means the whole local
// dataset. TEMPORARY: until the read queries the store itself, that is served
// by enumerating this identity's own documents across its three stores - the widest
// set this pipeline can reach. Every other `graphs` list names documents and is used
// as given, which is what the axis will mean once the enumeration is gone.
const wholeDataset = asked.graphs.includes(ENTIRE_DATASET);
// The current, STABLE snapshot (same reference until a transition rebuilds it).
let snapshot: ShapeQuery<UnionSubject> = {
let snapshot: ShapeQuery<T> = {
data: [],
isPending: true,
isSuccess: false,
@@ -284,21 +159,21 @@ export function useShapeQuery<T = UnionSubject>(
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`).
/** The shape query in force, or `null` between a teardown and the next open. */
let query: OrmSubscriptionHandle | null = null;
/** The change listener registered on {@link query}, so it can be taken off again. */
let onQueryChange: (() => void) | null = null;
/** 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;
/**
* Which open owns the question. A `refetch` supersedes the previous subscription, and
* a teardown supersedes all of them - so every callback checks it is still the current
* one before publishing. A stale answer is dropped, never merged.
*/
let generation = 0;
/** Has the CURRENT generation's initial payload arrived? Gates every publication:
* a change signal that beats `readyPromise` must not be read as an answer. */
let ready = false;
function emit(): void {
for (const l of listeners) {
@@ -310,22 +185,22 @@ export function useShapeQuery<T = UnionSubject>(
}
}
function setSnapshot(next: ShapeQuery<UnionSubject>): void {
function setSnapshot(next: ShapeQuery<T>): void {
snapshot = next;
emit();
}
/**
* Publish "the current answer is UNKNOWN" — the one snapshot every failure in the
* pipeline produces, wherever it was thrown.
* Publish "the current answer is UNKNOWN" - the one snapshot every failure on this
* path produces.
*
* `data` carries the LAST READ THAT ANSWERED rather than `[]`. A subscriber has
* already rendered that list and an observable cannot un-emit it; clearing it would
* put the empty answer back — in the field an application actually renders, and for
* put the empty answer back - in the field an application actually renders, and for
* the very case that must never read as empty. `isSuccess` stays false, so the array
* is never offered as a reply to the question that just failed, and a subscriber that
* gates on `isSuccess` shows nothing new while one that renders `data` keeps what it
* had. Before the first answer there is nothing to keep and `data` is `[]` — the
* had. Before the first answer there is nothing to keep and `data` is `[]` - the
* initial value, published under `isError`, never under `isSuccess`.
*/
function setUnknown(error: unknown): void {
@@ -338,169 +213,134 @@ export function useShapeQuery<T = UnionSubject>(
});
}
/** Resolve the logical scope → the current doc set: the CURRENT wallet's own
* entity documents for that scope, and nothing else.
/**
* The readable items of the live subscription, as a plain array.
*
* It PROPAGATES, and that is the whole point: the set it returns is what the rest
* of the pipeline calls "the scope", so a swallowed failure here does not degrade
* the answer, it INVENTS one — zero documents, a barrier trivially reached over
* them, and `isSuccess` published over a store nobody read. Nothing downstream can
* tell that set apart from a scope that is genuinely empty, because it IS the same
* set. The caller's error state exists for this.
*
* 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 (`readUnion`), not by appearing in
* a scope you did not put it in. */
async function resolveDocs(): Promise<Nuri[]> {
if (!wholeDataset) {
// The caller named the documents. `[]` or an absent `graphs` is upstream's "no
// objects are returned", and it reaches that answer here the same way: an empty
// doc set, a barrier trivially past it, and `isSuccess` over nothing.
return [...new Set(asked.graphs.filter(Boolean))].map((g) => toNuri(g, "useShapeQuery"));
}
const user = getCurrentUser();
const set = new Set<Nuri>();
resolving = true;
try {
if (user) {
for (const type of STORE_TYPES) {
for (const d of await listMyEntityDocs(user, type)) set.add(d);
}
}
} finally {
resolving = false;
}
return [...set];
* The ONE cast in this module, and it sits where the injected SDK meets the shape's
* promise: the subscription is typed by what `configure` received (`unknown` items,
* because the injection is a loose shape), while `shapeType` is what says they are
* `T`. Filtering happens first and on the raw items - the filter reads `@graph` and
* nothing else, so it needs no type at all.
*/
function readable(handle: OrmSubscriptionHandle): T[] {
const caps = getCaps();
const items = [...handle.signalObject];
// No cap issued yet -> the emulation is not in force and the set passes through,
// exactly as `surface/use-shape.ts` does. See `emulated-verifier/read-filter.ts`.
const kept = caps.isEnforcing() ? filterReadable(items, caps) : items;
return kept as T[];
}
/** Subscribe to the CONTAINER document (the scope index) so a change to the doc
* SET re-resolves. Idempotent per NURI.
*
* This one SWALLOWS, deliberately and unlike `resolveDocs` next door. It resolves
* no data and publishes no snapshot — it wires a change signal — so its failure
* cannot dress an unread store up as an empty one. And the answer is not lost with
* it: `resolveDocs` runs immediately after over the SAME account, so a store that
* cannot be reached surfaces there, as an error, one line later. What is lost is
* reactivity to a later change of the SET, which is a different (and much quieter)
* problem than the one this module was fixed for. */
async function ensureContainerSubs(): Promise<void> {
// A named `graphs` list IS the set; nothing a store records can change it, so there
// is no container to follow. Only the whole-dataset reading has one - three, in fact,
// one per store this identity places documents in.
if (!wholeDataset) return;
const containers: Nuri[] = [];
const user = getCurrentUser();
if (user) {
for (const type of STORE_TYPES) {
try {
containers.push(await userStoreDoc(user, type));
} catch (error) {
console.error("[use-shape-query] 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;
setUnknown(error);
return;
}
if (token !== refreshToken) return; // superseded by a newer refresh/reread
const data = subjects.filter((s) => matchesShape(s, typeIris) && matchesSubjects(s, asked.subjects));
const past = barrierReached(docs);
/** Publish what the subscription holds right now. Only ever called on a READY
* generation, so this snapshot is always an answer. */
function publish(handle: OrmSubscriptionHandle, gen: number): void {
if (gen !== generation || !ready) return;
setSnapshot({
data,
isPending: !past,
isSuccess: past,
data: readable(handle),
isPending: false,
isSuccess: true,
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.
/**
* The SDK's shape query, or the error that says it was never injected.
*
* This catch is the surface's ONE answer to "the pipeline could not answer" —
* scope resolution included, since `resolveDocs` propagates. It is reached only
* when the cycle is still the current one; a superseded cycle's failure is
* dropped, because a newer cycle owns the question by then. */
async function refresh(): Promise<void> {
const token = ++refreshToken;
* Upstream an application imports the ORM; here it hands it to `configure`, like the
* `ng` and the `useShape` beside it. Absent, there is nothing to read through - and
* that has to be said out loud rather than answered with an empty set, which is the
* one reading this surface must never produce.
*/
function shapeQuery(): OrmSubscriptionLike {
const injected = getConfig().ormSubscription;
if (!injected) {
throw new Error(
"[ng-eventually] useShapeQuery: no shape query was injected. Pass " +
"`ormSubscription: RdfOrmSubscription.getOrCreate` (from `@ng-org/orm`) to " +
"configure() - this package reads through the SDK's own shape query and has " +
"no read of its own to fall back on.",
);
}
return injected;
}
/** Take the current subscription down: stop listening, close it, forget it. Leaves
* the snapshot and the subscribers alone - this is not an answer, it is plumbing. */
function closeQuery(): void {
if (query === null) return;
const handle = query;
query = null;
ready = false;
if (onQueryChange) {
try {
handle.removeChangeListener(onQueryChange);
} catch {
/* ignore */
}
onQueryChange = null;
}
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;
setUnknown(error);
handle.close();
} catch {
/* ignore */
}
}
/** 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);
/**
* Open a shape query for `(shapeType, scope)` and wire its two signals.
*
* Supersedes whatever was open: the previous generation's callbacks are dropped on
* arrival rather than raced, so a slow answer can never clobber a newer question.
*/
function open(): void {
closeQuery();
const gen = ++generation;
let handle: OrmSubscriptionHandle;
try {
handle = shapeQuery()(shapeType, scope);
} catch (error) {
setUnknown(error);
return;
}
query = handle;
const onChange = (): void => publish(handle, gen);
onQueryChange = onChange;
handle.addChangeListener(onChange);
handle.readyPromise.then(
() => {
if (gen !== generation) return;
ready = true;
publish(handle, gen);
},
(error: unknown) => {
// Nothing upstream rejects this promise today (see the module header); the
// rejection path is wired all the same, because a state nobody handles is how
// a failure becomes silence the day the ORM starts reporting one.
if (gen !== generation) return;
setUnknown(error);
},
);
}
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
// consumer's `inbox.watch`) makes items readable that were not. Without
// this the view would stay stale until some unrelated push happened to fire -
// so re-publish whenever the keyring changes. 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();
if (query !== null) publish(query, generation);
});
void refresh();
open();
}
return {
getSnapshot(): ShapeQuery<T> {
return snapshot as unknown as ShapeQuery<T>;
return snapshot;
},
subscribe(onChange: () => void): () => void {
listeners.add(onChange);
@@ -508,24 +348,10 @@ export function useShapeQuery<T = UnionSubject>(
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();
// Last listener gone -> close the shape query. A later subscribe opens a
// fresh one; the ORM pools what it has just closed, so that is cheap.
generation += 1;
closeQuery();
if (capsUnsub) {
capsUnsub();
capsUnsub = null;
@@ -535,8 +361,11 @@ export function useShapeQuery<T = UnionSubject>(
};
},
refetch(): void {
start();
void refresh();
if (!started) {
start();
return;
}
open();
},
};
}