ng-eventually/packages/polyfill/test/continuous-inbox-observation.test.ts

/**
 * While I am connected, what lands in my inboxes is applied — no reload, nobody asked.
 *
 * ── The regime, and the one this replaces ─────────────────────────────────
 * Upstream, applying an inbox is what a SESSION does: a sealed message reaches the
 * recipient's own verifier as it arrives and is applied inline, and only the backlog handed
 * over at connection is marked apart (`from_queue`). This package had emulated the backlog
 * and nothing else — `inbox.processInbox` was called from exactly one place, at connection —
 * so a share deposited while its recipient sat connected in front of the application
 * converged when that person next RELOADED the page. Its stand-in was a twenty-second timer
 * in the DEPOSITOR's session, which does nothing if that tab closes and tells the connected
 * owner nothing either way.
 *
 * ── How a deposit gets here without a reconnection ────────────────────────
 * Bob makes his deposit through the published surface, under his own identity, naming a
 * PERSON (`inbox.share(doc, "alice")`) — he is handed no address, and the test hands him
 * none. What the test then does is what a broker does: it HOLDS what he wrote and delivers
 * it to this page later, once Alice is the one connected (`wallet-fake._deliver`). The
 * quads delivered are the ones the library itself produced under Bob; nothing is composed
 * by hand. That is the cross-session case verified against the real broker
 * (`e2e/reactivity-doc-subscribe.ts`): a second session's write reaches the first session's
 * subscription as a `Patch`.
 *
 * ── A deposit by the owner is a REAL case, not a shortcut ─────────────────
 * Two tests below have Alice deposit into her own document's inbox while she is connected.
 * That is not a stand-in for a stranger: it is the case a consuming application reported,
 * and a same-session write is pushed to that session's own subscription — verified against
 * the real broker the same day (`Patch@69ms` on the writer's own `sparqlUpdate`).
 */

import { test, expect, describe, afterAll, beforeEach } from "bun:test";
import { inbox as inboxSurface, storeRegistry } from "../src/index";
import { getCaps, setCurrentUser } from "../src/shared-wallet/bootstrap";
import { resolveAccount, userInbox } from "../src/shared-wallet/account-registry";
import { observationSettled } from "../src/emulated-verifier/inbox-observer";
import { cancelScheduledInboxProcessing } from "../src/emulated-verifier/inbox-processor";
import { connectedUser } from "../src/emulated-verifier/connect";
import { enumerateMyInboxes, myInboxes } from "../src/emulated-verifier/branch-registers";
import { setOpenTimeoutForTests } from "../src/emulated-verifier/open-repo";
import { bootPage, forgetEverything, signIn, type FakeWallet, type Quad } from "./wallet-fake";
import type { Nuri } from "../src/model/types";

const SHIM = "urn:ng-eventually:shim";
const INBOX = "urn:ng-eventually:inbox";

/** The reactive fake: a repo answers nothing until subscribed, and a commit pushes. */
const COLD = { unsyncedUntilSubscribed: true } as const;

let quads: Quad[];
let fake: FakeWallet;

function boot(): void {
  quads = [];
  fake = bootPage(quads, COLD);
}

/**
 * Let the page's pushes land, and everything they set off finish.
 *
 * Not a sleep with a number on it: each round YIELDS so the fake can deliver the push it
 * queued (on a macrotask, as the real RPC does), then WAITS on the work that push actually
 * started (`observationSettled` — the enumerations in flight and the drains behind them).
 * Several rounds because applying one push can queue the next: a Link filed on the private
 * store pushes to the register subscription, which re-enumerates.
 */
async function converge(): Promise<void> {
  for (let round = 0; round < 5; round += 1) {
    await new Promise((resolve) => setTimeout(resolve, 0));
    await observationSettled();
  }
}

/**
 * Take what has been written into `graph` OUT of this page's wallet and hand it back — the
 * broker holding a commit it has not delivered yet. Delivering it later
 * (`fake._deliver`) is the only way a deposit made in another session can arrive here
 * while Alice, and not its author, is the connected identity.
 */
function heldByTheBroker(graph: Nuri): Quad[] {
  const held: Quad[] = [];
  for (let i = quads.length - 1; i >= 0; i -= 1) {
    if (quads[i]!.g === graph) held.unshift(...quads.splice(i, 1));
  }
  return held;
}

/**
 * The documents a user has durably been GIVEN — the emulated `AddLink` records on their
 * User branch, named by the document each one opens. The record holds a `ReadCap` (the
 * reference plus its secret); the tests are about WHICH document arrived, so the secret is
 * dropped here rather than pinned to the stand-in value the emulation currently mints.
 */
