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

/**
 * While an identity is connected, its inboxes are WATCHED and what arrives is APPLIED.
 *
 * ── The regime this restores, and the one it replaces ─────────────────────
 * Upstream, applying an inbox is not something that happens at connection — it is what a
 * session DOES. A sealed message reaches the recipient's own verifier as it arrives
 * (`LocalBrokerMessage::Inbox` → `session.verifier.inbox(msg, from_queue)`,
 * `sdk/rust/src/local_broker.rs`), which unseals it and applies it inline
 * (`engine/verifier/src/inbox_processor.rs`); `from_queue` distinguishes the backlog handed
 * over at connection from the messages that follow, and BOTH go through the same door. So
 * the backlog is the special case, and the continuity is the rule.
 *
 * This package had emulated only the backlog. `inbox.processInbox` was called from exactly
 * one place — `connect.connectedUser`, at connection — and nothing anywhere applied a
 * deposit after that. A share deposited into an inbox whose owner was sitting connected in
 * front of it converged only when that person reloaded the page, and the deposit's own
 * session covered the gap by usurping the owner's identity on a twenty-second timer
 * (`emulated-verifier/inbox-processor.ts`) — which does nothing at all if the depositor
 * closes their tab, and tells the connected owner nothing either way.
 *
 * So: for as long as an identity is connected, every inbox it may read is subscribed to,
 * and every push over one of them applies what is in it.
 *
 * ── Watching is not applying ──────────────────────────────────────────────
 * `inbox.watch` looks like this and is not: it NOTIFIES an application that made a call and
 * named one inbox. Its read does absorb the caps it passes over into memory for the session
 * — that much is unavoidable, reading is what surfaces them — but it FILES nothing durably,
 * which only `processInbox` does. This applies, on the whole set, without anybody asking —
 * because processing an inbox is the library's job and not the app's, which is the same
 * ruling `connect.ts` opens with.
 *
 * ── Subscription, never a poll ────────────────────────────────────────────
 * Every half is push-driven. The inboxes themselves are subscribed to individually. WHICH
 * inboxes exist is itself a subscription, on two channels that do not overlap:
 *
 *   - the REGISTER — an inbox opened mid-session appends an `AddInboxCap` record to the
 *     User branch of the private store (`branch-registers.openDocumentInbox`), so a push on
 *     that document re-enumerates `myInboxes()`;
 *   - WHAT THIS IDENTITY HOLDS (`CapRegistry.onChange`) — the same signal `watchShape`
 *     re-reads on. It answers the case the register cannot: an identity that connects
 *     before it has an account at all. Provisioning is lazy here, so a person's FIRST visit
 *     connects with no stores, no private store to follow and no inbox to watch; the
 *     account appears later, the moment the application creates anything, and filing its
 *     caps is what says so. Without this channel a brand-new user was watched from their
 *     SECOND visit onwards — which is exactly the person most likely to be sent something.
 *
 * So a document whose inbox this identity opened five minutes into the session is watched
 * like the rest, and so is the very first inbox of a person who had none when they arrived.
 *
 * ── It belongs to ONE holder, and stops when that holder does ─────────────
 * Everything below resolves the CURRENT holder when it runs — `myInboxes`, `processInbox`
 * and everything under them ask `getCurrentUser()` at the moment they execute. So an
 * observation started for Alice must not still be firing when Bob is connected: it would
 * read Alice's registers under Bob, or file Alice's capabilities into Bob's ring. The
 * observation is therefore stamped with its holder, every step re-checks it, and switching
 * identity (or disconnecting) tears it down — the same rule, and the same reason,
 * `connect.connectedUser` carries `stillConnected()` for.
 */

import { getCaps, getCurrentUser } from "../shared-wallet/bootstrap";
import { accessLogPrefix, logStage, shortNuri } from "../shared-wallet/access-log";
import { enumerateMyInboxes, type InboxEnumeration, type InboxRegister } from "./branch-registers";
import { drainInboxSerially, drainsSettled } from "./inbox-drain";
import { lookupAccount } from "../shared-wallet/account-registry";
import { processInbox } from "../surface/inbox";
import { subscribeDoc, subscribeDocReportingSetupFailure, type Unsubscribe } from "../surface/subscribe";
import type { Nuri, PrincipalId } from "../model/types";

/**
 * One identity's live observation. Held whole rather than as loose module variables so
 * that a step started under it can ask "am I still the current one" by IDENTITY of the
 * object, not by comparing a name that a re-connection may have restored in between.
 */
