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, the reactive shape
 * read of the day subscribed to the documents of a scope (a pipeline retired in
 * `1.0.0-dev.6`), `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.
 *
 * ── What SHARING one subscription must not cost a caller ───────────────────
 * Three things a per-caller `doc_subscribe` gave for free, and that the fan-out has to give
 * back deliberately — each of them was lost when it was introduced on 2026-08-17:
 *
 *   - a caller is ITSELF, not its handler. One record per CALL, so passing the same function
 *     twice is two subscriptions and the first unsubscribe does not silence the second;
 *   - a caller that has LEFT hears nothing more, including from the push during which it
 *     left — a handler may tear another one down, and the pushes are re-checked against the
 *     live set rather than a copy taken before the first handler ran;
 *   - a setup that FAILED is not the end of the document. The attempt is forgotten so the
 *     next joiner makes its own (upstream, each caller's own `doc_subscribe` retried), and
 *     it is reported rather than left as silence — see {@link reportSetupFailure}.
 */

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;

/**
 * ONE call to {@link subscribeDocUnguarded}, and what that caller asked for.
 *
 * Identified by this record and never by the `onChange` function: two callers may
 * legitimately pass the SAME function — a module-level handler, a bound method, an arrow
 * that closes over nothing — and they are two subscriptions with two independent lifetimes.
 * Keyed by the function, the second call was swallowed by the set and the FIRST caller's
 * unsubscribe silenced the second, which had never asked to leave.
 */
interface DocListener {
  /** What this caller is handed on every push. */
  onChange: Listener;
  /**
   * Told when the one real `doc_subscribe` behind this listener could not be OPENED.
   *
   * Upstream `doc_subscribe` is async and REJECTS on a setup failure, so its caller learns.
   * This wrapper returns synchronously, so without this channel a caller cannot tell "this
   * document is quiet" from "this document is not subscribed at all" — the failure-as-absence
   * this package keeps closing. `null` for a caller that did not ask (the published
   * {@link subscribeDoc}, whose contract has no such argument); the failure is logged either
   * way.
   */
  onSetupFailed: ((error: unknown) => void) | null;
}

/**
 * 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<DocListener>;
  /** The platform's unsubscribe, once the async setup resolved. */
  realUnsub: (() => void) | null;
  /**
   * True from the moment a setup is kicked off — a later joiner must not kick off a second,
   * because a second `doc_subscribe` on a branch EVICTS the first (see the module header).
   * It therefore stays true once the setup has SUCCEEDED: the attempt still stands.
   *
   * Cleared on the two ways the attempt stops standing: the last listener leaving
   * ({@link releaseFanOut}), and the CURRENT setup FAILING ({@link reportSetupFailure}). Left
   * true on failure it stopped meaning "one is already running" and started meaning "this NURI
   * is finished" — no later joiner ever attempted it again, for the whole session, over one
   * transient rejection.
   */
  establishing: boolean;
  /**
   * WHICH establish this fan-out is waiting on — bumped by {@link beginEstablish} every time
   * one is kicked off, and captured by that call for the whole of its life.
   *
   * Two establishes can be in flight over the SAME entry: {@link resubscribeDocs} re-opens the
   * channel of a fan-out whose first attempt has not settled yet, and it re-opens it on the
   * entry rather than on a fresh one, precisely so the listeners are kept. So `entry` identity
   * — the only currency this module had — answers "is this fan-out still the one for this
   * NURI", and cannot answer "is this attempt still the one this fan-out is waiting on". A
   * counter can, and that second question is the one a LATE outcome has to ask: the first
   * attempt rejecting long after the second SUCCEEDED is not this document failing, it is one
   * superseded call finally answering.
   */
  attempt: number;
  /**
   * 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;
}

/**
 * Every document with at least one local listener.
 *
 * An entry in this map ALWAYS has listeners — that is the invariant the rest of the module
 * reads "is this fan-out still standing" off: a listener is added synchronously with the
 * entry that holds it ({@link subscribeDocUnguarded}), and the last one out takes the entry
 * with it ({@link releaseFanOut}, {@link resetDocSubscriptions}). So no caller asks after the
 * listener count separately; `fanOuts.get(nuri) === entry` already answers it.
 */
const fanOuts = new Map<Nuri, DocFanOut>();

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

/**
 * Hand one push to every listener of `entry` — the fan-out itself.
 *
 * Over a COPY, because a handler may unsubscribe itself or another from inside the push; and
 * re-checking each listener against the live set, because a copy alone only stops the walk
 * from breaking — it still delivers to whoever left DURING it. That contradicted what
 * {@link subscribeDoc} publishes ("no further `onChange` fires after unsubscribe"), on the
 * one ordering an application cannot control: whether its handler runs before or after the
 * one that tore it down.
 */
function fanOut(nuri: Nuri, entry: DocFanOut, resp: DocChange, type: DocChangeType): void {
  for (const listener of [...entry.listeners]) {
    if (!entry.listeners.has(listener)) continue;
    deliver(nuri, listener, resp, type);
  }
}

/**
 * Is `attempt` still the establish this fan-out is waiting on?
 *
 * Two independent ways to stop being it, and each is the only one that catches its own case:
 * the fan-out was RELEASED (last listener left, or `resetDocSubscriptions`), which drops it
 * from the map and leaves the counter untouched — so only the map half sees it; or it was
 * SUPERSEDED in place by {@link resubscribeDocs}, which keeps the very same entry in the map
 * and bumps the counter — so only the counter half sees it.
 */
function isCurrentAttempt(nuri: Nuri, entry: DocFanOut, attempt: number): boolean {
  return fanOuts.get(nuri) === entry && entry.attempt === attempt;
}

/**
 * Kick off the one real subscription for `entry` — the ONLY way an establish is started.
 *
 * It stamps the attempt (see {@link DocFanOut.attempt}) as it starts it, so an outcome that
 * arrives late can tell whether it is still the one being waited on. Being the only door is
 * what makes that true: a second call site that forgot to bump the counter would hand its
 * establish the number of the one it just superseded.
 */
function beginEstablish(nuri: Nuri, entry: DocFanOut, ng: { doc_subscribe?: unknown }): void {
  entry.establishing = true;
  entry.attempt += 1;
  void establish(nuri, entry, ng, entry.attempt);
}

/**
 * The one real subscription could not be OPENED: forget the attempt, and say so.
 *
 * Two halves, and the shape of every "failure disguised as an absence" this package has
 * closed. **Retryable** — `establishing` goes back to false, so the next `subscribeDoc` on
 * this NURI attempts it again instead of joining a fan-out that will never push. Left true,
 * one transient rejection made the document silent for every later joiner in the session.
 * **Visible** — the log always, plus the callers who asked to be told, so a caller whose job
 * depends on the subscription (the inbox observation) can report it and try again rather
 * than sit on a dead entry.
 *
 * ── Only for the attempt still being WAITED ON ─────────────────────────────
 * Both halves act on the fan-out as it stands NOW, so a SUPERSEDED attempt must do neither.
 * A rejection that arrives after {@link resubscribeDocs} has already re-opened the channel is
 * not this document failing — the channel is open and pushing. Told anyway, the inbox
 * observation dropped its entry and released it, which closed the WORKING subscription,
 * printed "this inbox could not be watched" about an inbox that was, and left the identity
 * with a silent queue. Clearing `establishing` was the same mistake one level down: it says
 * "nothing is running" while the second attempt is, so the next joiner opens a third — and a
 * second `doc_subscribe` on a branch EVICTS the one before it.
 *
 * Superseded, it is therefore LOGGED and nothing else: the call that replaced it owns the
 * outcome, and reports its own failure if it has one.
 */
function reportSetupFailure(nuri: Nuri, entry: DocFanOut, error: unknown, attempt: number): void {
  console.error("[subscribe] doc_subscribe failed for", nuri, error);
  if (!isCurrentAttempt(nuri, entry, attempt)) return;
  entry.establishing = false;
  for (const listener of [...entry.listeners]) {
    const tell = listener.onSetupFailed;
    if (tell === null || !entry.listeners.has(listener)) continue;
    try {
      tell(error);
    } catch (thrown) {
      console.error("[subscribe] onSetupFailed handler threw for", nuri, thrown);
    }
  }
}

/** 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.
 *
 * ── Everything here is gated on the ATTEMPT, never on the entry alone ──────
 * A superseded establish must not deliver, must not re-seed the barrier, and must not
 * install its unsubscribe: {@link resubscribeDocs} re-opens the channel ON THE SAME entry,
 * so entry identity says "this fan-out still stands" and says nothing about which call it is
 * waiting on. Gated on identity only, a first attempt that SUCCEEDS after a second one has
 * replaced it fans the PREVIOUS session's `State` out to every listener, re-seeds
 * `lastState` with it — the exact stale barrier {@link resubscribeDocs} cleared it to
 * prevent — and overwrites `realUnsub`, so the eventual teardown releases the dead channel
 * and leaves the live one subscribed upstream.
 *
 * ── A superseded attempt is RELEASED, not dropped ──────────────────────────
 * Dropping its unsubscribe would leak a live subscription. Upstream's `CancelFn` closes only
 * its OWN `tx` and never touches `branch_subscriptions` (`engine/verifier/src/verifier.rs:479-484`),
 * so releasing a loser cannot silence the winner — and cross-session there IS no winner to
 * silence: `resubscribeDocs` runs on a session change, the superseded call sits on the
 * previous session's verifier, and nothing else will ever close it.
 *
 * ── Why the guard sits BEFORE the call, and what it does NOT settle ────────
 * What it settles is a fact: superseded while the session id was still resolving, this opens
 * NO channel at all. That is the wider half of the window, and it closes it outright.
 *
 * What it leaves open is a BET, and it is written down here as one. For an attempt that DID
 * place its call before being superseded, this code releases the channel that call opened —
 * which is only ever releasing the upstream LOSER if the broker SERVED the two calls in the
 * order they were ISSUED. Issuing is all this module can order; service is the broker's, and
 * nothing found upstream promises the two agree:
 *
 *   - `LOCAL_BROKER` is behind an `async_std::sync::RwLock` held for the whole call
 *     (`sdk/rust/src/local_broker.rs:3057`), and that type IS `async_lock::RwLock`
 *     (async-std 1.13.2 `src/sync/mod.rs:181`, a re-export);
 *   - `RwLock::write()` takes the lock's internal `async_lock::Mutex` first (async-lock 3.4.1
 *     `src/rwlock/raw.rs:163-168`), and that mutex is documented as "eventual fairness" —
 *     fair ON AVERAGE, not FIFO (`src/mutex.rs:22-24`). Its hot loop re-runs a bare
 *     `compare_exchange` on every poll, so a call arriving later can take the lock ahead of
 *     one already waiting;
 *   - and the anti-starvation fallback that would eventually force fairness is
 *     `#[cfg(all(feature = "std", not(target_family = "wasm")))]` (`src/mutex.rs:578-581`) —
 *     compiled out on wasm, which is where this SDK runs.
 *
 * So the lock SERIALISES the calls; it does not ORDER them. Two attempts on the same branch
 * of the same verifier is a reachable state — the last listener leaving while an establish is
 * in flight releases the fan-out, and the next joiner opens a fresh one against the same
 * session — so this is not a cross-session-only concern.
 *
 * What breaks the bet is therefore a service order inverted relative to the issue order, and
 * this is what it costs when it happens: the superseded call inserts LAST, so it wins the
 * branch and closes the current attempt's sender (`verifier.rs:361`); this code then releases
 * its own channel, leaving the branch holding a closed sender that `push_app_response` drops
 * at the first push (`verifier.rs:258-261`). Nothing rejects and nothing logs. The fan-out
 * holds a `realUnsub` and believes itself subscribed, `establishing` stays true so no later
 * joiner re-opens it, and the document is silently dead for the rest of the session —
 * `test/subscribe.test.ts` drives exactly that and pins it.
 *
 * Releasing is still what to do: DROPPING the superseded unsubscribe instead leaks a live
 * subscription in the ordinary case, and buys nothing in the inverted one — the winner's own
 * callback is gated on an attempt that is no longer current, so it delivers to nobody either
 * way. The bet is on which failure mode is reachable, never on releasing being free.
 */
