ng-eventually/packages/client/src/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.
 */

import { getConfig, getStoreRegistryDeps } from "./polyfill";
import { assertMayReach } from "./reach";
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: { <Variant>: … } }` 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<string, unknown>);
  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<string> {
  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 {
  // 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);
}

/**
 * Subscribe as the PHYSICAL user — the shim's own documents. The machinery's
 * counterpart to {@link subscribeDoc}; never exported from the package.
 */
export function subscribePhysicalDoc(
  nuri: Nuri,
  onChange: (r: DocChange, type: DocChangeType) => void,
): Unsubscribe {
  return subscribeDocUnguarded(nuri, onChange);
}

function subscribeDocUnguarded(
  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);
      }
    }
  };
}