ng-eventually/packages/polyfill/test/inbox-consumption.test.ts

/**
 * An inbox is a QUEUE, and this suite is about the act that had been missing from it: a
 * deposit is delivered once and then it is gone.
 *
 * ── What each test would have looked like before ──────────────────────────
 * Until 2026-08-26 this package never consumed anything. Every drain re-read everything
 * ever deposited, which produced three symptoms at once: an inbox that grew without bound,
 * a malformed deposit re-reported at every push and every connection for as long as the
 * page lived, and an idempotence burden on every arm that existed only to survive the
 * replay. Each test below fails against that code — the deposit is still there — and fails
 * again if the removal is taken back out.
 *
 * ── Upstream is the specification, and so is its ORDER ────────────────────
 * The broker POPS: `take_first_msg` → `MSGS.take_first_value` is `get_first_key_value`
 * followed by `del`, so the message leaves the store BEFORE it is delivered; the verifier
 * then applies it and a failure is logged and swallowed, never re-enqueued
 * (`engine/verifier/src/verifier.rs`). Remove, apply, log, move on. So a deposit whose arm
 * refused is gone exactly like one that succeeded, and that is the point of the second test
 * below rather than an accident of it.
 *
 * ── Nobody is handed an inbox address ─────────────────────────────────────
 * A depositor names a document and a person (`inbox.share`, `inbox.postToDocument`); the
 * address is resolved inside. The addresses the TEST resolves are read off the wallet and
 * used only to count what is left in a queue — never given to an actor.
 */
import { test, expect, describe, afterAll, beforeEach } from "bun:test";
import { inbox as inboxSurface, storeRegistry } from "../src/index";
import { resolveAccount, userInbox } from "../src/shared-wallet/account-registry";
import { setCurrentUser } from "../src/shared-wallet/bootstrap";
import { openDocumentInbox } from "../src/emulated-verifier/branch-registers";
import {
  cancelScheduledInboxProcessing,
  runScheduledInboxProcessingNow,
} from "../src/emulated-verifier/inbox-processor";
import { bootPage, forgetEverything, signIn, type FakeWallet, type Quad } from "./wallet-fake";
import type { Nuri } from "../src/model/types";

const INBOX = "urn:ng-eventually:inbox";
/** A kind in this package's OWN namespace that no arm answers — upstream's `NotImplemented`. */
const UNARMED_KIND = `${INBOX}:nonesuch`;

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

beforeEach(() => {
  forgetEverything();
  quads = [];
  fake = bootPage(quads);
});

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

/** What is STILL QUEUED in `inbox` — one entry per deposit that has not been consumed. */
function stillQueuedIn(inbox: Nuri): string[] {
  return quads.filter((q) => q.g === inbox && q.p === `${INBOX}:payload`).map((q) => q.o);
}

/** How many deposits have been RETIRED — one `DELETE WHERE { <deposit> ?p ?o }` each. */
function removalCount(): number {
  return fake.sparql_update.mock.calls.filter((c) =>
    /DELETE\s+WHERE\s*\{\s*<[^>]+>\s+\?\w+\s+\?\w+\s*\}/.test(String(c[1]))
  ).length;
}

/** Capture what `console.error` is told while `body` runs — where a refusal goes. */
async function reportedDuring(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;
}

/**
 * Alice has been here once, so she is someone a deposit can NAME (`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).
 */
async function aliceHasBeenHere(): Promise<void> {
  await signIn("alice");
  await storeRegistry.createEntityDoc(await storeRegistry.protectedStore());
}

/** Bob shares a document of his with Alice, by name. Answers the document. */
async function bobSharesWithAlice(): Promise<Nuri> {
  await signIn("bob");
  const doc = await storeRegistry.createEntityDoc(await storeRegistry.protectedStore());
  await inboxSurface.share(doc, "alice");
  return doc;
}

/** The Links durably filed on a user's User branch — the emulated `AddLink` records. */
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 === "urn:ng-eventually:shim:link")
    .map((q) => q.o.split(":r:")[0]!);
}

// --- a deposit that was applied is GONE -----------------------------------

test("a Link that has been applied is no longer in the queue", async () => {
  await aliceHasBeenHere();
  const doc = await bobSharesWithAlice();
  const aliceInbox = await userInbox("alice", "protected");

  // Bob's deposit is waiting, and nothing of it has reached Alice yet.
  expect(stillQueuedIn(aliceInbox).length).toBe(1);
  expect(await documentsGivenTo("alice")).toEqual([]);

  // Alice connects: her own session drains her own inbox.
  await signIn("alice");

  // Applied — and CONSUMED. The second assertion is the one that was missing: before
  // 2026-08-26 the Link was filed and the deposit stayed, so this queue only ever grew.
  expect(await documentsGivenTo("alice")).toEqual([doc]);
  expect(stillQueuedIn(aliceInbox)).toEqual([]);
});

