/** * Reactive single-document subscription — the polyfill's typed wrapper over the * platform's `doc_subscribe` primitive. This is the canonical NextGraph reactive * read at the document granularity: subscribe once, get the initial state pushed, * then a push on every subsequent commit to that document — whether the write was * local (this session) or a broker-synced remote change. NO POLLING. * * ── Why call the REAL injected `ng` directly (never `makeNg`) ────────────── * Same hard constraint as `docs.ts`: the public `ng` is a JS `Proxy` over * `@ng-org/web`'s iframe-RPC proxy. `doc_subscribe` is a STREAMED method — the * `@ng-org/web` RPC strips the callback (by positional index) BEFORE it posts to * the iframe and drives it locally via a `MessageChannel` port (the function is * never structured-cloned, so no `DataCloneError`). Layering our own Proxy on top * risks re-wrapping that surface; reaching the real `ng` held in the config avoids * the double-proxy exactly as the raw `docs` primitives do. Do not import from * `./ng-proxy`. * * ── The primitive shape (verified against nextgraph-rs) ──────────────────── * `ng.doc_subscribe(repo_o: string, session_id, callback)` * (`sdk/js/lib-wasm/src/lib.rs:1907`) is **per-document** — one repo NURI, one * callback. It is `async`, resolving to a JS **unsubscribe function**. The * callback is invoked `callback(appResponse)` with a serialized `AppResponse`: * `{ V0: { State | Patch | TabInfo | ... } }`. It pushes an initial `State` * (plus a `TabInfo`) on subscribe, then a `Patch` per verified commit on the * branch. Returning `true` from the callback also cancels; we cancel by calling * the returned unsubscribe fn. * * ── Why per-document, never `orm_start_graph(graphs:[…])` ────────────────── * A single not-yet-synced repo in an ORM graph fan-out makes `RepoNotFound` abort * the WHOLE subscription (`initialize.rs:125-128`), so the readyPromise never * resolves → the ~75s hang. `doc_subscribe` is per-branch/per-doc and has no * fan-out: an absent doc breaks only its own subscription. {@link subscribeDocs} * builds a set of these with per-doc error isolation to preserve that property. */ import { getConfig, getStoreRegistryDeps } from "./polyfill"; import type { Nuri } from "./types"; /** * A push from the platform to a document subscriber. Loosely typed: the raw * serialized `AppResponse` (`{ V0: { State | Patch | TabInfo | ... } }`). The * consumer typically ignores the payload and uses the push purely as a * change SIGNAL (re-query on change — the read-model pattern), so this stays * permissive rather than modelling every AppResponse variant. */ export type DocChange = unknown; /** * The discriminant of a {@link DocChange} — the single variant key of the raw * `AppResponse` payload (`{ V0: { State | Patch | TabInfo | … } }`). It is NOT a * closed enum: the platform may push other variants, so this is a bare `string` * (e.g. `"State"`, `"Patch"`, `"TabInfo"`), or `undefined` when the shape can't * be read. Verified against the CONTRACT-3 e2e probe (`e2e/sdk-entry.ts`): the * variant is `Object.keys(resp.V0)[0]`. Exposed so a caller that needs the SYNC * BARRIER (the first `State`, per CONTRACT 3) can distinguish it from the earlier * `TabInfo`/`Patch` pushes — see `open-repo.ts`. Most callers ignore it and use * any push as a plain change signal. */ export type DocChangeType = string | undefined; /** * Extract the variant key from a raw {@link DocChange}. Reads `resp.V0` (case- * tolerant to `v0`) and returns its first key — the AppResponse variant name * (`"State"` / `"Patch"` / `"TabInfo"` / …). Returns `undefined` if the payload * is not a recognisable `{ V0: { : … } }` object. Inspects the variant * proplerly (no `any`-cast to force it) so a `State` push is identifiable. */ export function docChangeType(resp: DocChange): DocChangeType { if (!resp || typeof resp !== "object") return undefined; const outer = resp as { V0?: unknown; v0?: unknown }; const v0 = outer.V0 ?? outer.v0; if (!v0 || typeof v0 !== "object") return undefined; const keys = Object.keys(v0 as Record); return keys.length > 0 ? keys[0] : undefined; } /** An unsubscribe function — idempotent (calling it twice is a no-op). */ export type Unsubscribe = () => void; async function sessionId(): Promise { return (await getStoreRegistryDeps().getSession()).sessionId; } /** * Subscribe to ONE document. `onChange` fires on the initial state push and on * every subsequent change to that doc (local write OR broker-synced remote * change). Returns an unsubscribe function. * * The wrapper is synchronous-returning (an unsubscribe fn) even though the * underlying `ng.doc_subscribe` is async: the real unsubscribe is captured when * the promise resolves; if the caller unsubscribes before setup completes, the * cancellation is honoured as soon as the real unsubscribe is available (and no * further `onChange` fires after unsubscribe). * * `onChange` receives the raw payload AND its variant type ({@link docChangeType}, * e.g. `"State"`). The type is a NON-BREAKING second argument: existing callers * that ignore it (the change-signal pattern — `discovery.ts`, `inbox.ts`) are * unaffected; a caller that needs the sync barrier (`open-repo.ts`) reads it to * act only on the first `State`. * * Calls the REAL injected `ng.doc_subscribe` directly (never `makeNg`). */ export function subscribeDoc( nuri: Nuri, onChange: (r: DocChange, type: DocChangeType) => void, ): Unsubscribe { const { ng } = getConfig(); let stopped = false; let realUnsub: (() => void) | null = null; const cb = (resp: DocChange): void => { if (stopped) return; try { onChange(resp, docChangeType(resp)); } catch (error) { console.error("[subscribe] onChange handler threw for", nuri, error); } }; // Kick off the async subscription. Errors are isolated to this doc (they never // reject a shared batch — see subscribeDocs). If setup fails, this doc simply // never fires; the caller's unsubscribe stays a safe no-op. void (async () => { try { const sid = await sessionId(); const unsub = (await ng.doc_subscribe(nuri, sid, cb)) as (() => void) | undefined; if (stopped) { // Unsubscribed before setup resolved — cancel immediately. if (typeof unsub === "function") unsub(); return; } realUnsub = typeof unsub === "function" ? unsub : null; } catch (error) { console.error("[subscribe] doc_subscribe failed for", nuri, error); } })(); return () => { if (stopped) return; stopped = true; if (realUnsub) { try { realUnsub(); } catch (error) { console.error("[subscribe] unsubscribe failed for", nuri, error); } realUnsub = null; } }; } /** * Subscribe to a SET of documents, one {@link subscribeDoc} per NURI, with * PER-DOC error isolation. `onChange(nuri, r)` fires for whichever doc changed. * Returns a single unsubscribe that tears down all of them. * * The per-doc isolation is the point: a bad / not-yet-synced doc breaks only its * own subscription and NEVER aborts the others (this is precisely what avoids the * ORM fan-out hang — do NOT replace this with `orm_start_graph(graphs:[…])`). The * set is deduplicated; an empty set returns a no-op unsubscribe. */ export function subscribeDocs( nuris: Nuri[], onChange: (nuri: Nuri, r: DocChange, type: DocChangeType) => void, ): Unsubscribe { const unique = [...new Set(nuris.filter(Boolean))]; const unsubs = unique.map((nuri) => { // Each subscription is independent: subscribeDoc already isolates its own // async setup failure (logged, never thrown), so one bad doc cannot abort the // construction of the others here. try { return subscribeDoc(nuri, (r, type) => onChange(nuri, r, type)); } catch (error) { console.error("[subscribe] subscribeDocs: failed to subscribe", nuri, error); return () => {}; } }); return () => { for (const u of unsubs) { try { u(); } catch (error) { console.error("[subscribe] subscribeDocs: unsubscribe failed", error); } } }; }