interface Observation {
  /** WHO this observation belongs to. Every step re-checks it before acting. */
  holder: PrincipalId;
  /** One subscription per inbox observed, keyed by it. */
  inboxes: Map<Nuri, Unsubscribe>;
  /** The subscription on the register that says WHICH inboxes exist. */
  register: Unsubscribe | null;
  /** Unsubscribe from the held-caps change signal — the second "which inboxes" channel. */
  holdings: Unsubscribe | null;
  /** True while a re-enumeration is running, so its own effects do not restart it. */
  enumerating: boolean;
  /** A trigger that arrived mid-cycle: the running one repeats once rather than lose it. */
  enumerateAgain: boolean;
  /**
   * WHICH shortfall was reported last, or `null` when the last list came back WHOLE — so a
   * register that stays unreadable is reported once, not once per cycle.
   *
   * Every signal this observation listens to runs a full cycle, and a persistent shortfall
   * is re-read by each of them, so the condition is reported as many times as the identity
   * is nudged: one deposit produced three copies of the same line. That is what buries the
   * report that matters under the report that repeats. Cleared the moment a list comes back
   * WHOLE, so a shortfall that goes away and returns is a new occurrence and says so.
   *
   * ── The register, and not a bare "already said something" ─────────────────
   * There are TWO shortfalls, they are disjoint, and one is far worse than the other: the
   * account record failing leaves this identity with NOTHING watched, its own two inboxes
   * included, while the User branch failing leaves those two watched and only the
   * per-document inboxes missing ({@link InboxEnumeration.incomplete}). A boolean here made
   * the second silence the first — the condition CHANGED, to a materially different one, and
   * the log said nothing because something had already been said. Keyed by register, a
   * change of shortfall is what it is: news.
   */
  shortfallReported: InboxRegister | null;
  /** Enumerations and applications in flight — what {@link observationSettled} waits on. */
  pending: Set<Promise<void>>;
}

let observation: Observation | null = null;

/**
 * Where a failure to WATCH goes — distinct from a failure to APPLY, which
 * `inbox-drain.ts` reports.
 *
 * Not a rejection: nothing awaits this work, exactly as at connection. And not silence, for
 * the same reason as everywhere else in this package — an observation that cannot be
 * established leaves deposits unapplied for the whole session, and the only thing a person
 * would notice is that a share never arrives.
 */
function reportUnobserved(what: string, error: unknown): void {
  console.error(
    accessLogPrefix() + " connected, but " + what + " — deposits made from now on may not be " +
      "applied until the next connection:",
    error,
  );
}

/**
 * What a shortfall on `register` leaves unwatched, in the words the log uses.
 *
 * The two lines have to READ differently, not merely be counted differently: a reader who
 * sees the second one after the first has to be able to tell that the condition changed, and
 * two identical sentences make an escalation look like a repeat. Both keep the phrase the
 * report has always ended on, because it is the one thing this line is scanned for.
 */
function shortfallSaid(register: InboxRegister): string {
  return register === "account-record"
    ? "the inboxes to watch could not all be listed (the account record — your own two " +
        "inboxes, where a share addressed to you by name lands, are NOT among the ones watched)"
    : "the inboxes to watch could not all be listed (the register of the inboxes opened on " +
        "documents — your own two are watched)";
}

/** Is `obs` still the live observation, for the identity it belongs to? */
function current(obs: Observation): boolean {
  return observation === obs && getCurrentUser() === obs.holder;
}

/** Track a piece of in-flight work so {@link observationSettled} can wait for it. */
function track(obs: Observation, work: Promise<void>): void {
  obs.pending.add(work);
  void work.finally(() => obs.pending.delete(work));
}

/**
 * Apply whatever is in `inbox`, for the holder this observation belongs to.
 *
 * Through the shared per-inbox queue, so this cannot interleave with a drain the deferred
 * timer started on the same inbox (nor with another push of its own): two passes over one
 * queue apply the same records twice. A failure is reported there and stops nothing — the
 * next push over this inbox tries again, and the other inboxes were never involved.
 */
