SRCmail/lib/offline-replica/sync.ts

// One sync cycle. Request-scoped, bounded, single-flighted.
//
// PROCESS ARCHITECTURE - the decision that keeps most of the design review's
// findings out of scope. There is NO persistent background worker. A cycle is
// ordinary work inside an API route, using the request's own encrypted
// `jmap_stalwart_ctx` cookie via `lib/stalwart/credentials.ts`, exactly as the
// search index already does. That is deliberate, because the adversarial review
// of the original full-replica design killed four of its findings by removing the
// worker rather than fixing them:
//
//   C2 - "credentials are request-scoped, so no persistent worker can hold them".
//        Still true, and still fine, because there is no worker. A cycle only ever
//        reads an ALREADY-MINTED Authorization header off the request.
//   C3 - the OAuth-refresh mitigation being itself the bug. Avoided by
//        construction: nothing here touches the refresh-token cookie, so it cannot
//        rotate a token into a response nobody reads and log the user out.
//   C4 - a shared `registry.json` breaking the epoch fencing token. No registry,
//        no epochs: single-flight per account inside one process, and every piece
//        of state lives in the SQLite file under a real transaction.
//   H1 - "a server-side engine cannot read a renderer-only setting". The renderer
//        decides when to sync, so nothing materialises for an account that never
//        opted in. The retention POLICY does need to be durable server-side, so it
//        lives inside the encrypted store (written through PUT /api/offline/status),
//        not in renderer localStorage.
//   H4 - unbounded concurrent multi-account sync starving foreground activity.
//        One request, one account, one cycle, hard budgets below.
//
// H2 (key handoff via process environment) was already fixed by what shipped: the
// key crosses on an inherited file descriptor and is zeroed after each job.
// C1 (the native dependency breaking Alpine `docker build`s) is likewise already
// fixed and this file adds no new dependency.
//
// H3 is the one finding that genuinely COMES BACK. The webmail does local delta
// arithmetic on mailbox unread counts and totals for mark-read/move/delete, and a
// read-only offline cache sitting underneath that arithmetic needs a coherence
// story. The story is in `read.ts`: the replica is a FALLBACK, never a cache in
// front of the server. It is consulted only after a read has actually failed at
// the transport level, so an online session never sees a replica count.
//
// JOB ORDER within a cycle is A1 (Mailbox/changes) -> A2 (Email/changes) ->
// B (coverage enumeration) -> C1 (body queue drain) -> C2 (body backfill) ->
// retention. The three machines are logically independent but OPERATIONALLY
// SERIALISED. Either job order is safe; CONCURRENCY is not - coverage's
// query-then-apply pair interleaved with the delta path's apply resurrects a
// destroyed message as a zombie no future `/changes` page will ever re-report.
// "Run bodies in parallel, it's separate state" is forbidden for the same reason.

import type { Email, Mailbox } from '@/lib/jmap/types';
import { logger } from '@/lib/logger';
import type { JmapSessionInfo } from '@/lib/mail-index/jmap';
import {
  madeForwardProgress, planEmailFetches, planMailboxFetches, pageIsEmpty, advanceOneMs,
} from './apply';
import {
  backoffDelayMs, escalationApplies, movesCursor, nextRung, ReplicaSyncError, rungValue,
} from './errors';
import {
  AnchorNotFoundError, captureSnapshotStates, getEmailChanges, getEmails, getMailboxChanges,
  getMailboxes, maxObjectsInGet, queryAscending,
} from './jmap';
import {
  adjustForWindow, computeFloors, floorMovement, guardFloorAgainstClockJump,
} from './retention';
import { CURSOR_TYPES, mintEnumerationCommitment, type CursorType } from './states';
import { ReplicaStore, type RetentionPolicy } from './store';
import type { BodyQueueEntry, CoverageState, EnvelopeRow, MailboxRow } from './types';

// ── budgets ──────────────────────────────────────────────────────────────────
// A cycle runs inside an HTTP request the renderer is waiting on, so the wall
// clock matters more than page counts. Budget exhaustion is a NORMAL outcome
// reported as `unfinishedWork`, never an error.
export const BUDGET = {
  changesPagesPerCursor: 20,
  coveragePages: 12,
  coveragePageSize: 200,
  bodyItems: 60,
  bodyFetchChunk: 10,
  envelopeFetchChunk: 50,
  wallClockMs: 45_000,
} as const;

export const MAX_BODY_ATTEMPTS = 5;
/** Reconciles per rolling 24 h before throttling. Never a hard stop. */
export const MAX_RECONCILES_PER_DAY = 4;

/**
 * The reconcile stamp, derived from the DATA and not from the clock.
 *
 * Exported so it is testable on its own, because getting it wrong fails SILENTLY
 * in both directions. With a frozen or coarse clock, a plain `now` leaves
 * `cached_at < stamp` matching nothing and the sweep deletes nothing at all. And
 * because the pin routinely EXCEEDS `now`, any enumeration path that stamps with
 * `now` instead of the pin leaves its rows below the pin and gets them deleted by
 * the very sweep that just re-verified them against the server.
 */
export function reconcileStamp(now: number, maxCachedAt: number): number {
  return Math.max(now, maxCachedAt + 1);
}

export interface CycleContext {
  store: ReplicaStore;
  session: JmapSessionInfo;
  authHeader: string;
  jmapAccountId: string;
  policy: RetentionPolicy;
  now: number;
  deadline: number;
}

export interface CycleReport {
  ok: boolean;
  /** True when budgets ran out or a queue still has wanted work - chain another cycle. */
  unfinishedWork: boolean;
  bootstrapped: boolean;
  reconciled: boolean;
  mailboxesWritten: number;
  envelopesWritten: number;
  envelopesDeleted: number;
  bodiesWritten: number;
  bodiesEvicted: number;
  coveragePhase: CoverageState['phase'];
  resyncRequired: boolean;
  warnings: string[];
  errorClass?: string;
  error?: string;
  retryAfterMs?: number;
  durationMs: number;
}

