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

import { test, expect, mock, afterAll } from "bun:test";
import { subscribeDoc, subscribeDocs } from "../src/subscribe";
import {
  configure,
  configureStoreRegistry,
  resetConfig,
  resetStoreRegistry,
} from "../src/polyfill";
import type { RegistrySession } from "../src/store-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)` registers `cb` for `nuri`,
 * fires it once (initial State push), and returns an unsubscribe. `push(nuri)`
 * drives a later change to that doc's subscribers. A per-doc `failFor` set makes
 * `doc_subscribe` reject for chosen NURIs (a not-yet-synced doc).
 */
function makeFakeNg(failFor: Set<string> = new Set()) {
  const subs = new Map<string, Set<(r: unknown) => void>>();
  const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => {
    if (failFor.has(nuri)) throw new Error(`RepoNotFound: ${nuri}`);
    let set = subs.get(nuri);
    if (!set) {
      set = new Set();
      subs.set(nuri, set);
    }
    set.add(cb);
    // Initial State push, delivered async (as the real RPC does).
    queueMicrotask(() => cb({ V0: { State: { doc: nuri } } }));
    return () => set!.delete(cb);
  });
  const push = (nuri: string): void => {
    for (const cb of subs.get(nuri) ?? []) cb({ V0: { Patch: { doc: nuri } } });
  };
  const isSubscribed = (nuri: string): boolean => (subs.get(nuri)?.size ?? 0) > 0;
  return { doc_subscribe, push, isSubscribed, _subs: subs };
}

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