async function applyWhatArrived(obs: Observation, inbox: Nuri): Promise<void> {
  if (!current(obs)) return;
  await drainInboxSerially(inbox, async () => {
    // Re-checked INSIDE the queue: this run may have waited behind another one, and the
    // identity can have moved while it waited. Running it for the wrong holder is not a
    // near-miss — it reads someone else's registers and files into someone else's ring.
    //
    // This check is NECESSARY and it is not SUFFICIENT, and until 2026-08-17 this comment
    // claimed it was. It cannot be: the identity can move after it passes, while
    // `processInbox` is mid-read. The claim was that `processInbox` "resolves the holder at
    // each step, so the new holder is refused an inbox that is not theirs somewhere in the
    // middle" — but its ownership guard runs ONCE, at entry, and a switch after that reached
    // the filing with nobody left to refuse it. What it filed was the previous holder's cap,
    // into the new holder's ring, durably. The guard that makes this run safe is therefore
    // the one INSIDE `processInbox`, which captures the holder its guard authorised; this one
    // only spares the work when the switch is already visible.
    if (!current(obs)) return;
    try {
      await processInbox(inbox);
    } catch (error) {
      // An identity that moved MID-DRAIN can throw here too — the switch may land before
      // `processInbox`'s own ownership guard, which then refuses the new holder an inbox that
      // is not theirs. (Landing after it, the run ABANDONS quietly instead and returns; both
      // leave the deposit where it is.) Abandoning is what this package has always called
      // "not a failure", and reporting it would put a broker-looking error in the log every
      // time a page switches user. Nothing is lost: abandoning happens BEFORE the deposit it
      // stopped at is taken out of the queue (`emulated-verifier/inbox-consume.ts` asks
      // whether the holder still stands, then pops), so that deposit and every one behind it
      // are still there for its owner's next connection. What had already been popped was
      // applied for the holder that was authorised at the time. Same rule, same words, as the
      // drain loop in `connect.connectedUser`.
      if (!current(obs)) return;
      throw error;
    }
  });
}

/**
 * (Re-)enumerate the inboxes this identity may read and subscribe to the ones not yet
 * watched. Idempotent: an inbox already observed is left alone, so a re-enumeration costs
 * one read and nothing else.
 *
 * Never removes: the list only grows within a session (an `AddInboxCap` record is durable),
 * and dropping a subscription on a guess would silently stop applying an inbox.
 *
 * ── A list that came back SHORT still gets watched ────────────────────────
 * `enumerateMyInboxes` and not `myInboxes`: the list is read from two independent registers
 * and one of them being unreachable used to discard the other's answer too, so a broker
 * hiccup spanning sign-in left the identity connected with ZERO inboxes watched rather than
 * with one register's worth missing. Whatever came back is watched here and now — including,
 * crucially, the user's own two inboxes, which are where a share addressed to a PERSON
 * lands, and therefore the channel by which a read-only identity converges at all.
 *
 * The shortfall is reported, never passed over: an incomplete list is not a complete one,
 * and the inboxes it is missing are watched on the next enumeration — which is what the
 * register push and the holdings signal are for. Applying what lands in a user inbox is
 * itself one of those signals (it files a cap), so the ordinary case repairs itself.
 *
 * Once per OCCURRENCE, though, not once per enumeration — see
 * {@link Observation.shortfallReported}. The retry that repairs it is also what re-reads it,
 * so a condition that persists is re-discovered by every signal; reporting each discovery
 * says "it happened again" about the single thing that never stopped happening. An
 * occurrence is per REGISTER, because the two shortfalls are two conditions: the list going
 * from "missing the document inboxes" to "missing your own two" is not the same fault
 * continuing, it is a worse one starting.
 */
async function watchTheInboxes(obs: Observation): Promise<void> {
  if (!current(obs)) return;
  let listed: InboxEnumeration;
  try {
    listed = await enumerateMyInboxes();
  } catch (error) {
    // Not knowing WHICH inboxes exist AT ALL is the whole observation failing, not one queue.
    reportUnobserved("the inboxes to watch could not be listed", error);
    return;
  }
  if (!current(obs)) return;
  if (listed.incomplete === null) {
    obs.shortfallReported = null;
  } else if (obs.shortfallReported !== listed.incomplete.register) {
    obs.shortfallReported = listed.incomplete.register;
    reportUnobserved(shortfallSaid(listed.incomplete.register), listed.incomplete.error);
  }
  for (const inbox of listed.inboxes) {
    if (obs.inboxes.has(inbox)) continue;
    try {
      // The push is the signal; the drain is the work. The FIRST push is the initial
      // `State`, which drains an inbox connection has usually just drained — idempotent,
      // and the alternative (skip the first) would lose a deposit that landed in the gap
      // between the two.
      //
      // The failure channel is not decoration: `subscribeDoc` returns SYNCHRONOUSLY and its
      // one real `doc_subscribe` is opened afterwards, so a rejection there reaches nobody.
      // Without it, this map held an entry for an inbox that was never watched, the loop
      // above skipped it at every later enumeration, and the only observable difference from
      // a healthy session was that shares stopped arriving.
      obs.inboxes.set(
        inbox,
        subscribeDocReportingSetupFailure(
          inbox,
          () => track(obs, applyWhatArrived(obs, inbox)),
          (error) => watchFailed(obs, inbox, error),
        ),
      );
      logStage("OBSERVING " + shortNuri(inbox) + " for " + obs.holder);
    } catch (error) {
      reportUnobserved("this inbox could not be watched: " + shortNuri(inbox), error);
    }
  }
}

