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 three that must NOT happen: a stranger
 * connecting curates nothing, a document that is no index is left alone, and a
 * session that could not look for its indexes is still a working handle.
 */

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);
}

// --- 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");
});

// --- 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);
});