// ── row conversion ───────────────────────────────────────────────────────────

function toEnvelopeRow(jmapAccountId: string, email: Email): EnvelopeRow | null {
  // `received_at` is NOT NULL and drives every window, so a record without one
  // cannot be stored. In practice the server always sends it for a full envelope
  // fetch; skipping is the safe response to a malformed one.
  if (typeof email.receivedAt !== 'string' || email.receivedAt.length === 0) return null;
  return {
    jmapAccountId,
    id: email.id,
    threadId: email.threadId ?? null,
    receivedAt: email.receivedAt,
    size: typeof email.size === 'number' ? email.size : null,
    subject: email.subject ?? null,
    preview: email.preview ?? null,
    fromJson: email.from ? JSON.stringify(email.from) : null,
    toJson: email.to ? JSON.stringify(email.to) : null,
    ccJson: email.cc ? JSON.stringify(email.cc) : null,
    blobId: email.blobId ?? null,
    hasAttachment: email.hasAttachment === true,
    keywordsJson: JSON.stringify(email.keywords ?? {}),
    // The BARE JMAP mailbox ids, never the display layer's `<accountId>:<id>`
    // prefixed form. The store already keys every row by (jmapAccountId, id), so
    // a prefixed id would double-encode the account and break every lookup.
    mailboxIds: Object.entries(email.mailboxIds ?? {})
      .filter(([, v]) => v)
      .map(([k]) => k),
  };
}

function toMailboxRow(jmapAccountId: string, mailbox: Mailbox): MailboxRow {
  return {
    jmapAccountId,
    id: mailbox.originalId ?? mailbox.id,
    name: mailbox.name ?? '',
    parentId: mailbox.parentId ?? null,
    role: mailbox.role ?? null,
    sortOrder: typeof mailbox.sortOrder === 'number' ? mailbox.sortOrder : null,
    totalEmails: typeof mailbox.totalEmails === 'number' ? mailbox.totalEmails : null,
    unreadEmails: typeof mailbox.unreadEmails === 'number' ? mailbox.unreadEmails : null,
    totalThreads: typeof mailbox.totalThreads === 'number' ? mailbox.totalThreads : null,
    unreadThreads: typeof mailbox.unreadThreads === 'number' ? mailbox.unreadThreads : null,
    myRightsJson: mailbox.myRights ? JSON.stringify(mailbox.myRights) : null,
    isSubscribed: mailbox.isSubscribed !== false,
  };
}

/** The body tier as one opaque JSON blob. */
export function serialiseBody(email: Email): string {
  return JSON.stringify({
    sentAt: email.sentAt,
    bcc: email.bcc,
    replyTo: email.replyTo,
    textBody: email.textBody,
    htmlBody: email.htmlBody,
    bodyValues: email.bodyValues,
    attachments: email.attachments,
    messageId: email.messageId,
    inReplyTo: email.inReplyTo,
    references: email.references,
    headers: email.headers,
    bodyStructure: email.bodyStructure,
  });
}

// ── the cycle ────────────────────────────────────────────────────────────────

export async function runCycle(ctx: CycleContext): Promise<CycleReport> {
  const started = Date.now();
  const report: CycleReport = {
    ok: true,
    unfinishedWork: false,
    bootstrapped: false,
    reconciled: false,
    mailboxesWritten: 0,
    envelopesWritten: 0,
    envelopesDeleted: 0,
    bodiesWritten: 0,
    bodiesEvicted: 0,
    coveragePhase: 'never-run',
    resyncRequired: false,
    warnings: [],
    durationMs: 0,
  };

  const { store, jmapAccountId, now } = ctx;
  const flags = store.getFlags(now);
  const rawFloors = computeFloors(ctx.policy, now);

  // Intent vs glitch. Without this discriminator the clock guard also fires on a
  // legitimate user retention change, leaving a Settings edit unapplied until
  // some unrelated trigger happened to move the floor again.
  const policyChanged =
    flags.lastEnvelopeDays !== undefined && flags.lastEnvelopeDays !== ctx.policy.envelopeDays;
  const guarded = guardFloorAgainstClockJump(rawFloors.envelopeFrom, flags.lastWindowFloor, {
    policyChanged,
  });
  if (guarded.warning) {
    report.warnings.push(guarded.warning);
    logger.warn('offline-replica: retention floor suppressed', { warning: guarded.warning });
  }
  const envelopeFrom = guarded.envelopeFrom;
  // Clamp a second time after the guard: a suppressed envelope floor can end up
  // NEWER than the computed body floor, and a body without an envelope is an orphan.
  const bodyFrom = rawFloors.bodyFrom > envelopeFrom ? rawFloors.bodyFrom : envelopeFrom;

  try {
    let coverage = store.getCoverage(jmapAccountId);

    // ── bootstrap / reconcile ────────────────────────────────────────────────
    if (!coverage || coverage.phase === 'never-run') {
      await beginEnumeration(ctx, envelopeFrom, 'bootstrap');
      report.bootstrapped = true;
      coverage = store.getCoverage(jmapAccountId);
    } else if (flags.resyncRequired && coverage.phase !== 'reconciling') {
      if (reconcileBudgetAllows(store, flags, now)) {
        await beginEnumeration(ctx, envelopeFrom, 'reconcile');
        report.reconciled = true;
        coverage = store.getCoverage(jmapAccountId);
      } else {
        // Throttle, never stop: a hard stop would trade a reconcile loop for a
        // permanent wedge.
        report.warnings.push('reconcile throttled: more than 4 rebuilds in the last 24h');
      }
    }

    // ── A1 / A2: drain the two /changes cursors ──────────────────────────────
    let anyCursorPending = false;
    for (const type of CURSOR_TYPES) {
      if (Date.now() > ctx.deadline) { report.unfinishedWork = true; break; }
      const drained = await drainCursor(ctx, type, bodyFrom, report);
      if (drained.pending) anyCursorPending = true;
    }
    if (anyCursorPending) report.unfinishedWork = true;

    // ── B: coverage enumeration ──────────────────────────────────────────────
    coverage = store.getCoverage(jmapAccountId);
    if (coverage && (coverage.phase === 'scanning' || coverage.phase === 'reconciling')) {
      const scanned = await runCoverage(ctx, coverage, bodyFrom, guarded.evictionAllowed, report);
      if (scanned.unfinished) report.unfinishedWork = true;
    }

    // ── C1 / C2: bodies ──────────────────────────────────────────────────────
    const drainedBodies = await drainBodyQueue(ctx, report);
    const backfilled = await backfillBodies(ctx, bodyFrom, rawFloors.maxBodyBytes, drainedBodies);
    if (backfilled > 0) report.unfinishedWork = true;
    if (store.countWantedBodies(jmapAccountId, Date.now()) > 0) report.unfinishedWork = true;

    // ── retention ────────────────────────────────────────────────────────────
    applyRetention(ctx, {
      envelopeFrom,
      bodyFrom,
      maxBodyBytes: rawFloors.maxBodyBytes,
      evictionAllowed: guarded.evictionAllowed,
      previousFloor: flags.lastWindowFloor,
      report,
    });

    const finalCoverage = store.getCoverage(jmapAccountId);
    const finalFlags = store.getFlags(now);
    report.coveragePhase = finalCoverage?.phase ?? 'never-run';
    report.resyncRequired = finalFlags.resyncRequired;

    store.transaction(() => {
      store.patchFlags(now, {
        // The USED floor, never the rejected one.
        lastWindowFloor: guarded.nextLastWindowFloor,
        lastEnvelopeDays: ctx.policy.envelopeDays,
        lastMaxBodyBytes: rawFloors.maxBodyBytes,
        lastCycleAt: now,
        lastCycleOk: true,
        lastCycleError: undefined,
      });
    });
  } catch (error) {
    report.ok = false;
    if (error instanceof ReplicaSyncError) {
      report.errorClass = error.cls;
      report.error = error.message;
      report.retryAfterMs = error.retryAfterMs;
      // "Offline is not an error" - a transport failure leaves every cursor
      // exactly where it was and is simply retried later.
      report.unfinishedWork = error.cls !== 'Fatal' && error.cls !== 'Auth';
    } else {
      report.error = error instanceof Error ? error.message : String(error);
      report.unfinishedWork = false;
    }
    try {
      store.transaction(() => {
        store.patchFlags(now, { lastCycleAt: now, lastCycleOk: false, lastCycleError: report.error });
      });
    } catch { /* the store may be the thing that failed */ }
  }

  report.durationMs = Date.now() - started;
  return report;
}

