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

/**
 * Inbox — a generic deposit + read/materialize mechanism the consumer reuses for
 * its own purposes (same `inbox.post` API, same watcher — see the discovery-model
 * decision). The mechanism itself knows no application domain: the consumer
 * supplies the inbox document NURI and interprets the `payload`. (An example
 * consumer mapping, purely illustrative: a consumer might use one inbox for a
 * registration deposit and another for submitting a reference to an index.)
 *
 * ── Real target vs this emulation ─────────────────────────────────────────
 * In real NextGraph, a message is sealed to the recipient's key and queued into
 * their inbox; the recipient's own verifier unseals each queued message and
 * applies it inline as it processes the inbox — there is no separate curator
 * process. A future `inbox_post_link` is the intended way to seal a link into an
 * inbox from the sender side; it is not exposed yet.
 *
 * Here, on one shared wallet where everything is readable, both sides run in-lib:
 *   - `post` appends a deposit `{ from, payload, ts }` as RDF into the inbox
 *     document (in the shared wallet) via the `docs.sparqlUpdate` primitive;
 *   - `read` / `watch` read the deposits back via `docs.sparqlQuery` and expose
 *     them. This in-lib read stands in for the recipient's own inbox processing
 *     until the sealed-inbox path (`inbox_post_link`) is available.
 *
 * All NextGraph I/O routes through the `docs` primitives (the real injected `ng`,
 * never `makeNg`), so this module imports no `@ng-org` package.
 */

import { depositInto, sparqlQuery } from "./docs";
import { subscribeDoc } from "./subscribe";
import { ensureRepoOpen } from "./open-repo";
import { getCaps, getCurrentUser, getStoreRegistryDeps } from "./polyfill";
import { addLink, isOwnInbox } from "./store-registry";
import { escapeLiteral } from "./sparql";
import { hasReadCap } from "./nuri";
import {
  accessLogPrefix,
  enabled as accessLogEnabled,
  logAccess,
  logStage,
  shortNuri,
} from "./access-log";
import type { Nuri, PrincipalId, ReadCap } from "./types";

// --- deposit model --------------------------------------------------------

/** One deposit as materialized from an inbox document. */
export interface Deposit {
  /** The sender, if identified; `null` when the deposit was anonymous. */
  from: PrincipalId | null;
  /** The consumer-defined payload (opaque here — JSON-serialized in storage). */
  payload: unknown;
  /** Deposit timestamp (ms epoch). Caller may pass one for determinism. */
  ts: number;
}

/** Options for {@link post}. `from` and `ts` are both optional. */
export interface PostOptions {
  /**
   * Who is depositing. Omit (or pass `null`) for an ANONYMOUS deposit; pass a
   * principal id to identify the sender. Defaults to the current polyfill user
   * ({@link getCurrentUser}) when the property is entirely absent, so callers
   * that want anonymity must pass `from: null` explicitly.
   */
  from?: PrincipalId | null;
  /** The payload to deposit (interpreted only by the consumer). */
  payload: unknown;
  /** Optional deposit timestamp (ms epoch). Omitted → `Date.now()`. Passing it
   *  keeps tests deterministic. */
  ts?: number;
}

const SHIM = "urn:ng-eventually:inbox";
const P = {
  type: `${SHIM}:Deposit`,
  from: `${SHIM}:from`,
  payload: `${SHIM}:payload`,
  ts: `${SHIM}:ts`,
} as const;

// --- session access (shared with the storeRegistry) -----------------------

/** The inbox documents live in the shared wallet, so we reuse the registry's
 *  injected session provider for the sessionId. Disappears at migration. */
async function sessionId(): Promise<string> {
  return (await getStoreRegistryDeps().getSession()).sessionId;
}

// --- diagnostic logging helper ---------------------------------------------

/**
 * Best-effort, length-capped JSON rendering of a deposit payload for the
 * inbox diagnostic log (see {@link enabled}/{@link logAccess}). This module
 * stays domain-agnostic (see module header) — it never interprets payload
 * fields, it only dumps them verbatim so the consumer's own shape (e.g. a
 * Festipod participation: `{ participantId, eventId, … }`) is visible in the
 * log without this module knowing that shape. Capped so one oversized payload
 * can't blow up a log line; a payload that fails to stringify (e.g. a
 * circular structure a caller mistakenly passed) falls back to `String()`.
 */
function summarizePayload(payload: unknown): string {
  let s: string;
  try {
    s = JSON.stringify(payload) ?? String(payload);
  } catch {
    s = String(payload);
  }
  return s.length > 200 ? s.slice(0, 200) + "…" : s;
}