async function documentsGivenTo(id: string): Promise<string[]> {
  const store = (await resolveAccount(id))?.docPrivate;
  if (!store) return [];
  return quads
    .filter((q) => q.g === store && q.p === `${SHIM}:link`)
    .map((q) => q.o.split(":r:")[0]!);
}

/** How many times this inbox's deposits have been read — i.e. how often it was processed. */
function depositReadsOf(inbox: Nuri): number {
  return fake.sparql_query.mock.calls.filter(
    (c) => c[3] === inbox && String(c[1]).includes(`${INBOX}:payload`),
  ).length;
}

/** The inbox recorded for a note, read off the WALLET — no application can ask the package. */
function inboxOnTheNote(note: Nuri): Nuri {
  const record = quads.find((q) => q.p === `${SHIM}:inboxCap` && q.o.startsWith(note + " "));
  if (!record) throw new Error("no AddInboxCap record was written for the note");
  return record.o.split(" ")[1] as Nuri;
}

/** Capture what `console.error` is told while `body` runs. */
async function whileWatchingTheLog(body: () => Promise<void>): Promise<string[]> {
  const lines: string[] = [];
  const real = console.error;
  console.error = ((...args: unknown[]) => {
    lines.push(args.map((a) => String(a)).join(" "));
  }) as typeof console.error;
  try {
    await body();
  } finally {
    console.error = real;
  }
  return lines;
}

/**
 * Bob makes a document and shares it with Alice by NAME, then the broker holds his deposit
 * back. Returns the document he shared and the quads still in transit.
 *
 * Alice has to have been here before, and to have DONE something: `inbox.share` refuses a
 * recipient nobody has ever been, and connecting does not provision — an identity acquires
 * its account the first time it creates anything. That is the model and not a fixture
 * detail: upstream a deposit is sealed to an inbox key somebody had to hand you, so you
 * cannot address a name you invented. Her first visit here is not the one under test; every
 * test below connects her again afterwards, and the deposit arrives strictly after that.
 */
async function bobSharesWithAlice(): Promise<{ doc: Nuri; inTransit: Quad[] }> {
  await signIn("alice");
  await storeRegistry.createEntityDoc("protected"); // her first visit — now she exists
  await signIn("bob");
  const doc = await storeRegistry.createEntityDoc("protected");
  await inboxSurface.share(doc, "alice");
  // Test-side inspection only — the address is read off the wallet to say WHICH document
  // is in transit, and is never handed to an actor.
  const aliceInbox = await userInbox("alice", "protected");
  return { doc, inTransit: heldByTheBroker(aliceInbox) };
}

beforeEach(() => {
  forgetEverything();
  boot();
});

afterAll(() => {
  cancelScheduledInboxProcessing();
  forgetEverything();
});

describe("a deposit that arrives while its recipient is connected", () => {
  test("is applied, without anyone reconnecting", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();

    await signIn("alice");
    // The honest baseline: as far as this page is concerned Alice's inbox is empty, so
    // connecting applied nothing. Whatever the next lines prove, they do not prove it twice.
    expect(getCaps().capForHolder("alice", doc)).toBeUndefined();

    fake._deliver(inTransit);
    await converge();

    expect(getCaps().capForHolder("alice", doc)).toBeDefined();
    expect(await documentsGivenTo("alice")).toContain(doc);
  });

  test("is applied DURABLY — the same as if she had reconnected to find it", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();
    await signIn("alice");
    fake._deliver(inTransit);
    await converge();

    // A cap held only in memory is a cap lost at the next reload, and the whole point of
    // applying rather than merely reading is that it survives.
    expect(await documentsGivenTo("alice")).toEqual([doc]);
  });

  test("does not need the depositor's tab to stay open — no timer is involved", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();
    await signIn("alice");
    // Whatever the deposit armed in Bob's session is dropped here, exactly as a closed tab
    // drops it. What follows is the connected owner's own doing, or it does not happen.
    cancelScheduledInboxProcessing();

    fake._deliver(inTransit);
    await converge();

    expect(getCaps().capForHolder("alice", doc)).toBeDefined();
  });
});