// ── bootstrap / reconcile ────────────────────────────────────────────────────

function reconcileBudgetAllows(
  store: ReplicaStore,
  flags: { reconcilesInWindow: number; reconcileWindowStartedAt: number },
  now: number,
): boolean {
  const dayMs = 24 * 60 * 60 * 1000;
  if (now - flags.reconcileWindowStartedAt > dayMs) {
    store.transaction(() => {
      store.patchFlags(now, { reconcilesInWindow: 0, reconcileWindowStartedAt: now });
    });
    return true;
  }
  return flags.reconcilesInWindow < MAX_RECONCILES_PER_DAY;
}

/**
 * THE MANDATORY ORDER. Step 1 must precede step 3.
 *
 *  1. capture both cursors, in one request, BEFORE touching any data, and seed
 *     them inside one EnumerationCommitment that writes the coverage row in the
 *     SAME transaction - so a seed is never durable without the durable promise to
 *     enumerate that justifies it.
 *  2. full `Mailbox/get` - cheap, always complete, no paging.
 *  3. the seeded cursors are LIVE FROM HERE. Each cycle runs the delta jobs and
 *     only then the scan, so a wide-window rebuild does not stall incoming mail.
 *  4. when the scan reaches the target: `coveredFrom = sweepFloor`, phase complete.
 */
async function beginEnumeration(
  ctx: CycleContext,
  envelopeFrom: string,
  kind: 'bootstrap' | 'reconcile',
): Promise<void> {
  const { store, jmapAccountId, now } = ctx;

  // Step 0 for a reconcile: PIN THE FLOOR. Every later step reads `sweepFloor`,
  // never a live `targetFrom`. Widening retention while a reconcile runs would
  // otherwise make the sweep delete against the new wide window when the
  // enumeration only covered the old narrow one - permanently, since `coveredFrom`
  // then claims the wider range and `/changes` cannot re-deliver old mail.
  const existing = store.getCoverage(jmapAccountId);
  const sweepFloor = kind === 'reconcile' ? (existing?.deferredTargetFrom ?? envelopeFrom) : envelopeFrom;

  const snapshots = await captureSnapshotStates(ctx.session, ctx.authHeader, jmapAccountId);

  // Derive the reconcile stamp from the DATA, not the clock: with a frozen or
  // coarse clock `cached_at < stamp` matches nothing and the sweep deletes nothing.
  const stampedAt = kind === 'reconcile'
    ? reconcileStamp(now, store.maxEnvelopeCachedAt(jmapAccountId))
    : undefined;

  store.transaction(() => {
    for (const type of CURSOR_TYPES) {
      store.seedCursor(
        { jmapAccountId, type },
        mintEnumerationCommitment({
          jmapAccountId,
          snapshot: type === 'Mailbox' ? snapshots.mailbox : snapshots.email,
          targetFrom: envelopeFrom,
          sweepFloor,
          kind,
        }),
        now,
      );
    }
    if (stampedAt !== undefined) store.patchCoverage(jmapAccountId, { reconcileStampedAt: stampedAt });
    if (kind === 'reconcile') {
      const flags = store.getFlags(now);
      store.patchFlags(now, { reconcilesInWindow: flags.reconcilesInWindow + 1 });
    }
  });

  // Step 2: every mailbox, in full.
  const mailboxes = await getMailboxes(ctx.session, ctx.authHeader, jmapAccountId, null);
  const rows = mailboxes.map((m) => toMailboxRow(jmapAccountId, m));
  store.transaction(() => { store.upsertMailboxes(rows); });
}