// --- SPARQL result helpers ------------------------------------------------

/** Tolerant extraction of SPARQL SELECT bindings across possible shapes. */
function readBindings(result: unknown): Array<Record<string, { value: string }>> {
  if (!result) return [];
  if (Array.isArray(result)) return result as Array<Record<string, { value: string }>>;
  const anyRes = result as {
    results?: { bindings?: Array<Record<string, { value: string }>> };
  };
  return anyRes.results?.bindings ?? [];
}

// --- deposit (client side) ------------------------------------------------

/**
 * Deposit a payload into `targetInbox`.
 *
 * Appends `{ from, payload, ts }` into the inbox document via `docs.sparqlUpdate`
 * (the real injected `ng`). Each deposit is a fresh RDF subject in the inbox
 * graph, so concurrent deposits don't collide.
 *
 * `from` is bound to the current identity — it is authenticated, not
 * caller-supplied. Omit it to stamp the current identity; pass `null` to deposit
 * anonymously (a legitimate choice — identified if known, anonymous otherwise).
 * A `from` naming another identity is rejected as a spoof: in the target the
 * broker seals the sender from the wallet's own key, so a client cannot forge
 * another's identity. This check is redundant once the seal enforces it, but
 * until then it closes the spoof the shared wallet would otherwise allow.
 */
export async function post(targetInbox: Nuri, opts: PostOptions): Promise<void> {
  const current = getCurrentUser();
  let from: PrincipalId | null;
  if (opts.from === undefined) {
    from = current; // default: stamp the current identity
  } else if (opts.from === null) {
    from = null; // explicit anonymous deposit
  } else if (opts.from === current) {
    from = opts.from; // identifying as self — allowed
  } else {
    throw new Error(
      "[ng-eventually] inbox.post: `from` must be the current identity or null " +
        "(anonymous) — depositing as another principal is a spoof.",
    );
  }
  const ts = opts.ts ?? Date.now();
  const sid = await sessionId();

  // A unique subject per deposit (in the inbox graph) — no collisions.
  const subject = `${SHIM}:deposit:${ts}:${Math.random().toString(36).slice(2)}`;
  const payloadLiteral = escapeLiteral(JSON.stringify(opts.payload ?? null));
  const fromTriple =
    from == null ? "" : ` ;\n        <${P.from}> "${escapeLiteral(from)}"`;

  // NO explicit `GRAPH <…>` wrapper — write the anchored DEFAULT graph:
  // `sparqlUpdate(sid, update, targetInbox)` scopes the write to that repo's
  // default graph (same shape as read-model.ts readDoc/readUnion). This is the
  // CANONICAL, always-safe shape and the one the anchored default-graph read
  // queries. (Not a round-trip necessity on the current broker: the e2e harness
  // `packages/client/e2e/` verified that an anchored `GRAPH <plainNuri>` write
  // ALSO round-trips here — it resolves to the same repo graph, no phantom graph.
  // The no-GRAPH form is kept as a simplicity/safety convention; re-verify with
  // that harness if the broker version changes.)
  const update = `
    INSERT DATA {
      <${subject}> a <${P.type}> ;
        <${P.payload}> "${payloadLiteral}" ;
        <${P.ts}> "${ts}"${fromTriple} .
    }`;
  // A deposit crosses the boundary on purpose — see docs.depositInto.
  await depositInto(sid, update, targetInbox, "deposit");
  // Domain-level diagnostic (on top of docs.ts's generic access-path WRITE log):
  // who deposited WHAT into which inbox — the decoded payload, not just the
  // triple-write. Gated by the same access-log flag; skip the JSON work when off.
  if (accessLogEnabled()) {
    logAccess(
      "WRITE",
      targetInbox,
      "inbox deposit",
      " from=" + (from ?? "anonymous") + " payload=" + summarizePayload(opts.payload ?? null),
    );
  }
}

// --- cap delivery ---------------------------------------------------------

/**
 * A **Link** — the deposit that carries a ReadCap. The word is upstream's, and it
 * is the same one at all three stages: `InboxMsgContent::Link` is the message
 * (`engine/net/src/types.rs:4249-4261`, declared but payload-less so far),
 * `AddLink { read_cap }` is where the recipient files it (`repo/types.rs:1934-1950`),
 * `RemoveLink` withdraws it. So giving access is: deposit a Link, and on connection
 * the recipient processes their inbox and files it.
 *
 * It travels the SAME channel as any other deposit, which is why key ROTATION needs
 * no special case on the surface — a re-delivered cap is just another Link.
 */
