/** * 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 { 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 { 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 { 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): Promise { 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 attempted again", 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. The inbox that could not be // opened was not written off for the session: it is subscribed to on the next // enumeration, and its initial push 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", () => { test("still leaves the identity watched — being connected is what is observed", async () => { const { doc, inTransit } = await bobSharesWithAlice(); // The broker cannot answer for one of Alice's own stores, 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"))?.docPublic; if (store === undefined) throw new Error("the fixture did not give Alice a public store"); fake._failReadsOn.add(store); setCurrentUser("alice"); let rejected = false; 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); // The hiccup passes. Alice never touched the page. 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. Watching used to be the LAST line of the connection work, so a restore // that rejected skipped it and left her connected with nothing observing her inboxes: one // hiccup at sign-in, and every share made afterwards was lost to her for the session. expect(getCaps().capForHolder("alice", doc)).toBeDefined(); expect(await documentsGivenTo("alice")).toContain(doc); }); });