ng-eventually/packages/client/src/read-model.ts

/**
 * read-model — the listing primitive of the polyfill: read a bounded, by-need set
 * of documents, each with its own anchored `sparql_query`, and return the triples
 * grouped per subject. This is the mechanism documented in docs/read-model.md.
 *
 * ── Why per-doc anchored, rather than an anchorless union-scan ─────────────
 * An anchored `sparql_query(sid, "SELECT ?s ?p ?o WHERE { ?s ?p ?o }", base, doc)`
 * is restricted to the anchor repo's graph: `resolve_target_for_sparql(Repo)` →
 * `Some(repo_graph_name)`, which becomes the query's default graph. A body with no
 * `GRAPH` wrapper reads only that default graph → only that doc's triples, O(1) per
 * doc, independent of how many other graphs the local store holds.
 *
 * The footgun this avoids: an anchorless query (`anchor` undefined → `UserSite` →
 * `set_default_graph_as_union`) spans EVERY named graph currently in the session
 * store. On a shared / bloated wallet that accumulates across runs, that is
 * O(wallet size) → the observed ~90s timeouts. So the read path never union-scans
 * all graphs — it reads exactly the bounded by-need set, one anchored query per doc.
 *
 * NB (verified, docs/read-model.md § probe step 4): an explicit `GRAPH ?g { … }`
 * body iterates the named graphs regardless of the default graph, so an anchor does
 * not bound such a body. The per-doc read therefore uses a default-graph body (no
 * `GRAPH` wrapper) so the anchor's one-repo restriction actually applies.
 *
 * ── 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
 * is instead a set of one-shot anchored `sparql_query`s. There is no reactive
 * union query, so reactivity is assembled by re-querying on a change signal.
 *
 * ── Generic by construction ───────────────────────────────────────────────
 * No application domain here: the consumer passes the doc NURIs to read (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 the real multi-store migration the per-doc anchored read is unchanged (native
 * SPARQL, anchored to one repo); only bringing a repo into the session (open by cap)
 * changes — the anchored query already resolves a same-session repo directly.
 */

import { docCreate, sparqlUpdate, sparqlQuery } from "./docs";
import { getCaps, getCurrentUser, 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 a doc, with its properties (predicate → values). */
export interface UnionSubject {
  /** The subject IRI (`?s`) — in the polyfill, the doc's own NURI. */
  subject: string;
  /** The graph (doc NURI) the subject was read from. */
  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;
}

/**
 * Read one doc with an anchored default-graph query, tolerant per-doc.
 *
 * The anchor (`doc` NURI) restricts the query to that repo's graph as the default
 * graph (`resolve_target_for_sparql(Repo)` → `Some(repo_graph_name)`); a body with
 * no `GRAPH` wrapper reads exactly that default graph → only this doc's triples.
 * This is O(1) in the doc's own size and independent of the rest of the (possibly
 * bloated / shared) session store — it never iterates other graphs.
 *
 * All same-session repos (every doc `doc_create`d this session — in the mono-wallet
 * polyfill, all of them) are already in `self.repos`, so the anchored query resolves
 * the repo directly with no separate open. A genuinely-absent repo throws
 * `RepoNotFound` here, in isolation, and the doc is skipped — never aborting the
 * others (unlike the ORM fan-out). Returns the doc's rows, or `[]` on failure.
 *
 * At the real multi-store migration this becomes a real sync: opening a per-user
 * store repo by cap is a native broker fetch (`verifier.rs:1423` `OpenRepo` TODO).
 */
async function readDoc(
  sid: string,
  doc: Nuri,
): Promise<Array<Record<string, { value: string } | undefined>>> {
  try {
    const nuri = assertNuri(doc);
    // Anchored to `nuri` → default graph = this repo. No `GRAPH ?g` wrapper, so
    // the anchor's one-repo restriction applies (an explicit `GRAPH ?g` body would
    // iterate all named graphs regardless of the anchor — see docs § probe step 4).
    const res = await sparqlQuery(
      sid,
      "SELECT ?s ?p ?o WHERE { ?s ?p ?o }",
      undefined,
      nuri,
    );
    return bindings(res);
  } catch (error) {
    console.error("[read-model] read failed for", doc, error);
    return [];
  }
}

/**
 * Read a BOUNDED, by-need set of docs — each with its OWN anchored query — and
 * return the triples grouped per subject. `docs` are the NURIs to read (the
 * consumer resolves them by need — index for public, own scope docs for mine).
 * Docs that fail are skipped (see {@link readDoc}); a failing doc never aborts the
 * batch.
 *
 * Never an anchorless union-scan over all graphs (which is O(wallet size) and wrong
 * on a shared / bloated wallet — the footgun this path exists to avoid). Each doc is
 * read with an anchored default-graph query, O(1) per doc, independent of wallet
 * size — a non-empty wallet no longer matters. Reads run in parallel via `Promise.all`.
 */
export async function readUnion(docs: Nuri[]): Promise<UnionSubject[]> {
  const sid = await sessionId();
  const unique = [...new Set(docs.filter(Boolean))];
  if (unique.length === 0) return [];

  // One anchored query per doc, in parallel, tolerant (a bad doc yields []).
  const perDoc = await Promise.all(
    unique.map(async (d) => ({ doc: assertNuri(d), rows: await readDoc(sid, d) })),
  );

  // Cap gate (defence-in-depth). A doc whose read policy the current user may not
  // satisfy is dropped. Isolation holds both by construction (the app only resolves
  // docs it is entitled to) and by filter here. Generic: the lib owns the cap
  // registry; a doc under no policy (`!governsRead`) flows through unchanged. In this
  // polyfill each subject IRI is its own document NURI, so the cap key is the doc NURI.
  const caps = getCaps();
  const user = getCurrentUser();

  const bySubject = new Map<string, UnionSubject>();
  for (const { doc, rows } of perDoc) {
    if (caps.governsRead(doc) && !caps.canRead(doc, user)) continue;
    // Anchored to `doc`, so every row belongs to `doc`; the subject is the doc NURI
    // (writeEntity invariant). Pin subject/graph to the doc NURI (the anchor), which
    // is stable regardless of the repo_graph_name overlay suffix the store carries.
    for (const row of rows) {
      const p = row.p?.value;
      const o = row.o?.value;
      if (!p || o === undefined) continue;
      let entry = bySubject.get(doc);
      if (!entry) {
        entry = { subject: doc, graph: doc, props: {} };
        bySubject.set(doc, entry);
      }
      (entry.props[p] ??= []).push(o);
    }
  }
  return [...bySubject.values()];
}