// ── A1 / A2: the delta drain ─────────────────────────────────────────────────

async function drainCursor(
  ctx: CycleContext,
  type: CursorType,
  bodyFrom: string,
  report: CycleReport,
): Promise<{ pending: boolean }> {
  const { store, jmapAccountId } = ctx;
  const key = { jmapAccountId, type };
  let cursor = store.getCursor(key);
  if (!cursor) return { pending: false };
  if (cursor.invalidatedAt) {
    // Serving `/changes` from an invalidated cursor is forbidden. The reconcile
    // is what clears it.
    return { pending: false };
  }

  const cap = maxObjectsInGet(ctx.session);
  let pages = 0;

  while (pages < BUDGET.changesPagesPerCursor) {
    if (Date.now() > ctx.deadline) {
      store.transaction(() => { store.patchCursor(key, { drainPending: true }); });
      return { pending: true };
    }
    cursor = store.getCursor(key);
    if (!cursor) return { pending: false };

    let page;
    try {
      const maxChanges = rungValue(cursor.maxChangesRung, cap);
      page = type === 'Email'
        ? await getEmailChanges(ctx.session, ctx.authHeader, jmapAccountId, cursor.state, maxChanges)
        : await getMailboxChanges(ctx.session, ctx.authHeader, jmapAccountId, cursor.state, maxChanges);
    } catch (error) {
      handleDrainError(ctx, key, cursor.state, error, report);
      // Any cursor failing means the cycle is unfinished for escalation
      // purposes, but never that the cursor moved.
      return { pending: true };
    }
    pages++;

    if (page.oldState !== cursor.state) {
      // Re-issue once before escalating: a transient anomaly is far more common
      // than a genuine invalidation, and a full rebuild is expensive.
      let reissued;
      try {
        const maxChanges = rungValue(cursor.maxChangesRung, cap);
        reissued = type === 'Email'
          ? await getEmailChanges(ctx.session, ctx.authHeader, jmapAccountId, cursor.state, maxChanges)
          : await getMailboxChanges(ctx.session, ctx.authHeader, jmapAccountId, cursor.state, maxChanges);
      } catch (error) {
        handleDrainError(ctx, key, cursor.state, error, report);
        return { pending: true };
      }
      if (reissued.oldState !== cursor.state) {
        invalidate(ctx, key, 'oldStateMismatch', report);
        return { pending: false };
      }
      // Use the RE-ISSUED page. Advancing to the original response's newState
      // would skip whatever the re-issue reported - the same silent-gap shape as
      // the cursor-provenance bug, reintroduced one level down.
      page = reissued;
      report.warnings.push(`${type}/changes reported a transient oldState mismatch`);
    }

    if (pageIsEmpty(page)) {
      // An empty page still advances. Skipping it re-requests forever.
      store.transaction(() => {
        store.advanceCursor(key, page.newState);
        store.patchCursor(key, { consecutiveFailures: 0, lastFailedState: undefined, maxChangesRung: 0 });
      });
      if (!page.hasMoreChanges) {
        store.transaction(() => { store.patchCursor(key, { drainPending: false }); });
        return { pending: false };
      }
      continue;
    }

    if (type === 'Mailbox') {
      await applyMailboxPage(ctx, page, report);
    } else {
      await applyEmailPage(ctx, page, bodyFrom, report);
    }

    store.transaction(() => {
      store.advanceCursor(key, page.newState);
      store.patchCursor(key, {
        consecutiveFailures: 0,
        lastFailedState: undefined,
        maxChangesRung: 0,
        drainPending: page.hasMoreChanges,
      });
    });

    if (!page.hasMoreChanges) return { pending: false };
  }

  // Budget exhaustion is normal, not an error. It is also the answer to a server
  // whose `hasMoreChanges` never goes false.
  store.transaction(() => { store.patchCursor(key, { drainPending: true }); });
  return { pending: true };
}

function handleDrainError(
  ctx: CycleContext,
  key: { jmapAccountId: string; type: CursorType },
  failedState: string,
  error: unknown,
  report: CycleReport,
): void {
  const { store } = ctx;
  const cls = error instanceof ReplicaSyncError ? error.cls : 'ServerTransient';
  const message = error instanceof Error ? error.message : String(error);
  report.warnings.push(`${key.type}/changes ${cls}: ${message}`);
  if (error instanceof ReplicaSyncError && error.retryAfterMs) {
    report.retryAfterMs = Math.max(report.retryAfterMs ?? 0, error.retryAfterMs);
  }

  if (movesCursor(cls)) {
    invalidate(ctx, key, 'cannotCalculateChanges', report);
    return;
  }

  const cursor = store.getCursor(key);
  if (!cursor) return;
  // The ladder only counts failures at the SAME sinceState: a failure at a new
  // position means progress was made, so the ladder restarts.
  const sameSpot = cursor.lastFailedState === failedState;
  const failures = sameSpot ? cursor.consecutiveFailures + 1 : 1;
  store.transaction(() => {
    store.patchCursor(key, {
      consecutiveFailures: failures,
      lastFailedState: failedState,
      maxChangesRung: escalationApplies(cls) && failures >= 2
        ? nextRung(cursor.maxChangesRung)
        : cursor.maxChangesRung,
    });
  });
}

/**
 * RFC 8620 s5.2 says the client MUST invalidate its cache. The LITERAL reading -
 * delete everything, now - would empty a user's offline mail exactly when they may
 * be offline and depending on it. So: mark the cursor unusable, set the STICKY
 * resync flag, and leave every record READABLE. The accepted cost is that between
 * detection and the sweep, a server-deleted message can still show locally.
 *
 * An invalidation of EITHER cursor reconciles the account as a whole - splitting
 * it is not worth the reasoning burden when `Mailbox/get` is one cheap call.
 */
