8338b0efd7
Le commentaire de `namedDocumentsArrived` decrivait `open-repo` comme gardant trois valeurs, et disait que l'attente "resout sur le State ou sur le repli, et chacun pose sa propre valeur" - plus vrai depuis que rien ne les stocke.
731 lines
38 KiB
TypeScript
731 lines
38 KiB
TypeScript
/**
|
|
* 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
|
|
* 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 -------
|
|
* 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
|
|
* `.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
|
|
* scope still syncing reads `{ data: [], isPending: true, isSuccess: false }`.
|
|
*
|
|
* -- 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, one anchored read over
|
|
* each of them, and a filter of the result 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.
|
|
*
|
|
* 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.
|
|
*
|
|
* -- 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.
|
|
*
|
|
* -- 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.
|
|
*
|
|
* -- `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.
|
|
*
|
|
* 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 named document is REACHED before the query opens, since `1.0.0-dev.9` --------
|
|
* `scope.graphs` does not only narrow a search: each entry NAMES a document, and naming
|
|
* one is the same act `docs.sparqlQuery`'s anchor is. So each named document is put
|
|
* through the same two steps that call makes - the (emulated) public store is asked for
|
|
* its cap, and the document is refused out loud if the cap is not held - plus the
|
|
* cold-start open every other read path in this library crosses (`ensureRepoOpen`).
|
|
*
|
|
* What it repairs is this surface's own version of the failure it exists against. A named
|
|
* document whose cap the holder does not hold used to reach the engine, come back, and be
|
|
* emptied by the capability filter below, publishing `isSuccess` with `data: []` - "this
|
|
* document holds nothing of that shape" for "you cannot read this document", while the
|
|
* anchored read of the SAME document refuses. A named document this session has never
|
|
* synced published the same snapshot for a document that is not empty. Both are now
|
|
* `isError`, which is what `isError` has always meant here: the current answer is UNKNOWN.
|
|
*
|
|
* **What is deliberately NOT guarded, and each is an answer rather than a failure:**
|
|
* - `graphs: "did:ng:i"` or `""` - the whole local dataset. The application named no
|
|
* document, so there is no cap question to ask; the capability filter is the right
|
|
* and only answer, exactly as the broker delivering only what your keys open is.
|
|
* - `graphs: []` - no objects, by the SDK's own definition. `isSuccess` with `data: []`.
|
|
* - a named document held and synced that holds nothing of the shape - `isSuccess` with
|
|
* `data: []`, which is the distinction this whole surface exists to publish.
|
|
* - the INERT case: before the first cap is issued nothing is enforcing, the public
|
|
* store is not asked at all, and `assertMayReach` passes for everything.
|
|
*
|
|
* A scope naming SEVERAL documents fails as a whole when one of them cannot be reached,
|
|
* rather than answering the others. A short list is the same lie as an empty one, and it
|
|
* is the tolerance `readUnion` was removed for in `1.0.0-dev.8`.
|
|
*
|
|
* The barrier is why the open below is asynchronous: it prolongs `isPending`, and nothing
|
|
* else about the states moves.
|
|
*
|
|
* -- A refusal that could not be re-tried would be worse than the defect ------------
|
|
* The first version of this barrier was TERMINAL, and an adversarial pass found it. The
|
|
* keyring signal below is written as `if (query !== null) publish(...)` because until the
|
|
* barrier existed there was only ever one reason to have no query - nothing had been
|
|
* tried yet - and a cap arriving could only ever change what an OPEN query was allowed to
|
|
* show. The barrier added a second reason: it aborts BEFORE the query is created, so the
|
|
* one channel that makes a late cap converge went straight past the state that needed it
|
|
* most. A page subscribing while its keyring was still being restored, or a document
|
|
* shared with someone while they watched, published `isError` and stayed there for the
|
|
* life of the view. Trading a silent empty answer for a stuck error is not a stricter
|
|
* surface, it is a worse one - and it broke a guarantee this package publishes, which
|
|
* `emulated-verifier/read-filter.ts` and `surface/inbox.ts` both restate.
|
|
*
|
|
* So the signal re-ATTEMPTS a refused scope. Two things make that sound, and both were
|
|
* arrived at by getting them wrong first:
|
|
*
|
|
* - **the decision is a LEVEL, not a memory.** Which documents the holder may reach is
|
|
* re-read on every publication ({@link publish}), never carried over from the barrier.
|
|
* So a cap arriving turns a refusal into the document through the channel that
|
|
* already existed, and a query opened while somebody else was connected cannot come
|
|
* back to this subscriber as a success it is not entitled to. Nothing is remembered,
|
|
* so there is no event to miss.
|
|
* - **an attempt that RACED a change notices, rather than waiting.** A keyring change
|
|
* landing while the barrier is crossing fires its listener when there is no query to
|
|
* re-publish and no refusal on record - and that same change is often what makes the
|
|
* walk refuse. The edge is simply gone. So the attempt samples `CapRegistry.version()`
|
|
* before it starts and compares after it fails; a mismatch means it decided on stale
|
|
* ground and it makes another attempt itself.
|
|
*
|
|
* This is the third shape this recovery has had, and the first two failed the same way -
|
|
* each recorded that it had refused and then waited for a later event. Round 2 gated the
|
|
* record on a condition that could not become true; round 3 set the record after the
|
|
* window in which the deciding event arrives. An edge that is missed is not recoverable
|
|
* by construction, which is why the answer here is to stop depending on edges.
|
|
*
|
|
* -- Why there is no cleverer gate than "attempt again" ----------------------------
|
|
* The obvious one is to re-open only once the refusal has been lifted - "does the holder
|
|
* hold every named document's cap now". It was written that way first, and it is CIRCULAR
|
|
* for a whole class: for a document in a public store the only channel that can ever file
|
|
* that cap is `fetchReadCap`, the very call the gate was there to withhold. A scope
|
|
* refused because the store could not be ASKED then waited on an answer that only the
|
|
* action it gated could produce, and no number of cap arrivals ever released it. A second
|
|
* adversarial pass proved it with the broker healing and ten caps landing afterwards.
|
|
*
|
|
* Attempting is the cheaper question anyway, because the two ways to be refused are
|
|
* repaired by opposite things and the memo already tells them apart:
|
|
*
|
|
* - **the store could not be asked.** A failure is deliberately NOT memoised
|
|
* (`public-store.ts`), so a re-attempt genuinely re-asks the network - which is the
|
|
* whole repair for this class, not a cost to avoid.
|
|
* - **the store answered and there is no cap to be had.** An absence IS memoised, so a
|
|
* re-attempt asks nobody: it re-runs the barrier against what is already known and
|
|
* republishes the same refusal. It costs one snapshot, and it stops the day the cap
|
|
* arrives from somewhere else, which is the only thing that can repair this class.
|
|
*
|
|
* It is self-limiting either way: the moment an attempt succeeds there is a query again,
|
|
* and this branch is never reached. What it is NOT is a timer - nothing here polls, and
|
|
* a scope refused in a session where the keyring never moves again stays `isError` until
|
|
* the caller's own `refetch()`, which is what `isError` has always meant on this surface.
|
|
*
|
|
* -- 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
|
|
* 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, getConfig } from "../shared-wallet/bootstrap";
|
|
import { filterReadable } from "../emulated-verifier/read-filter";
|
|
import { ensureRepoOpen, getSyncState } from "../emulated-verifier/open-repo";
|
|
import { assertMayReach } from "../emulated-verifier/reach";
|
|
import { isNuri } from "../model/nuri";
|
|
import type { Nuri, 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";
|
|
import type { BaseType, ShapeType } from "@ng-org/shex-orm";
|
|
|
|
/**
|
|
* A minimal TanStack-`useQuery`-shaped read snapshot. `data` is ALWAYS an array
|
|
* (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; `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 until the shape query's initial payload has arrived (`readyPromise`).
|
|
* Mutually exclusive with `isSuccess`. */
|
|
isPending: boolean;
|
|
/** 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 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 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 closes the underlying shape query. */
|
|
subscribe(onChange: () => void): () => void;
|
|
/** 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;
|
|
}
|
|
|
|
/**
|
|
* The whole local dataset, as the SDK spells it. `normalizeScope` maps `""` onto this
|
|
* very string (`@ng-org/orm`'s `types.ts`), so both forms are the same marker and neither
|
|
* names a document.
|
|
*/
|
|
const WHOLE_DATASET = "did:ng:i";
|
|
|
|
/**
|
|
* The DOCUMENTS a scope NAMES - the only entries a capability question can be asked
|
|
* about.
|
|
*
|
|
* Flattened the way the SDK flattens `graphs` (a bare string is one entry), then reduced
|
|
* to what identifies a document: the whole-dataset marker is dropped because it names
|
|
* none, and so is anything that is not a NextGraph reference - an ordinary IRI names no
|
|
* repo this library could hold a cap for, and the engine is left to match it or not.
|
|
* Deduplicated, so a scope repeating a document opens it once.
|
|
*
|
|
* `""` is named beside the marker although `isNuri` would drop it anyway. It is the SDK's
|
|
* own alias for `did:ng:i`, and a reader should be able to see that here rather than have
|
|
* to work out that a guard two files away happens to exclude it.
|
|
*/
|
|
function namedDocuments(scope: Scope): Nuri[] {
|
|
const out: Nuri[] = [];
|
|
for (const g of [scope.graphs].flat()) {
|
|
if (typeof g !== "string" || g === "" || g === WHOLE_DATASET) continue;
|
|
if (isNuri(g) && !out.includes(g)) out.push(g);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
/**
|
|
* Are these two failures the same one, for the purpose of deciding whether a subscriber
|
|
* has anything new to look at?
|
|
*
|
|
* By MESSAGE, because every attempt builds its own `Error` object and identity would
|
|
* therefore always answer "different". A message is what an application shows and what a
|
|
* maintainer greps for, so two failures that read alike are alike here.
|
|
*/
|
|
function sameFailure(a: unknown, b: unknown): boolean {
|
|
if (a === b) return true;
|
|
return a instanceof Error && b instanceof Error && a.message === b.message;
|
|
}
|
|
|
|
/**
|
|
* Create a reactive, `useQuery`-shaped observable over one SHEX `shapeType` in one
|
|
* `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 extends BaseType = BaseType>(
|
|
shapeType: ShapeType<T>,
|
|
scope: Scope,
|
|
): ShapeObservable<T> {
|
|
// The current, STABLE snapshot (same reference until a transition rebuilds it).
|
|
let snapshot: ShapeQuery<T> = {
|
|
data: [],
|
|
isPending: true,
|
|
isSuccess: false,
|
|
isError: false,
|
|
error: undefined,
|
|
};
|
|
|
|
const listeners = new Set<() => void>();
|
|
let started = false;
|
|
/** 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;
|
|
/**
|
|
* 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;
|
|
/**
|
|
* Is the BARRIER what left this observable without a query - rather than the shape
|
|
* query failing to open, or nothing having been tried yet?
|
|
*
|
|
* Only the first is worth attempting again: a `configure` with no `ormSubscription` is
|
|
* not repaired by a cap arriving, and retrying it on every keyring change would churn
|
|
* for a misconfiguration no amount of retrying repairs.
|
|
*
|
|
* It records no HOLDER, and an earlier version did. Keying the retry on "the same
|
|
* holder that was refused" looks like the way to stop a re-attempt acting for somebody
|
|
* who never asked, and it is a trap twice over: the anonymous ring is a holder like any
|
|
* other, so a refusal taken before the identity settles - which is where components
|
|
* mount - never matched again once it did; and an identity CAN change under a live page
|
|
* (`shared-wallet/access-gate.ts` adopts the identifier a revived barrier confirms), so
|
|
* the rule left that case terminal too. What actually has to hold is that no snapshot
|
|
* ever claims success for a document the CURRENT holder cannot reach, and that is
|
|
* enforced where it belongs, on every publication - see {@link publish}.
|
|
*/
|
|
let barrierRefused = false;
|
|
|
|
function emit(): void {
|
|
for (const l of listeners) {
|
|
try {
|
|
l();
|
|
} catch (error) {
|
|
console.error("[use-shape-query] listener threw", error);
|
|
}
|
|
}
|
|
}
|
|
|
|
function setSnapshot(next: ShapeQuery<T>): void {
|
|
snapshot = next;
|
|
emit();
|
|
}
|
|
|
|
/**
|
|
* 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
|
|
* 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
|
|
* initial value, published under `isError`, never under `isSuccess`.
|
|
*/
|
|
function setUnknown(error: unknown): void {
|
|
// A standing failure that has not changed is not news. This surface documents its
|
|
// snapshot as STABLE until it actually changes, and a re-attempt that lands on the
|
|
// same refusal changes nothing an application renders - so handing out a fresh object
|
|
// carrying a fresh `Error` would break that promise and re-render every bound view
|
|
// for nothing. Measured: eight unrelated cap arrivals over a failing fetch produced
|
|
// eight notifications and eight distinct `Error`s for one unchanged state.
|
|
//
|
|
// Compared by MESSAGE rather than by identity, because each attempt builds its own
|
|
// `Error`. A failure that says something different IS news and is published.
|
|
if (snapshot.isError && sameFailure(snapshot.error, error)) return;
|
|
setSnapshot({
|
|
data: snapshot.data,
|
|
isPending: false,
|
|
isSuccess: false,
|
|
isError: true,
|
|
error,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Publish "the question is still open" - no authority yet, and nothing to say about the
|
|
* data either way.
|
|
*
|
|
* `data` is carried over for the same reason {@link setUnknown} carries it: an observable
|
|
* cannot un-emit a list a subscriber has already rendered. Before the first answer there
|
|
* is nothing to carry and it stays `[]`. Idempotent, so a document that keeps failing to
|
|
* sync does not wake every bound view on each attempt.
|
|
*/
|
|
function setNotYetAnswered(): void {
|
|
if (snapshot.isPending && !snapshot.isSuccess && !snapshot.isError) return;
|
|
setSnapshot({
|
|
data: snapshot.data,
|
|
isPending: true,
|
|
isSuccess: false,
|
|
isError: false,
|
|
error: undefined,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* The readable items of the live subscription, as a plain array.
|
|
*
|
|
* 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[];
|
|
}
|
|
|
|
/**
|
|
* The refusal standing against the CURRENT holder for the first document the scope
|
|
* NAMES that it cannot reach, or `null` when it reaches all of them.
|
|
*
|
|
* It goes through `assertMayReach` and catches, rather than deciding again: the refusal
|
|
* is worded in exactly one place and a second wording here would drift from it. Scope
|
|
* order, like the barrier.
|
|
*/
|
|
function namedRefusal(): unknown {
|
|
for (const doc of namedDocuments(scope)) {
|
|
try {
|
|
assertMayReach(doc, "useShapeQuery");
|
|
} catch (error) {
|
|
return error;
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* Has every document the scope NAMES actually ARRIVED - not merely "may this holder
|
|
* reach it", which is a different question and the only one this module used to ask?
|
|
*
|
|
* `emulated-verifier/open-repo.ts` DERIVES four values for exactly this reader - since
|
|
* `1.0.0-dev.10` nothing stores them, they are read off the document's one real
|
|
* subscription - and which of them counts as an answer is the whole of the decision:
|
|
*
|
|
* - `"synced"` - a `State` has arrived on the channel standing NOW. THE truth signal:
|
|
* zero rows means zero objects, and that is the success this surface exists to publish.
|
|
* Read off the channel rather than remembered, so it cannot survive that channel's
|
|
* teardown - which is what used to let a second `configure(...)` answer a success for a
|
|
* document nothing was subscribed to.
|
|
* - `"timed-out"` - opened, and no barrier to show for it: the bounded fallback fired, the
|
|
* subscribe was refused, or the channel was taken away. Sync unconfirmed. The document
|
|
* may hold anything and this read has no standing to call it empty.
|
|
* - `"syncing"` - the open is still inside its bounded wait; refused for the same reason,
|
|
* and if it is ever seen here the barrier demonstrably was not crossed.
|
|
* - `"unknown"` - "never opened there, or the fake-ng no-op path: no `State` semantics".
|
|
* **Passes, deliberately.** There is no barrier to have missed where there are no
|
|
* barrier semantics, and an `ng` without `doc_subscribe` is the whole unit suite and
|
|
* every early path. Treating it as not-synced is the inert regression this line of
|
|
* work has already shipped twice.
|
|
*/
|
|
function namedDocumentsArrived(): boolean {
|
|
for (const doc of namedDocuments(scope)) {
|
|
const state = getSyncState(doc);
|
|
if (state === "timed-out" || state === "syncing") return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Publish what the subscription holds right now. Only ever called on a READY
|
|
* generation, so this snapshot is always an answer.
|
|
*
|
|
* **The named documents are re-checked HERE, every time, and that is what makes the
|
|
* refusal self-healing.** The barrier decides once, before the query opens; this
|
|
* decides on each publication, against what the holder holds at that instant - so a cap
|
|
* arriving turns `isError` into the document through the channel that already existed,
|
|
* and a query opened while somebody else was connected can never come back to this
|
|
* subscriber as `isSuccess` holding nothing. A level, not an edge: there is no event to
|
|
* miss, because nothing is remembered.
|
|
*
|
|
* The capability FILTER below still applies on top, and the two are not the same
|
|
* question: the filter bounds what a scope may show per `@graph`, this refuses a scope
|
|
* the caller NAMED and cannot read. Under `did:ng:i` nothing is named and the filter is
|
|
* the whole of the answer, exactly as before.
|
|
*/
|
|
function publish(handle: OrmSubscriptionHandle, gen: number): void {
|
|
if (gen !== generation || !ready) return;
|
|
const refusal = namedRefusal();
|
|
if (refusal !== null) {
|
|
setUnknown(refusal);
|
|
return;
|
|
}
|
|
// NOT AUTHORITATIVE YET, so say the question is still open rather than say "empty". A
|
|
// named document whose sync barrier was never crossed may hold anything, and
|
|
// publishing the ORM's view of it as `isSuccess` is the original defect in the one case
|
|
// the barrier was added to close.
|
|
//
|
|
// It PUBLISHES pending rather than returning. Returning left whatever snapshot was
|
|
// standing, and that was wrong in the case that matters: a refusal that has since been
|
|
// LIFTED stayed on display, so an application was handed `isError` reading "you do not
|
|
// hold this document's cap" about a document it now holds. Whatever the state was, it
|
|
// is not the answer to the question just asked.
|
|
if (!namedDocumentsArrived()) {
|
|
setNotYetAnswered();
|
|
return;
|
|
}
|
|
setSnapshot({
|
|
data: readable(handle),
|
|
isPending: false,
|
|
isSuccess: true,
|
|
isError: false,
|
|
error: undefined,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* The SDK's shape query, or the error that says it was never injected.
|
|
*
|
|
* 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 {
|
|
handle.close();
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Cross the barrier on every document the scope NAMES, before the query opens - the
|
|
* same two steps `docs.sparqlQuery` takes on its anchor, in the same order.
|
|
*
|
|
* `ensureRepoOpen` asks the (emulated) public store for the document's cap, and opens
|
|
* and syncs the repo when it may be touched; it declines SILENTLY when it may not,
|
|
* which is right for a caller that has another way to find out. This one has none, so
|
|
* `assertMayReach` says it out loud immediately after - without that, an unheld
|
|
* document would simply produce no objects and read as a document holding none.
|
|
*
|
|
* The opens run TOGETHER, and every outcome is then read in SCOPE ORDER, so the
|
|
* document reported is the first one the caller named that failed - whatever it failed
|
|
* of. Two earlier shapes were wrong about this and the difference is not cosmetic, it
|
|
* is which document the error names:
|
|
*
|
|
* - one at a time: correct order, unacceptable cost. Each open waits on the sync
|
|
* barrier, bounded by `OPEN_TIMEOUT_MS` (8s) for a document that never pushes a
|
|
* `State`, so a scope naming N of them paid N timeouts before answering anything.
|
|
* - `Promise.all`: right cost, wrong order. It SHORT-CIRCUITS on the first rejection
|
|
* to arrive, which is a race - a fetch that rejects does not wait for the sync
|
|
* barrier, so it overtakes a refusal on a document named before it, and the error
|
|
* named the wrong document while a comment beside it claimed otherwise.
|
|
*
|
|
* `allSettled` has neither fault: nothing is discarded, nothing rejects early, and no
|
|
* rejection escapes as an unhandled one. The opens that succeeded are memoised per
|
|
* session exactly as they would have been, so a re-subscribe costs a map hit.
|
|
*/
|
|
async function reachNamedDocuments(): Promise<void> {
|
|
const docs = namedDocuments(scope);
|
|
if (docs.length === 0) return;
|
|
const outcomes = await Promise.allSettled(docs.map((doc) => ensureRepoOpen(doc)));
|
|
for (const [i, doc] of docs.entries()) {
|
|
const outcome = outcomes[i];
|
|
if (outcome && outcome.status === "rejected") throw outcome.reason;
|
|
assertMayReach(doc, "useShapeQuery");
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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. The
|
|
* barrier above runs first, which is why the body is asynchronous - every step past an
|
|
* `await` re-checks that it is still the current generation before touching anything,
|
|
* so a `refetch` or a teardown during the barrier leaves no query behind.
|
|
*/
|
|
function open(): void {
|
|
closeQuery();
|
|
const gen = ++generation;
|
|
barrierRefused = false;
|
|
// Sampled SYNCHRONOUSLY, before anything is awaited: what the keyring looked like when
|
|
// this attempt decided to trust it. See the comparison in the catch below.
|
|
const keyringAtStart = getCaps().version();
|
|
void (async () => {
|
|
try {
|
|
await reachNamedDocuments();
|
|
} catch (error) {
|
|
if (gen !== generation) return;
|
|
// THE EDGE THAT WOULD OTHERWISE BE MISSED, and it is the one that matters most: a
|
|
// keyring change landing WHILE the barrier was crossing fired its listener at a
|
|
// moment when there was no query to re-publish and no refusal on record yet, so
|
|
// the listener did nothing - and that same change is what made the walk refuse.
|
|
// Waiting for another event would be waiting for one that may never come; this
|
|
// attempt simply notices it raced and makes another, which re-samples.
|
|
//
|
|
// It terminates: an attempt is repeated only because the keyring moved, moves are
|
|
// finite events, and the moves an attempt causes itself are the caps it files -
|
|
// at most one per named document, after which that document is held and asks
|
|
// nothing.
|
|
if (getCaps().version() !== keyringAtStart) {
|
|
open();
|
|
return;
|
|
}
|
|
barrierRefused = true;
|
|
setUnknown(error);
|
|
return;
|
|
}
|
|
// Superseded while the barrier was crossing: drop this open rather than leaving a
|
|
// subscription nothing will ever close.
|
|
if (gen !== generation) return;
|
|
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 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 (query !== null) {
|
|
publish(query, generation);
|
|
return;
|
|
}
|
|
// No query in force. Since `1.0.0-dev.9` there is a second way to be here, and it
|
|
// is the one this signal exists for: the BARRIER refused the scope, so the query
|
|
// was never created and there is nothing for `publish` to re-filter. A view left
|
|
// like that would be TERMINAL - the very case this channel was added to prevent,
|
|
// arriving by a door that did not exist when the channel was written.
|
|
//
|
|
// So ATTEMPT AGAIN. There is deliberately no further test of whether the refusal
|
|
// has been lifted, and none of who is asking: see the module header for why the
|
|
// obvious gate is circular, why an attempt is the cheaper question anyway, and why
|
|
// the holder belongs in {@link publish} rather than here.
|
|
if (barrierRefused) open();
|
|
});
|
|
open();
|
|
}
|
|
|
|
return {
|
|
getSnapshot(): ShapeQuery<T> {
|
|
return snapshot;
|
|
},
|
|
subscribe(onChange: () => void): () => void {
|
|
listeners.add(onChange);
|
|
start();
|
|
return () => {
|
|
listeners.delete(onChange);
|
|
if (listeners.size === 0) {
|
|
// 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;
|
|
}
|
|
started = false;
|
|
}
|
|
};
|
|
},
|
|
refetch(): void {
|
|
if (!started) {
|
|
start();
|
|
return;
|
|
}
|
|
open();
|
|
},
|
|
};
|
|
}
|