/** * Inbox — a generic deposit + read/materialize mechanism the consumer reuses for * its own purposes (same `inbox.post` API, same watcher — see the discovery-model * decision). The mechanism itself knows no application domain: the consumer * supplies the inbox document NURI and interprets the `payload`. (An example * consumer mapping, purely illustrative: a consumer might use one inbox for a * registration deposit and another for submitting a reference to an index.) * * ── Real target vs this emulation ───────────────────────────────────────── * In real NextGraph, a message is sealed to the recipient's key and queued into * their inbox; the recipient's own verifier unseals each queued message and * applies it inline as it processes the inbox — there is no separate curator * process. A future `inbox_post_link` is the intended way to seal a link into an * inbox from the sender side; it is not exposed yet. * * Here, on one shared wallet where everything is readable, both sides run in-lib: * - `post` appends a deposit `{ from, payload, ts }` as RDF into the inbox * document (in the shared wallet) via the `docs.sparqlUpdate` primitive; * - `read` / `watch` read the deposits back via `docs.sparqlQuery` and expose * them. This in-lib read stands in for the recipient's own inbox processing * until the sealed-inbox path (`inbox_post_link`) is available. * * All NextGraph I/O routes through the `docs` primitives (the real injected `ng`, * never `makeNg`), so this module imports no `@ng-org` package. */ import { sparqlUpdate, sparqlQuery } from "./docs"; import { subscribeDoc } from "./subscribe"; import { getCurrentUser, getStoreRegistryDeps } from "./polyfill"; import { escapeLiteral } from "./sparql"; import type { Nuri, PrincipalId } from "./types"; // --- deposit model -------------------------------------------------------- /** One deposit as materialized from an inbox document. */ export interface Deposit { /** The sender, if identified; `null` when the deposit was anonymous. */ from: PrincipalId | null; /** The consumer-defined payload (opaque here — JSON-serialized in storage). */ payload: unknown; /** Deposit timestamp (ms epoch). Caller may pass one for determinism. */ ts: number; } /** Options for {@link post}. `from` and `ts` are both optional. */ export interface PostOptions { /** * Who is depositing. Omit (or pass `null`) for an ANONYMOUS deposit; pass a * principal id to identify the sender. Defaults to the current polyfill user * ({@link getCurrentUser}) when the property is entirely absent, so callers * that want anonymity must pass `from: null` explicitly. */ from?: PrincipalId | null; /** The payload to deposit (interpreted only by the consumer). */ payload: unknown; /** Optional deposit timestamp (ms epoch). Omitted → `Date.now()`. Passing it * keeps tests deterministic. */ ts?: number; } const SHIM = "urn:ng-eventually:inbox"; const P = { type: `${SHIM}:Deposit`, from: `${SHIM}:from`, payload: `${SHIM}:payload`, ts: `${SHIM}:ts`, } as const; // --- session access (shared with the storeRegistry) ----------------------- /** The inbox documents live in the shared wallet, so we reuse the registry's * injected session provider for the sessionId. Disappears at migration. */ async function sessionId(): Promise { return (await getStoreRegistryDeps().getSession()).sessionId; } // --- SPARQL result helpers ------------------------------------------------ /** Tolerant extraction of SPARQL SELECT bindings across possible shapes. */ function readBindings(result: unknown): Array> { if (!result) return []; if (Array.isArray(result)) return result as Array>; const anyRes = result as { results?: { bindings?: Array> }; }; return anyRes.results?.bindings ?? []; } // --- deposit (client side) ------------------------------------------------ /** * Deposit a payload into `targetInbox`. * * Appends `{ from, payload, ts }` into the inbox document via `docs.sparqlUpdate` * (the real injected `ng`). Each deposit is a fresh RDF subject in the inbox * graph, so concurrent deposits don't collide. * * `from` is bound to the current identity — it is authenticated, not * caller-supplied. Omit it to stamp the current identity; pass `null` to deposit * anonymously (a legitimate choice — identified if known, anonymous otherwise). * A `from` naming another identity is rejected as a spoof: in the target the * broker seals the sender from the wallet's own key, so a client cannot forge * another's identity. This check is redundant once the seal enforces it, but * until then it closes the spoof the shared wallet would otherwise allow. */ export async function post(targetInbox: Nuri, opts: PostOptions): Promise { const current = getCurrentUser(); let from: PrincipalId | null; if (opts.from === undefined) { from = current; // default: stamp the current identity } else if (opts.from === null) { from = null; // explicit anonymous deposit } else if (opts.from === current) { from = opts.from; // identifying as self — allowed } else { throw new Error( "[ng-eventually] inbox.post: `from` must be the current identity or null " + "(anonymous) — depositing as another principal is a spoof.", ); } const ts = opts.ts ?? Date.now(); const sid = await sessionId(); // A unique subject per deposit (in the inbox graph) — no collisions. const subject = `${SHIM}:deposit:${ts}:${Math.random().toString(36).slice(2)}`; const payloadLiteral = escapeLiteral(JSON.stringify(opts.payload ?? null)); const fromTriple = from == null ? "" : ` ;\n <${P.from}> "${escapeLiteral(from)}"`; // NO explicit `GRAPH <…>` wrapper — write the anchored DEFAULT graph: // `sparqlUpdate(sid, update, targetInbox)` scopes the write to that repo's // default graph (same shape as read-model.ts readDoc/readUnion). This is the // CANONICAL, always-safe shape and the one the anchored default-graph read // queries. (Not a round-trip necessity on the current broker: the e2e harness // `packages/client/e2e/` verified that an anchored `GRAPH ` write // ALSO round-trips here — it resolves to the same repo graph, no phantom graph. // The no-GRAPH form is kept as a simplicity/safety convention; re-verify with // that harness if the broker version changes.) const update = ` INSERT DATA { <${subject}> a <${P.type}> ; <${P.payload}> "${payloadLiteral}" ; <${P.ts}> "${ts}"${fromTriple} . }`; await sparqlUpdate(sid, update, targetInbox); } // --- read -------------------------------------------------------------- /** * Read every deposit currently in `targetInbox`, sorted by `ts` ascending. In * real NextGraph the recipient's own verifier applies queued messages inline as * it processes the inbox; here this read stands in for that until the * sealed-inbox path is available. The consumer interprets each deposit's * `payload`. */ export async function read(targetInbox: Nuri): Promise { const sid = await sessionId(); // NO explicit `GRAPH <…>` clause — read the anchored DEFAULT graph (see the // note in `post`). The anchor (`targetInbox`) scopes the query to that repo's // default graph, exactly where `post` writes. const query = ` SELECT ?payload ?ts ?from WHERE { ?d a <${P.type}> ; <${P.payload}> ?payload ; <${P.ts}> ?ts . OPTIONAL { ?d <${P.from}> ?from } }`; const result = await sparqlQuery(sid, query, undefined, targetInbox); const deposits: Deposit[] = []; for (const row of readBindings(result)) { const rawPayload = row.payload?.value ?? "null"; let payload: unknown; try { payload = JSON.parse(rawPayload); } catch { payload = rawPayload; // tolerate a non-JSON literal } const tsRaw = row.ts?.value ?? "0"; const ts = Number.parseInt(tsRaw, 10) || 0; const fromValue = row.from?.value; deposits.push({ from: fromValue ? fromValue : null, payload, ts }); } deposits.sort((a, b) => a.ts - b.ts); return deposits; } /** Alias for {@link read} — the name that reads as "process the inbox now". */ export const materialize = read; /** * Subscription over an inbox — **event-driven, not polled**. Subscribes to the * inbox document via {@link subscribeDoc} (the platform's `doc_subscribe` push): * `onDeposits` fires once on the initial state push and again on every subsequent * change to the inbox document — a local deposit OR a broker-synced remote one. * Returns an unsubscribe function. * * On each push it re-reads the full deposit list ({@link read}) and invokes * `onDeposits` only when the deposit count changed (grew), keeping the same * "fires on change" contract the polling watcher had — same callback signature * and same behaviour, just event-driven instead of `setInterval`. * * The `intervalMs` option is accepted for signature compatibility but IGNORED: * there is no polling. (The inbox document is a single doc, so this is immune to * the ORM fan-out hang — see {@link subscribeDoc}.) */ export function watch( targetInbox: Nuri, onDeposits: (deposits: Deposit[]) => void, _opts?: { intervalMs?: number }, ): () => void { let stopped = false; let lastCount = -1; // Re-read on every push; fire onDeposits only when the set changed (grew). const refresh = async (): Promise => { if (stopped) return; try { const deposits = await read(targetInbox); if (!stopped && deposits.length !== lastCount) { lastCount = deposits.length; onDeposits(deposits); } } catch (error) { console.error("[inbox] watch read failed:", error); } }; // Subscribe to the inbox document: the initial State push fires the first read // (so onDeposits fires once immediately, as before), each later Patch a re-read. const unsubscribe = subscribeDoc(targetInbox, () => void refresh()); return () => { stopped = true; unsubscribe(); }; }