const LINK_KIND = "urn:ng-eventually:inbox:link";

/** Links observed during the last read of an inbox, awaiting durable filing. */
const seenByInbox = new Map<Nuri, ReadCap[]>();
function capsSeenIn(inbox: Nuri): ReadCap[] {
  return seenByInbox.get(inbox) ?? [];
}


/** The cap a deposit carries, if it is a Link rather than consumer data. */
function capOfPayload(payload: unknown): ReadCap | null {
  const p = payload as { kind?: unknown; cap?: unknown } | null;
  if (!p || typeof p !== "object" || p.kind !== LINK_KIND) return null;
  return typeof p.cap === "string" && hasReadCap(p.cap) ? p.cap : null;
}

/**
 * Share ONE document's read cap with ONE recipient, addressed by their inbox.
 *
 * The unit of sharing is the DOCUMENT: never hand over a store's cap, which would
 * give away everything the store contains, present and future. The recipient needs
 * no dedicated operation to receive it — the cap arrives as a deposit that their
 * existing {@link watch} absorbs into what they hold (see {@link read}).
 *
 * Reaching several recipients means calling this once per inbox, which is what the
 * real model does too: each delivery is sealed to one recipient.
 *
 * Upstream this path is a GAP, not a disagreement — verified at both ends:
 *   - the field exists, `ContactDetails.read_cap: Option<ReadCap>`
 *     (`engine/net/src/types.rs:4233`), but building a message that carries one is
 *     `read_cap: if with_readcap { unimplemented!() }` (`types.rs:3786`);
 *   - and the receiver ignores it: `InboxMsgContent::ContactDetails` writes only
 *     `ng:site`/`ng:protected` + `ng:*_inbox` into a fresh contact document
 *     (`engine/verifier/src/inbox_processor.rs:778-830`), never `details.read_cap`.
 *
 * Do NOT read `InboxMsgContent::Link` as the intended channel either: it is a **unit
 * variant carrying nothing** (`engine/net/src/types.rs:4251`).
 *
 * The shape is right; the implementation is absent at both ends, so we emulate it
 * meanwhile.
 */
export async function shareCap(cap: ReadCap, toInbox: Nuri): Promise<void> {
  if (!hasReadCap(cap)) {
    throw new Error(
      "[ng-eventually] inbox.shareCap: expected a ReadCap (a NURI carrying `:r:`), " +
        `got a bare reference — naming is not reading: ${JSON.stringify(cap)}`,
    );
  }
  await post(toInbox, { payload: { kind: LINK_KIND, cap } });
}

// --- the read guard ------------------------------------------------------

/**
 * Refuse to READ an inbox that is not the current wallet's.
 *
 * Depositing into someone else's inbox is the one legitimate cross-wallet act (it
 * is how a link reaches another wallet at all — see {@link post} / {@link shareCap});
 * READING one is not, and it is not symmetric with it. Since caps travel as
 * deposits, an unguarded read let anyone who knew an inbox NURI collect the caps
 * addressed to its owner, which defeats directed sharing entirely.
 *
 * Anonymous owns no inbox, so it can read none — an identity has to be established
 * first. At migration this disappears: an inbox is sealed to its owner's key, and
 * the guard is the cryptography.
 */
async function assertOwnInbox(targetInbox: Nuri, op: string): Promise<void> {
  if (getCurrentUser() === null) {
    throw new Error(
      `[ng-eventually] inbox.${op}: no identity is set, so no inbox belongs to this ` +
        "session — call setCurrentUser() first. Depositing (post/shareCap) stays open.",
    );
  }
  if (!(await isOwnInbox(targetInbox))) {
    throw new Error(
      `[ng-eventually] inbox.${op}: refusing to read an inbox that does not belong to ` +
        "the connected wallet. You may DEPOSIT into anyone's inbox; you may only READ " +
        "your own — otherwise the caps addressed to its owner would be collectable by " +
        `whoever knows its NURI: ${JSON.stringify(targetInbox)}`,
    );
  }
}

// --- read --------------------------------------------------------------

/**
 * Read every deposit currently in `targetInbox`, sorted by `ts` ascending. In
 * real NextGraph the recipient's own verifier applies queued messages inline as
 * it processes the inbox; here this read stands in for that until the
 * sealed-inbox path is available. The consumer interprets each deposit's
 * `payload`.
 *
 * Cap deliveries ({@link shareCap}) are applied inline and NOT returned: they land
 * in what the current holder holds, like the verifier applying a queued message.
 * That is why receiving a cap needs no dedicated operation — a consumer already
 * watching its inbox gets them, and the resulting change re-triggers the
 * reads that were empty for want of that cap.
 */