async function establish(
  nuri: Nuri,
  entry: DocFanOut,
  ng: { doc_subscribe?: unknown },
  attempt: number,
): 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();
    // Superseded (or torn down) while the session id was resolving: open NOTHING. Cheaper
    // than opening a channel to close it, and it is what keeps a superseded attempt from
    // ever reaching the broker AFTER the attempt that replaced it.
    if (!isCurrentAttempt(nuri, entry, attempt)) return;
    const unsub = (await doc_subscribe(nuri, sid, (resp: DocChange): void => {
      // A push that belongs to a torn-down fan-out, or to an establish that has been
      // superseded — either way the real unsubscribe has not taken effect yet, and neither
      // this fan-out's listeners nor its barrier are this call's to touch any more.
      if (!isCurrentAttempt(nuri, entry, attempt)) return;
      const type = docChangeType(resp);
      // Remember the barrier for whoever joins next; a later `State` replaces it.
      if (type === "State") entry.lastState = { resp, type };
      fanOut(nuri, entry, resp, type);
    })) as (() => void) | undefined;
    if (!isCurrentAttempt(nuri, entry, attempt)) {
      // Everyone left, the fan-out was reset, or a later establish replaced this one —
      // release the channel this call just opened, because nothing else holds it. (A
      // released fan-out is out of the map, which is why the listener count is not asked
      // about separately: the last one out already took the entry with it.)
      if (typeof unsub === "function") unsub();
      return;
    }
    entry.realUnsub = typeof unsub === "function" ? unsub : null;
  } catch (error) {
    reportSetupFailure(nuri, entry, error, attempt);
  }
}

