ng-helpers/test/inbox-processing.test.ts

import { expect, mock, test } from "bun:test";
import { indexing, type Indexing } from "../src/indexing";
import { coalescing } from "../src/coalescing";
import type { Nuri } from "../src/port";
import { FakeNextGraph, publishObject } from "./fake-nextgraph";

/**
 * WHEN an index is curated — the engagement `createIndex` makes about what becomes
 * of what it created: the index is curated at its creator's next connection, and on
 * each deposit while the creator is connected.
 *
 * Nothing below calls curation, because there is nothing to call. Every test here
 * drives the two acts an application really has — connecting (obtaining a handle)
 * and depositing — and asserts what the index holds afterwards.
 *
 * The case space is the creator's presence crossed with the deposit's timing:
 * away when it was made, connected when it was made, and connected on an index
 * that a previous session created. Plus the two that must NOT happen: a stranger
 * connecting curates nothing, and a document that is no index is left alone.
 *
 * And crossing all of it, the three ways connecting can FAIL — it cannot look for
 * its indexes, it cannot go through one, it cannot watch one. Each has its own test
 * below, because each is a failure wearing the shape of an absence: the session
 * carries on, the handle works, and an index quietly holds less than it should. The
 * engagement is that none of them denies anything and none of them loses a deposit,
 * which is only worth anything if it is exercised rather than asserted.
 */

const PUBLISHED_AT = "http://schema.org/datePublished";

function hardcodedInAppSource(nuri: Nuri): Nuri {
  return nuri;
}

/**
 * Alice creates an index, and then her page closes. That is the state most of these
 * tests start from: an index exists, its creator is away, and nothing is watching
 * it — so a deposit made now can only be seen at her next connection.
 */
async function aliceCreatesAnIndexAndLeaves(network: FakeNextGraph): Promise<Nuri> {
  const alice = await indexing(network.portFor("alice"));
  const index = await alice.createIndex(PUBLISHED_AT);
  network.disconnect("alice");
  return hardcodedInAppSource(index);
}

/**
 * Runs `body` with this package's log stream captured, and reports HOW MANY
 * failures it put there alongside whatever the body produced.
 *
 * The count, never the text: what a log line reads is for a human and nothing
 * promises it, so a test that pinned the words would pin the one thing that is
 * free to change. What is worth pinning is that a failure was reported AT ALL —
 * harmless is not the same as invisible, and the whole risk here is a failure
 * passing for an absence.
 */
async function capturingReports<T>(
  body: () => Promise<T>,
): Promise<{ result: T; reports: number }> {
  const reported = mock((..._args: unknown[]) => {});
  const original = console.error;
  console.error = reported;
  try {
    return { result: await body(), reports: reported.mock.calls.length };
  } finally {
    console.error = original;
  }
}

// --- the deposits that piled up while the creator was away ----------------

test("an index is curated at its creator's next connection, with nobody asking", async () => {
  const network = new FakeNextGraph();
  const index = await aliceCreatesAnIndexAndLeaves(network);

  // Bob deposits while Alice is away: her session is never told, and the deposit
  // waits in the inbox where only she can see it.
  const bobPort = network.portFor("bob");
  const article = await publishObject(bobPort, PUBLISHED_AT, "2026-03-04");
  await (await indexing(bobPort)).refer(index, article);

  const carol = await indexing(network.portFor("carol"));
  expect(await carol.read(index)).toEqual([]);

  // Alice comes back. This is the whole of it: obtaining her handle IS the trigger.
  const alice = await indexing(network.portFor("alice"));

  expect(await alice.read(index)).toEqual([{ object: article, value: "2026-03-04" }]);
  expect(await carol.read(index)).toEqual([{ object: article, value: "2026-03-04" }]);
});

test("a whole backlog is caught up, across every index the creator owns", async () => {
  const network = new FakeNextGraph();
  const alicePort = network.portFor("alice");
  const bobPort = network.portFor("bob");

  const first = await aliceCreatesAnIndexAndLeaves(network);
  const second = await aliceCreatesAnIndexAndLeaves(network);
  // An ordinary public document of Alice's, which is no index at all.
  await publishObject(alicePort, PUBLISHED_AT, "2026-01-01");

  const bob = await indexing(bobPort);
  const early = await publishObject(bobPort, PUBLISHED_AT, "2026-01-02");
  const late = await publishObject(bobPort, PUBLISHED_AT, "2026-05-06");
  await bob.refer(first, early);
  await bob.refer(first, late);
  await bob.refer(second, late);

  const alice = await indexing(alicePort);

  expect((await alice.read(first)).map((e) => e.value)).toEqual(["2026-01-02", "2026-05-06"]);
  expect(await alice.read(second)).toEqual([{ object: late, value: "2026-05-06" }]);
});