/**
 * The subscription on `inbox` could not be OPENED — report it, and forget it.
 *
 * **Forget it**, because the entry this observation holds is the whole record of "already
 * watched": leaving a dead one in place is how one rejection at connection turned into a
 * session-long silence. Removed, the next enumeration subscribes again as if it had never
 * been attempted.
 *
 * ── And that is ALL it does: recovery waits for the next SIGNAL ───────────
 * Until 2026-08-17 this also re-enumerated on the spot, once per inbox. That call runs
 * synchronously out of the very rejection it is reacting to, so it asks the broker that has
 * just refused, in the same turn, with nothing having changed — and spends the one extra
 * attempt it was allowed doing it. Deleting it left the whole suite green, which is the
 * measure of what it achieved.
 *
 * What actually repairs this is an event, and this module already listens to the two that
 * exist: the register push (`watchTheRegister`) and the held-caps signal
 * (`CapRegistry.onChange`), each of which re-enters {@link enumerate} and finds this inbox
 * unwatched. Applying anything at all fires the second one, so a session that is doing
 * something catches up on its own. A session that is doing nothing waits for its next
 * connection — where an unapplied deposit has always waited, since an inbox is not consumed
 * by being unread. Polling for it is ruled out here for the same reason it is in
 * `inbox-processor.ts`: this regime is push-driven, and a timer is what it replaced.
 */
function watchFailed(obs: Observation, inbox: Nuri, error: unknown): void {
  if (!current(obs)) return;
  const stale = obs.inboxes.get(inbox);
  obs.inboxes.delete(inbox);
  if (stale) {
    try {
      stale();
    } catch (thrown) {
      console.error(accessLogPrefix() + " releasing a failed inbox observation failed:", thrown);
    }
  }
  reportUnobserved("this inbox could not be watched: " + shortNuri(inbox), error);
}

/**
 * Follow the register that says which inboxes exist, so one opened MID-SESSION is picked
 * up. See the module header: `openDocumentInbox` appends its record to the User branch of
 * the private store, and a same-session write pushes to its own subscribers.
 *
 * Idempotent, and RETRIED rather than given up on: an identity that connects before it has
 * an account has no private store to follow yet, and acquires one the first time the
 * application creates anything. Re-entered from the holdings signal, it catches up then.
 */
async function watchTheRegister(obs: Observation): Promise<void> {
  if (!current(obs) || obs.register !== null) return;
  let store: Nuri | undefined;
  try {
    // `lookupAccount`, not `resolveAccount`: the tolerant form answers `null` for a read
    // that FAILED exactly as for one that found nothing, and the two could not mean more
    // different things here — "this identity has no account yet" is a state to wait
    // quietly through, "the broker did not answer" is a failure to report. The same
    // distinction `connect.connectedUser` and `inbox.share` are built on.
    store = (await lookupAccount(obs.holder))?.docPrivate;
  } catch (error) {
    reportUnobserved("the register of your inboxes could not be reached", error);
    return;
  }
  // No account yet — nothing to follow, and asking for one would PROVISION it, which
  // connecting deliberately does not do (`connect.connectedUser`). The holdings signal
  // brings us back here the moment there is something to follow.
  if (!store || !current(obs) || obs.register !== null) return;
  try {
    obs.register = subscribeDoc(store, () => track(obs, enumerate(obs)));
  } catch (error) {
    reportUnobserved("an inbox opened later in this session will not be watched", error);
  }
}

/**
 * The whole "which inboxes am I watching" cycle: catch up on the register, then on the
 * inboxes. Both halves are idempotent, so this is what every trigger runs.
 *
 * A trigger that arrives while a cycle is running does not start a second one — the cycle
 * FILES caps for the inboxes it lists (`myInboxes` → `fileOwnInbox`), which is one of the
 * signals that runs it, so overlapping runs would pile up on their own effects. It is
 * REMEMBERED instead of dropped: the running cycle repeats once more at the end, because a
 * signal that arrived mid-cycle may be about something that cycle had already read past.
 *
 * That terminates rather than ping-ponging: filing a cap this identity already holds fires
 * nothing (`caps.file` notifies only on a genuinely new one), so the repeat that finds
 * nothing new sets no flag and the loop ends.
 */
