ng-eventually/packages/client/test/inbox.test.ts

import { test, expect, mock, beforeEach, afterAll } from "bun:test";
import { post, read, materialize, watch } from "../src/inbox";
import { walletInbox, resetRegistryCache } from "../src/store-registry";
import type { Deposit } from "../src/inbox";
import {
  configure,
  configureStoreRegistry,
  resetStoreRegistry,
  resetConfig,
  setCurrentUser,
} from "../src/polyfill";
import type { RegistrySession } from "../src/store-registry";

// This suite injects a fake `ng` via configure() and reuses the storeRegistry's
// injected session provider (inbox docs live in the shared wallet). Restore the
// un-configured state at the end so docs.test.ts's guard still sees null config.
afterAll(() => {
  resetConfig();
  resetStoreRegistry();
  setCurrentUser(null);
});

// NOTE ORDER: the "not configured → throw" case runs first — it exercises the
// registry-deps guard before any configureStoreRegistry() call.

test("throws a clear error when configureStoreRegistry() was not called", async () => {
  resetStoreRegistry();
  await expect(post("did:ng:o:inbox", { payload: { hi: 1 } })).rejects.toThrow(
    /configureStoreRegistry\(\) must be called before use/,
  );
  await expect(read("did:ng:o:inbox")).rejects.toThrow(
    /configureStoreRegistry\(\) must be called before use/,
  );
});

// --- A stateful fake `ng`: parses the inbox INSERT DATA and answers the read
// SELECT over an in-memory quad store.

interface Quad { g: string; s: string; p: string; o: string }

const INBOX = "urn:ng-eventually:inbox";

/** Reverse of the lib's escapeLiteral: single left-to-right pass over `\x`. */
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;
}

function makeFakeNg() {
  const quads: Quad[] = [];

  // Reactive subscriptions: doc_subscribe registers a callback per anchor and
  // fires an initial State push; a matching sparql_update pushes a Patch to that
  // anchor's subscribers. This mirrors the real broker's local-push behaviour so
  // inbox.watch (now event-driven, no polling) can be tested without a timer.
  const subs = new Map<string, Set<(r: unknown) => void>>();
  const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => {
    let set = subs.get(nuri);
    if (!set) {
      set = new Set();
      subs.set(nuri, set);
    }
    set.add(cb);
    queueMicrotask(() => cb({ V0: { State: { doc: nuri } } })); // initial push
    return () => set!.delete(cb);
  });
  const pushTo = (anchor: string): void => {
    for (const cb of subs.get(anchor) ?? []) cb({ V0: { Patch: { doc: anchor } } });
  };

  const doc_create = mock(async (..._a: unknown[]) => "did:ng:o:new");

  // Parses one deposit: `<subj> a <Deposit> ; <payload> "..." ; <ts> "..." [; <from> "..."] .`
  //
  // The REAL broker keys triples by the ANCHORED repo's default graph, not by an
  // explicit `GRAPH <…>` IRI (repo_graph_name(repo_id, overlay_id)). So this mock
  // keys stored quads by the ANCHOR arg (a[2]) — the default graph of the anchored
  // repo — and REJECTS any explicit `GRAPH <…>` wrapper, so the old wrong shape
  // does NOT round-trip and can never regress silently.
  const sparql_update = mock(async (...a: unknown[]) => {
    const query = a[1] as string;
    const anchor = a[2] as string | undefined;
    if (/GRAPH\s*</.test(query)) return undefined; // explicit-GRAPH write → dropped
    if (!anchor) return undefined;
    const g = anchor;
    const body = query.replace(/^\s*INSERT DATA\s*\{/, "").replace(/\}\s*$/, "");
    const sm = body.match(/<([^>]+)>/);
    if (!sm) return undefined;
    const s = sm[1]!;
    const after = body.slice(body.indexOf(sm[0]) + sm[0].length);
    // predicate/object pairs: `a <type>` or `<p> "literal"`.
    const pairRe = /(?:a|<([^>]+)>)\s+(?:"((?:[^"\\]|\\.)*)"|<([^>]+)>)/g;
    let m: RegExpExecArray | null;
    while ((m = pairRe.exec(after)) !== null) {
      const p = m[1] ?? `${INBOX}:Deposit`; // `a` → rdf:type-ish
      // Un-escape the SPARQL literal so payload JSON round-trips. Single pass
      // over `\x` sequences (reverses the lib's escapeLiteral without the
      // double-processing that chained .replace() would cause).
      const rawLit = m[2];
      const o = rawLit !== undefined ? unescapeLiteral(rawLit) : (m[3] ?? "");
      quads.push({ g, s, p, o });
    }
    // A write to `g` (the anchored default graph) pushes a Patch to that doc's
    // subscribers — the local-push the real broker performs on a verified commit.
    pushTo(g);
    return undefined;
  });

  const sparql_query = mock(async (...a: unknown[]) => {
    const anchor = a[3] as string | undefined;
    const bySubject = new Map<string, Record<string, string>>();
    for (const q of quads) {
      if (q.g !== anchor) continue;
      if (q.p === `${INBOX}:Deposit`) {
        // rdf:type marker — ensure the subject exists.
        if (!bySubject.has(q.s)) bySubject.set(q.s, {});
        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);
    }
    const bindings = [...bySubject.values()]
      .filter((r) => r.payload !== undefined && r.ts !== undefined)
      .map((r) => {
        const row: Record<string, { value: string }> = {
          payload: { value: r.payload! },
          ts: { value: r.ts! },
        };
        if (r.from !== undefined) row.from = { value: r.from };
        return row;
      });
    return { results: { bindings } };
  });

  return { doc_create, doc_subscribe, sparql_update, sparql_query, _quads: quads };
}