export async function read(targetInbox: Nuri): Promise<Deposit[]> {
  await assertOwnInbox(targetInbox, "read");
  const sid = await sessionId();
  // NOTE: cold-start repo opening is done by the COLD DIRECT readers that need it
  // (a cold reader that opens the repo before reading), NOT here — `inbox.watch`
  // already holds the repo open via its own `subscribeDoc`, so opening a second
  // bootstrap subscription from inside a watch's re-read would be redundant and can
  // race the watch's own initial-`State` delivery. Keeping `read` a pure anchored
  // read leaves both callers correct: the watch path stays event-driven, and the
  // cold direct-read path opens the repo explicitly before calling `read`.
  // NO explicit `GRAPH <…>` clause — read the anchored DEFAULT graph (see the
  // note in `post`). The anchor (`targetInbox`) scopes the query to that repo's
  // default graph, exactly where `post` writes.
  const query = `
    SELECT ?payload ?ts ?from WHERE {
      ?d a <${P.type}> ;
         <${P.payload}> ?payload ;
         <${P.ts}> ?ts .
      OPTIONAL { ?d <${P.from}> ?from }
    }`;
  const result = await sparqlQuery(sid, query, undefined, targetInbox, "inboxRead");
  const deposits: Deposit[] = [];
  for (const row of readBindings(result)) {
    const rawPayload = row.payload?.value ?? "null";
    let payload: unknown;
    try {
      payload = JSON.parse(rawPayload);
    } catch {
      payload = rawPayload; // tolerate a non-JSON literal
    }
    const tsRaw = row.ts?.value ?? "0";
    const ts = Number.parseInt(tsRaw, 10) || 0;
    const fromValue = row.from?.value;
    deposits.push({ from: fromValue ? fromValue : null, payload, ts });
  }
  deposits.sort((a, b) => a.ts - b.ts);
  // Links are infrastructure, not consumer data: they never reach the caller. They
  // are only KEPT here (in memory, for this session) — FILING them durably is
  // `processInbox`'s job, because reading an inbox must not quietly write to a
  // user's store. Filing fires the registry's change signal, which is what makes a
  // view that was empty for want of that cap re-read instead of staying stale.
  const delivered: Deposit[] = [];
  const links: ReadCap[] = [];
  for (const d of deposits) {
    const cap = capOfPayload(d.payload);
    if (cap) {
      getCaps().learn(cap);
      links.push(cap);
      continue;
    }
    delivered.push(d);
  }
  if (links.length > 0) seenByInbox.set(targetInbox, links);
  // Domain-level diagnostic (on top of docs.ts's generic access-path READ log
  // of raw triple-rows): how many DEPOSITS were found, and the decoded data of
  // each — the exact visibility needed to trace materialization at the owner
  // side. Gated by the same access-log flag; skip the JSON work when off.
  if (accessLogEnabled()) {
    logAccess(
      "READ",
      targetInbox,
      "inbox materialize",
      " → " + delivered.length + " message(s)" +
        (deposits.length !== delivered.length
          ? " (+" + (deposits.length - delivered.length) + " cap deliver(y/ies) absorbed)"
          : ""),
    );
    for (const d of delivered) {
      logAccess(
        "READ",
        targetInbox,
        "inbox message",
        " ts=" + d.ts + " from=" + (d.from ?? "anonymous") + " payload=" + summarizePayload(d.payload),
      );
    }
  }
  return delivered;
}

/** Alias for {@link read} — the name that reads as "process the inbox now". */
export const materialize = read;

/**
 * COLD, BARRIER-GATED read of `targetInbox` — the reliable "process the inbox at
 * (re)connection" read. Opens/subscribes the inbox repo and AWAITS its first
 * `State` (the deterministic sync barrier — after it, presence is guaranteed and
 * absence definitive, {@link ensureRepoOpen}) BEFORE the anchored {@link read}.
 *
 * Why this over a plain {@link read}: on a FRESH session over the persistent
 * wallet (a (re)connection / new page), the inbox repo is not yet in the verifier's
 * `self.repos`, so a plain anchored `read` resolves an unopened repo and silently
 * returns 0 deposits — even for a deposit a remote session already synced to the
 * broker. Gating on the sync barrier makes the read see the synced deposits. This
 * is the same cold-read heal any cold direct reader needs.
 *
 * NOT for the `watch` path: {@link watch} already holds the repo open via its own
 * `subscribeDoc`, so opening a second bootstrap subscription from inside a watch
 * re-read would be redundant and could race the watch's own initial-`State`
 * delivery. Use this from a COLD reader (materialize-at-connection), like
 * `discovery.readIndex` does. Idempotent per session (no polling); a no-op open on
 * the unit fake-ng path (no `doc_subscribe`) so `bun test` is unaffected.
 */