describe("an inbox opened in the middle of a session", () => {
  test("is watched too — what lands in it is processed without reconnecting", async () => {
    await signIn("alice");
    // The note and its inbox come into existence AFTER connecting, so nothing the
    // connection enumerated could have included them.
    const note = await storeRegistry.createEntityDoc("public");
    await storeRegistry.openDocumentInbox(note);
    await converge();

    const inbox = inboxOnTheNote(note);
    const readsBefore = depositReadsOf(inbox);

    // A message is left on the NOTE — the depositor names the document, never an address.
    await inboxSurface.postToDocument(note, { payload: { text: "j'apporte le café" }, from: null, ts: 1 });
    await converge();

    // Processing an inbox IS reading its queue and applying what is this library's to
    // apply; for a document inbox nothing is (a Link only ever reaches a person's inbox),
    // so the read is the whole of the consequence — and it can come from nowhere else:
    // depositing reads the shim, not the queue, and the deferred window has not closed.
    expect(depositReadsOf(inbox)).toBeGreaterThan(readsBefore);
  });

  test("the messages left on it are readable, on the document its owner named", async () => {
    await signIn("alice");
    const note = await storeRegistry.createEntityDoc("public");
    await storeRegistry.openDocumentInbox(note);
    await converge();

    await inboxSurface.postToDocument(note, { payload: { text: "à demain" }, from: null, ts: 2 });
    await converge();

    const left = await inboxSurface.readForDocument(note);
    expect(left.map((d) => (d.payload as { text: string }).text)).toEqual(["à demain"]);
  });
});

describe("switching identity", () => {
  test("stops the observation — the previous identity's inbox is no longer applied", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();
    await signIn("alice");

    // Alice steps away and Bob takes the page. A session belongs to one person.
    await signIn("bob");

    fake._deliver(inTransit);
    await converge();

    // Nothing was applied for Alice — she is not connected, and her queue keeps its
    // deposit for the next time she is.
    expect(getCaps().capForHolder("alice", doc)).toBeUndefined();
    expect(await documentsGivenTo("alice")).toEqual([]);
    // …and emphatically nothing was filed for Bob either: work started for one holder must
    // never file for another. (Bob's own cap on the document is not evidence of that — he
    // made it. What would be evidence is a Link, and there is none.)
    expect(await documentsGivenTo("bob")).toEqual([]);
  });

  test("and disconnecting stops it too", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();
    await signIn("alice");

    setCurrentUser(null); // no identity is acting — anonymous holds nothing and owns no inbox

    fake._deliver(inTransit);
    await converge();

    expect(await documentsGivenTo("alice")).toEqual([]);
  });

  test("MID-DRAIN files nothing for the identity that arrives", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();
    // Test-side inspection only: the address is used to recognise the read in flight, and
    // is never handed to an actor.
    const aliceInbox = await userInbox("alice", "protected");
    await signIn("alice");

    // The broker takes its time over the deposits read, and the page switches user INSIDE
    // that window — a person clicking "sign in as Bob" while a push is being applied. The
    // ownership guard has already passed by then; it ran at the start of the read.
    const answering = fake.sparql_query.getMockImplementation()!;
    let switched = false;
    fake.sparql_query.mockImplementation(async (...args: unknown[]) => {
      const answer = await answering(...args);
      if (!switched && args[3] === aliceInbox && String(args[1]).includes(`${INBOX}:payload`)) {
        switched = true;
        setCurrentUser("bob");
      }
      return answer;
    });

    fake._deliver(inTransit);
    await converge();

    // Bob was GIVEN nothing. (His own cap on the document is not evidence either way — he
    // made it; what would be evidence is a Link, and there must be none.) Filed here, the
    // cap addressed to Alice becomes a capability Bob holds at his next sign-in, and its
    // real recipient is left with nothing at all.
    expect(await documentsGivenTo("bob")).toEqual([]);
    // …and Alice has lost nothing: an inbox is not consumed by a drain that abandoned it,
    // so what was deposited for her is still there when she is the one connected.
    await signIn("alice");
    await converge();
    expect(getCaps().capForHolder("alice", doc)).toBeDefined();
  });

  test("and Alice coming back finds the deposit still there to apply", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();
    await signIn("alice");
    await signIn("bob");
    fake._deliver(inTransit);
    await converge();

    // An inbox is not consumed by being ignored: connecting again drains what was left.
    await signIn("alice");
    await converge();

    expect(getCaps().capForHolder("alice", doc)).toBeDefined();
  });
});

