ng-eventually/packages/polyfill/src/emulated-verifier/inbox-processor.ts

/**
 * The inbox-processing SERVICE this deployment has not got — emulated by a timer.
 *
 * ── What is missing, and what stands in for it ────────────────────────────
 * Upstream a deposit does not wait for anybody: the broker routes the sealed message
 * (`inboxes: PubKey → RepoId`, `engine/verifier/src/verifier.rs:105,1677`) and the
 * recipient's verifier applies it when it processes the inbox
 * (`engine/verifier/src/inbox_processor.rs`). A real NextGraph deployment will have
 * something running continuously that does this; today, in this library, the ONLY thing
 * that drains an inbox is its owner connecting (`connect.connectedUser` →
 * `inbox.processInbox`), which can be hours away.
 *
 * So: a successful deposit arms a ONE-SHOT timer that processes the TARGET's inbox, twenty
 * seconds later. Alice deposits into Bob's inbox; twenty seconds later this package drains
 * Bob's inbox, while the connected session is still Alice's.
 *
 * ── That is identity usurpation, and it is deliberate ─────────────────────
 * It is only possible because one shared wallet holds every virtual identity, and it is
 * acceptable for exactly one reason: an application learns nothing false from it. What an
 * application observes is that **deposits converge** — a share becomes readable without the
 * recipient having to re-open the page — and that stays true when a real service takes over.
 * What must never leak is the MECHANISM: nothing here is published, there is no way for a
 * caller to ask for it, to name another user's inbox, or to turn it off. Publishing any of
 * that would publish a call that processes someone else's inbox, and the confidentiality of
 * the whole shared-wallet emulation rests on that being unreachable.
 *
 * ── Its accepted limit, stated rather than papered over ───────────────────
 * A timer lives in a page. If the page goes away before it fires, the deposit simply waits
 * for its owner's next connection — exactly where it waited before. There is deliberately
 * no persistence, no retry and no longer window to hide that: a stand-in for an absent
 * service should look like what it is, and the fallback (the owner connects and drains) is
 * the real path, not a repair.
 */

import { accessLogPrefix, logStage, shortNuri } from "../shared-wallet/access-log";
import type { Nuri } from "../model/types";

/**
 * How long a deposit waits before the emulated service picks it up.
 *
 * Not configurable, on purpose. A knob would be a published question — *how fast does
 * delivery converge?* — that the target does not ask a caller, because upstream the answer
 * belongs to the deployment. An application must be written so that the number does not
 * matter, and the surest way to keep it that way is to give nobody a way to change it.
 */
const PROCESSING_DELAY_MS = 20_000;

/** A drain waiting to happen: the timer, and what it will do. */
interface Scheduled {
  handle: ReturnType<typeof setTimeout>;
  process: () => Promise<void>;
}

/**
 * At most ONE pending drain per inbox — the coalescing this map exists for.
 *
 * Several deposits inside the window must produce ONE drain, not several: processing writes
 * what it applies (`branch-registers.addLink`, on the owner's User branch), so two runs over
 * the same inbox race each other or apply the same Link twice. The FIRST deposit arms the
 * window and later ones join it, rather than each pushing it further out — a busy inbox must
 * still converge, not be starved by its own traffic.
 */
const scheduled = new Map<Nuri, Scheduled>();

/**
 * Drains in flight, per inbox — so a run that starts while another is still going CHAINS
 * behind it instead of interleaving with it. Same reason `connect.connectedUser` drains its
 * inboxes sequentially: two concurrent passes over one queue apply the same records twice.
 */
const running = new Map<Nuri, Promise<void>>();

/**
 * Keep a pending drain from holding a runtime open.
 *
 * A browser's `setTimeout` answers a number and has no `unref`; a Node/Bun runtime answers a
 * timer object that does, and without it a process would sit for twenty seconds after its
 * last deposit waiting on work nobody asked for. Probed rather than assumed, because the two
 * platforms genuinely differ and this library runs on both.
 */
function releaseFromTheEventLoop(handle: ReturnType<typeof setTimeout>): void {
  const maybe: { unref?: unknown } = handle as unknown as { unref?: unknown };
  if (typeof maybe.unref === "function") (maybe.unref as () => void).call(handle);
}