export async function readSynced(targetInbox: Nuri): Promise<Deposit[]> {
  // Marks the cold, connection-triggered entry point in the trace — the BARRIER
  // line (open-repo.ts) and the "inbox materialize"/"inbox message" lines below
  // (from the read() this wraps) follow right after, so a live session shows
  // the whole owner-reconnect sequence together.
  logStage("READSYNCED " + shortNuri(targetInbox) + " (cold, barrier-gated)");
  await assertOwnInbox(targetInbox, "readSynced");
  await ensureRepoOpen(targetInbox);
  return read(targetInbox);
}

/**
 * PROCESS an inbox: read it, and **apply** what it contains.
 *
 * Applying a {@link shareCap} Link means filing it durably — `storeRegistry.addLink`,
 * the emulated `AddLink { read_cap }` on the User branch of the private store — so
 * the cap survives the session. Upstream this is what a verifier does when it
 * processes queued messages: an inbox is a **queue you consume**, not a store you
 * re-read. Re-reading an inbox every session to recover caps is using a queue as a
 * database, and it is the thing this replaces.
 *
 * Idempotent: `addLink` ignores a Link it already holds, so processing twice (a
 * second tab, a reconnect) costs nothing. Returns the consumer deposits, exactly as
 * {@link read} does — Links are never surfaced.
 */
export async function processInbox(targetInbox: Nuri): Promise<Deposit[]> {
  const deposits = await readSynced(targetInbox);
  // `readSynced` already put every Link in memory for this session; now make
  // them durable. Reading the raw deposits again would mean re-parsing, so the caps
  // are taken from what the read just observed.
  for (const cap of capsSeenIn(targetInbox)) await addLink(cap);
  seenByInbox.delete(targetInbox);
  return deposits;
}

/**
 * Subscription over an inbox — **event-driven, not polled**. Subscribes to the
 * inbox document via {@link subscribeDoc} (the platform's `doc_subscribe` push):
 * `onDeposits` fires once on the initial state push and again on every subsequent
 * change to the inbox document — a local deposit OR a broker-synced remote one.
 * Returns an unsubscribe function.
 *
 * On each push it re-reads the full deposit list ({@link read}) and invokes
 * `onDeposits` only when the deposit count changed (grew), keeping the same
 * "fires on change" contract the polling watcher had — same callback signature
 * and same behaviour, just event-driven instead of `setInterval`.
 *
 * The `intervalMs` option is accepted for signature compatibility but IGNORED:
 * there is no polling. (The inbox document is a single doc, so this is immune to
 * the ORM fan-out hang — see {@link subscribeDoc}.)
 */
export function watch(
  targetInbox: Nuri,
  onDeposits: (deposits: Deposit[]) => void,
  _opts?: { intervalMs?: number },
): () => void {
  let stopped = false;
  let lastCount = -1;

  // Re-read on every push; fire onDeposits only when the set changed (grew).
  const refresh = async (): Promise<void> => {
    if (stopped) return;
    try {
      const deposits = await read(targetInbox);
      const changed = deposits.length !== lastCount;
      // Owner-side processing decision: did this push actually grow the
      // deposit set (→ onDeposits fires, the polyfill's stand-in for
      // materialization) or was it a no-op push (→ skipped)? This is the
      // exact line to check for the "must reconnect an extra time" symptom:
      // a push whose read still sees the OLD count means the barrier/read
      // raced the write, not that watch itself failed to fire.
      if (accessLogEnabled()) {
        logAccess(
          "READ",
          targetInbox,
          "inbox watch",
          " → " + deposits.length + " message(s)" + (changed ? " (materializing)" : " (unchanged, skip)"),
        );
      }
      if (!stopped && changed) {
        lastCount = deposits.length;
        onDeposits(deposits);
      }
    } catch (error) {
      console.error(accessLogPrefix() + " watch read failed:", error);
    }
  };

  // Subscribe to the inbox document: the initial State push fires the first read
  // (so onDeposits fires once immediately, as before), each later Patch a re-read.
  // The ownership guard runs inside `read`, so a watch on someone else's inbox
  // yields nothing but logged refusals rather than their deposits.
  const unsubscribe = subscribeDoc(targetInbox, () => void refresh());
  return () => {
    stopped = true;
    unsubscribe();
  };
}