function invalidate(
  ctx: CycleContext,
  key: { jmapAccountId: string; type: CursorType },
  reason: 'cannotCalculateChanges' | 'oldStateMismatch',
  report: CycleReport,
): void {
  const { store, now } = ctx;
  store.transaction(() => {
    store.patchCursor(key, { invalidatedAt: now, invalidatedReason: reason });
    store.patchFlags(now, { resyncRequired: true });
  });
  report.resyncRequired = true;
  report.warnings.push(`${key.type} cursor invalidated (${reason}); a rebuild is queued`);
  logger.warn('offline-replica: cursor invalidated', { type: key.type, reason });
}

async function applyMailboxPage(
  ctx: CycleContext,
  page: Parameters<typeof planMailboxFetches>[0],
  report: CycleReport,
): Promise<void> {
  const { store, jmapAccountId } = ctx;
  const plan = planMailboxFetches(page);

  if (plan.fullIds.length > 0) {
    const mailboxes = await getMailboxes(ctx.session, ctx.authHeader, jmapAccountId, plan.fullIds);
    const rows = mailboxes.map((m) => toMailboxRow(jmapAccountId, m));
    store.transaction(() => { report.mailboxesWritten += store.upsertMailboxes(rows); });
  }

  if (plan.countOnlyIds.length > 0) {
    const props = ['totalEmails', 'unreadEmails', 'totalThreads', 'unreadThreads'];
    const mailboxes = await getMailboxes(
      ctx.session, ctx.authHeader, jmapAccountId, plan.countOnlyIds, props,
    );
    store.transaction(() => {
      for (const m of mailboxes) {
        store.patchMailboxCounts(jmapAccountId, m.originalId ?? m.id, {
          totalEmails: typeof m.totalEmails === 'number' ? m.totalEmails : undefined,
          unreadEmails: typeof m.unreadEmails === 'number' ? m.unreadEmails : undefined,
          totalThreads: typeof m.totalThreads === 'number' ? m.totalThreads : undefined,
          unreadThreads: typeof m.unreadThreads === 'number' ? m.unreadThreads : undefined,
        });
      }
    });
  }

  if (plan.destroyIds.length > 0) {
    // The mailbox row ONLY. Never email records.
    store.transaction(() => { store.deleteMailboxes(jmapAccountId, plan.destroyIds); });
  }
}

async function applyEmailPage(
  ctx: CycleContext,
  page: Parameters<typeof planEmailFetches>[0],
  bodyFrom: string,
  report: CycleReport,
): Promise<void> {
  const { store, jmapAccountId, now } = ctx;
  // Presence is tested in BULK, before either fetch is issued - an `updated` id we
  // do not hold is filtered out rather than fetched and then discarded.
  const present = store.whichEnvelopesExist(jmapAccountId, page.updated);
  const plan = planEmailFetches(page, present);

  // CREATES: full envelope tier, and a body enqueue when inside the body window.
  // Bodies are NEVER fetched inline - they are 10-500 KB against an envelope's
  // ~1 KB, and the queue is what keeps a page a small, quickly-committable unit.
  for (let i = 0; i < plan.createIds.length; i += BUDGET.envelopeFetchChunk) {
    const chunk = plan.createIds.slice(i, i + BUDGET.envelopeFetchChunk);
    const { list } = await getEmails(ctx.session, ctx.authHeader, jmapAccountId, chunk, 'envelope');
    const rows = list.map((e) => toEnvelopeRow(jmapAccountId, e)).filter((r): r is EnvelopeRow => r !== null);
    const queue: BodyQueueEntry[] = rows
      .filter((r) => r.receivedAt >= bodyFrom)
      .map((r) => ({ emailId: r.id, jmapAccountId, receivedAt: r.receivedAt, attempts: 0 }));
    store.transaction(() => {
      report.envelopesWritten += store.upsertEnvelopes(rows, now);
      store.enqueueBodies(queue);
    });
  }

  // UPDATES: three properties, never a body.
  for (let i = 0; i < plan.updateIds.length; i += BUDGET.envelopeFetchChunk) {
    const chunk = plan.updateIds.slice(i, i + BUDGET.envelopeFetchChunk);
    const { list } = await getEmails(ctx.session, ctx.authHeader, jmapAccountId, chunk, 'mutable');
    store.transaction(() => {
      for (const e of list) {
        store.patchEnvelopeMutable(jmapAccountId, e.id, {
          keywordsJson: JSON.stringify(e.keywords ?? {}),
          mailboxIds: Object.entries(e.mailboxIds ?? {}).filter(([, v]) => v).map(([k]) => k),
        });
      }
    });
  }

  // DESTROYS LAST. Ids are never reused, so a destroy always refers to the same
  // record as any create/update of that id in the same page, and destroy-last
  // converges. The reverse order would resurrect a dead id, spend a fetch and get
  // `notFound`. A destroy for an id we never held is a harmless no-op.
  if (plan.destroyIds.length > 0) {
    store.transaction(() => {
      report.envelopesDeleted += store.deleteEmails(jmapAccountId, plan.destroyIds);
    });
  }
}

// ── B: coverage enumeration ──────────────────────────────────────────────────

