ng-eventually/packages/polyfill/src/surface/subscribe.ts

/**
 * 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.
 *
 * ── ONE real subscription per document, fanned out here ────────────────────
 * A branch has room for exactly ONE subscriber upstream, and a second `doc_subscribe`
 * does not join it — it EVICTS the first. `branch_subscriptions: HashMap<BranchId,
 * Sender<AppResponse>>` holds one sender per branch, and `create_branch_subscription`
 * inserts into it and closes whatever it displaced
 * (`engine/verifier/src/verifier.rs:create_branch_subscription`), for the document's own
 * branch AND for its Header branch.
 *
 * Verified against the real broker on 2026-08-17: with two `subscribeDoc` calls on one
 * document, a write fired the SECOND callback (`Patch`) and the first stopped firing
 * entirely — after having fired on a write moments earlier. Silently: nothing rejects,
 * nothing logs, the stale unsubscribe still "works".
 *
 * That made every internal subscriber a hazard to every other one. `ensureRepoOpen`
 * holds a bootstrap subscription per document for the whole session, `watchShape`
 * subscribes to the documents of a scope, `inbox.watch` to an inbox — so opening a
 * repo killed the watch on it, and the app-visible result was a view that never
 * re-read and an inbox that never notified. Both were reported as "the layer does not
 * notify of its own actions", which is NOT what happens: a same-session write DOES
 * push (verified the same day — `Patch@69ms` on the writer's own `sparqlUpdate`,
 * `e2e/reactivity-doc-subscribe.ts`). The push arrived; there was no longer anybody
 * on the other end.
 *
 * So this module keeps at most one real subscription per NURI and fans its pushes out
 * to every local listener. A listener that joins after the initial `State` is replayed
 * the remembered one, because that is what its own `doc_subscribe` would have handed
 * it — without the replay, joining a document somebody else already opened would never
 * fire, and `inbox.watch`'s "fires once immediately" would silently stop being true.
 * The real subscription is torn down when the LAST listener leaves.
 */

import { getConfig, getStoreRegistryDeps } from "../shared-wallet/bootstrap";
import { assertMayReach } from "../emulated-verifier/reach";
import { toNuri } from "../model/nuri";
import type { Nuri, NuriLike } from "../model/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.
 */
// @provenance DocChange kind=aligned level=1 ref=engine/net/src/app_protocol.rs:AppResponseV0 — the raw serialized response, left `unknown` rather than modelling every 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/polyfill-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.
 */