// --- the deposits that arrive while the creator is looking ----------------

test("a deposit made while the creator is connected is curated as it lands", async () => {
  const network = new FakeNextGraph();
  const alice = await indexing(network.portFor("alice"));
  const index = hardcodedInAppSource(await alice.createIndex(PUBLISHED_AT));

  const bobPort = network.portFor("bob");
  const article = await publishObject(bobPort, PUBLISHED_AT, "2026-03-04");
  await (await indexing(bobPort)).refer(index, article);

  // The index was created in THIS session, so the store search never saw it: what
  // brings it under observation is `createIndex` itself.
  await network.deliverNotifications();

  expect(await alice.read(index)).toEqual([{ object: article, value: "2026-03-04" }]);
});

test("an index from a previous session is watched too, not merely caught up once", async () => {
  const network = new FakeNextGraph();
  const index = await aliceCreatesAnIndexAndLeaves(network);
  const bobPort = network.portFor("bob");
  const bob = await indexing(bobPort);

  // Alice comes back to an index she created before, with nothing waiting in it.
  const alice = await indexing(network.portFor("alice"));
  expect(await alice.read(index)).toEqual([]);

  // …and only now does Bob deposit. Nothing but the watch can carry this one.
  const article = await publishObject(bobPort, PUBLISHED_AT, "2026-07-08");
  await bob.refer(index, article);
  await network.deliverNotifications();

  expect(await alice.read(index)).toEqual([{ object: article, value: "2026-07-08" }]);
});

test("a burst of deposits settles on the same index, whatever order they are told in", async () => {
  const network = new FakeNextGraph();
  const alice = await indexing(network.portFor("alice"));
  const index = hardcodedInAppSource(await alice.createIndex(PUBLISHED_AT));

  const bobPort = network.portFor("bob");
  const bob = await indexing(bobPort);
  for (const date of ["2026-03-04", "2026-01-31", "2025-12-25"]) {
    await bob.refer(index, await publishObject(bobPort, PUBLISHED_AT, date));
  }

  await network.deliverNotifications();

  expect((await alice.read(index)).map((e) => e.value)).toEqual([
    "2025-12-25",
    "2026-01-31",
    "2026-03-04",
  ]);
});

// --- what connecting must NOT do -----------------------------------------

test("connecting curates nothing for anyone but the creator", async () => {
  const network = new FakeNextGraph();
  const index = await aliceCreatesAnIndexAndLeaves(network);

  const bobPort = network.portFor("bob");
  const bob = await indexing(bobPort);
  const article = await publishObject(bobPort, PUBLISHED_AT, "2026-03-04");
  await bob.refer(index, article);

  // Bob connects again, and Carol connects: neither owns the index, so neither can
  // read its inbox — and connecting must not try, nor fail, nor write anything.
  const carol = await indexing(network.portFor("carol"));
  await indexing(bobPort);
  await network.deliverNotifications();

  expect(await carol.read(index)).toEqual([]);
});

test("a public document that is no index is left alone — no inbox, no entry", async () => {
  const network = new FakeNextGraph();
  const alicePort = network.portFor("alice");
  const ordinary = await alicePort.createPublicDocument();

  await indexing(alicePort);

  // Had connecting treated every public document as an index, it would have opened
  // an inbox on this one — which is exactly what makes a deposit possible.
  const bob = await indexing(network.portFor("bob"));
  await expect(bob.refer(ordinary, "did:ng:o:doc-9")).rejects.toThrow(/has no inbox/);
});

test("a session that could not look for its indexes is still a working handle", async () => {
  const network = new FakeNextGraph();
  const index = await aliceCreatesAnIndexAndLeaves(network);
  const bobPort = network.portFor("bob");
  const article = await publishObject(bobPort, PUBLISHED_AT, "2026-03-04");
  await (await indexing(bobPort)).refer(index, article);

  network.breakListing("broker unreachable");
  const reported = mock((..._args: unknown[]) => {});
  const original = console.error;
  console.error = reported;
  let alice: Indexing;
  try {
    alice = await indexing(network.portFor("alice"));
  } finally {
    console.error = original;
  }

  // Reading an index and depositing into one need none of that work, so nothing is
  // denied — but the failure is on the log, because a silent one teaches nobody.
  expect(await alice.read(index)).toEqual([]);
  await alice.refer(index, article);
  expect(reported).toHaveBeenCalledTimes(1);
  expect(String(reported.mock.calls[0]?.[0])).toContain("public store could not be listed");
});

