ng-eventually/packages/polyfill/test/subscribe.test.ts

import { test, expect, mock, afterAll } from "bun:test";
import {
  docChangeType,
  resubscribeDocs,
  subscribeDoc,
  subscribeDocReportingSetupFailure,
  subscribeDocs,
  type Unsubscribe,
} from "../src/surface/subscribe";
import { configure } from "../src/index";
import { configureStoreRegistry } from "../src/shared-wallet/bootstrap";
import { resetConfig, resetStoreRegistry } from "../src/shared-wallet/bootstrap";
import type { RegistrySession } from "../src/shared-wallet/account-registry";

// subscribeDoc/subscribeDocs wrap the REAL injected `ng.doc_subscribe`. This
// suite injects a fake `ng` whose `doc_subscribe` records the callback per doc
// and hands back an unsubscribe, so we can assert routing + isolation without a
// broker. Restore the un-configured state at the end.
afterAll(() => {
  resetConfig();
  resetStoreRegistry();
});

const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" };

/**
 * A fake reactive `ng`: `doc_subscribe(nuri, sid, cb)` takes over `nuri`, fires `cb` once
 * (initial State push), and returns an unsubscribe. `push(nuri)` drives a later change to
 * that doc's subscriber. A per-doc `failFor` set makes `doc_subscribe` reject for chosen
 * NURIs (a not-yet-synced doc).
 *
 * ── ONE subscriber per document, and a second one EVICTS it ───────────────
 * Not a simplification — it is what the broker does. A branch holds a single sender
 * (`branch_subscriptions: HashMap<BranchId, Sender<AppResponse>>`) and
 * `create_branch_subscription` closes whatever it displaces, silently: the evicted
 * unsubscribe still returns cleanly and nothing anywhere errors. Confirmed against the real
 * broker on 2026-08-17 — with two subscriptions on one document, a write fired the second
 * callback and the first, which had been firing moments earlier, went quiet for good.
 *
 * A Set of callbacks here would model a world where every subscriber coexists. It is
 * exactly the assumption that cost this package a view that never re-read and an inbox that
 * never notified, and a fake that holds it cannot fail on either.
 */
function makeFakeNg(failFor: Set<string> = new Set(), hangFor: Set<string> = new Set()) {
  const subs = new Map<string, (r: unknown) => void>();
  // A call the broker has neither answered nor refused yet, so a LATER call can overtake it
  // and this one can settle afterwards. Held on an object rather than in a `let` so its
  // type survives being written from one closure and read from another.
  const hung: {
    reject: ((error: unknown) => void) | null;
    resolve: ((unsub: () => void) => void) | null;
    cb: ((r: unknown) => void) | null;
    live: boolean;
  } = { reject: null, resolve: null, cb: null, live: false };
  const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => {
    if (failFor.has(nuri)) throw new Error(`RepoNotFound: ${nuri}`);
    if (hangFor.has(nuri)) {
      return await new Promise((resolve, reject) => {
        hung.reject = reject;
        hung.resolve = resolve as (unsub: () => void) => void;
        hung.cb = cb;
      });
    }
    subs.set(nuri, cb); // whoever held this branch is dropped, without a word
    // Initial State push, delivered async (as the real RPC does) — and only while this
    // callback still holds the branch.
    queueMicrotask(() => {
      if (subs.get(nuri) === cb) cb({ V0: { State: { doc: nuri } } });
    });
    return () => {
      if (subs.get(nuri) === cb) subs.delete(nuri);
    };
  });
  const push = (nuri: string): void => {
    subs.get(nuri)?.({ V0: { Patch: { doc: nuri } } });
  };
  const isSubscribed = (nuri: string): boolean => subs.has(nuri);
  /** The call that was left hanging finally answers — with a refusal. */
  const rejectHung = (): void => {
    const reject = hung.reject;
    hung.reject = null;
    reject?.(new Error("RepoNotFound: late"));
  };
  /**
   * The call that was left hanging finally answers — SUCCESSFULLY, and from the session it
   * was placed against, which by now is the PREVIOUS one.
   *
   * Modelled apart from `subs` on purpose: a session is a verifier of its own, with its own
   * `branch_subscriptions`, so this channel neither evicted nor was evicted by the one the
   * new session opened (`engine/verifier/src/verifier.rs:361` inserts into `self`). It is
   * genuinely live and genuinely pushing, and nothing but its own unsubscribe closes it —
   * which is the whole reason a superseded attempt has to be released rather than dropped.
   */
  const resolveHung = (): void => {
    const resolve = hung.resolve;
    hung.resolve = null;
    hung.reject = null;
    hung.live = true;
    resolve?.(() => {
      hung.live = false;
    });
  };
  /**
   * The PREVIOUS session pushes over the channel it just handed back.
   *
   * Unconditionally, even once that channel has been released: a cancel does not reach back
   * and un-send what the broker already dispatched, which is precisely why the callback
   * carries a guard of its own instead of trusting the unsubscribe to have taken effect.
   */
  const pushHung = (payload: unknown): void => {
    hung.cb?.(payload);
  };
  /** Is the previous session's channel still open upstream? */
  const hungIsLive = (): boolean => hung.live;
  return {
    doc_subscribe,
    push,
    isSubscribed,
    rejectHung,
    resolveHung,
    pushHung,
    hungIsLive,
    _subs: subs,
  };
}