// @provenance DocChangeType kind=aligned level=1 ref=engine/net/src/app_protocol.rs:AppResponseV0 — the variant discriminant; a bare string because the enum is open (State | Patch | TabInfo | …)
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: { <Variant>: … } }` object. Inspects the variant
 * proplerly (no `any`-cast to force it) so a `State` push is identifiable.
 */
// @provenance docChangeType kind=aligned level=1 ref=engine/net/src/app_protocol.rs:AppResponseV0 — reads the variant key out of the payload — a convenience over a verified shape, not an upstream call
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<string, unknown>);
  return keys.length > 0 ? keys[0] : undefined;
}

/** An unsubscribe function — idempotent (calling it twice is a no-op). */
// @provenance Unsubscribe kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — upstream resolves to this function; ours is returned synchronously
export type Unsubscribe = () => void;

async function sessionId(): Promise<string | number> {
  return (await getStoreRegistryDeps().getSession()).sessionId;
}

// --- one real subscription per document -----------------------------------

/** What a listener is handed on every push. */
type Listener = (r: DocChange, type: DocChangeType) => void;

/**
 * The single real `doc_subscribe` behind every local listener on one document.
 * See the module header for why there can only be one.
 */
interface DocFanOut {
  /** Every local listener on this document. The last one to leave tears it down. */
  listeners: Set<Listener>;
  /** The platform's unsubscribe, once the async setup resolved. */
  realUnsub: (() => void) | null;
  /** True from the moment setup is kicked off — a later joiner must not kick off a second. */
  establishing: boolean;
  /**
   * The most recent `State` push, replayed to a listener that joins later.
   * Its own `doc_subscribe` would have pushed one; the fan-out owes it the same.
   */
  lastState: { resp: DocChange; type: DocChangeType } | null;
}

const fanOuts = new Map<Nuri, DocFanOut>();

/** Hand one push to one listener, isolating a throwing handler from the others. */
function deliver(nuri: Nuri, listener: Listener, resp: DocChange, type: DocChangeType): void {
  try {
    listener(resp, type);
  } catch (error) {
    console.error("[subscribe] onChange handler threw for", nuri, error);
  }
}

/** Drop a fan-out: forget the remembered state and release the real subscription. */
function releaseFanOut(nuri: Nuri, entry: DocFanOut): void {
  if (fanOuts.get(nuri) === entry) fanOuts.delete(nuri);
  entry.lastState = null;
  entry.establishing = false;
  const unsub = entry.realUnsub;
  entry.realUnsub = null;
  if (!unsub) return;
  try {
    unsub();
  } catch (error) {
    console.error("[subscribe] unsubscribe failed for", nuri, error);
  }
}

/**
 * Open the one real subscription for `entry` and route its pushes to every listener.
 * Errors are isolated to this document (they never reject a shared batch — see
 * {@link subscribeDocs}); a failed setup simply leaves the document silent.
 */
async function establish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unknown }): Promise<void> {
  // No reactive primitive on the injected `ng` (the fake in the unit suite): there is
  // nothing to call, so this document simply never pushes — the same documented no-op
  // `openRepoUnguarded` takes for the same injection, and not a failure to report. A real
  // `@ng-org/web` always exposes it.
  if (typeof ng.doc_subscribe !== "function") return;
  const doc_subscribe = ng.doc_subscribe as (...a: unknown[]) => unknown;
  try {
    const sid = await sessionId();
    const unsub = (await doc_subscribe(nuri, sid, (resp: DocChange): void => {
      // A push that arrives after the last listener left belongs to a torn-down
      // fan-out — the real unsubscribe may not have taken effect yet.
      if (fanOuts.get(nuri) !== entry) return;
      const type = docChangeType(resp);
      // Remember the barrier for whoever joins next; a later `State` replaces it.
      if (type === "State") entry.lastState = { resp, type };
      // A copy: a handler may unsubscribe itself (or another) from inside the push.
      for (const listener of [...entry.listeners]) deliver(nuri, listener, resp, type);
    })) as (() => void) | undefined;
    if (fanOuts.get(nuri) !== entry || entry.listeners.size === 0) {
      // Everyone left (or the fan-out was reset) before setup resolved — cancel now.
      if (typeof unsub === "function") unsub();
      return;
    }
    entry.realUnsub = typeof unsub === "function" ? unsub : null;
  } catch (error) {
    console.error("[subscribe] doc_subscribe failed for", nuri, error);
  }
}

/**
 * Drop every fan-out without notice — what a page that re-`configure()`s does.
 *
 * A subscription belongs to the injected SDK that opened it, so when that is replaced
 * (or removed) its live subscriptions are void: their pushes would come from a verifier
 * nobody is talking to any more. Called by `configure`/`resetConfig`, and by nothing on
 * the reactive path — a listener is never dropped while its SDK is still there.
 */
// @provenance resetDocSubscriptions kind=invention level=none ref=none — per-session bookkeeping reset, like `resetOpenedRepos`; upstream a verifier owns its own subscriptions and nothing resets them from outside
export function resetDocSubscriptions(): void {
  for (const [nuri, entry] of [...fanOuts]) {
    entry.listeners.clear();
    releaseFanOut(nuri, entry);
  }
  fanOuts.clear();
}

/**
 * Re-open every live subscription against the CURRENT session, keeping its listeners.
 *
 * Called when the session id changes under the page (`open-repo.resetOpenedRepos`), and
 * this is the one thing the fan-out owes that a per-caller subscription got for free. A new
 * session is a new verifier with an empty `self.repos`, so the channel behind a fan-out is
 * dead — and its remembered `State` is the old verifier's, which would let a bootstrap open
 * JOIN it, be handed that stale barrier at once, and read a repo the new session has never
 * synced. Zero rows, "synced", no error: the silent staleness this module exists to remove.
 *
 * So the barrier is forgotten and the channel re-opened, rather than the listeners dropped:
 * a `watchShape` that was following a document keeps following it across the change.
 */
// @provenance resubscribeDocs kind=invention level=none ref=none — per-session bookkeeping, like `resetOpenedRepos`; upstream a session's subscriptions die with it and nothing carries them over
export function resubscribeDocs(): void {
  // Un-configured (a torn-down suite): there is no SDK to re-open anything against, and
  // `resetDocSubscriptions` has already emptied this map on that path.
  let ng: { doc_subscribe?: unknown };
  try {
    ng = getConfig().ng as { doc_subscribe?: unknown };
  } catch {
    return;
  }
  for (const [nuri, entry] of [...fanOuts]) {
    if (entry.listeners.size === 0) continue;
    const stale = entry.realUnsub;
    entry.realUnsub = null;
    // The barrier belonged to the session that is gone. Whoever joins next waits for a
    // real one, exactly as they would have on a fresh page.
    entry.lastState = null;
    if (stale) {
      try {
        stale();
      } catch (error) {
        console.error("[subscribe] unsubscribe failed for", nuri, error);
      }
    }
    entry.establishing = true;
    void establish(nuri, entry, ng);
  }
}

/**
 * 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`).
 */