describe("a deposit that cannot be applied", () => {
  test("is reported, and stops neither the observation nor the next deposit", async () => {
    const first = await bobSharesWithAlice();
    const second = await bobSharesWithAlice();
    const aliceInbox = await userInbox("alice", "protected");

    await signIn("alice");

    // The broker cannot answer for Alice's inbox — the deposit lands, applying it does not.
    fake._failReadsOn.add(aliceInbox);
    const reported = await whileWatchingTheLog(async () => {
      fake._deliver(first.inTransit);
      await converge();
    });

    expect(reported.filter((l) => /could not apply what is in this inbox/.test(l)).length)
      .toBeGreaterThan(0);
    // Prefixed by the connected identity, like every other polyfill-layer line.
    expect(reported.find((l) => /could not apply what is in this inbox/.test(l)))
      .toContain("[alice][polyfill]");
    expect(getCaps().capForHolder("alice", first.doc)).toBeUndefined();

    // The broker recovers. The observation is still running — one unapplicable item denies
    // nothing — and the deposit that failed was never consumed, so both land now.
    fake._failReadsOn.delete(aliceInbox);
    fake._deliver(second.inTransit);
    await converge();

    expect(getCaps().capForHolder("alice", second.doc)).toBeDefined();
    expect(getCaps().capForHolder("alice", first.doc)).toBeDefined();
  });

  test("because its inbox could not be WATCHED is reported, and the next signal re-attempts it", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();
    const aliceInbox = await userInbox("alice", "protected");

    // A repo that never pushes its initial `State` is what a refused subscription looks like
    // from the bootstrap open's side, and it waits out its bounded fallback before giving up.
    // Eight seconds of it, twice, is the production wait and not a unit test's.
    setOpenTimeoutForTests(20);

    // The broker will not open a channel on Alice's inbox. Everything else about her session
    // works — which is the point: the only symptom of a watch that was never established is
    // that shares stop arriving.
    const opening = fake.doc_subscribe!.getMockImplementation()!;
    let refusing = true;
    fake.doc_subscribe!.mockImplementation(async (...args: unknown[]) => {
      if (refusing && args[0] === aliceInbox) throw new Error(`RepoNotFound: ${String(args[0])}`);
      return opening(...args);
    });

    const reported = await whileWatchingTheLog(async () => {
      await signIn("alice");
      await converge();
    });

    // Said out loud, in this package's own words and under the connected identity — not left
    // as the absence of a push.
    expect(reported.filter((l) => /could not be watched/.test(l)).length).toBeGreaterThan(0);
    expect(reported.find((l) => /could not be watched/.test(l))).toContain("[alice][polyfill]");

    // Nothing is watching, so the deposit that lands now cannot be applied — and is not.
    fake._deliver(inTransit);
    await converge();
    expect(getCaps().capForHolder("alice", doc)).toBeUndefined();

    // The broker recovers and Alice does something ordinary — which is a SIGNAL, not a
    // coincidence: creating anything files caps, and the held-caps channel re-enters the
    // enumeration. That is the whole of the repair, and it is deliberately the whole of it:
    // a re-attempt fired from the rejection itself asks the broker that has just refused, in
    // the same turn, with nothing having changed. The inbox that could not be opened was not
    // written off for the session — the failed entry is forgotten, so this enumeration
    // subscribes again as if it had never been attempted, and the initial push of that new
    // subscription finds the deposit still waiting.
    refusing = false;
    await storeRegistry.createEntityDoc("protected");
    await converge();
    expect(getCaps().capForHolder("alice", doc)).toBeDefined();
    expect(await documentsGivenTo("alice")).toContain(doc);
  });

  test("never rejects into the application — nobody asked for this work", async () => {
    const { inTransit } = await bobSharesWithAlice();
    const aliceInbox = await userInbox("alice", "protected");
    await signIn("alice");
    fake._failReadsOn.add(aliceInbox);

    const unhandled: unknown[] = [];
    const onUnhandled = (e: unknown): void => {
      unhandled.push(e);
    };
    process.on("unhandledRejection", onUnhandled);
    try {
      await whileWatchingTheLog(async () => {
        fake._deliver(inTransit);
        await converge();
      });
    } finally {
      process.off("unhandledRejection", onUnhandled);
      fake._failReadsOn.delete(aliceInbox);
    }
    expect(unhandled).toEqual([]);
  });
});