function inject(failFor?: Set<string>, hangFor?: Set<string>) {
  const ng = makeFakeNg(failFor, hangFor);
  configure({ ng: ng as any, useShape: (() => {}) as any });
  // Synchronous fake store → no sync lag; disable the anti-fork retry backoff.
  configureStoreRegistry({ getSession: async () => SESSION });
  return ng;
}

const A = "did:ng:o:docA";
const B = "did:ng:o:docB";

const tick = () => new Promise((r) => setTimeout(r, 5));

test("subscribeDoc calls ng.doc_subscribe with (nuri, sessionId, callback)", async () => {
  const ng = inject();
  const onChange = mock(() => {});
  subscribeDoc(A, onChange);
  await tick();
  expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
  const call = ng.doc_subscribe.mock.calls[0]!;
  expect(call[0]).toBe(A);
  expect(call[1]).toBe("sid-1"); // sessionId from the injected session
  expect(typeof call[2]).toBe("function"); // the callback
});

test("subscribeDoc routes the initial State push and every later change", async () => {
  const ng = inject();
  const seen: unknown[] = [];
  subscribeDoc(A, (r) => seen.push(r));
  await tick();
  expect(seen).toHaveLength(1); // initial State push
  ng.push(A);
  ng.push(A);
  expect(seen).toHaveLength(3); // + 2 patches
});

test("subscribeDoc unsubscribe stops further callbacks", async () => {
  const ng = inject();
  const seen: unknown[] = [];
  const stop = subscribeDoc(A, (r) => seen.push(r));
  await tick();
  expect(seen).toHaveLength(1);
  stop();
  expect(ng.isSubscribed(A)).toBe(false); // real unsubscribe was invoked
  ng.push(A); // ignored — no subscriber
  expect(seen).toHaveLength(1);
});

test("subscribeDoc unsubscribe BEFORE async setup resolves cancels cleanly", async () => {
  const ng = inject();
  const seen: unknown[] = [];
  const stop = subscribeDoc(A, (r) => seen.push(r));
  stop(); // before the microtask/promise setup resolved
  await tick();
  // The subscription was cancelled the moment setup resolved: no callbacks, and
  // no lingering subscriber.
  expect(seen).toHaveLength(0);
  expect(ng.isSubscribed(A)).toBe(false);
});

test("subscribeDocs fans out one subscription per doc and reports the source nuri", async () => {
  const ng = inject();
  const seen: Array<[string, unknown]> = [];
  subscribeDocs([A, B], (nuri, r) => seen.push([nuri, r]));
  await tick();
  // Two initial pushes, one per doc.
  expect(seen.map((s) => s[0]).sort()).toEqual([A, B]);
  ng.push(B);
  expect(seen.filter((s) => s[0] === B)).toHaveLength(2); // initial + patch
  expect(seen.filter((s) => s[0] === A)).toHaveLength(1); // isolated: A didn't fire
});

test("subscribeDocs isolates a failing doc — the others still fire", async () => {
  const ng = inject(new Set([A])); // A's subscription throws (RepoNotFound)
  const seen: Array<[string, unknown]> = [];
  subscribeDocs([A, B], (nuri, r) => seen.push([nuri, r]));
  await tick();
  // A failed to subscribe (logged, not thrown); B is unaffected and fired.
  expect(seen.map((s) => s[0])).toEqual([B]);
  ng.push(B);
  expect(seen.filter((s) => s[0] === B)).toHaveLength(2);
});

test("subscribeDocs unsubscribe tears down all subscriptions", async () => {
  const ng = inject();
  const stop = subscribeDocs([A, B], () => {});
  await tick();
  expect(ng.isSubscribed(A)).toBe(true);
  expect(ng.isSubscribed(B)).toBe(true);
  stop();
  expect(ng.isSubscribed(A)).toBe(false);
  expect(ng.isSubscribed(B)).toBe(false);
});