async function runCoverage(
  ctx: CycleContext,
  coverage: CoverageState,
  bodyFrom: string,
  evictionAllowed: boolean,
  report: CycleReport,
): Promise<{ unfinished: boolean }> {
  const { store, jmapAccountId, now } = ctx;
  const isReconcile = coverage.phase === 'reconciling';
  const floor = isReconcile ? (coverage.sweepFloor ?? coverage.targetFrom) : coverage.targetFrom;
  // A reconcile stamps with its PINNED value, not `now`. The pin is
  // max(now, maxCachedAt + 1) so it routinely EXCEEDS now - stamping with `now`
  // would leave rows below the pin and get them deleted by the very sweep that
  // just re-verified them against the server.
  const stamp = isReconcile ? coverage.reconcileStampedAt ?? now : now;

  let scanCursor = coverage.scanCursor ?? floor;
  let lastPageIds: string[] = [];
  let pages = 0;

  while (pages < BUDGET.coveragePages) {
    if (Date.now() > ctx.deadline) return { unfinished: true };
    pages++;

    let ids: string[];
    try {
      ({ ids } = await queryAscending(
        ctx.session, ctx.authHeader, jmapAccountId, scanCursor, BUDGET.coveragePageSize,
      ));
    } catch (error) {
      if (error instanceof AnchorNotFoundError) {
        ids = [];
      } else {
        store.transaction(() => {
          store.patchCoverage(jmapAccountId, {
            consecutiveFailures: coverage.consecutiveFailures + 1,
          });
        });
        throw error;
      }
    }

    if (ids.length === 0) {
      // The scan has reached the present.
      finishEnumeration(ctx, floor, stamp, isReconcile, evictionAllowed, report);
      return { unfinished: !evictionAllowed && isReconcile };
    }

    const { list } = await getEmails(ctx.session, ctx.authHeader, jmapAccountId, ids, 'envelope');
    const rows = list.map((e) => toEnvelopeRow(jmapAccountId, e)).filter((r): r is EnvelopeRow => r !== null);
    const maxReceivedAt = rows.reduce<string | null>(
      (max, r) => (max === null || r.receivedAt > max ? r.receivedAt : max),
      null,
    );
    const queue: BodyQueueEntry[] = rows
      .filter((r) => r.receivedAt >= bodyFrom)
      .map((r) => ({ emailId: r.id, jmapAccountId, receivedAt: r.receivedAt, attempts: 0 }));

    let nextScanCursor = scanCursor;
    let gapMarker: NonNullable<CoverageState['gapMarkers']>[number] | null = null;

    if (madeForwardProgress(maxReceivedAt, scanCursor)) {
      nextScanCursor = maxReceivedAt as string;
    } else if (ids.length >= BUDGET.coveragePageSize) {
      // A FULL page whose every row shares one millisecond. Try the anchor rung
      // first: resume from the id after the last one we saw.
      const anchorId = lastPageIds.length > 0 ? lastPageIds[lastPageIds.length - 1] : ids[ids.length - 1];
      let recovered = false;
      try {
        const anchored = await queryAscending(
          ctx.session, ctx.authHeader, jmapAccountId, scanCursor, BUDGET.coveragePageSize,
          { anchor: anchorId, anchorOffset: 1 },
        );
        if (anchored.ids.length > 0) {
          const fetched = await getEmails(
            ctx.session, ctx.authHeader, jmapAccountId, anchored.ids, 'envelope',
          );
          const anchoredRows = fetched.list
            .map((e) => toEnvelopeRow(jmapAccountId, e))
            .filter((r): r is EnvelopeRow => r !== null);
          const anchoredMax = anchoredRows.reduce<string | null>(
            (max, r) => (max === null || r.receivedAt > max ? r.receivedAt : max),
            null,
          );
          store.transaction(() => {
            // The PINNED stamp here too, for the same reason.
            report.envelopesWritten += store.upsertEnvelopes(anchoredRows, stamp);
            store.enqueueBodies(
              anchoredRows
                .filter((r) => r.receivedAt >= bodyFrom)
                .map((r) => ({ emailId: r.id, jmapAccountId, receivedAt: r.receivedAt, attempts: 0 })),
            );
          });
          if (anchoredMax !== null && anchoredMax > scanCursor) {
            nextScanCursor = anchoredMax;
            recovered = true;
          }
        }
      } catch (error) {
        if (!(error instanceof AnchorNotFoundError)) throw error;
      }
      if (!recovered) {
        // Last resort: advance one millisecond, WARN, and leave a durable trace.
        // This rung CAN skip messages sharing the boundary millisecond, so it is
        // never normal-path behaviour and always leaves a record so a support
        // question has an answer. A 200-message single-millisecond cluster is a
        // corrupt server, not a case to design for.
        const to = advanceOneMs(scanCursor);
        gapMarker = { from: scanCursor, to, reason: 'tie-cluster-skip', at: now };
        nextScanCursor = to;
        report.warnings.push(
          `coverage skipped a tie cluster at ${scanCursor}; some messages sharing that ` +
            `millisecond may be missing from the offline store`,
        );
        logger.warn('offline-replica: tie-cluster skip', { at: scanCursor });
      }
    } else {
      // A partial page with no forward progress means we are at the tail.
      //
      // COMMIT THE ROWS BEFORE FINISHING. Finishing runs the reconcile sweep, and
      // the sweep deletes anything still stamped below the pin - so finishing
      // first would delete this page's own records (taking their bodies and queue
      // rows with them) and then re-insert them bodyless, costing a re-download of
      // every body at the tail on every reconcile.
      store.transaction(() => {
        report.envelopesWritten += store.upsertEnvelopes(rows, stamp);
        store.enqueueBodies(queue);
      });
      finishEnumeration(ctx, floor, stamp, isReconcile, evictionAllowed, report);
      return { unfinished: false };
    }

    const advanceTo = nextScanCursor;
    const marker = gapMarker;
    store.transaction(() => {
      report.envelopesWritten += store.upsertEnvelopes(rows, stamp);
      store.enqueueBodies(queue);
      // CURSOR LAST, inside the same transaction as the records it accounts for.
      store.patchCoverage(jmapAccountId, {
        scanCursor: advanceTo,
        seen: coverage.seen + rows.length,
        consecutiveFailures: 0,
        ...(marker
          ? { gapMarkers: [...(coverage.gapMarkers ?? []), marker].slice(-32) }
          : {}),
      });
    });
    scanCursor = advanceTo;
    lastPageIds = ids;
  }

  return { unfinished: true };
}

/**
 * The only place the reconcile pins are released, and it clears them all at once.
 */