const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" };
/** Resolved per test: an inbox BELONGS to a wallet, and only its owner may read it. */
let TARGET: `did:ng:${string}`;

function inject() {
  const ng = makeFakeNg();
  configure({ ng: ng as any, useShape: (() => {}) as any });
  // Synchronous fake store → no sync lag; disable the anti-fork retry backoff.
  configureStoreRegistry({ getSession: async () => SESSION });
  setCurrentUser(null);
  return ng;
}

let fake: ReturnType<typeof makeFakeNg>;
beforeEach(async () => {
  fake = inject();
  resetRegistryCache();
  setCurrentUser("alice");
  TARGET = await walletInbox("alice");
});

test("post writes via the real injected ng.sparql_update (not makeNg), scoped to the inbox", async () => {
  setCurrentUser("alice"); // `from` is bound to the current identity
  // Count from HERE: resolving this wallet's own inbox already wrote to the shim.
  const before = fake.sparql_update.mock.calls.length;
  await post(TARGET, { from: "alice", payload: { kind: "join" }, ts: 100 });
  expect(fake.sparql_update.mock.calls.length).toBe(before + 1);
  const call = fake.sparql_update.mock.calls[before]!;
  expect(call[0]).toBe("sid-1"); // sessionId from the injected session
  expect(call[2]).toBe(TARGET); // anchored to the target inbox
  // The write targets the anchored DEFAULT graph — NO explicit `GRAPH <…>`
  // wrapper (which the real broker would route to a phantom graph).
  expect(call[1] as string).not.toContain("GRAPH <");
});

test("post → read round-trips payload, from and ts", async () => {
  setCurrentUser("alice"); // `from` is bound to the current identity
  await post(TARGET, { from: "alice", payload: { kind: "join", n: 3 }, ts: 100 });
  const deposits = await read(TARGET);
  expect(deposits).toHaveLength(1);
  expect(deposits[0]).toEqual({ from: "alice", payload: { kind: "join", n: 3 }, ts: 100 });
});

// (c) `from` is BOUND to the current identity — a spoof (naming another
// principal) is REJECTED; identifying as self or anonymous (null) is allowed.
test("(c) post rejects a spoofed `from` (naming another principal); self/null allowed", async () => {
  setCurrentUser("alice");
  // SPOOF: alice tries to deposit AS bob → rejected.
  await expect(post(TARGET, { from: "bob", payload: { x: 1 }, ts: 1 })).rejects.toThrow(
    /spoof|current identity/i,
  );
  // Identifying as self → allowed.
  await post(TARGET, { from: "alice", payload: { x: 2 }, ts: 2 });
  // Explicit anonymous → allowed.
  await post(TARGET, { from: null, payload: { x: 3 }, ts: 3 });
  const froms = (await read(TARGET)).map((d) => d.from);
  expect(froms).toEqual(["alice", null]);
});

test("from is optional — omitting it defaults to the current user", async () => {
  setCurrentUser("bob");
  await post(TARGET, { payload: { hi: 1 }, ts: 200 });
  const deposits = await read(TARGET);
  expect(deposits[0]!.from).toBe("bob");
});

test("from: null makes an anonymous deposit even when a current user is set", async () => {
  setCurrentUser("bob");
  await post(TARGET, { from: null, payload: { hi: 1 }, ts: 200 });
  const deposits = await read(TARGET);
  expect(deposits[0]!.from).toBeNull();
});

test("read returns deposits sorted by ts ascending and materialize is an alias", async () => {
  await post(TARGET, { from: null, payload: "second", ts: 300 });
  await post(TARGET, { from: null, payload: "first", ts: 100 });
  await post(TARGET, { from: null, payload: "third", ts: 500 });
  const deposits = await materialize(TARGET);
  expect(deposits.map((d) => d.payload)).toEqual(["first", "second", "third"]);
});

test("read is scoped to one inbox — deposits in another inbox are not returned", async () => {
  await post(TARGET, { from: null, payload: "mine", ts: 1 });
  await post("did:ng:o:other-inbox", { from: null, payload: "theirs", ts: 2 });
  const deposits = await read(TARGET);
  expect(deposits.map((d) => d.payload)).toEqual(["mine"]);
});

test("payload with quotes/newlines/backslashes survives the round-trip", async () => {
  const payload = { text: 'a "quoted"\nline\\path\ttab' };
  await post(TARGET, { from: null, payload, ts: 1 });
  const deposits = await read(TARGET);
  expect(deposits[0]!.payload).toEqual(payload);
});

test("watch fires immediately then on each new deposit, and unsubscribe stops it", async () => {
  const seen: Deposit[][] = [];
  const stop = watch(TARGET, (d) => seen.push(d), { intervalMs: 5 });
  // Give the immediate tick a chance to run (empty inbox → still fires once).
  await new Promise((r) => setTimeout(r, 20));
  expect(seen.length).toBeGreaterThanOrEqual(1);
  expect(seen[seen.length - 1]).toEqual([]);

  await post(TARGET, { from: null, payload: "x", ts: 1 });
  await new Promise((r) => setTimeout(r, 20));
  const last = seen[seen.length - 1]!;
  expect(last.map((d) => d.payload)).toEqual(["x"]);

  stop();
  const countAfterStop = seen.length;
  await post(TARGET, { from: null, payload: "y", ts: 2 });
  await new Promise((r) => setTimeout(r, 20));
  expect(seen.length).toBe(countAfterStop); // no more callbacks after unsubscribe
});