test("subscribeDocs deduplicates repeated NURIs", async () => {
  const ng = inject();
  subscribeDocs([A, A, A], () => {});
  await tick();
  expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
});

// --- two subscribers on ONE document ---------------------------------------
//
// A branch has room for exactly one subscriber and a second `doc_subscribe` evicts the
// first (see `makeFakeNg`). Everything inside this package subscribes — `ensureRepoOpen`
// holds a bootstrap subscription per document for the session, `watchShape` follows the
// documents of a scope, `inbox.watch` follows an inbox, and the inbox observation follows
// every inbox — so any two of them on the same document used to silence one another.
// Nothing rejected and nothing logged; the view simply stopped re-reading.
//
// So the package opens ONE real subscription per document and fans it out. These are the
// tests that say so.

test("two subscribers on one document BOTH keep firing", async () => {
  const ng = inject();
  const first: unknown[] = [];
  const second: unknown[] = [];
  subscribeDoc(A, (r) => first.push(r));
  await tick();
  expect(first).toHaveLength(1); // its initial State

  subscribeDoc(A, (r) => second.push(r));
  await tick();

  ng.push(A);
  // The one that was there first is not silenced by the one that came second.
  expect(first).toHaveLength(2);
  expect(second.length).toBeGreaterThanOrEqual(1);
});

test("one real doc_subscribe serves every subscriber of a document", async () => {
  const ng = inject();
  subscribeDoc(A, () => {});
  subscribeDoc(A, () => {});
  subscribeDoc(A, () => {});
  await tick();
  // Three callers, one branch taken. A second call would have evicted the first caller.
  expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
});

test("a subscriber that joins LATE still gets its initial State", async () => {
  inject();
  subscribeDoc(A, () => {});
  await tick(); // the initial State has come and gone

  const late: unknown[] = [];
  subscribeDoc(A, (r) => late.push(r));
  await tick();

  // Its own `doc_subscribe` would have pushed it a State; joining an open one owes it the
  // same, or "fires once immediately" quietly stops being true for whoever arrives second.
  expect(late).toHaveLength(1);
  expect(docChangeType(late[0])).toBe("State");
});

test("the real subscription is released only when the LAST subscriber leaves", async () => {
  const ng = inject();
  const seen: unknown[] = [];
  const stopFirst = subscribeDoc(A, () => {});
  const stopSecond = subscribeDoc(A, (r) => seen.push(r));
  await tick();

  stopFirst();
  expect(ng.isSubscribed(A)).toBe(true); // somebody is still listening
  ng.push(A);
  expect(seen.length).toBeGreaterThanOrEqual(2); // and still hearing

  stopSecond();
  expect(ng.isSubscribed(A)).toBe(false); // now nobody is
});

// --- what sharing one subscription must not COST a caller --------------------
//
// Three things a per-caller `doc_subscribe` gave for free. The fan-out took each of them
// away when it was introduced, and each is invisible from the caller's side: nothing
// rejects, nothing logs, the document simply stops speaking to somebody.

test("two subscriptions with the SAME handler are two subscriptions", async () => {
  const ng = inject();
  const seen: unknown[] = [];
  // One function, two callers. A module-level handler, a bound method or a shared arrow
  // makes this ordinary rather than exotic — and keyed by the function, the second caller
  // was never registered at all, so the first one's departure took it with it.
  const handler = (r: unknown): void => {
    seen.push(r);
  };
  const stopFirst = subscribeDoc(A, handler);
  const stopSecond = subscribeDoc(A, handler);
  await tick();
  const afterInitial = seen.length;

  stopFirst(); // only the FIRST caller has left
  expect(ng.isSubscribed(A)).toBe(true);
  ng.push(A);
  expect(seen.length).toBeGreaterThan(afterInitial); // the second one is still listening

  stopSecond();
  expect(ng.isSubscribed(A)).toBe(false); // …and now the last one has gone
});

test("a subscriber torn down inside another's handler does not receive that push", async () => {
  const ng = inject();
  const seen: unknown[] = [];
  let stopSecond: Unsubscribe | null = null;
  let armed = false;
  // The first handler tears the second one down mid-push. Which of the two runs first is an
  // ordering no application controls, and `subscribeDoc` publishes that no `onChange` fires
  // after unsubscribe — so the answer must not depend on it.
  const stopFirst = subscribeDoc(A, () => {
    if (armed) stopSecond?.();
  });
  stopSecond = subscribeDoc(A, (r) => seen.push(r));
  await tick();
  seen.length = 0;

  armed = true;
  ng.push(A);

  expect(seen).toEqual([]);
  stopFirst();
});