describe("a connection whose own work FAILED", () => {
  /**
   * Aimed at the PRIVATE store, and that is the whole test.
   *
   * It used to fail reads on `docPublic`, which makes the restore reject and leaves the
   * enumeration of the inboxes untouched — so it proved that watching survives a failure that
   * was never going to threaten it. The private store is the one the connection restores from
   * AND the register that says which inboxes exist, so failing it is the case that actually
   * decides: listing the inboxes reads it, and one throw used to discard the two user inboxes
   * that had ALREADY been listed before it. The identity was then connected with nothing
   * watched at all, and a person who only reads — who never creates anything, so never fires
   * a signal — had no way back for the rest of the session.
   */
  test("still leaves the identity watched — including when the failing store is the register", async () => {
    const { doc, inTransit } = await bobSharesWithAlice();
    // The broker cannot answer for Alice's private store, so the RESTORE fails and the
    // connection rejects. She is connected regardless: `setCurrentUser` is synchronous and
    // took effect before any of this ran, and nothing signs her back out.
    const store = (await resolveAccount("alice"))?.docPrivate;
    if (store === undefined) throw new Error("the fixture did not give Alice a private store");
    fake._failReadsOn.add(store);

    setCurrentUser("alice");
    let rejected = false;
    const reported = await whileWatchingTheLog(async () => {
      try {
        await connectedUser();
      } catch {
        rejected = true;
      }
      await converge();
    });
    // The caller is still TOLD, and that rule is not what changes here: failing to reach the
    // registers rejects, exactly as before.
    expect(rejected).toBe(true);
    // …and so is the log, about the half of the list that could not be read. A short list
    // that says nothing is a failure wearing the face of an absence, which is the one thing
    // this package will not do — the inboxes it names are watched, the ones it does not are
    // owed a next enumeration, and both facts have to be legible.
    expect(reported.filter((l) => /could not all be listed/.test(l)).length).toBeGreaterThan(0);
    expect(reported.find((l) => /could not all be listed/.test(l))).toContain("[alice][polyfill]");

    // The hiccup passes. Alice never touched the page — no sign-in, no document created,
    // nothing that could stand in for the watching she is owed.
    fake._failReadsOn.delete(store);
    fake._deliver(inTransit);
    await converge();

    // What she is owed is not the restore she lost — it is that a deposit made while she sits
    // there converges. Her own two inboxes are where a share addressed to her by NAME lands,
    // and they were listed before the register threw; watching them is what makes this
    // session behave like every other one.
    expect(getCaps().capForHolder("alice", doc)).toBeDefined();
    expect(await documentsGivenTo("alice")).toContain(doc);
  });

});

/**
 * The list the observation works from, asked directly.
 *
 * It is built from two independent registers — the account record, which names the user's own
 * two store inboxes, and the User branch, which names one per document it opened an inbox on
 * — and the two fail independently. What a caller may do with a half-read list depends
 * entirely on being TOLD it is half-read, so both halves of that answer are pinned here
 * rather than only through the behaviour above.
 */
describe("listing the inboxes when one of the two registers cannot be read", () => {
  test("comes back as what WAS listed plus the failure — never as a short list", async () => {
    await signIn("alice");
    // A document inbox: a record on the User branch of the private store, which is the
    // register the broker is about to stop answering for.
    const note = await storeRegistry.createEntityDoc("public");
    await storeRegistry.openDocumentInbox(note);
    await converge();
    const store = (await resolveAccount("alice"))?.docPrivate;
    if (store === undefined) throw new Error("the fixture did not give Alice a private store");

    const whole = await enumerateMyInboxes();
    expect(whole.incomplete).toBeNull();
    expect(whole.inboxes).toContain(inboxOnTheNote(note));

    fake._failReadsOn.add(store);
    const partial = await enumerateMyInboxes();
    fake._failReadsOn.delete(store);

    // Her own two inboxes were in hand before the second register threw. Discarding them
    // with it is what left an identity connected with ZERO inboxes watched.
    expect(partial.inboxes).toEqual([
      await userInbox("alice", "public"),
      await userInbox("alice", "protected"),
    ]);
    // …and the shortfall travels WITH them: an answer that came back short while looking
    // complete is a failure disguised as an absence, which is the fault this package keeps
    // closing. The document inbox is missing from the list and that fact is legible.
    expect(partial.incomplete).not.toBeNull();
    expect(String(partial.incomplete?.error)).toContain("RepoNotFound");
    expect(partial.inboxes).not.toContain(inboxOnTheNote(note));
  });

  test("still REJECTS for the caller that cannot use a partial list", async () => {
    await signIn("alice");
    await storeRegistry.openDocumentInbox(await storeRegistry.createEntityDoc("public"));
    await converge();
    const store = (await resolveAccount("alice"))?.docPrivate;
    if (store === undefined) throw new Error("the fixture did not give Alice a private store");

    fake._failReadsOn.add(store);
    try {
      // `connect.connectedUser` drains this list, and a queue missing from it is a delivered
      // share silently never applied. Not knowing which queues exist is the session failing
      // to establish, and that contract is not what the partial answer above relaxes.
      await expect(myInboxes()).rejects.toThrow(/RepoNotFound/);
    } finally {
      fake._failReadsOn.delete(store);
    }
  });
});