ng-eventually/packages/polyfill/test/deferred-inbox-processing.test.ts

/**
 * 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 { readDocumentSubjects } from "../src/surface/read-document";
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;
    }
    // Retiring a DEPOSIT: every triple of one subject, whatever predicates it carries
    // (`emulated-verifier/inbox-consume.removeDeposit`). Modelled rather than ignored — a
    // fake that let this delete fall through would leave every deposit in place, and a
    // suite asserting convergence would pass on a queue that never shrinks.
    const delDeposit = query.match(/^\s*DELETE\s+WHERE\s*\{\s*<([^>]+)>\s+\?\w+\s+\?\w+\s*\}/);
    if (delDeposit) {
      const subject = delDeposit[1]!;
      for (let i = quads.length - 1; i >= 0; i--) {
        const q = quads[i]!;
        if (q.s === subject) 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+<urn:ng-eventually:shim:Account>/);
      const only = subjM ? subjM[1]! : null;
      const bySubject = new Map<string, Record<string, string>>();
      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<string, Record<string, string>>();
      for (const q of quads) {
        if (q.g !== anchor) continue;
        const rec = bySubject.get(q.s) ?? {};
        // The deposit's own subject — the key a drain names to retire it.
        rec.d = 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<string, { value: string }> = {
                d: { value: r.d! },
                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+<urn:ng-eventually:shim:inboxOwner>/);
      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+<urn:ng-eventually:shim:docInbox/);
      const subj = sm ? sm[1]! : null;
      return {
        results: {
          bindings: quads
            .filter((q) => 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<typeof makeFakeNg>;
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<void> {
  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<void> {
  await sparqlUpdate(SESSION.sessionId, `INSERT DATA { <${doc}> <${p}> "${o}" }`, doc, "test");
}

/** The values `p` carries in `doc`, as the CURRENT holder reads them - one document,
 *  one anchored query, which is all `readUnion` ever did per entry. */
async function readValues(doc: Nuri, p: string): Promise<string[]> {
  const subjects = await readDocumentSubjects(doc);
  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<string[]> {
  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();
  // …and the read says so out loud. It used to answer `[]`, because `readUnion` dropped a
  // document it could not attempt; the anchored read refuses it instead, which asserts the
  // same fact more strongly - Alice is not merely shown nothing, she is told no.
  await expect(readValues(carolDoc, SECRET)).rejects.toThrow(/does not hold this document/i);
  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();

  // The inbox is ALREADY empty — the drain consumed the deposit as it applied it. Until
  // 2026-08-26 this test had to empty the queue by hand to reach that state, which is
  // exactly the shortcut that fabricates a state the system never produced; now it is
  // asserted instead. It matters here because what Bob recovers below must come from his
  // own User branch, where the drain filed it: with the deposit still in the queue he
  // would simply be draining it himself on connection, which is the behaviour this
  // service exists to get ahead of.
  const bobInbox = await userInbox("bob", "protected");
  expect(fake.quads.filter((q) => q.g === bobInbox && q.p === `${INBOX}:payload`)).toEqual([]);
  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;
  }

  // The wording is the SHARED reporter's since 2026-08-17 (`emulated-verifier/inbox-drain.ts`):
  // the timer is no longer the only thing that drains an inbox — the continuous observation
  // does too, through the same queue — so the line names the act and not the schedule. Since
  // 2026-08-26 it also stops promising a retry it cannot make: a deposit already taken is
  // consumed, and only what is STILL QUEUED waits for the next attempt.
  const reported = errors.filter((line) => /could not finish draining this inbox/.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",
    // The ORM's one-shot read and its subscription-free write, WRAPPED (`surface/orm.ts`).
    // Neither reaches another identity's queue: the first is filtered by what the holder
    // holds, the second is refused unless the holder owns the document it names.
    "getObjects",
    "inbox",
    "init",
    "initNg",
    "insertObject",
    "ng",
    "storeRegistry",
    "subscribeDoc",
    "subscribeDocs",
    "useShape",
    "useShapeQuery",
  ]);
  expect(Object.keys(polyfill.inbox).sort()).toEqual([
    "post",
    "postToDocument",
    "processInbox",
    "read",
    "readForDocument",
    "readSynced",
    // Added 2026-08-17. It names a DOCUMENT, like every other door an application has
    // here: the two reads it composes were published separately, so reaching the synced
    // one meant holding an inbox address. It processes nobody else's queue — the owner
    // guard is `readSynced`'s, unchanged.
    "readSyncedForDocument",
    "share",
    "watch",
  ]);
  expect(Object.keys(polyfill.docs).sort()).toEqual([
    "docCreate",
    "sparqlQuery",
    "sparqlUpdate",
  ]);
  // `1.0.0-dev.5`: `listMyEntityDocs`, `resolveScopeGraph` and `resolveWriteGraph` left,
  // and the three store accessors arrived in their place — an application names a store by
  // the reference one of them hands it, never by a word this package made up.
  expect(Object.keys(polyfill.storeRegistry).sort()).toEqual([
    "createEntityDoc",
    "openDocumentInbox",
    "privateStore",
    "protectedStore",
    "publicStore",
  ]);
});