test("a setup that FAILED does not poison the document for the next subscriber", async () => {
  const failFor = new Set([A]);
  const ng = inject(failFor);
  const stopFirst = subscribeDoc(A, () => {});
  await tick();
  expect(ng.doc_subscribe).toHaveBeenCalledTimes(1); // …and it rejected

  // The document syncs; the broker can serve it now. A caller arriving after a transient
  // rejection must get a real subscription, exactly as its own `doc_subscribe` would have.
  failFor.delete(A);
  const seen: unknown[] = [];
  const stopSecond = subscribeDoc(A, (r) => seen.push(r));
  await tick();

  expect(ng.doc_subscribe).toHaveBeenCalledTimes(2); // attempted again, not joined to a corpse
  expect(seen.length).toBeGreaterThan(0); // its initial State
  ng.push(A);
  expect(seen.length).toBeGreaterThan(1); // and the changes that follow
  stopFirst();
  stopSecond();
});

test("a caller that asks to be told learns its subscription could not be opened", async () => {
  inject(new Set([A]));
  const failures: unknown[] = [];
  const stop = subscribeDocReportingSetupFailure(
    A,
    () => {},
    (error) => failures.push(error),
  );
  await tick();
  // Upstream `doc_subscribe` is async and rejects, so its caller learns. This wrapper
  // returns synchronously, and a caller whose job depends on the subscription (the inbox
  // observation) cannot otherwise tell "quiet" from "never opened".
  expect(failures).toHaveLength(1);
  expect(String(failures[0])).toContain("RepoNotFound");
  stop();
});

/**
 * Two establishes over ONE fan-out, and the first one answering last.
 *
 * `resubscribeDocs` re-opens the channel of a fan-out whose first `doc_subscribe` has not
 * settled yet — that is the whole point of it, since the session it was opened against is
 * gone — and it re-opens it on the SAME entry so the listeners are kept. The two calls
 * therefore race, and the broker is under no obligation to answer them in order.
 */
async function hangingThenReopened(): Promise<{
  ng: ReturnType<typeof makeFakeNg>;
  failures: unknown[];
  seen: unknown[];
  stop: Unsubscribe;
}> {
  const hangFor = new Set([A]);
  const ng = inject(new Set(), hangFor);
  const failures: unknown[] = [];
  const seen: unknown[] = [];
  const stop = subscribeDocReportingSetupFailure(
    A,
    (r) => seen.push(r),
    (error) => failures.push(error),
  );
  // Awaited before the broker is allowed to answer: `establish` resolves the session id
  // first, so the call this has to leave hanging has not been placed yet.
  await tick();
  hangFor.delete(A);
  return { ng, failures, seen, stop };
}

test("a SUPERSEDED setup rejecting late is not reported as this document failing", async () => {
  const { ng, failures, seen, stop } = await hangingThenReopened();
  expect(ng.isSubscribed(A)).toBe(false); // the first call has not answered

  resubscribeDocs(); // the session rotated: a second establish, on the same fan-out
  await tick();
  expect(ng.isSubscribed(A)).toBe(true);
  const before = seen.length;
  expect(before).toBeGreaterThan(0); // …and it is pushing

  ng.rejectHung(); // …and only now does the first call refuse
  await tick();

  // Nothing failed: the document is subscribed and pushing. Told otherwise, the caller whose
  // job depends on this subscription (the inbox observation) releases the entry it holds —
  // which closes the WORKING channel and reports an inbox that is watched as unwatchable.
  expect(failures).toEqual([]);
  expect(ng.isSubscribed(A)).toBe(true);
  ng.push(A);
  expect(seen.length).toBeGreaterThan(before);
  stop();
});

test("a SUPERSEDED setup rejecting late does not let the next joiner evict the live channel", async () => {
  const { ng, stop } = await hangingThenReopened();
  resubscribeDocs();
  await tick();
  expect(ng.doc_subscribe).toHaveBeenCalledTimes(2);

  ng.rejectHung();
  await tick();

  // The second establish still stands, so this joiner must join it. Counting the late
  // rejection as "nothing is running any more" opens a THIRD `doc_subscribe` — and a second
  // subscribe on a branch evicts the one before it, so the joiner's own call is what silences
  // everybody already listening.
  const late: unknown[] = [];
  const stopLate = subscribeDoc(A, (r) => late.push(r));
  await tick();
  expect(ng.doc_subscribe).toHaveBeenCalledTimes(2);
  expect(late.length).toBeGreaterThan(0); // replayed the barrier, as any late joiner is
  ng.push(A);
  expect(late.length).toBeGreaterThan(1);
  stopLate();
  stop();
});