function finishEnumeration(
  ctx: CycleContext,
  floor: string,
  stamp: number,
  isReconcile: boolean,
  evictionAllowed: boolean,
  report: CycleReport,
): void {
  const { store, jmapAccountId, now } = ctx;
  const coverage = store.getCoverage(jmapAccountId);
  if (!coverage) return;

  if (isReconcile && !evictionAllowed) {
    // The sweep is a delete against the retention floor, so it is subject to the
    // same rule as eviction: a suspect clock reading may not drive deletion.
    // Leave the reconcile open; it completes on a cycle whose floor is trustworthy.
    report.warnings.push(
      'deferring the reconcile sweep: the retention floor came from a suppressed clock anomaly',
    );
    return;
  }

  store.transaction(() => {
    if (isReconcile) {
      report.envelopesDeleted += store.sweep(jmapAccountId, floor, stamp);
      // A give-up recorded during whatever went wrong must not outlive it, or a
      // transient outage would permanently deny a body with no path back.
      store.clearBodyGiveUps(jmapAccountId);
      store.patchFlags(now, { resyncRequired: false });
      for (const type of CURSOR_TYPES) {
        store.patchCursor({ jmapAccountId, type }, {
          invalidatedAt: undefined,
          invalidatedReason: undefined,
          consecutiveFailures: 0,
          lastFailedState: undefined,
          maxChangesRung: 0,
        });
      }
    }
    const deferred = coverage.deferredTargetFrom;
    store.patchCoverage(jmapAccountId, {
      coveredFrom: floor,
      scanCursor: null,
      sweepFloor: undefined,
      reconcileStampedAt: undefined,
      deferredTargetFrom: undefined,
      ...(deferred ? { targetFrom: deferred, phase: 'scanning' as const } : { phase: 'complete' as const }),
    });
  });
}

// ── C1: drain the durable body queue ─────────────────────────────────────────

async function drainBodyQueue(ctx: CycleContext, report: CycleReport): Promise<number> {
  const { store, jmapAccountId } = ctx;
  let fetched = 0;
  const wanted = store.takeBodyQueue(jmapAccountId, BUDGET.bodyItems, Date.now());
  if (wanted.length === 0) return 0;

  for (let i = 0; i < wanted.length; i += BUDGET.bodyFetchChunk) {
    if (Date.now() > ctx.deadline) { report.unfinishedWork = true; break; }
    const chunk = wanted.slice(i, i + BUDGET.bodyFetchChunk);
    let result;
    try {
      result = await getEmails(
        ctx.session, ctx.authHeader, jmapAccountId, chunk.map((e) => e.emailId), 'body',
      );
    } catch (error) {
      // Body failures NEVER touch a cursor: the body jobs are separate state, so
      // the delta path keeps its position and the queue simply retries later.
      const message = error instanceof Error ? error.message : String(error);
      store.transaction(() => {
        for (const e of chunk) {
          store.bumpBodyAttempt(jmapAccountId, e.emailId, Date.now() + backoffDelayMs(e.attempts), message);
        }
      });
      report.warnings.push(`body fetch failed: ${message}`);
      report.unfinishedWork = true;
      break;
    }

    const byId = new Map(result.list.map((e) => [e.id, e]));
    const notFound = new Set(result.notFound);

    store.transaction(() => {
      for (const entry of chunk) {
        const email = byId.get(entry.emailId);
        if (email) {
          // Conditional on the envelope still existing, so a body fetched just
          // before its envelope was destroyed cannot land as an orphan.
          const wrote = store.putBodyIfEnvelopeExists(jmapAccountId, entry.emailId, serialiseBody(email));
          store.dequeueBodies(jmapAccountId, [entry.emailId]);
          if (wrote) { fetched++; report.bodiesWritten++; }
          continue;
        }
        if (notFound.has(entry.emailId)) {
          // The message is gone, so the entry can NEVER succeed and must not burn
          // five attempts. A durable give-up, not a row deletion - deleting it
          // would let the backfill pass re-insert it every cycle forever.
          store.markBodyGaveUp(jmapAccountId, [
            { emailId: entry.emailId, receivedAt: entry.receivedAt, reason: 'notFound' },
          ]);
          continue;
        }
        // Present in neither list nor notFound: a transient miss.
        const attempts = entry.attempts + 1;
        if (attempts >= MAX_BODY_ATTEMPTS) {
          store.markBodyGaveUp(jmapAccountId, [
            {
              emailId: entry.emailId,
              receivedAt: entry.receivedAt,
              reason: 'attempts',
              lastError: `gave up after ${attempts} attempts`,
            },
          ]);
        } else {
          store.bumpBodyAttempt(
            jmapAccountId, entry.emailId, Date.now() + backoffDelayMs(attempts),
            'the server returned neither the record nor notFound',
          );
        }
      }
    });
  }
  return fetched;
}

// ── C2: body backfill ───────────────────────────────────────────────────────

/**
 * Notices envelopes that never had a body enqueued - a widened body window, or a
 * queue row lost to a purge. Runs EVEN WHEN C1 found nothing, because that is its
 * whole job.
 *
 * It is cap-aware and excludes durable give-ups. Both matter: the headroom check
 * avoids paying for a download with nowhere to go, and the give-up exclusion is
 * what makes termination provable. Without it, the size cap sheds the oldest
 * bodies, those envelopes are still inside the body WINDOW, so this pass
 * re-enqueues them, they download again, and the cap sheds them again - unbounded
 * data use with no termination, because there is always an envelope without a body.
 *
 * A heuristic frontier instead of a durable mark does NOT work, and the reason is
 * worth recording: the evictor sheds down TO the cap, which leaves headroom, so
 * this pass refills and the two trade the same bytes back and forth. Only the
 * durable mark makes the store monotone.
 */
