ng-eventually/packages/polyfill/test/open-repo.test.ts

/**
 * open-repo.test.ts — behavioral tests for ensureRepoOpen (src/open-repo.ts).
 *
 * Core invariant: on a fresh session over a persistent wallet, a scope-index
 * or entity repo is NOT yet in `self.repos`, so an anchored sparql_query returns
 * 0 rows. `ensureRepoOpen(nuri)` calls `doc_subscribe(nuri, …)` FIRST (which
 * pushes the repo into the session), then the anchored read returns data.
 *
 * Fake design:
 *   - sparql_query returns EMPTY for a nuri UNTIL doc_subscribe has been called
 *     for that nuri (tracked in a Set).
 *   - doc_subscribe is a mock that records calls, fires the callback once
 *     (simulating the initial State push), then returns an unsubscribe fn.
 *
 * We test ensureRepoOpen via `readDocumentSubjects` (from `surface/read-document`)
 * because that is the production caller — it opens the document before its anchored
 * read, and `emulated-verifier/index-deposit.ts` depends on that ordering to tell an
 * absence from an unsynced repo. It was `readUnion` until `1.0.0-dev.8`; the integration
 * path is the same one, minus the batch.
 */

import { describe, it, expect, mock, beforeEach, afterAll } from "bun:test";
import { ensureRepoOpen, getSyncState, resetOpenedRepos, setOpenTimeoutForTests } from "../src/emulated-verifier/open-repo";
import { readDocumentSubjects } from "../src/surface/read-document";
import { docChannel, subscribeDoc } from "../src/surface/subscribe";
import { configure } from "../src/index";
import { configureStoreRegistry, setCurrentUser } from "../src/shared-wallet/bootstrap";
import { resetCaps, resetConfig, resetStoreRegistry } from "../src/shared-wallet/bootstrap";
import { resetRegistryCache } from "../src/shared-wallet/account-registry";

afterAll(() => {
  resetConfig();
  resetStoreRegistry();
  resetRegistryCache();
  resetOpenedRepos();
});

// The reach guard and the cap registry are process-wide: once ANY cap exists the
// boundary applies to every reader. A suite that declares none must start from an
// empty one, or it inherits another suite's enforcement.
beforeEach(() => {
  resetOpenedRepos();
  resetRegistryCache();
  resetCaps();
  setCurrentUser(null);
});

const SESSION = { sessionId: "sid-or", privateStoreId: "PRIV-OR" };
const TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type";
const FP = "http://festipod.org/";

// ---------------------------------------------------------------------------
// Fake ng builder: tracks which nuris have been doc_subscribe-d.
// sparql_query returns rows only AFTER the corresponding nuri is subscribed.
// ---------------------------------------------------------------------------

function makeFakeNgWithSubscribe(
  triplesByDoc: Record<string, Array<[string, string]>>,
  options: { pushState?: boolean; subscribeFails?: boolean } = {},
) {
  // A broker that accepts the subscription and never pushes is the state the bounded
  // fallback exists for, and the only way to hold an open genuinely IN FLIGHT.
  const pushState = options.pushState !== false;
  const subscribed = new Set<string>();
  const subscribeCallOrder: string[] = [];

  // doc_subscribe: record the call, fire callback immediately (initial push), return unsub
  const doc_subscribe = mock(async (nuri: string, _sid: string, cb: (r: unknown) => void) => {
    subscribed.add(nuri);
    subscribeCallOrder.push(nuri);
    // A broker that REFUSES the subscription - a real failure, and one `subscribe.ts`
    // deliberately makes retryable by offering `onSetupFailed`.
    if (options.subscribeFails) throw new Error("BrokerError: cannot subscribe");
    // Simulate initial State push (synchronously deferred so the subscription
    // setup promise path in ensureRepoOpen can resolve it).
    if (pushState) setTimeout(() => cb({ V0: { State: {} } }), 0);
    return () => {}; // unsubscribe fn
  });

  const sparql_query = mock(async (_sid: string, _query: string, _base: unknown, anchor: unknown) => {
    const doc = anchor as string | undefined;
    if (!doc) return { results: { bindings: [] } };
    // Only return data if the repo has been subscribed (i.e. opened)
    if (!subscribed.has(doc)) return { results: { bindings: [] } };
    const triples = triplesByDoc[doc];
    if (!triples) return { results: { bindings: [] } };
    const bindings = triples.map(([p, o]) => ({
      s: { value: doc },
      p: { value: p },
      o: { value: o },
    }));
    return { results: { bindings } };
  });

  const doc_create = mock(async () => "did:ng:o:new");
  const sparql_update = mock(async () => undefined);

  return { doc_subscribe, sparql_query, doc_create, sparql_update, subscribed, subscribeCallOrder };
}