test("an index whose catch-up failed is still a working handle, and loses no deposit", async () => {
  const network = new FakeNextGraph();
  const stalled = await aliceCreatesAnIndexAndLeaves(network);
  const healthy = await aliceCreatesAnIndexAndLeaves(network);

  const bobPort = network.portFor("bob");
  const bob = await indexing(bobPort);
  const article = await publishObject(bobPort, PUBLISHED_AT, "2026-03-04");
  await bob.refer(stalled, article);
  await bob.refer(healthy, article);

  // The broker answers about the document and not about its inbox. Two repos
  // upstream, read with two capabilities, so this is a partial failure and not a
  // contrived one — and it is what makes the catch-up fail on THIS index alone.
  network.breakInboxReadsOf(stalled, "broker unreachable");

  const { result: alice, reports } = await capturingReports(() =>
    indexing(network.portFor("alice")),
  );

  // Obtaining the handle RESOLVED — reaching this line at all is the assertion.
  // Reading the index it could not go through still works…
  expect(await alice.read(stalled)).toEqual([]);
  // …and so does depositing into it: neither ever depended on that work.
  await alice.refer(stalled, article);
  // The session is not poisoned either: the other index was caught up normally.
  expect(await alice.read(healthy)).toEqual([{ object: article, value: "2026-03-04" }]);
  expect(reports).toBe(1);

  // And nothing was lost. The deposits never left the inbox, so the first
  // connection that can read it puts them in — which is the whole reason a failed
  // run is allowed to be this quiet.
  network.healInboxReadsOf(stalled);
  network.disconnect("alice");
  const back = await indexing(network.portFor("alice"));
  expect(await back.read(stalled)).toEqual([{ object: article, value: "2026-03-04" }]);
});

test("an index that could not be watched is still caught up, and the rest still notices", async () => {
  const network = new FakeNextGraph();
  const unwatched = await aliceCreatesAnIndexAndLeaves(network);
  const watched = await aliceCreatesAnIndexAndLeaves(network);

  const bobPort = network.portFor("bob");
  const bob = await indexing(bobPort);
  const waiting = await publishObject(bobPort, PUBLISHED_AT, "2026-01-01");
  await bob.refer(unwatched, waiting);

  // The subscription is refused; reading that same inbox still works. A watch is
  // held open where a read is one question and one answer, so one can be turned
  // down while the other is served.
  network.breakWatchingOf(unwatched, "the broker refused the subscription");

  const { result: alice, reports } = await capturingReports(() =>
    indexing(network.portFor("alice")),
  );

  // The watch failed and the catch-up ran ANYWAY — the backlog is in. That is the
  // order the code goes to some trouble to hold: failing to watch must not cost
  // the deposits that were already waiting.
  expect(await alice.read(unwatched)).toEqual([{ object: waiting, value: "2026-01-01" }]);
  expect(reports).toBe(1);

  // What the failure costs, exactly and no more: a deposit made from now on is not
  // NOTICED on that index…
  const late = await publishObject(bobPort, PUBLISHED_AT, "2026-02-02");
  await bob.refer(unwatched, late);
  await bob.refer(watched, late);
  await network.deliverNotifications();
  expect((await alice.read(unwatched)).map((e) => e.value)).toEqual(["2026-01-01"]);
  // …while every other index of the very same session goes on noticing its own.
  expect(await alice.read(watched)).toEqual([{ object: late, value: "2026-02-02" }]);

  // "Until the next connection" is the whole of the damage, and the next
  // connection is where it ends.
  network.healWatchingOf(unwatched);
  network.disconnect("alice");
  const back = await indexing(network.portFor("alice"));
  expect((await back.read(unwatched)).map((e) => e.value)).toEqual(["2026-01-01", "2026-02-02"]);
});

// --- the primitive that keeps a burst from piling up ----------------------

test("coalescing never runs twice at once, and grants exactly one more run", async () => {
  const trace: string[] = [];
  const ask = coalescing(async () => {
    trace.push("start");
    // Yields, so the asks below really do arrive while a run is in flight — which
    // is the only situation this primitive exists for.
    await Promise.resolve();
    trace.push("end");
  });

  const first = ask();
  const during = [ask(), ask(), ask()];
  await Promise.all([first, ...during]);

  // Three asks during one run earn ONE more run between them, not three — and not
  // none, since a deposit that landed after the first run read the inbox would
  // otherwise wait for the next connection.
  expect(trace).toEqual(["start", "end", "start", "end"]);
  // Runs never overlap: no "start" ever follows a "start".
  expect(trace.join(" ")).not.toContain("start start");

  // And an ask that arrives once everything is quiet is a run of its own.
  await ask();
  expect(trace.filter((step) => step === "start")).toHaveLength(3);
});