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 { sparqlUpdate, sparqlQuery } from "./docs";
import { subscribeDoc } from "./subscribe";
import { getCurrentUser, getStoreRegistryDeps } from "./polyfill";
import { escapeLiteral } from "./sparql";
import type { Nuri, PrincipalId } 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;
}

// --- 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} .
    }`;
  await sparqlUpdate(sid, update, targetInbox, "deposit");
}

// --- 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`.
 */
export async function read(targetInbox: Nuri): Promise<Deposit[]> {
  const sid = await sessionId();
  // 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);
  return deposits;
}

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

/**
 * 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);
      if (!stopped && deposits.length !== lastCount) {
        lastCount = deposits.length;
        onDeposits(deposits);
      }
    } catch (error) {
      console.error("[inbox] 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.
  const unsubscribe = subscribeDoc(targetInbox, () => void refresh());
  return () => {
    stopped = true;
    unsubscribe();
  };
}