async function backfillBodies(
  ctx: CycleContext,
  bodyFrom: string,
  maxBodyBytes: number,
  alreadyFetched: number,
): Promise<number> {
  const { store, jmapAccountId } = ctx;
  const budget = BUDGET.bodyItems - alreadyFetched;
  if (budget <= 0) return 0;

  const used = store.bodyBytesTotal(jmapAccountId);
  let headroom = maxBodyBytes - used;
  if (headroom <= 0) return 0;

  const candidates = store.envelopesWithoutBody(jmapAccountId, bodyFrom, budget * 2);
  if (candidates.length === 0) return 0;
  const gaveUp = new Set(store.listBodyGiveUps(jmapAccountId, 5_000));

  const entries: BodyQueueEntry[] = [];
  for (const c of candidates) {
    if (entries.length >= budget) break;
    if (gaveUp.has(c.id)) continue;
    // Always allow at least one, or a single oversized message would stall the
    // queue forever.
    if (entries.length > 0 && c.size > headroom) break;
    headroom -= c.size;
    entries.push({ emailId: c.id, jmapAccountId, receivedAt: c.receivedAt, attempts: 0 });
    if (headroom <= 0) break;
  }
  if (entries.length === 0) return 0;

  // The INSERTED count, not the attempted count. Reporting the latter made the
  // mobile engine treat every cycle as having unfinished work for as long as any
  // envelope lacked a body, chaining a new cycle every few seconds indefinitely.
  return store.transaction(() => store.enqueueBodies(entries));
}

// ── retention ────────────────────────────────────────────────────────────────

function applyRetention(
  ctx: CycleContext,
  args: {
    envelopeFrom: string;
    bodyFrom: string;
    maxBodyBytes: number;
    evictionAllowed: boolean;
    previousFloor: string | undefined;
    report: CycleReport;
  },
): void {
  const { store, jmapAccountId, now } = ctx;
  const { envelopeFrom, bodyFrom, maxBodyBytes, evictionAllowed, previousFloor, report } = args;
  const coverage = store.getCoverage(jmapAccountId);

  // A retention change arriving DURING a reconcile is deferred to after the
  // sweep. Applying a widen mid-reconcile would make the sweep delete everything
  // between the old and new floors permanently; a narrow is deferred too, because
  // evicting below a floor the sweep is about to use races it for no benefit.
  if (coverage?.phase === 'reconciling') {
    if (coverage.targetFrom !== envelopeFrom) {
      store.transaction(() => {
        store.patchCoverage(jmapAccountId, { deferredTargetFrom: envelopeFrom });
      });
    }
    return;
  }

  const movement = floorMovement(previousFloor, envelopeFrom);
  const adjustment = adjustForWindow(movement, envelopeFrom);

  if (adjustment.evictBelow) {
    if (!evictionAllowed) {
      report.warnings.push(
        `skipping retention eviction below ${adjustment.evictBelow}: the floor came from a ` +
          `suppressed clock anomaly`,
      );
    } else {
      store.transaction(() => {
        report.envelopesDeleted += store.evictEnvelopesBelow(jmapAccountId, adjustment.evictBelow as string);
        // `coveredFrom` follows the floor, but ONLY when it was already set:
        // otherwise a narrow claims a range that was never enumerated, and delta
        // sync cannot re-deliver pre-existing mail to repair that.
        if (coverage?.coveredFrom !== null && coverage?.coveredFrom !== undefined) {
          store.patchCoverage(jmapAccountId, { coveredFrom: adjustment.evictBelow as string });
        }
        store.patchCoverage(jmapAccountId, { targetFrom: envelopeFrom });
      });
    }
  } else if (adjustment.rescanFrom) {
    // A WIDEN is not a resync: the target moves back, coverage re-enters scanning,
    // and the cursors are untouched.
    store.transaction(() => {
      store.patchCoverage(jmapAccountId, {
        targetFrom: adjustment.rescanFrom as string,
        scanCursor: adjustment.rescanFrom as string,
        phase: 'scanning',
      });
    });
    report.unfinishedWork = true;
  }

  // Body narrow: delete and merely DEQUEUE, never mark - the backfill's
  // `receivedAfter` already excludes out-of-window bodies, and a later widen must
  // be free to re-fetch them.
  const stale = store.bodiesBelow(jmapAccountId, bodyFrom, 2_000);
  if (stale.length > 0 && evictionAllowed) {
    store.transaction(() => {
      report.bodiesEvicted += store.deleteBodies(jmapAccountId, stale);
      store.dequeueBodies(jmapAccountId, stale);
    });
  }

  // Orphan bodies: invisible to cap eviction, which only walks the body table.
  const orphans = store.orphanBodies(jmapAccountId, 500);
  if (orphans.length > 0) {
    store.transaction(() => {
      report.bodiesEvicted += store.deleteBodies(jmapAccountId, orphans);
      store.dequeueBodies(jmapAccountId, orphans);
    });
  }

  // A cap RAISE revives bodies shed for space: a durable refusal recorded under
  // one policy must not outlive that policy.
  const flags = store.getFlags(now);
  if (flags.lastMaxBodyBytes !== undefined && maxBodyBytes > flags.lastMaxBodyBytes) {
    store.transaction(() => { store.clearBodyGiveUps(jmapAccountId, 'shed-by-cap'); });
  }

  // The MB cap: oldest bodies first, envelopes always survive, so the message
  // stays listed and openable when back online.
  let used = store.bodyBytesTotal(jmapAccountId);
  if (used > maxBodyBytes) {
    const oldest = store.oldestBodies(jmapAccountId, 5_000);
    const shed: Array<{ emailId: string; receivedAt: string; reason: 'shed-by-cap' }> = [];
    for (const b of oldest) {
      if (used <= maxBodyBytes) break;
      used -= b.bytes;
      shed.push({ emailId: b.emailId, receivedAt: '', reason: 'shed-by-cap' });
    }
    if (shed.length > 0) {
      // MARKED, not merely dequeued. See backfillBodies' comment for the loop
      // this closes.
      const withDates = shed.map((s) => {
        const row = store.getEnvelopeRaw(jmapAccountId, s.emailId);
        return { ...s, receivedAt: typeof row?.received_at === 'string' ? row.received_at : new Date(now).toISOString() };
      });
      store.transaction(() => {
        report.bodiesEvicted += store.deleteBodies(jmapAccountId, shed.map((s) => s.emailId));
        store.markBodyGaveUp(jmapAccountId, withDates);
      });
    }
  }
}