// --- a deposit whose ARM FAILED is gone too, and said so once --------------

describe("a deposit no arm can apply", () => {
  /**
   * Alice opens a document with an inbox on it; Bob deposits something that declares
   * itself to this package and that no arm answers. He names her DOCUMENT — the address is
   * resolved inside, and he never learns it.
   */
  async function bobDepositsSomethingUnarmed(): Promise<Nuri> {
    await signIn("alice");
    // PUBLIC, so Bob can read it — the address rides on the document, so a depositor who
    // cannot read it cannot reach its owner. That is the model, not a fixture detail.
    const note = await storeRegistry.createEntityDoc(await storeRegistry.publicStore());
    await openDocumentInbox(note);
    await signIn("bob");
    await inboxSurface.postToDocument(note, { payload: { kind: UNARMED_KIND } });
    return note;
  }

  test("is reported AND consumed — upstream logs the failure and the message is already gone", async () => {
    const note = await bobDepositsSomethingUnarmed();
    const noteInbox = await (async () => {
      const rec = quads.find(
        (q) => q.p === "urn:ng-eventually:shim:inboxCap" && q.o.startsWith(note + " "),
      );
      if (!rec) throw new Error("no AddInboxCap record was written for the note");
      return rec.o.split(" ")[1] as Nuri;
    })();
    expect(stillQueuedIn(noteInbox).length).toBe(1);

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

    // It said so — a message that could not be applied is never silent.
    expect(reported.some((l) => /no arm applies deposits of kind/.test(l))).toBe(true);
    // …and it is GONE. This is the whole change: upstream the broker had already removed
    // the message before the arm ran, so `log_err!` is the END of it, not a promise to
    // try again. The old wording ("it stays in it and the next attempt will try again")
    // described a deposit that would be re-reported forever.
    expect(stillQueuedIn(noteInbox)).toEqual([]);
    expect(reported.some((l) => /will not be retried/.test(l))).toBe(true);
  });

  test("is not reported a SECOND time, because a second drain finds an empty queue", async () => {
    const note = await bobDepositsSomethingUnarmed();
    await signIn("alice");

    // Alice connects again — a reload, a second tab, any of the things that drain. Before
    // consumption this reported the same unapplicable deposit again, and would have gone
    // on doing so at every connection for as long as it sat there.
    const reportedAgain = await reportedDuring(async () => {
      await signIn("alice");
    });

    expect(reportedAgain.filter((l) => /no arm applies deposits of kind/.test(l))).toEqual([]);
    void note;
  });
});

// --- a second drain of a drained inbox does nothing ------------------------

test("draining an inbox twice applies once and reports nothing the second time", async () => {
  await aliceHasBeenHere();
  const doc = await bobSharesWithAlice();
  const aliceInbox = await userInbox("alice", "protected");

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

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

  // Nothing to do, nothing said, and nothing filed twice. Note WHY that last one is not
  // just `addLink`'s de-duplication doing the work: the queue is empty, so no arm ran.
  expect(reported).toEqual([]);
  expect(stillQueuedIn(aliceInbox)).toEqual([]);
  expect(await documentsGivenTo("alice")).toEqual([doc]);
});

// --- ONE primitive, exercised through BOTH triggers -----------------------

/**
 * The two triggers must consume through the SAME act, so this asserts on that act rather
 * than on each trigger — a fault injected at the ONE removal, and the same consequence
 * demanded of both paths.
 *
 * The consequence is `inbox-consume`'s own contract: a removal that fails STOPS the apply,
 * because applying a deposit that stayed in the queue is the replay this exists to end. A
 * trigger carrying its own copy of the removal would not be bound by it.
 */
