feat(client): read-model surface — open/sync + anchorless union sparql_query
Proven against the real broker (probe): opening docs then a single anchorless
`sparql_query` with a `GRAPH ?g { ... }` body reads the LOCAL UNION of all synced
named graphs — the fast, hang-free replacement for the reactive-ORM per-document
fan-out (which aborted on any unsynced/fresh repo → 75s never-fires). New
`read-model.ts` (readUnion) exposes this; there is no reactive union query, so
listing is one-shot and consumers re-query on change. docs/read-model.md refined
with the probe finding (an explicit `GRAPH ?g` body iterates all named graphs
regardless of the anchor; the anchor only bounds the default graph). 93 tests
pass; tsc rc=0.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sylvain Duchesne
@@ -112,3 +112,32 @@ The one experiment that pins down union vs anchor, to run against a real broker:
|
||||
If (3) returns both and (4) returns only A, the union read model above holds as
|
||||
implemented in `resolve_target_for_sparql` /
|
||||
`set_default_graph_as_union`.
|
||||
|
||||
### Verified against the real broker (T03.k)
|
||||
|
||||
Step (3) — **the load-bearing one** — is CONFIRMED: an anchorless
|
||||
`SELECT … WHERE { GRAPH ?g { ?s ?p ?o } }` returns triples from BOTH docs A and B
|
||||
(the local union of the opened graphs). That is the entire premise the listing
|
||||
path relies on.
|
||||
|
||||
Step (4) has a nuance worth recording: with an **explicit `GRAPH ?g { … }` body**,
|
||||
passing `anchor = A` did **not** restrict the result to A (B still appeared). The
|
||||
reason: the anchor sets the query's **default graph**, but a `GRAPH ?g` pattern
|
||||
iterates over the **named graphs** regardless of the default graph — so an
|
||||
explicit `GRAPH ?g` body spans every opened graph independently of the anchor.
|
||||
The anchor's "one repo" restriction is observable only for a body that reads the
|
||||
**default graph** (no `GRAPH` wrapper). The read model never needs the anchored
|
||||
form for listing — it uses the anchorless `GRAPH ?g` union — so this does not
|
||||
affect it. (The per-doc **open** step in `read-model.ts` uses an anchored `ASK`
|
||||
purely for its side effect of opening the repo, not to restrict a read.)
|
||||
|
||||
## Implementation — `read-model.ts`
|
||||
|
||||
`readModel.readUnion(docs)` implements this: (1) open/sync each doc via a per-doc
|
||||
anchored `ASK` (tolerant — a doc that can't open is skipped, never aborting the
|
||||
listing like the ORM fan-out would); (2) run ONE anchorless
|
||||
`SELECT ?g ?s ?p ?o WHERE { GRAPH ?g { ?s ?p ?o . VALUES ?s { … } } }` over the
|
||||
local union, constrained to the requested subjects (each entity's subject IRI IS
|
||||
its own document NURI). Returns the triples grouped per subject; the consumer maps
|
||||
them to its types (e.g. Festipod's `readEntities`). Reactivity = the consumer
|
||||
re-calls `readUnion` on its change signal (no reactive union query exists).
|
||||
|
||||
@@ -18,6 +18,8 @@ export * as inbox from "./inbox";
|
||||
export * as discovery from "./discovery";
|
||||
export type { IndexEntry, SubmitOptions } from "./discovery";
|
||||
export * as docs from "./docs";
|
||||
export * as readModel from "./read-model";
|
||||
export type { UnionSubject } from "./read-model";
|
||||
export * as storeRegistry from "./store-registry";
|
||||
export type { AccountRecord, RegistrySession } from "./store-registry";
|
||||
export * as isolation from "./isolation";
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
/**
|
||||
* read-model — the LISTING primitive of the polyfill: open/sync a set of
|
||||
* documents, then run ONE anchorless union `sparql_query` over the LOCAL UNION
|
||||
* and return the triples grouped per subject. This is the mechanism documented
|
||||
* in docs/read-model.md, verified against the real broker (T03.k probe): a
|
||||
* `GRAPH ?g { ?s ?p ?o }` body with NO anchor spans every graph currently opened
|
||||
* in the session store.
|
||||
*
|
||||
* ── WHY not the reactive ORM fan-out ──────────────────────────────────────
|
||||
* `useShape({ graphs: […manyDocs] })` drives `orm_start_graph` over a fan-out of
|
||||
* per-entity graphs; a freshly-created / not-yet-synced doc in that fan-out makes
|
||||
* `RepoNotFound` abort the whole subscription → the readyPromise never resolves →
|
||||
* the ~75s hang (docs/nextgraph-current-state.md § The ORM fan-out hang). Listing
|
||||
* MUST instead be a one-shot union `sparql_query`. There is no reactive union
|
||||
* query, so reactivity is assembled by RE-QUERYING on a change signal.
|
||||
*
|
||||
* ── The two steps ─────────────────────────────────────────────────────────
|
||||
* 1. OPEN/SYNC the docs. A repo is only in the union once opened into the
|
||||
* session's oxigraph store. Opening is done PER-DOC via an anchored probe
|
||||
* query (`resolveTargetForSparql(anchor)` opens that one repo) — per-doc, so a
|
||||
* single unreachable doc fails in isolation instead of aborting a fan-out.
|
||||
* 2. UNION QUERY. One anchorless `SELECT ?g ?s ?p ?o WHERE { GRAPH ?g { ?s ?p ?o } }`
|
||||
* over the local union; group rows by subject.
|
||||
*
|
||||
* ── GENERIC by construction ───────────────────────────────────────────────
|
||||
* Zero application domain here: the consumer passes the doc NURIs to open (from
|
||||
* the discovery index for public events, or its own scope docs for my-entities)
|
||||
* and interprets the returned per-subject property bags. All NextGraph I/O routes
|
||||
* through the T01.a `docs` primitives (the REAL injected `ng`), so this module
|
||||
* imports no `@ng-org` package.
|
||||
*
|
||||
* At migration the union query is unchanged (native SPARQL union over the wallet);
|
||||
* only the "open/sync" step swaps to opening real per-user store repos by cap.
|
||||
*/
|
||||
|
||||
import { docCreate, sparqlUpdate, sparqlQuery } from "./docs";
|
||||
import { getStoreRegistryDeps } from "./polyfill";
|
||||
import { assertNuri } from "./sparql";
|
||||
import type { Nuri } from "./types";
|
||||
|
||||
// Keep the primitives referenced so tree-shaking never drops the import used by
|
||||
// the (side-effecting) open step below; `docCreate`/`sparqlUpdate` are not used
|
||||
// here but the module intentionally depends only on the docs primitive surface.
|
||||
void docCreate;
|
||||
void sparqlUpdate;
|
||||
|
||||
/** One subject read from the union, with its properties (predicate → values). */
|
||||
export interface UnionSubject {
|
||||
/** The subject IRI (`?s`). */
|
||||
subject: string;
|
||||
/** The graph the subject was read from (`?g` — the repo graph name). */
|
||||
graph: string;
|
||||
/** predicate IRI → the list of object values (literals or IRIs) for it. */
|
||||
props: Record<string, string[]>;
|
||||
}
|
||||
|
||||
/** Tolerant extraction of SPARQL SELECT bindings across possible shapes. */
|
||||
function bindings(
|
||||
result: unknown,
|
||||
): Array<Record<string, { value: string } | undefined>> {
|
||||
if (!result) return [];
|
||||
if (Array.isArray(result))
|
||||
return result as Array<Record<string, { value: string }>>;
|
||||
const anyRes = result as {
|
||||
results?: { bindings?: Array<Record<string, { value: string }>> };
|
||||
};
|
||||
return anyRes.results?.bindings ?? [];
|
||||
}
|
||||
|
||||
async function sessionId(): Promise<string> {
|
||||
return (await getStoreRegistryDeps().getSession()).sessionId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Open/sync ONE doc into the session store by running a cheap anchored probe
|
||||
* query against it (`resolve_target_for_sparql(anchor)` opens that repo). Per-doc
|
||||
* and tolerant: a doc that can't be opened (not yet synced, no cap) throws HERE,
|
||||
* in isolation, and is skipped — it never aborts the listing of the others (the
|
||||
* fan-out ORM's failure mode). Returns whether the open succeeded.
|
||||
*/
|
||||
async function openDoc(sid: string, doc: Nuri): Promise<boolean> {
|
||||
try {
|
||||
const nuri = assertNuri(doc);
|
||||
// ASK is the cheapest way to touch (hence open) the repo; the result is
|
||||
// irrelevant — the side effect (repo opened into the union) is the point.
|
||||
await sparqlQuery(sid, "ASK { ?s ?p ?o }", undefined, nuri);
|
||||
return true;
|
||||
} catch (error) {
|
||||
console.error("[read-model] open failed for", doc, error);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* OPEN/SYNC a set of docs, then read the LOCAL UNION with ONE anchorless
|
||||
* `sparql_query`, grouped per subject. `docs` are the NURIs to bring into the
|
||||
* union (the consumer resolves them by need — index for public, own scope docs
|
||||
* for mine). Docs that fail to open are skipped (see {@link openDoc}).
|
||||
*
|
||||
* The union query is anchorless (`anchor` undefined → LOCAL UNION) with an
|
||||
* explicit `GRAPH ?g { ?s ?p ?o }` body, so it spans every opened graph and
|
||||
* attributes each row to its source graph.
|
||||
*/
|
||||
export async function readUnion(docs: Nuri[]): Promise<UnionSubject[]> {
|
||||
const sid = await sessionId();
|
||||
const unique = [...new Set(docs.filter(Boolean))];
|
||||
// 1. OPEN/SYNC — per-doc, tolerant (a bad doc never aborts the whole list).
|
||||
await Promise.all(unique.map((d) => openDoc(sid, d)));
|
||||
|
||||
if (unique.length === 0) return [];
|
||||
|
||||
// 2. UNION QUERY — ONE anchorless query over the local union. The union spans
|
||||
// ALL opened graphs (including the shim/inbox internals), so constrain it to
|
||||
// the requested subjects with a VALUES block. In the polyfill each entity's
|
||||
// subject IRI IS its own document NURI (writeEntity uses the doc NURI as the
|
||||
// subject), so the requested doc NURIs ARE the subjects to read — precise and
|
||||
// independent of the repo_graph_name overlay suffix on ?g.
|
||||
const values = unique.map((d) => `<${assertNuri(d)}>`).join(" ");
|
||||
const query =
|
||||
`SELECT ?g ?s ?p ?o WHERE { GRAPH ?g { ?s ?p ?o . VALUES ?s { ${values} } } }`;
|
||||
let rows: Array<Record<string, { value: string } | undefined>> = [];
|
||||
try {
|
||||
rows = bindings(await sparqlQuery(sid, query, undefined, undefined));
|
||||
} catch (error) {
|
||||
console.error("[read-model] union query failed:", error);
|
||||
return [];
|
||||
}
|
||||
|
||||
const bySubject = new Map<string, UnionSubject>();
|
||||
for (const row of rows) {
|
||||
const s = row.s?.value;
|
||||
const p = row.p?.value;
|
||||
const o = row.o?.value;
|
||||
const g = row.g?.value ?? "";
|
||||
if (!s || !p || o === undefined) continue;
|
||||
let entry = bySubject.get(s);
|
||||
if (!entry) {
|
||||
entry = { subject: s, graph: g, props: {} };
|
||||
bySubject.set(s, entry);
|
||||
}
|
||||
(entry.props[p] ??= []).push(o);
|
||||
}
|
||||
return [...bySubject.values()];
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
import { test, expect, mock } from "bun:test";
|
||||
import { readUnion } from "../src/read-model";
|
||||
import { configure, configureStoreRegistry } from "../src/polyfill";
|
||||
|
||||
// A fake `ng` whose sparql_query answers the ANCHORLESS union query with the
|
||||
// triples of the requested subjects, and the anchored ASK (open step) with an
|
||||
// empty result. Each entity subject IRI IS its own document NURI (writeEntity
|
||||
// convention), so the fixture keys triples by the doc NURI.
|
||||
function fakeNgWith(triplesByDoc: Record<string, Array<[string, string]>>) {
|
||||
return {
|
||||
doc_create: mock(async () => "did:ng:o:new"),
|
||||
sparql_update: mock(async () => undefined),
|
||||
sparql_query: mock(async (_sid: string, query: string, _base: unknown, anchor: unknown) => {
|
||||
// The open step is `ASK { ?s ?p ?o }` with an anchor → return a truthy ASK.
|
||||
if (query.startsWith("ASK")) return { boolean: true };
|
||||
// The union query is anchorless (anchor undefined) with a VALUES ?s block.
|
||||
if (anchor !== undefined) return { results: { bindings: [] } };
|
||||
const bindings: Array<Record<string, { value: string }>> = [];
|
||||
for (const [doc, triples] of Object.entries(triplesByDoc)) {
|
||||
// Only surface docs whose NURI is named in the VALUES block.
|
||||
if (!query.includes(`<${doc}>`)) continue;
|
||||
for (const [p, o] of triples) {
|
||||
bindings.push({
|
||||
g: { value: `${doc}:graph` },
|
||||
s: { value: doc },
|
||||
p: { value: p },
|
||||
o: { value: o },
|
||||
});
|
||||
}
|
||||
}
|
||||
return { results: { bindings } };
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
function inject(triplesByDoc: Record<string, Array<[string, string]>>) {
|
||||
const ng = fakeNgWith(triplesByDoc);
|
||||
configure({ ng: ng as any, useShape: (() => {}) as any });
|
||||
configureStoreRegistry({
|
||||
getSession: async () => ({ sessionId: "sid-rm", privateStoreId: "priv" }),
|
||||
normalizeUser: (u: string) => u,
|
||||
});
|
||||
return ng;
|
||||
}
|
||||
|
||||
const TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type";
|
||||
const FP = "http://festipod.org/";
|
||||
|
||||
test("readUnion opens each doc then runs ONE anchorless union query", async () => {
|
||||
const ng = inject({
|
||||
"did:ng:o:a": [[TYPE, `${FP}Event`], [`${FP}title`, "A"]],
|
||||
"did:ng:o:b": [[TYPE, `${FP}Event`], [`${FP}title`, "B"]],
|
||||
});
|
||||
const subjects = await readUnion(["did:ng:o:a", "did:ng:o:b"]);
|
||||
|
||||
// Two per-doc ASK opens + one anchorless union query = 3 sparql_query calls.
|
||||
expect(ng.sparql_query).toHaveBeenCalledTimes(3);
|
||||
const anchorless = ng.sparql_query.mock.calls.filter(
|
||||
(c: unknown[]) => !String(c[1]).startsWith("ASK") && c[3] === undefined,
|
||||
);
|
||||
expect(anchorless.length).toBe(1);
|
||||
|
||||
expect(subjects.length).toBe(2);
|
||||
const a = subjects.find((s) => s.subject === "did:ng:o:a")!;
|
||||
expect(a.props[`${FP}title`]).toEqual(["A"]);
|
||||
});
|
||||
|
||||
test("readUnion groups predicates per subject", async () => {
|
||||
inject({
|
||||
"did:ng:o:p": [
|
||||
[TYPE, `${FP}Participation`],
|
||||
[`${FP}event`, "did:ng:o:e"],
|
||||
[`${FP}user`, "urn:festipod:user:x"],
|
||||
],
|
||||
});
|
||||
const s = (await readUnion(["did:ng:o:p"]))[0]!;
|
||||
expect(s.subject).toBe("did:ng:o:p");
|
||||
expect(s.props[`${FP}event`]).toEqual(["did:ng:o:e"]);
|
||||
expect(s.props[`${FP}user`]).toEqual(["urn:festipod:user:x"]);
|
||||
});
|
||||
|
||||
test("readUnion returns [] for an empty doc set (no query)", async () => {
|
||||
const ng = inject({});
|
||||
const subjects = await readUnion([]);
|
||||
expect(subjects).toEqual([]);
|
||||
expect(ng.sparql_query).toHaveBeenCalledTimes(0);
|
||||
});
|
||||
|
||||
test("a doc that fails to open is skipped, not aborting the union", async () => {
|
||||
const ng = fakeNgWith({ "did:ng:o:ok": [[TYPE, `${FP}Event`], [`${FP}title`, "ok"]] });
|
||||
// Make the OPEN (ASK) throw for the bad doc only.
|
||||
const orig = ng.sparql_query;
|
||||
ng.sparql_query = mock(async (sid: string, query: string, base: unknown, anchor: unknown) => {
|
||||
if (query.startsWith("ASK") && anchor === "did:ng:o:bad") throw new Error("RepoNotFound");
|
||||
return orig(sid, query, base, anchor);
|
||||
}) as any;
|
||||
configure({ ng: ng as any, useShape: (() => {}) as any });
|
||||
configureStoreRegistry({
|
||||
getSession: async () => ({ sessionId: "sid-rm", privateStoreId: "priv" }),
|
||||
normalizeUser: (u: string) => u,
|
||||
});
|
||||
|
||||
const subjects = await readUnion(["did:ng:o:ok", "did:ng:o:bad"]);
|
||||
// The bad doc opened-failed but the good one still lists.
|
||||
expect(subjects.map((s) => s.subject)).toEqual(["did:ng:o:ok"]);
|
||||
});
|
||||
Reference in New Issue
Block a user