async function enumerate(obs: Observation): Promise<void> {
  if (!current(obs)) return;
  if (obs.enumerating) {
    obs.enumerateAgain = true;
    return;
  }
  obs.enumerating = true;
  try {
    do {
      obs.enumerateAgain = false;
      await watchTheRegister(obs);
      await watchTheInboxes(obs);
    } while (obs.enumerateAgain && current(obs));
  } finally {
    obs.enumerating = false;
  }
}

/**
 * Start observing the connected identity's inboxes. Called at the end of the connection
 * work (`connect.connectedUser`), and idempotent: calling it again for the identity already
 * being observed changes nothing.
 *
 * ── It resolves when the WATCHING is in place, not when the applying is done ──
 * Connecting publishes one guarantee — *the work it starts, it finishes* — and "your
 * inboxes are now being watched" is part of that work, so this is awaited rather than
 * fired. What it does NOT wait for is the applying: subscribing pushes an initial `State`,
 * which drains each inbox again, and making a person's sign-in wait on a second pass over
 * a backlog `connectedUser` has just applied would buy nothing.
 *
 * It never rejects. Failing to WATCH is reported where failures go and denies nobody their
 * session — the same rule the drain loop above it follows, and for the same reason: what
 * has not been DELIVERED stays in its queue for the next attempt. (Delivered, not applied:
 * since 2026-08-26 a deposit is popped before its arm runs, so one whose arm then failed is
 * gone — see `emulated-verifier/inbox-consume.ts`. Failing to WATCH takes nothing out.)
 */
// @provenance startObservingInboxes kind=aligned level=1 ref=engine/verifier/src/verifier.rs:inbox — upstream a connected verifier is handed each inbox message as it arrives and applies it; here the arrival signal is a document push, and the set of inboxes is the User branch's `AddInboxCap` records
export async function startObservingInboxes(): Promise<void> {
  const holder = getCurrentUser();
  if (holder === null) return;
  if (observation !== null && observation.holder === holder) return;
  stopObservingInboxes();
  const obs: Observation = {
    holder,
    inboxes: new Map(),
    register: null,
    holdings: null,
    enumerating: false,
    enumerateAgain: false,
    shortfallReported: null,
    pending: new Set(),
  };
  observation = obs;
  // The second "which inboxes exist" channel — see the module header. It is what makes a
  // person's FIRST session watched at all: they connect holding nothing, and their account
  // (with its inboxes) appears the moment the application creates something.
  obs.holdings = getCaps().onChange(() => track(obs, enumerate(obs)));
  const setup = enumerate(obs);
  track(obs, setup);
  await setup;
}

/**
 * Stop observing — on an identity change, on disconnection, and when the library is
 * un-configured.
 *
 * Work already inside the drain queue is not cancelled (nothing can un-issue a write
 * already sent); it is neutralised instead, because every step re-checks that its
 * observation is still the current one and returns rather than filing for the wrong holder.
 */
// @provenance stopObservingInboxes kind=aligned level=1 ref=engine/verifier/src/verifier.rs:Verifier — a verifier's inbox processing lives and dies with its session; switching identity is another session, so nothing of the previous one keeps applying
export function stopObservingInboxes(): void {
  const obs = observation;
  if (obs === null) return;
  observation = null;
  for (const unsub of obs.inboxes.values()) {
    try {
      unsub();
    } catch (error) {
      console.error(accessLogPrefix() + " stopping an inbox observation failed:", error);
    }
  }
  obs.inboxes.clear();
  for (const unsub of [obs.register, obs.holdings]) {
    if (!unsub) continue;
    try {
      unsub();
    } catch (error) {
      console.error(accessLogPrefix() + " stopping an inbox observation failed:", error);
    }
  }
  obs.register = null;
  obs.holdings = null;
}

/**
 * Wait until the observation has nothing in flight — the enumerations it is running and the
 * drains they started.
 *
 * A seam for a suite that has to observe convergence without waiting for it, in the same
 * spirit as `runScheduledInboxProcessingNow`: it names no inbox and starts no work, so it
 * can only wait for what a push has already caused. Never exported from the package.
 */
// @provenance observationSettled kind=invention level=none ref=none — a test seam over the observation; upstream nothing exposes "has the verifier finished applying"
export async function observationSettled(): Promise<void> {
  for (;;) {
    const obs = observation;
    const inFlight = obs ? [...obs.pending] : [];
    if (inFlight.length === 0) break;
    await Promise.all(inFlight);
  }
  await drainsSettled();
}