/**
 * 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:
 * an `inbox.watch` 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]) {
    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);
      }
    }
    beginEstablish(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);
}

/**
 * {@link subscribeDoc}, for a caller that must be TOLD when the subscription could not be
 * opened. Same guard, same fan-out; the only difference is that a setup failure reaches
 * `onSetupFailed` instead of only the log.
 *
 * Internal, and deliberately not the published shape: upstream `doc_subscribe` is async and
 * rejects, so a failure has a caller to reach; ours returns synchronously, and the published
 * signature has no room for it (`.project/concepts/app-contract/polyfill-surface/`). An
 * application uses a subscription as a change SIGNAL and has nothing to do with the answer,
 * so it keeps the two-argument call. The inbox observation does have something to do with it
 * — an inbox it believes it is watching and is not leaves every deposit unapplied for the
 * session — so it asks.
 */
// @provenance subscribeDocReportingSetupFailure kind=aligned level=2 ref=sdk/js/lib-wasm/src/lib.rs:doc_subscribe — restores to an internal caller what upstream's async `doc_subscribe` gives every caller: the setup failure. One delta remains, the synchronous unsubscribe
export function subscribeDocReportingSetupFailure(
  nuriLike: NuriLike,
  onChange: (r: DocChange, type: DocChangeType) => void,
  onSetupFailed: (error: unknown) => void,
): Unsubscribe {
  const nuri = toNuri(nuriLike, "subscribeDoc");
  assertMayReach(nuri, "subscribeDoc");
  return subscribeDocUnguarded(nuri, onChange, onSetupFailed);
}

/**
 * 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,
  onSetupFailed?: (error: unknown) => 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,
      attempt: 0,
      lastState: null,
    };
    fanOuts.set(nuri, entry);
  }
  const joined = entry;
  // THIS call's subscription — see {@link DocListener} for why it is not the function.
  const listener: DocListener = { onChange, onSetupFailed: onSetupFailed ?? null };
  joined.listeners.add(listener);

  // 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(listener)) {
        deliver(nuri, listener, resp, type);
      }
    });
  }

  if (!joined.establishing) {
    beginEstablish(nuri, joined, ng as { doc_subscribe?: unknown });
  }

  let stopped = false;
  return () => {
    if (stopped) return;
    stopped = true;
    joined.listeners.delete(listener);
    // 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);
      }
    }
  };
}