/**
 * Where a deferred drain's failure goes.
 *
 * It cannot throw: nothing awaits it — the application never asked for this work, and an
 * unhandled rejection would take down whatever runtime it is in over a drain it did not
 * request. But "nowhere" is not the alternative. A drain that swallows its failure is
 * indistinguishable from one that succeeded, which is precisely the defect family removed
 * from this package on 2026-08-13. So it goes to the package's own log stream — the same
 * `console.error` with the same `[<identity>][polyfill]` prefix that `startConnect` and
 * `inbox.watch` use for their un-awaitable failures, and it is NOT gated by the access-log
 * flag: a diagnostic may be opt-in, a failure may not.
 */
function reportFailure(inbox: Nuri, error: unknown): void {
  console.error(
    accessLogPrefix() + " deferred inbox processing failed for " + shortNuri(inbox) + ":",
    error,
  );
}

/** Run the drain of `inbox`, behind any run of the same inbox still in flight. */
function fire(inbox: Nuri, process: () => Promise<void>): void {
  scheduled.delete(inbox);
  const previous = running.get(inbox) ?? Promise.resolve();
  const next = previous
    // A previous run that FAILED must not cancel this one: its failure was reported where
    // failures go, and the deposits it did not apply are exactly what this run is for.
    .catch(() => undefined)
    .then(process)
    .catch((error: unknown) => reportFailure(inbox, error))
    .then(() => {
      if (running.get(inbox) === next) running.delete(inbox);
    });
  running.set(inbox, next);
}

/**
 * Arm the emulated service for `inbox` — called by a deposit that LANDED, and by nothing
 * else. A write that failed has nothing to converge.
 *
 * `process` is what to do when the window closes; the scheduler itself knows nothing about
 * inboxes' contents, so the drain stays where the inbox's own vocabulary lives.
 *
 * Idempotent inside the window: a second deposit into the same inbox joins the pending run.
 */
// @provenance scheduleInboxProcessing kind=divergent level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream the RECIPIENT's own verifier applies its inbox; here another identity's session drains it on a timer. Deliberate, unpublished, and only possible on one shared wallet
export function scheduleInboxProcessing(inbox: Nuri, process: () => Promise<void>): void {
  if (scheduled.has(inbox)) return;
  const handle = setTimeout(() => fire(inbox, process), PROCESSING_DELAY_MS);
  releaseFromTheEventLoop(handle);
  scheduled.set(inbox, { handle, process });
}

/**
 * Close every open window NOW and wait for the drains to finish.
 *
 * **The emulation's clock, not a door.** It cannot name an inbox, so it can only bring
 * forward work a deposit has already armed — the same runs, at the same holder, with the
 * same effects, minus the twenty seconds. That is what makes it usable by a suite that has
 * to observe convergence without waiting for it, and useless as a way to reach anyone's
 * inbox. Never exported from the package.
 *
 * Resolves rather than rejects, like the timer path it stands in for: a drain's failure is
 * reported where failures go, and a caller of this is not the party that asked for the work.
 */
// @provenance runScheduledInboxProcessingNow kind=divergent level=1 ref=engine/verifier/src/verifier.rs:inbox — the same drain, run at once; a test seam over the divergence above
export async function runScheduledInboxProcessingNow(): Promise<void> {
  for (const [inbox, entry] of [...scheduled]) {
    clearTimeout(entry.handle);
    fire(inbox, entry.process);
  }
  await Promise.all([...running.values()]);
}

/**
 * Drop every pending drain without running it — what an un-configured library must do.
 *
 * Called by `resetConfig`: once the injected `ng` is gone, a drain cannot read or write
 * anything, so leaving armed timers behind would produce a burst of failures about a
 * session that no longer exists. In-flight runs are not cancelled (nothing can un-issue a
 * write already sent); they end where they end.
 */
// @provenance cancelScheduledInboxProcessing kind=divergent level=1 ref=engine/verifier/src/verifier.rs:inbox — cancels a pending drain — upstream there is no pending anything to cancel
export function cancelScheduledInboxProcessing(): void {
  for (const entry of scheduled.values()) clearTimeout(entry.handle);
  scheduled.clear();
}

/** Trace one completed drain — diagnostics, so gated by the access-log flag. */
// @provenance traceProcessed kind=invention level=none ref=none — a diagnostic line in this package's own log stream
export function traceProcessed(inbox: Nuri, holder: string, applied: number): void {
  logStage(
    "DEFERRED PROCESS " + shortNuri(inbox) + " for " + holder + " → " + applied + " link(s) applied",
  );
}