// @provenance subscribeDoc kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — three deliberate deltas: the unsubscribe is returned synchronously, the callback gets a pre-extracted `type`, and N callers of ONE document share ONE upstream subscription because a branch holds a single sender and a second subscribe evicts the first
export function subscribeDoc(
  nuriLike: NuriLike,
  onChange: (r: DocChange, type: DocChangeType) => void,
): Unsubscribe {
  const nuri = toNuri(nuriLike, "subscribeDoc");
  // RULE 1 — a subscription IS an access: the push carries the document's state.
  // Guarding the read paths while leaving this open would be a door beside the gate.
  assertMayReach(nuri, "subscribeDoc");
  return subscribeDocUnguarded(nuri, onChange);
}

/**
 * The unguarded core. Exported for ONE importer — `shared-wallet/physical.ts`, which
 * owns the machinery's entire privileged door — and for nobody else. It is not
 * re-exported by either entry point; the `Unguarded` suffix is the warning.
 */
// @provenance subscribeDocUnguarded kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — the same call without the reach guard, and the place the one-subscription-per-document fan-out is kept; internal, never published
export function subscribeDocUnguarded(
  nuri: Nuri,
  onChange: (r: DocChange, type: DocChangeType) => void,
): Unsubscribe {
  // Resolved here, synchronously, so calling this before `configure()` still throws at
  // the call rather than inside a background task nobody awaits.
  const { ng } = getConfig();

  let entry = fanOuts.get(nuri);
  if (!entry) {
    entry = { listeners: new Set(), realUnsub: null, establishing: false, lastState: null };
    fanOuts.set(nuri, entry);
  }
  const joined = entry;
  joined.listeners.add(onChange);

  // Joining a document somebody else already opened: hand this listener the `State` its
  // own `doc_subscribe` would have pushed it. Asynchronously, like the real push, so a
  // caller that sets itself up after this call still sees it.
  if (joined.lastState) {
    const { resp, type } = joined.lastState;
    queueMicrotask(() => {
      if (fanOuts.get(nuri) === joined && joined.listeners.has(onChange)) {
        deliver(nuri, onChange, resp, type);
      }
    });
  }

  if (!joined.establishing) {
    joined.establishing = true;
    void establish(nuri, joined, ng as { doc_subscribe?: unknown });
  }

  let stopped = false;
  return () => {
    if (stopped) return;
    stopped = true;
    joined.listeners.delete(onChange);
    // The LAST one out releases the real subscription — while anybody is still
    // listening, tearing it down would silence them (and re-opening it later is not
    // free: a second `doc_subscribe` evicts whoever else has the branch by then).
    if (joined.listeners.size === 0) releaseFanOut(nuri, joined);
  };
}

/**
 * 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.
 */
// @provenance subscribeDocs kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — client-side composition with per-doc error isolation; `orm_start_graph` is deliberately unused — it aborts wholesale on one RepoNotFound
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);
      }
    }
  };
}