function inject(ng: ReturnType<typeof makeFakeNgWithSubscribe>) {
  configure({ ng: ng as any, useShape: (() => {}) as any });
  configureStoreRegistry({
    getSession: async () => SESSION,
    normalizeId: (u: string) => u,
  });
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

describe("ensureRepoOpen", () => {
  it("calls doc_subscribe BEFORE the anchored read returns data", async () => {
    const ng = makeFakeNgWithSubscribe({
      "did:ng:o:a": [[TYPE, `${FP}Event`], [`${FP}title`, "Alpha"]],
    });
    inject(ng);

    // Directly call ensureRepoOpen then verify read sees data
    await ensureRepoOpen("did:ng:o:a");

    // doc_subscribe was called for the nuri
    expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
    expect(ng.subscribeCallOrder[0]).toBe("did:ng:o:a");

    // sparql_query was called AFTER subscribe (ensureRepoOpen guarantees ordering)
    const result = await readDocumentSubjects("did:ng:o:a");
    expect(result.length).toBe(1);
    expect(result[0]!.props[`${FP}title`]).toEqual(["Alpha"]);
  });

  it("WITHOUT doc_subscribe, sparql_query returns 0 rows (verifies fake mechanics)", async () => {
    const ng = makeFakeNgWithSubscribe({
      "did:ng:o:a": [[TYPE, `${FP}Event`], [`${FP}title`, "Alpha"]],
    });
    inject(ng);

    // Do NOT call ensureRepoOpen — subscribed Set remains empty
    // Query directly (bypass `readDocumentSubjects`, which opens the repo internally)
    const result = await ng.sparql_query("sid-or", "SELECT ?s ?p ?o WHERE { ?s ?p ?o }", undefined, "did:ng:o:a");
    const bindings = (result as any).results.bindings;
    expect(bindings.length).toBe(0); // not subscribed → 0 rows (confirms fake design)
  });

  it("idempotence: a 2nd ensureRepoOpen for the same nuri does NOT re-subscribe", async () => {
    const ng = makeFakeNgWithSubscribe({
      "did:ng:o:b": [[TYPE, `${FP}Event`]],
    });
    inject(ng);

    await ensureRepoOpen("did:ng:o:b");
    await ensureRepoOpen("did:ng:o:b"); // second call

    // doc_subscribe must have been called exactly ONCE
    expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
  });

  it("no-op when the fake ng has no doc_subscribe (unit fake path)", async () => {
    // Fake ng WITHOUT doc_subscribe
    const noSubscribeNg = {
      doc_create: mock(async () => "did:ng:o:new"),
      sparql_update: mock(async () => undefined),
      sparql_query: mock(async () => ({ results: { bindings: [] } })),
    };
    configure({ ng: noSubscribeNg as any, useShape: (() => {}) as any });
    configureStoreRegistry({
      getSession: async () => SESSION,
      normalizeId: (u: string) => u,
    });

    // Must not throw; nuri is added to opened Set (guard skips subscribe)
    await expect(ensureRepoOpen("did:ng:o:c")).resolves.toBeUndefined();

    // Calling again should also be a no-op (idempotent, already in opened)
    await expect(ensureRepoOpen("did:ng:o:c")).resolves.toBeUndefined();
  });

  // An open ENDS, whatever happens to the registry underneath it.
  //
  // A reset - which is what a session change does, `syncSession` - tears down the
  // subscription an in-flight open is waiting on. Until 2026-09-21 it then left that open
  // hanging: its `State` could no longer arrive, so the caller waited out the whole
  // 8-second fallback and only then proceeded, with its sync state CLEARED. Cleared reads
  // as `"unknown"`, and `unknown` is the LAX answer - `emulated-verifier/index-deposit.ts`
  // accepts it as "a zero-row read means the document is empty" - so a read taken right
  // after a session change could have an absence read into it. Both halves are the point:
  // it must END, and it must end as `"timed-out"`.
  it("a reset ENDS the opens it abandons, and marks them timed-out rather than unknown", async () => {
    const ng = makeFakeNgWithSubscribe({}, { pushState: false });
    inject(ng);

    const inFlight = ensureRepoOpen("did:ng:o:abandoned");
    await new Promise((r) => setTimeout(r, 5)); // let the subscribe land
    expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);

    resetOpenedRepos();

    // Ends at once. Without the fix it resolves only when the 8s fallback fires, which is
    // longer than any test's patience and longer than any page's.
    await expect(
      Promise.race([
        inFlight.then(() => "ended"),
        new Promise((r) => setTimeout(() => r("still waiting"), 200)),
      ]),
    ).resolves.toBe("ended");
    // ...and NOT as a barrier that was reached.
    expect(getSyncState("did:ng:o:abandoned")).toBe("timed-out");
  });

  // A reset must INVALIDATE the opens it abandons, not merely end them.
  //
  // The continuation of an in-flight open records the repo as opened. Settling that open
  // from the reset (2026-09-21) made the continuation run right after the maps were
  // cleared, so it wrote a stale `opened` entry into the FRESH registry - and every later
  // `ensureRepoOpen` for that nuri then short-circuited on it. The repo was never
  // subscribed in the new session, and a read over it answered as though it had been.
  // Before that change the same line ran only when the 8s fallback fired, so a re-open
  // inside that window did re-subscribe; ending the open promptly is right, but it moved
  // the write into the window where it does damage.
  it("a reset INVALIDATES the opens it abandons, so the new session subscribes again", async () => {
    const ng = makeFakeNgWithSubscribe({}, { pushState: false });
    inject(ng);

    const abandoned = ensureRepoOpen("did:ng:o:reopened");
    await new Promise((r) => setTimeout(r, 5));
    expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);

    resetOpenedRepos();
    await abandoned; // its continuation runs here, on the cleared registry
    await new Promise((r) => setTimeout(r, 5));

    // The new session has subscribed NOTHING yet, so this must open the repo for real.
    // NOT awaited: this broker never pushes a `State`, so the open runs to its bounded
    // fallback - and what is under test is that the subscribe HAPPENS, not how it ends.
    const reopened = ensureRepoOpen("did:ng:o:reopened");
    await new Promise((r) => setTimeout(r, 5));
    expect(ng.doc_subscribe).toHaveBeenCalledTimes(2);
    // Released so the fallback's timer does not outlive the test with a live handle.
    resetOpenedRepos();
    await reopened;
  });

  // A subscribe the broker REFUSED leaves nothing behind, so it can be tried again.
  //
  // `subscribe.ts` reports a failed establish through `onSetupFailed` precisely so a
  // caller can retry; `open-repo` ignored it, waited out the bounded fallback, and then
  // recorded the repo as OPEN. Every later open short-circuited on that record, so
  // `refetch()` issued no new subscribe and the read had no exit at all.
  it("a subscribe the broker REFUSED is not recorded as open, so it can be retried", async () => {
    const ng = makeFakeNgWithSubscribe({}, { subscribeFails: true });
    inject(ng);
    setOpenTimeoutForTests(30);

    await ensureRepoOpen("did:ng:o:refused-sub");
    expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
    // Not authoritative: no `State` was ever going to arrive.
    expect(getSyncState("did:ng:o:refused-sub")).toBe("timed-out");

    // The retry a caller is entitled to make.
    await ensureRepoOpen("did:ng:o:refused-sub");
    expect(ng.doc_subscribe).toHaveBeenCalledTimes(2);
  });

  // -- ONE answer to "is this document open in this session" ---------------
  //
  // Both of these are the same defect: a record of openness kept HERE, which outlives the
  // subscription it describes because `surface/subscribe.ts` drops its listeners without
  // telling anybody. `configure()` is PUBLISHED and calls `resetDocSubscriptions()`, so this
  // is a sequence an application performs, not a montage.

  it("a second configure() leaves NOTHING claiming the document is open or synced", async () => {
    const ngA = makeFakeNgWithSubscribe({ "did:ng:o:reconf": [[TYPE, `${FP}Event`]] });
    inject(ngA);
    await ensureRepoOpen("did:ng:o:reconf");
    expect(getSyncState("did:ng:o:reconf")).toBe("synced");

    // A second SDK. The subscription behind that `"synced"` is gone with the first one, so the
    // barrier it reports was crossed against a verifier nobody is talking to any more. Until
    // 2026-09-24 this module kept saying `"synced"` - which is AUTHORITY, a zero-row read over
    // it being a definitive absence - and its own `held` entry then short-circuited every
    // re-open for the rest of the session, so `refetch()` had no exit either.
    const ngB = makeFakeNgWithSubscribe({ "did:ng:o:reconf": [[TYPE, `${FP}Event`]] });
    inject(ngB);
    expect(getSyncState("did:ng:o:reconf")).toBe("timed-out");

    // ...and the re-open actually happens, against the SDK that is now injected.
    await ensureRepoOpen("did:ng:o:reconf");
    expect(ngB.doc_subscribe).toHaveBeenCalledTimes(1);
    expect(getSyncState("did:ng:o:reconf")).toBe("synced");
  });

  // The second face, and the one the failure channel cannot reach.
  //
  // `subscribe.ts` reports a refused establish only for the attempt it is still WAITING ON, so
  // a rejection that arrives after `resetDocSubscriptions` released the fan-out is logged and
  // nothing else. A record written when the subscribe was REQUESTED therefore stood for the
  // whole session with nothing able to clear it: measured on the previous code, zero subscribes
  // under the new SDK, for ever. The repair is that nothing here records openness at all.
  it("an establish still outstanding when the subscription layer is torn down leaves no claim", async () => {
    let refuse: ((error: Error) => void) | null = null;
    const doc_subscribe = mock(
      async () =>
        await new Promise<never>((_resolve, reject) => {
          refuse = reject;
        }),
    );
    const ngA = {
      doc_subscribe,
      sparql_query: mock(async () => ({ results: { bindings: [] } })),
      doc_create: mock(async () => "did:ng:o:new"),
      sparql_update: mock(async () => undefined),
    };
    inject(ngA as unknown as ReturnType<typeof makeFakeNgWithSubscribe>);
    setOpenTimeoutForTests(20);

    // Stops waiting on the bounded fallback, with the subscribe STILL outstanding: at this
    // instant nothing is true yet, and that is precisely when the old code recorded a hold.
    await ensureRepoOpen("did:ng:o:outstanding");
    expect(doc_subscribe).toHaveBeenCalledTimes(1);

    const ngB = makeFakeNgWithSubscribe({});
    inject(ngB); // configure() -> resetDocSubscriptions(), mid-establish
    // The first SDK finally refuses - into a fan-out that no longer exists, so no caller is
    // told. Nothing may depend on being told.
    refuse!(new Error("BrokerError: late refusal"));
    await new Promise((r) => setTimeout(r, 5));

    setOpenTimeoutForTests(20);
    await ensureRepoOpen("did:ng:o:outstanding");
    expect(ngB.doc_subscribe).toHaveBeenCalledTimes(1);
    expect(getSyncState("did:ng:o:outstanding")).toBe("synced");
  });

  // The other half of the same question, and it is why "the channel is open" is not the whole
  // answer either.
  //
  // A reset releases the bootstrap listener, which is what kept the repo open for the session.
  // If something ELSE was following the document, the fan-out survives and `resubscribeDocs`
  // re-opens its channel - so the channel is `"open"` again while this module holds nothing.
  // Skipping the work there would leave the document held by a listener that is free to leave,
  // and the read would not wait for the new session's barrier either.
  it("a reset leaves the document to be HELD again, even though its channel is open", async () => {
    const ng = makeFakeNgWithSubscribe({});
    inject(ng);
    const doc = "did:ng:o:alsowatched";

    // Somebody else follows this document - an inbox observation, or an application's own
    // subscription. It is not this module's listener and it does not hold the repo for it.
    const appUnsub = subscribeDoc(doc, () => {});
    await ensureRepoOpen(doc);
    expect(getSyncState(doc)).toBe("synced");

    // The session changes: the bootstrap listener goes, the other one stays, and the fan-out's
    // channel is re-opened against the new session with its barrier forgotten.
    resetOpenedRepos();
    await new Promise((r) => setTimeout(r, 5));
    expect(docChannel(doc).state).toBe("open");

    await ensureRepoOpen(doc);
    expect(getSyncState(doc)).toBe("synced");

    // The other listener leaves. The document must still be open, because the bootstrap open is
    // holding it - which it can only do if the call above actually took a listener out.
    appUnsub();
    expect(docChannel(doc).state).toBe("open");
    expect(getSyncState(doc)).toBe("synced");
  });

  it("the anchored document read triggers doc_subscribe then returns data (integration path)", async () => {
    const ng = makeFakeNgWithSubscribe({
      "did:ng:o:p": [[TYPE, `${FP}Participation`], [`${FP}event`, "did:ng:o:e"]],
    });
    inject(ng);

    const subjects = await readDocumentSubjects("did:ng:o:p");

    // doc_subscribe was called as part of the `ensureRepoOpen` inside the read
    expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
    expect(subjects.length).toBe(1);
    expect(subjects[0]!.props[`${FP}event`]).toEqual(["did:ng:o:e"]);
  });
});