/** * The inbox-processing service this deployment has not got, emulated by a timer. * * A deposit used to sit in an inbox until its owner next connected — hours, on a page * nobody reloads. `emulated-verifier/inbox-processor.ts` arms a one-shot timer on the * TARGET's inbox instead, so the deposit converges while the depositor's session is still * the one in the page. That is identity usurpation, and every test here is about the two * things that make it acceptable: the result lands in the OWNER's space and nowhere else, * and the depositor's session comes out of it exactly as it went in. * * ── What the actors hand each other, and what they do not ───────────────── * Nobody is handed an inbox address. Each depositor calls `inbox.share(doc, toUser)` and * names a document and a person, which is all an application has; the address is resolved * inside. The one NURI the TEST resolves for itself (`userInbox`) is used only to count * reads and to inspect what landed — never given to an actor. * * ── Time is closed, not slept through ──────────────────────────────────── * `runScheduledInboxProcessingNow()` fires the windows a deposit has ALREADY armed — the * same runs, the same holder, the same effects, minus the twenty seconds. It cannot name an * inbox, so it fabricates nothing: every state asserted below is one the timer produces on * its own. */ import { test, expect, mock, afterEach } from "bun:test"; import * as polyfill from "../src/index"; import { configure } from "../src/index"; import { configureStoreRegistry, getCaps, resetCaps, resetConfig, resetStoreRegistry, setCurrentUser, getCurrentUser, } from "../src/shared-wallet/bootstrap"; import { createEntityDoc, resetRegistryCache, resolveAccount, resolveWriteGraph, userInbox, } from "../src/shared-wallet/account-registry"; import type { RegistrySession } from "../src/shared-wallet/account-registry"; import { connectedUser } from "../src/emulated-verifier/connect"; import { cancelScheduledInboxProcessing, runScheduledInboxProcessingNow, } from "../src/emulated-verifier/inbox-processor"; import { resetOpenedRepos } from "../src/emulated-verifier/open-repo"; import { share } from "../src/surface/inbox"; import { readUnion } from "../src/surface/read-model"; import { sparqlUpdate } from "../src/surface/docs"; import type { Nuri } from "../src/model/types"; const SESSION: RegistrySession = { sessionId: "sid-deferred", privateStoreId: "PRIV-DEFERRED" }; const SHIM = "urn:ng-eventually:shim"; const INBOX = "urn:ng-eventually:inbox"; const SECRET = "urn:e2e:secret"; interface Quad { g: string; s: string; p: string; o: string } /** What the broker refuses to do. Mutable after `inject`, so a test builds a healthy world * first and breaks only the one call it is about — the fault is the broker's, never a * reach into the library to make one of its own functions reject. */ interface Faults { /** The anchored read of THIS inbox document's deposits throws. */ depositsRead: Nuri | null; } function unescapeLiteral(s: string): string { let out = ""; for (let i = 0; i < s.length; i++) { if (s[i] === "\\" && i + 1 < s.length) { const next = s[++i]; out += next === "n" ? "\n" : next === "r" ? "\r" : next === "t" ? "\t" : next!; } else out += s[i]; } return out; } /** A stateful fake `ng`: the shim SPARQL, the inbox SPARQL, and the anchored per-doc read. */ function makeFakeNg(faults: Faults) { const quads: Quad[] = []; let docCounter = 0; const doc_create = mock(async () => `did:ng:o:doc${++docCounter}`); const sparql_update = mock(async (...a: unknown[]) => { const query = a[1] as string; const anchor = a[2] as string | undefined; if (!anchor) return undefined; const del = query.match(/^\s*DELETE\s+WHERE\s*\{\s*<([^>]+)>\s+<([^>]+)>\s+\?/); if (del) { const [s0, p0] = [del[1]!, del[2]!]; for (let i = quads.length - 1; i >= 0; i--) { const q = quads[i]!; if (q.g === anchor && q.s === s0 && q.p === p0) quads.splice(i, 1); } return undefined; } const gm = query.match(/GRAPH\s+<([^>]+)>\s*\{([\s\S]*)\}/); const body = gm ? gm[2]! : query.replace(/^\s*INSERT DATA\s*\{/, "").replace(/\}\s*$/, ""); const sm = body.match(/<([^>]+)>/); if (!sm) return undefined; const s = sm[1]!; const pairRe = /(?:a|<([^>]+)>)\s+(?:"((?:[^"\\]|\\.)*)"|<([^>]+)>)/g; let m: RegExpExecArray | null; const after = body.slice(body.indexOf(sm[0]) + sm[0].length); while ((m = pairRe.exec(after)) !== null) { const p = m[1] ?? (query.includes(`${INBOX}:Deposit`) ? `${INBOX}:Deposit` : `${SHIM}:Account`); const o = m[2] !== undefined ? unescapeLiteral(m[2]) : (m[3] ?? ""); quads.push({ g: anchor, s, p, o }); } return undefined; }); const sparql_query = mock(async (...a: unknown[]) => { const query = a[1] as string; const anchor = a[3] as string | undefined; const byPred = (pred: string, v: string) => ({ results: { bindings: quads .filter((q) => q.g === anchor && q.p === pred) .map((q) => ({ [v]: { value: q.o } })), }, }); if (query.includes(`<${SHIM}:shimDoc>`)) return byPred(`${SHIM}:shimDoc`, "shimDoc"); if (query.includes(`<${SHIM}:id>`)) { const subjM = query.match(/<([^>]+)>\s+a\s+/); const only = subjM ? subjM[1]! : null; const bySubject = new Map>(); for (const q of quads) { if (q.g !== anchor) continue; if (only !== null && q.s !== only) continue; const rec = bySubject.get(q.s) ?? {}; if (q.p === `${SHIM}:id`) rec.id = q.o; if (q.p === `${SHIM}:docPublic`) rec.docPublic = q.o; if (q.p === `${SHIM}:docProtected`) rec.docProtected = q.o; if (q.p === `${SHIM}:docPrivate`) rec.docPrivate = q.o; bySubject.set(q.s, rec); } return { results: { bindings: [...bySubject.values()].filter((r) => r.id).map((r) => ({ id: { value: r.id! }, docPublic: { value: r.docPublic ?? "" }, docProtected: { value: r.docProtected ?? "" }, docPrivate: { value: r.docPrivate ?? "" }, })), }, }; } if (query.includes(`<${INBOX}:payload>`)) { // The one failure this fake can inject, at the platform boundary: the anchored read // of an inbox document throws, exactly as the engine hard-errors for a repo it // cannot resolve (`RepoNotFound`, `engine/verifier/src/request_processor.rs:264`). if (faults.depositsRead !== null && faults.depositsRead === anchor) { throw new Error("RepoNotFound"); } const bySubject = new Map>(); for (const q of quads) { if (q.g !== anchor) continue; const rec = bySubject.get(q.s) ?? {}; if (q.p === `${INBOX}:payload`) rec.payload = q.o; if (q.p === `${INBOX}:ts`) rec.ts = q.o; if (q.p === `${INBOX}:from`) rec.from = q.o; bySubject.set(q.s, rec); } return { results: { bindings: [...bySubject.values()] .filter((r) => r.payload !== undefined && r.ts !== undefined) .map((r) => { const row: Record = { payload: { value: r.payload! }, ts: { value: r.ts! }, }; if (r.from !== undefined) row.from = { value: r.from }; return row; }), }, }; } if (query.includes(`<${SHIM}:inboxCap>`)) return byPred(`${SHIM}:inboxCap`, "c"); if (query.includes(`<${SHIM}:inboxAddress>`)) return byPred(`${SHIM}:inboxAddress`, "a"); if (query.includes(`<${SHIM}:readCap>`)) return byPred(`${SHIM}:readCap`, "c"); if (query.includes(`<${SHIM}:link>`)) return byPred(`${SHIM}:link`, "c"); if (query.includes(`${SHIM}:isInbox`)) return byPred(`${SHIM}:isInbox`, "i"); // Shim `inboxOwner` SELECT — WHOSE inbox a NURI is, the emulated routing entry the // deferred service resolves its holder from. Keyed by the INBOX (the subject), as // upstream's `inboxes: PubKey → RepoId` is keyed by the inbox's pubkey. if (query.includes(`<${SHIM}:inboxOwner>`)) { const sm = query.match(/<([^>]+)>\s+/); const subj = sm ? sm[1]! : null; return { results: { bindings: quads .filter((q) => q.g === anchor && q.p === `${SHIM}:inboxOwner` && q.s === subj) .map((q) => ({ u: { value: q.o } })), }, }; } if (query.includes(`${SHIM}:docInbox`)) { const pm = query.match(/<(urn:ng-eventually:shim:docInbox:[a-z]+)>/); const pred = pm ? pm[1]! : ""; const sm = query.match(/<([^>]+)>\s+ q.g === anchor && q.p === pred && (subj === null || q.s === subj)) .map((q) => ({ d: { value: q.o } })), }, }; } if (query.includes(`<${SHIM}:exposedReadCap>`)) return byPred(`${SHIM}:exposedReadCap`, "c"); if (query.includes(`<${SHIM}:contains>`)) return byPred(`${SHIM}:contains`, "e"); return { results: { bindings: quads .filter((q) => q.g === anchor) .map((q) => ({ s: { value: q.s }, p: { value: q.p }, o: { value: q.o } })), }, }; }); return { doc_create, sparql_update, sparql_query, quads }; } let fake: ReturnType; let faults: Faults; function inject() { faults = { depositsRead: null }; fake = makeFakeNg(faults); configure({ ng: fake as never, useShape: (() => {}) as never }); configureStoreRegistry({ getSession: async () => SESSION }); resetRegistryCache(); resetOpenedRepos(); resetCaps(); setCurrentUser(null); } afterEach(() => { cancelScheduledInboxProcessing(); resetConfig(); resetStoreRegistry(); resetCaps(); resetRegistryCache(); resetOpenedRepos(); setCurrentUser(null); }); /** * Bob has been in the page once — which is what makes him someone a share can NAME * (`inbox.share` refuses a recipient nobody has ever signed in as). Nothing of his crosses * to the depositors afterwards: they name the person, never his address. */ async function bobSignsInOnce(): Promise { setCurrentUser("bob"); await resolveWriteGraph("bob", "protected"); setCurrentUser(null); } /** Write one triple into `doc`, as the consumer's write path would. */ async function write(doc: Nuri, p: string, o: string): Promise { await sparqlUpdate(SESSION.sessionId, `INSERT DATA { <${doc}> <${p}> "${o}" }`, doc, "test"); } /** The values `p` carries in `docs`, as the CURRENT holder reads them. */ async function readValues(docs: Nuri[], p: string): Promise { const subjects = await readUnion(docs); return subjects.flatMap((s) => s.props[p] ?? []); } /** How many times an inbox document's deposits have been read — one per drain. */ function depositReadsOf(inbox: Nuri): number { return fake.sparql_query.mock.calls.filter( (c) => c[3] === inbox && String(c[1]).includes(`${INBOX}:payload`), ).length; } /** The Links durably filed on a user's User branch — the emulated `AddLink` records. */ async function linksFiledFor(id: string): Promise { const record = await resolveAccount(id); const store = record?.docPrivate; if (!store) return []; return fake.quads.filter((q) => q.g === store && q.p === `${SHIM}:link`).map((q) => q.o); } // --- the drain acts for the OWNER, and touches nothing of the depositor's --- test("a deposit is drained for the inbox's OWNER, while the depositor still holds the session", async () => { inject(); await bobSignsInOnce(); // Carol shares one of her documents with Bob. Her deposit is in Bob's inbox, unapplied. setCurrentUser("carol"); const carolDoc = await createEntityDoc("carol", "protected"); await write(carolDoc, SECRET, "carol's content"); await share(carolDoc, "bob"); // Alice, later, shares one of hers with Bob too. Neither of them is ever handed an // address: `share` names a document and a person. setCurrentUser("alice"); const aliceDoc = await createEntityDoc("alice", "protected"); await write(aliceDoc, SECRET, "alice's content"); await share(aliceDoc, "bob"); // Before the window closes: nothing has been applied for anyone. const caps = getCaps(); expect(caps.capForHolder("bob", carolDoc)).toBeUndefined(); expect(caps.capForHolder("bob", aliceDoc)).toBeUndefined(); // The service picks the deposits up. Alice is the connected identity throughout. await runScheduledInboxProcessingNow(); // The session is untouched — and this is the assertion that matters, because the drain // read an inbox holding a Link for CAROL's document. Had it filed against whoever was // connected, Alice would now hold a capability nobody gave her. expect(getCurrentUser()).toBe("alice"); expect(caps.capForHolder("alice", carolDoc)).toBeUndefined(); expect(await readValues([carolDoc], SECRET)).toEqual([]); expect(await linksFiledFor("alice")).toEqual([]); // …and both Links landed in Bob's space instead. expect(caps.capForHolder("bob", carolDoc)).toBeDefined(); expect(caps.capForHolder("bob", aliceDoc)).toBeDefined(); expect((await linksFiledFor("bob")).length).toBe(2); }); test("what the drain filed for the owner is DURABLE: he restores it with his inbox emptied", async () => { inject(); await bobSignsInOnce(); setCurrentUser("alice"); const aliceDoc = await createEntityDoc("alice", "protected"); await write(aliceDoc, SECRET, "the-protected-content"); await share(aliceDoc, "bob"); await runScheduledInboxProcessingNow(); // EMPTY the inbox, as a consumed queue would be, and drop every in-memory cap — so what // Bob recovers below can only have come from his own User branch, where the drain filed // it. Without that, he would simply be draining the queue himself on connection, which // is the behaviour this service exists to get ahead of. const bobInbox = await userInbox("bob", "protected"); for (let k = fake.quads.length - 1; k >= 0; k--) { if (fake.quads[k]!.g === bobInbox) fake.quads.splice(k, 1); } resetCaps(); setCurrentUser("alice"); await createEntityDoc("alice", "private"); // re-arms the emulation: a cap exists again setCurrentUser("bob"); expect(getCaps().capFor(aliceDoc)).toBeUndefined(); // he holds nothing yet await connectedUser(); expect(getCaps().capFor(aliceDoc)).toBeDefined(); expect(await readValues([aliceDoc], SECRET)).toEqual(["the-protected-content"]); }); // --- coalescing -------------------------------------------------------------- test("several deposits inside the window produce exactly ONE drain, and lose nothing", async () => { inject(); await bobSignsInOnce(); setCurrentUser("alice"); const docs: Nuri[] = []; for (const marker of ["one", "two", "three"]) { const doc = await createEntityDoc("alice", "protected"); await write(doc, SECRET, marker); docs.push(doc); } const bobInbox = await userInbox("bob", "protected"); const readsBefore = depositReadsOf(bobInbox); for (const doc of docs) await share(doc, "bob"); // The window is open, not running: depositing processes nothing by itself. expect(depositReadsOf(bobInbox)).toBe(readsBefore); await runScheduledInboxProcessingNow(); // ONE pass over the queue for three deposits. Two passes would race each other on the // very writes the pass makes (`addLink`, on the owner's User branch). expect(depositReadsOf(bobInbox)).toBe(readsBefore + 1); // …and coalescing loses nothing: the one pass reads the whole queue. for (const doc of docs) expect(getCaps().capForHolder("bob", doc)).toBeDefined(); expect((await linksFiledFor("bob")).length).toBe(3); }); test("a deposit after the window has closed arms a NEW one", async () => { inject(); await bobSignsInOnce(); setCurrentUser("alice"); const first = await createEntityDoc("alice", "protected"); const second = await createEntityDoc("alice", "protected"); const bobInbox = await userInbox("bob", "protected"); const readsBefore = depositReadsOf(bobInbox); await share(first, "bob"); await runScheduledInboxProcessingNow(); await share(second, "bob"); await runScheduledInboxProcessingNow(); expect(depositReadsOf(bobInbox)).toBe(readsBefore + 2); expect(getCaps().capForHolder("bob", second)).toBeDefined(); }); // --- failure ---------------------------------------------------------------- test("a drain that fails says so in the package's log, and rejects into nobody", async () => { inject(); await bobSignsInOnce(); setCurrentUser("alice"); const doc = await createEntityDoc("alice", "protected"); await write(doc, SECRET, "content"); const bobInbox = await userInbox("bob", "protected"); // The deposit itself must still succeed — what breaks is the drain that follows it. await share(doc, "bob"); faults.depositsRead = bobInbox; const errors: string[] = []; const realError = console.error; console.error = ((...args: unknown[]) => { errors.push(args.map((a) => String(a)).join(" ")); }) as typeof console.error; try { // Resolves. The application never asked for this work, so it must not be handed a // rejection for it — and an unhandled one would take the runtime down. await expect(runScheduledInboxProcessingNow()).resolves.toBeUndefined(); } finally { console.error = realError; } const reported = errors.filter((line) => /deferred inbox processing failed/.test(line)); expect(reported.length).toBe(1); // Prefixed by the CONNECTED identity, like every other polyfill-layer line — which is // what makes a drain running under someone else's session legible in a live trace. expect(reported[0]).toContain("[alice][polyfill]"); expect(reported[0]).toContain("RepoNotFound"); // Nothing was applied on a drain that could not read, and nothing was applied for the // depositor either: a failure leaves the deposit where it was, waiting for its owner. expect(getCaps().capForHolder("bob", doc)).toBeUndefined(); expect(await linksFiledFor("alice")).toEqual([]); }); test("an inbox the shim records no owner for is reported, not drained for whoever is connected", async () => { inject(); await bobSignsInOnce(); setCurrentUser("alice"); const doc = await createEntityDoc("alice", "protected"); await write(doc, SECRET, "content"); const bobInbox = await userInbox("bob", "protected"); await share(doc, "bob"); // A wallet written before inboxes carried a routing entry — the record simply is not // there. Drained for the connected identity, this is exactly how a capability addressed // to Bob ends up in Alice's hands, so it has to refuse and say so. for (let k = fake.quads.length - 1; k >= 0; k--) { if (fake.quads[k]!.p === `${SHIM}:inboxOwner`) fake.quads.splice(k, 1); } resetRegistryCache(); // …and the session's memo of it goes too const errors: string[] = []; const realError = console.error; console.error = ((...args: unknown[]) => { errors.push(args.map((a) => String(a)).join(" ")); }) as typeof console.error; try { await runScheduledInboxProcessingNow(); } finally { console.error = realError; } expect(errors.some((l) => /records no owner for this inbox/.test(l))).toBe(true); expect(getCaps().capForHolder("alice", doc)).toBeDefined(); // hers, she created it expect(await linksFiledFor("alice")).toEqual([]); // and nothing of Bob's was filed for her expect(depositReadsOf(bobInbox)).toBe(0); // the queue was never even read }); // --- the surface ------------------------------------------------------------ test("nothing about the deferred service reaches the published surface", () => { // The published entry, as an application really sees it at runtime. A "process now", a // delay knob, or any way to name another user's inbox would show up here — and each of // them would publish a call that processes somebody else's queue, which is the one thing // the shared-wallet emulation cannot survive. expect(Object.keys(polyfill).sort()).toEqual([ "configure", "docChangeType", "docs", "ensureIdentity", "inbox", "init", "initNg", "ng", "readUnion", "storeRegistry", "subscribeDoc", "subscribeDocs", "useShape", "watchShape", ]); expect(Object.keys(polyfill.inbox).sort()).toEqual([ "post", "postToDocument", "processInbox", "read", "readForDocument", "readSynced", "share", "watch", ]); expect(Object.keys(polyfill.docs).sort()).toEqual([ "docCreate", "sparqlQuery", "sparqlUpdate", ]); expect(Object.keys(polyfill.storeRegistry).sort()).toEqual([ "createEntityDoc", "listMyEntityDocs", "openDocumentInbox", "resolveScopeGraph", "resolveWriteGraph", ]); });