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 = new Set()) { const subs = new Map 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) { 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 }); 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); });