/** A `State` that could only have come from the session `resubscribeDocs` walked away from. */
const OLD_SESSION_STATE = { V0: { State: { doc: A, from: "the previous session" } } };

test("a SUPERSEDED setup that SUCCEEDS pushes to nobody, re-seeds nothing, and is released", async () => {
  const { ng, seen, stop } = await hangingThenReopened();
  resubscribeDocs(); // the session rotated: a second establish, on the SAME fan-out
  await tick();
  expect(ng.isSubscribed(A)).toBe(true);
  const afterReopen = seen.length;
  expect(afterReopen).toBeGreaterThan(0); // the current session is pushing

  // …and only now does the FIRST call answer — successfully, from the session that is gone.
  ng.resolveHung();
  await tick();
  ng.pushHung(OLD_SESSION_STATE);
  await tick();

  // Its channel was released as it resolved: nothing else would ever close it, because it
  // sits on the previous session's verifier where the new subscription never displaced it.
  expect(ng.hungIsLive()).toBe(false);

  // Its push reaches nobody. The listeners are following the establish that replaced it, and
  // handing them a verifier nobody is talking to any more is the staleness, not the cure.
  expect(seen).not.toContainEqual(OLD_SESSION_STATE);
  expect(seen.length).toBe(afterReopen);

  // Nor is the abandoned `State` left as the barrier: a joiner is replayed the CURRENT
  // session's, which is what `resubscribeDocs` cleared `lastState` to guarantee. Replayed the
  // other one, a bootstrap open calls a repo this session has never synced "synced".
  const late: unknown[] = [];
  const stopLate = subscribeDoc(A, (r) => late.push(r));
  await tick();
  expect(late.length).toBeGreaterThan(0);
  expect(late).not.toContainEqual(OLD_SESSION_STATE);

  // And `realUnsub` is still the live call's, so the last listener out leaves NOTHING
  // subscribed upstream — neither session's channel.
  stopLate();
  stop();
  await tick();
  expect(ng.isSubscribed(A)).toBe(false);
  expect(ng.hungIsLive()).toBe(false);
});

test("a SUPERSEDED setup that FAILS is logged, even though it is reported to nobody", async () => {
  const { ng, failures, stop } = await hangingThenReopened();
  resubscribeDocs();
  await tick();

  const logged: string[] = [];
  const realError = console.error;
  console.error = ((...a: unknown[]) => {
    logged.push(a.map(String).join(" "));
  }) as typeof console.error;
  try {
    ng.rejectHung();
    await tick();
  } finally {
    console.error = realError;
  }

  // Told to nobody — the call that replaced it owns the outcome — but never passed over in
  // silence: a rejection that reaches no caller AND no log is a failure that never happened,
  // and this one is the only trace that the abandoned session refused at all.
  expect(failures).toEqual([]);
  expect(logged.some((l) => l.startsWith("[subscribe] doc_subscribe failed for"))).toBe(true);
  stop();
});

test("an establish SUPERSEDED while the session id resolves never reaches the broker at all", async () => {
  const ng = makeFakeNg();
  configure({ ng: ng as unknown as Parameters<typeof configure>[0]["ng"], useShape: (() => {}) as never });
  // A session lookup this test holds open, so the supersession lands while the first
  // establish is between "started" and "placed its call" — the window `resubscribeDocs`
  // actually opens, since it fires from a `resetOpenedRepos` that a session read preceded.
  const gate: { release: (() => void) | null } = { release: null };
  const held = new Promise<void>((r) => {
    gate.release = r;
  });
  configureStoreRegistry({
    getSession: async () => {
      await held;
      return SESSION;
    },
  });

  const seen: unknown[] = [];
  const stop = subscribeDoc(A, (r) => seen.push(r));
  await tick();
  expect(ng.doc_subscribe).toHaveBeenCalledTimes(0); // still resolving the session

  resubscribeDocs(); // superseded before it ever placed its call
  gate.release?.();
  await tick();

  // ONE call, and it is the current attempt's. This is what makes "release a superseded
  // attempt" safe rather than a coin flip: every attempt that OWNS a channel placed its call
  // before the attempt replacing it existed, so the broker — which serialises the whole call
  // under one lock — served it first, and whoever inserts last holds the branch. The
  // superseded one is therefore always the evicted one, never the live channel.
  expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
  expect(ng.isSubscribed(A)).toBe(true);
  expect(seen.length).toBeGreaterThan(0);
  stop();
  expect(ng.isSubscribed(A)).toBe(false);
});