describe("both triggers consume through the one primitive", () => {
  /** Make retiring a deposit fail — the single `DELETE WHERE { <deposit> ?p ?o }`. */
  function breakTheRemoval(): void {
    const real = fake.sparql_update;
    fake.sparql_update = (async (...a: unknown[]) => {
      if (/DELETE\s+WHERE\s*\{\s*<[^>]+>\s+\?\w+\s+\?\w+\s*\}/.test(String(a[1]))) {
        throw new Error("the broker refused to retire this deposit");
      }
      return real(...a);
    }) as typeof fake.sparql_update;
  }

  test("the OWNER's own drain applies nothing when the deposit cannot be retired", async () => {
    await aliceHasBeenHere();
    await bobSharesWithAlice();
    const aliceInbox = await userInbox("alice", "protected");

    breakTheRemoval();
    await reportedDuring(async () => {
      await signIn("alice");
    });

    // Nothing applied, and the deposit is still queued — so the next drain will try it
    // again, which is the only safe answer when the pop did not happen.
    expect(await documentsGivenTo("alice")).toEqual([]);
    expect(stillQueuedIn(aliceInbox).length).toBe(1);
  });

  test("the DEFERRED stand-in applies nothing either, by the same rule", async () => {
    await aliceHasBeenHere();
    // Alice is NOT connected: Bob's session stands in for her absent one.
    await bobSharesWithAlice();
    const aliceInbox = await userInbox("alice", "protected");

    breakTheRemoval();
    await reportedDuring(async () => {
      await runScheduledInboxProcessingNow();
    });

    expect(await documentsGivenTo("alice")).toEqual([]);
    expect(stillQueuedIn(aliceInbox).length).toBe(1);
  });

  test("and with the removal working, the DEFERRED stand-in consumes exactly as the owner does", async () => {
    await aliceHasBeenHere();
    const doc = await bobSharesWithAlice();
    const aliceInbox = await userInbox("alice", "protected");

    await runScheduledInboxProcessingNow();

    // Same two facts the owner's own drain produces above — one act, two triggers.
    expect(await documentsGivenTo("alice")).toEqual([doc]);
    expect(stillQueuedIn(aliceInbox)).toEqual([]);
  });
});

// --- a pop whose effect did not land must SAY so ---------------------------

test("an I/O failure while filing a Link STOPS the drain, so the rest of the queue survives", async () => {
  await aliceHasBeenHere();
  // Three deposits waiting for an absent Alice, so "what happened to the OTHERS" is a
  // question this can answer at all.
  await bobSharesWithAlice();
  await bobSharesWithAlice();
  await bobSharesWithAlice();
  const aliceInbox = await userInbox("alice", "protected");
  expect(stillQueuedIn(aliceInbox).length).toBe(3);

  // The `AddLink` write fails — a broker that will not take the commit.
  const real = fake.sparql_update;
  fake.sparql_update = (async (...a: unknown[]) => {
    if (/urn:ng-eventually:shim:link/.test(String(a[1]))) {
      throw new Error("the broker refused the AddLink commit");
    }
    return real(...a);
  }) as typeof fake.sparql_update;

  const reported = await reportedDuring(async () => {
    await runScheduledInboxProcessingNow();
  });

  // Reported ungated, at the granularity of the INBOX — a genuine I/O failure is not an
  // unapplicable item, and the drain stops rather than working through the queue.
  expect(reported.some((l) => /could not finish draining this inbox/.test(l))).toBe(true);
  expect(await documentsGivenTo("alice")).toEqual([]);

  // The load-bearing assertion. One deposit was popped before the write failed and is lost;
  // the other TWO are untouched and wait for the next attempt. Between 2026-08-26's first
  // and second corrections this read 0: `applyLink` caught the throw and returned `false`,
  // so the loop went on popping — and most of what `addLink` does before its write is
  // deposit-independent infrastructure that fails for the whole queue at once. Catching it
  // per deposit therefore destroyed all three while dutifully reporting each one.
  expect(stillQueuedIn(aliceInbox).length).toBe(2);

  // And there is no WEDGE. Stopping the drain could have traded destruction for permanent
  // blockage — a deposit that always throws would sit at the head of the queue for ever and
  // no later one would be reached. It cannot, and the ORDER is why: the pop happens BEFORE
  // the arm, so the deposit that failed is already gone and the next attempt starts at the
  // one behind it. A failing broker costs one deposit per attempt, not the queue.
  //
  // The attempts are ALICE CONNECTING, not the timer: a deferred window is armed by a
  // deposit and is not re-armed by a drain that failed, so the owner's own next connection
  // is the recovery path — which is what the emulated stand-in has always documented.
  await signIn("alice");
  expect(stillQueuedIn(aliceInbox).length).toBe(1);
  await signIn("alice");
  expect(stillQueuedIn(aliceInbox).length).toBe(0);
});

// --- delivered ONCE means once, across both triggers ----------------------

test("a deposit the deferred drain consumed is not re-applied by the owner's own next pass", async () => {
  await aliceHasBeenHere();
  await bobSharesWithAlice();
  const aliceInbox = await userInbox("alice", "protected");

  // Alice READS her inbox without processing it — which is what `inbox.watch` does on every
  // push, and it is the only step that records what this session decoded. She is set as the
  // holder WITHOUT connecting, deliberately: connecting drains, and a drain would consume
  // the deposit before the read could see it.
  setCurrentUser("alice");
  expect((await inboxSurface.read(aliceInbox)).length + stillQueuedIn(aliceInbox).length)
    .toBeGreaterThan(0);
  setCurrentUser("bob");

  // The deferred stand-in then drains the same inbox through its own path, which never
  // touches that record. Before 2026-08-26 the record was only ever WRITTEN when something
  // was declared, so it survived the queue it described: Alice's own next pass worked from
  // a list of deposits that were already gone, re-popping and re-applying every one.
  await runScheduledInboxProcessingNow();

  const removalsBefore = removalCount();
  await signIn("alice");

  expect(removalCount()).toBe(removalsBefore);
  expect(stillQueuedIn(aliceInbox)).toEqual([]);
});

// --- the gap between deciding to deliver and delivering --------------------

test("an identity that moves DURING the pop does not get the deposit applied for it", async () => {
  await aliceHasBeenHere();
  // Alice opens an index and Bob deposits a bare reference into it. That arm is the one
  // that reads AMBIENT state, so it is the one an identity switch can mis-file.
  await signIn("alice");
  const index = await storeRegistry.createEntityDoc(await storeRegistry.publicStore());
  await openDocumentInbox(index);
  const object = await storeRegistry.createEntityDoc(await storeRegistry.publicStore());
  await signIn("bob");
  await inboxSurface.postToDocument(index, { payload: object });

  const indexInbox = (() => {
    const rec = quads.find(
      (q) => q.p === "urn:ng-eventually:shim:inboxCap" && q.o.startsWith(index + " "),
    );
    if (!rec) throw new Error("no AddInboxCap record was written for the index");
    return rec.o.split(" ")[1] as Nuri;
  })();

  // Alice signs in, so the reference is CLAIMED — her own session may apply it. Then the
  // identity moves in the one place nothing used to look: inside the await that removes the
  // deposit. `claimsDeposit` had already said yes, and the arm below reads whoever holds the
  // page now, so before the re-check this filed Alice's deposit against Bob's registers.
  const real = fake.sparql_update;
  fake.sparql_update = (async (...a: unknown[]) => {
    const out = await real(...a);
    if (/DELETE\s+WHERE\s*\{\s*<[^>]+>\s+\?\w+\s+\?\w+\s*\}/.test(String(a[1]))) {
      setCurrentUser("bob");
    }
    return out;
  }) as typeof fake.sparql_update;

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

  // The deposit is gone — it was popped before the switch was visible — and that is said
  // out loud rather than papered over. What must NOT have happened is the arm running for
  // the wrong holder.
  expect(stillQueuedIn(indexInbox)).toEqual([]);
  expect(reported.some((l) => /consumed but not applied/.test(l))).toBe(true);
});

test("an arm that fails WHILE the page switches user is still reported, not swallowed", async () => {
  await aliceHasBeenHere();
  await bobSharesWithAlice();
  const aliceInbox = await userInbox("alice", "protected");

  // Both at once: the `AddLink` write fails AND the identity moves. `connect.connectedUser`
  // drops an error from a drain whose holder has gone — an ownership refusal there is
  // abandoning, not a failure — and that guard cannot tell the two apart. So for one day
  // this exact pair destroyed a deposit and printed nothing at all.
  const real = fake.sparql_update;
  fake.sparql_update = (async (...a: unknown[]) => {
    if (/urn:ng-eventually:shim:link/.test(String(a[1]))) {
      setCurrentUser("bob");
      throw new Error("the broker refused the AddLink commit");
    }
    return real(...a);
  }) as typeof fake.sparql_update;

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

  // The deposit is gone, and SOMETHING said so. Reported by the only code that knows a
  // deposit was consumed, before the error is handed to anyone who might drop it.
  expect(stillQueuedIn(aliceInbox)).toEqual([]);
  expect(reported.some((l) => /was consumed and could not be applied/.test(l))).toBe(true);
});

test("a deposit that cannot be retired is refused, not applied again at every drain", async () => {
  await aliceHasBeenHere();
  const doc = await bobSharesWithAlice();
  const aliceInbox = await userInbox("alice", "protected");

  // A broker that answers the deposits query WITHOUT binding the deposit's subject. `?d` is
  // bound by the query so this should not happen, but `depositsFrom` defaults it to "" and
  // an empty subject would issue `DELETE WHERE { <> ?p ?o }`. The load-bearing half is that
  // it must not be APPLIED either: a deposit applied but never retired is the replay this
  // whole change exists to end, repeated at every drain for as long as the row sits there.
  const realQuery = fake.sparql_query;
  fake.sparql_query = (async (...a: unknown[]) => {
    const res = await realQuery(...a);
    if (a[3] === aliceInbox && String(a[1]).includes(`${INBOX}:payload`)) {
      const r = res as { results?: { bindings?: Array<Record<string, unknown>> } };
      for (const row of r.results?.bindings ?? []) delete row.d;
    }
    return res;
  }) as typeof fake.sparql_query;

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

  expect(reported.some((l) => /carries no subject/.test(l))).toBe(true);
  expect(await documentsGivenTo("alice")).toEqual([]);
  expect(stillQueuedIn(aliceInbox).length).toBe(1);
  void doc;
});