SRCmail/lib/offline-replica/store.ts

// The replica store: the encrypted SQLite file, and every read/write against it.
//
// SYNCHRONOUS ON PURPOSE. `@signalapp/sqlcipher` is a synchronous binding, so
// `transaction()` here takes a synchronous callback and no `await` can ever
// appear inside a `BEGIN ... COMMIT`. That removes an entire hazard class by
// construction: no network call, no timer and no other request can interleave
// with a half-applied transaction. Every JMAP fetch happens OUTSIDE a
// transaction and the results are applied inside one.

import fs from 'node:fs';
import path from 'node:path';
import { loadSqlcipher, type SqlcipherDatabase } from '@/lib/mail-index/binding';
import { dbSiblings, indexDbPath } from '@/lib/mail-index/paths';
import {
  coverageStateKey, cursorStateKey, FLAGS_KEY, POLICY_KEY, REPLICA_DDL,
  REPLICA_RECORD_TABLES, REPLICA_SCHEMA_VERSION, REPLICA_TABLES, REPLICA_VERSION_KEY,
} from './schema';
import {
  coveragePhaseForCommitment, type ChangesState, type CursorType, type EnumerationCommitment,
} from './states';
import {
  defaultFlags, type BodyGiveUpReason, type BodyQueueEntry, type CoverageState,
  type EnvelopeRow, type FlagsPatch, type MailboxRow, type ReplicaFlags, type SyncCursor,
} from './types';

export class ReplicaUnavailableError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'ReplicaUnavailableError';
  }
}

export interface CursorKey {
  jmapAccountId: string;
  type: CursorType;
}

/** Retention policy, persisted server-side inside the encrypted store. */
export interface RetentionPolicy {
  envelopeDays: number;
  bodyDays: number;
  maxBodyMB: number;
}

export const DEFAULT_POLICY: RetentionPolicy = {
  // Envelopes are ~1 KB, so a wide window costs kilobytes per message and means
  // a message never falls out of the offline LIST because of a body size cap.
  envelopeDays: 180,
  bodyDays: 30,
  maxBodyMB: 250,
};

export const POLICY_LIMITS = {
  envelopeDays: { min: 7, max: 3650 },
  bodyDays: { min: 1, max: 3650 },
  maxBodyMB: { min: 16, max: 20_000 },
} as const;

export function clampPolicy(raw: Partial<RetentionPolicy> | null | undefined): RetentionPolicy {
  const pick = (
    value: unknown,
    fallback: number,
    { min, max }: { min: number; max: number },
  ): number => {
    const n = typeof value === 'number' && Number.isFinite(value) ? Math.round(value) : fallback;
    return Math.min(Math.max(n, min), max);
  };
  const envelopeDays = pick(raw?.envelopeDays, DEFAULT_POLICY.envelopeDays, POLICY_LIMITS.envelopeDays);
  const bodyDays = pick(raw?.bodyDays, DEFAULT_POLICY.bodyDays, POLICY_LIMITS.bodyDays);
  return {
    envelopeDays,
    // The body window can never be wider than the envelope window: a body with
    // no envelope is an orphan by construction.
    bodyDays: Math.min(bodyDays, envelopeDays),
    maxBodyMB: pick(raw?.maxBodyMB, DEFAULT_POLICY.maxBodyMB, POLICY_LIMITS.maxBodyMB),
  };
}

/**
 * `PRAGMA cipher_version` must return a non-empty STRING.
 *
 * Checking the row COUNT instead passes vacuously: a non-SQLCipher binding
 * returns ZERO ROWS for this pragma, and `PRAGMA key = ...` is silently accepted
 * and does nothing on plain SQLite - no error, a working database, and the mail
 * sitting on disk in cleartext.
 */
function assertEncrypted(db: SqlcipherDatabase, dbPath: string): void {
  const rows = db.pragma('cipher_version');
  const value =
    Array.isArray(rows) && rows.length > 0 && rows[0] && typeof rows[0] === 'object'
      ? (rows[0] as Record<string, unknown>).cipher_version
      : undefined;
  if (typeof value !== 'string' || value.trim().length === 0) {
    db.close();
    throw new ReplicaUnavailableError(
      `Refusing to use ${path.basename(dbPath)}: the SQLite binding reports no SQLCipher ` +
        `support (PRAGMA cipher_version returned ${JSON.stringify(rows)}), so the offline ` +
        `replica would be written in cleartext.`,
    );
  }
}

function num(v: unknown): number | null {
  return typeof v === 'number' && Number.isFinite(v) ? v : null;
}

function str(v: unknown): string | null {
  return typeof v === 'string' ? v : null;
}

export interface OpenReplicaOptions {
  storeDir: string;
  accountId: string;
  /** Raw 32-byte key, from the main process's key service. */
  key: Buffer;
}

export class ReplicaStore {
  private constructor(
    private readonly db: SqlcipherDatabase,
    readonly dbPath: string,
  ) {}

  static open({ storeDir, accountId, key }: OpenReplicaOptions): ReplicaStore {
    const Database = loadSqlcipher();
    if (!Database) {
      throw new ReplicaUnavailableError(
        '@signalapp/sqlcipher is not installed for this platform (it is an optional dependency).',
      );
    }
    if (key.length !== 32) {
      throw new ReplicaUnavailableError(`Replica key must be 32 bytes, got ${key.length}.`);
    }

    // The SAME file as the search index. See schema.ts for why.
    const dbPath = indexDbPath(storeDir, accountId);
    fs.mkdirSync(path.dirname(dbPath), { recursive: true, mode: 0o700 });

    const connect = (): SqlcipherDatabase => {
      const db = new Database(dbPath);
      // The key pragma must be the FIRST statement on the connection. Hex form
      // means SQLCipher uses these 32 bytes as the raw key with no KDF.
      db.pragma(`key = "x'${key.toString('hex')}'"`);
      assertEncrypted(db, dbPath);
      return db;
    };

    let db = connect();
    let version: number | null;
    try {
      db.pragma('journal_mode = WAL');
      db.pragma('synchronous = NORMAL');
      // The index and the replica are two connections to one file. WAL lets a
      // writer and readers coexist, but two WRITERS get SQLITE_BUSY immediately
      // without this - and both are driven by the same renderer push handler, so
      // they genuinely do overlap.
      db.pragma('busy_timeout = 8000');
      version = readVersion(db);
    } catch {
      // A wrong key surfaces here, not at open: SQLCipher reads the header
      // lazily. The replica is derived data, so there is nothing to recover and
      // never anything to prompt the user for.
      db.close();
      for (const f of dbSiblings(dbPath)) {
        try { fs.rmSync(f, { force: true }); } catch { /* best effort */ }
      }
      db = connect();
      db.pragma('journal_mode = WAL');
      db.pragma('synchronous = NORMAL');
      db.pragma('busy_timeout = 8000');
      version = null;
    }

    if (version !== null && version !== REPLICA_SCHEMA_VERSION) version = null;

    if (version === null) {
      // ALL-OR-NOTHING. Records must never survive while the version row is
      // gone: a cursor that outlives its records is the one state no amount of
      // syncing repairs, because `/changes` cannot re-deliver mail that already
      // existed when the cursor was captured.
      db.exec('BEGIN');
      try {
        for (const table of REPLICA_TABLES) db.exec(`DROP TABLE IF EXISTS ${table}`);
        db.exec(REPLICA_DDL);
        db.exec('CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT NOT NULL)');
        db.prepare('INSERT OR REPLACE INTO meta (k, v) VALUES (?, ?)').run([
          REPLICA_VERSION_KEY,
          String(REPLICA_SCHEMA_VERSION),
        ]);
        db.exec('COMMIT');
      } catch (error) {
        db.exec('ROLLBACK');
        db.close();
        throw error;
      }
    } else {
      // The tables exist per the version row, but `CREATE TABLE IF NOT EXISTS`
      // is cheap and covers a partially-created file from an interrupted open.
      db.exec(REPLICA_DDL);
    }

    return new ReplicaStore(db, dbPath);
  }

  close(): void {
    try { this.db.close(); } catch { /* already closed */ }
  }

  /**
   * One SQLite transaction. The callback is SYNCHRONOUS, so nothing can
   * interleave and no `await` can sit inside `BEGIN ... COMMIT`.
   */
  transaction<T>(fn: () => T): T {
    this.db.exec('BEGIN');
    try {
      const out = fn();
      this.db.exec('COMMIT');
      return out;
    } catch (error) {
      try { this.db.exec('ROLLBACK'); } catch { /* the commit may have failed */ }
      throw error;
    }
  }

  // ── raw sync_state access ────────────────────────────────────────────────

  private readState<T>(k: string): T | null {
    const row = this.db.prepare('SELECT v FROM replica_sync_state WHERE k = ?').get([k]);
    if (!row || typeof row.v !== 'string') return null;
    try {
      return JSON.parse(row.v) as T;
    } catch {
      // A corrupt state blob is a resync signal, not something to guess at.
      return null;
    }
  }

  private writeState(k: string, value: unknown): void {
    this.db
      .prepare('INSERT INTO replica_sync_state (k, v) VALUES (?, ?) ON CONFLICT(k) DO UPDATE SET v = excluded.v')
      .run([k, JSON.stringify(value)]);
  }

  // ── policy ───────────────────────────────────────────────────────────────

  getPolicy(): RetentionPolicy {
    return clampPolicy(this.readState<Partial<RetentionPolicy>>(POLICY_KEY));
  }

  setPolicy(policy: RetentionPolicy): void {
    this.writeState(POLICY_KEY, clampPolicy(policy));
  }

  // ── flags ────────────────────────────────────────────────────────────────

  getFlags(now: number): ReplicaFlags {
    return this.readState<ReplicaFlags>(FLAGS_KEY) ?? defaultFlags(now);
  }

  patchFlags(now: number, patch: FlagsPatch): void {
    const current = this.getFlags(now);
    this.writeState(FLAGS_KEY, { ...current, ...patch });
  }

  // ── cursors ──────────────────────────────────────────────────────────────

  getCursor(key: CursorKey): SyncCursor | null {
    return this.readState<SyncCursor>(cursorStateKey(key.jmapAccountId, key.type));
  }

  /**
   * The delta path's ONLY cursor write. The signature is what makes the mobile
   * client's D4 a compile error here: a `SnapshotState` cannot be passed.
   *
   * Throws when the cursor does not exist. A cursor is born from `seedCursor`
   * and nowhere else; creating one here would be a silent cursor-from-nowhere,
   * which is the exact class of bug the branded types exist to prevent.
   */
  advanceCursor(key: CursorKey, next: ChangesState): void {
    const k = cursorStateKey(key.jmapAccountId, key.type);
    const current = this.readState<SyncCursor>(k);
    if (!current) {
      throw new Error(
        `advanceCursor: no cursor for ${key.type}/${key.jmapAccountId}; seed it first`,
      );
    }
    this.writeState(k, { ...current, state: next, updatedAt: Date.now() } satisfies SyncCursor);
  }

  /**
   * Bootstrap / reconcile only. Writes the snapshot state AND the `CoverageState`
   * it justifies in the SAME transaction, so a seed is never durable without the
   * durable commitment to enumerate that justifies it.
   *
   * Call inside `transaction()`.
   */
  seedCursor(key: CursorKey, commitment: EnumerationCommitment, now: number): void {
    if (commitment.jmapAccountId !== key.jmapAccountId) {
      throw new Error('seedCursor: commitment is for a different JMAP account');
    }
    const seeded: SyncCursor = {
      type: key.type,
      jmapAccountId: key.jmapAccountId,
      state: commitment.snapshot,
      drainPending: false,
      consecutiveFailures: 0,
      maxChangesRung: 0,
      updatedAt: now,
    };
    this.writeState(cursorStateKey(key.jmapAccountId, key.type), seeded);

    const existing = this.getCoverage(key.jmapAccountId);
    const next: CoverageState = {
      jmapAccountId: key.jmapAccountId,
      // Records stay readable during a reconcile, so what was already covered
      // stays claimed until the reconcile finishes and sets the pinned floor.
      coveredFrom: existing?.coveredFrom ?? null,
      scanCursor: null,
      targetFrom: commitment.targetFrom,
      sweepFloor: commitment.sweepFloor,
      deferredTargetFrom: undefined,
      gapMarkers: existing?.gapMarkers,
      phase: coveragePhaseForCommitment(commitment),
      seen: 0,
      consecutiveFailures: 0,
      updatedAt: now,
    };
    this.writeState(coverageStateKey(key.jmapAccountId), next);
  }

  /** Field-level patch. `state` is deliberately NOT patchable - see advance/seed. */
  patchCursor(
    key: CursorKey,
    patch: Partial<Omit<SyncCursor, 'type' | 'jmapAccountId' | 'state'>>,
  ): void {
    const k = cursorStateKey(key.jmapAccountId, key.type);
    const current = this.readState<SyncCursor>(k);
    if (!current) return;
    this.writeState(k, { ...current, ...patch, updatedAt: Date.now() });
  }

  // ── coverage ─────────────────────────────────────────────────────────────

  getCoverage(jmapAccountId: string): CoverageState | null {
    return this.readState<CoverageState>(coverageStateKey(jmapAccountId));
  }

  patchCoverage(jmapAccountId: string, patch: Partial<CoverageState>): void {
    const k = coverageStateKey(jmapAccountId);
    const current = this.readState<CoverageState>(k);
    if (!current) return;
    this.writeState(k, { ...current, ...patch, updatedAt: Date.now() });
  }

  // ── mailboxes ────────────────────────────────────────────────────────────

  upsertMailboxes(rows: readonly MailboxRow[]): number {
    if (rows.length === 0) return 0;
    const stmt = this.db.prepare(`
      INSERT INTO replica_mailbox (jmap_account_id, id, name, parent_id, role, sort_order,
        total_emails, unread_emails, total_threads, unread_threads, my_rights_json, is_subscribed)
      VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
      ON CONFLICT(jmap_account_id, id) DO UPDATE SET
        name = excluded.name, parent_id = excluded.parent_id, role = excluded.role,
        sort_order = excluded.sort_order, total_emails = excluded.total_emails,
        unread_emails = excluded.unread_emails, total_threads = excluded.total_threads,
        unread_threads = excluded.unread_threads, my_rights_json = excluded.my_rights_json,
        is_subscribed = excluded.is_subscribed
    `);
    for (const r of rows) {
      stmt.run([
        r.jmapAccountId, r.id, r.name, r.parentId, r.role, r.sortOrder,
        r.totalEmails, r.unreadEmails, r.totalThreads, r.unreadThreads,
        r.myRightsJson, r.isSubscribed ? 1 : 0,
      ]);
    }
    return rows.length;
  }

  /**
   * Patches ONLY the four count columns.
   *
   * `Mailbox/changes` reports `updatedProperties` as an upper bound of what may
   * have changed (RFC 8621 s2.2), and counts move on every delivery and every
   * read. On a busy account this is the difference between patching four
   * integers and re-fetching every folder object.
   */
  patchMailboxCounts(
    jmapAccountId: string,
    id: string,
    counts: {
      totalEmails?: number | null; unreadEmails?: number | null;
      totalThreads?: number | null; unreadThreads?: number | null;
    },
  ): void {
    const sets: string[] = [];
    const params: unknown[] = [];
    for (const [column, value] of [
      ['total_emails', counts.totalEmails], ['unread_emails', counts.unreadEmails],
      ['total_threads', counts.totalThreads], ['unread_threads', counts.unreadThreads],
    ] as const) {
      if (value !== undefined) { sets.push(`${column} = ?`); params.push(value); }
    }
    if (sets.length === 0) return;
    this.db
      .prepare(`UPDATE replica_mailbox SET ${sets.join(', ')} WHERE jmap_account_id = ? AND id = ?`)
      .run([...params, jmapAccountId, id]);
  }

  /**
   * Deletes the mailbox row ONLY. Never touches email records.
   *
   * Deletion provenance: if the server destroyed the messages too,
   * `Email/changes` reports them `destroyed`; if it moved them, their
   * `mailboxIds` update arrives as `updated`. Truth arrives on the Email stream
   * either way. Inferring deletion from a mailbox disappearing is how a client
   * loses mail the server still has.
   */
  deleteMailboxes(jmapAccountId: string, ids: readonly string[]): number {
    if (ids.length === 0) return 0;
    const stmt = this.db.prepare('DELETE FROM replica_mailbox WHERE jmap_account_id = ? AND id = ?');
    let n = 0;
    for (const id of ids) n += stmt.run([jmapAccountId, id]).changes;
    return n;
  }

  listMailboxes(jmapAccountId: string): MailboxRow[] {
    return this.db
      .prepare('SELECT * FROM replica_mailbox WHERE jmap_account_id = ? ORDER BY sort_order ASC, name ASC')
      .all([jmapAccountId])
      .map((r) => ({
        jmapAccountId: String(r.jmap_account_id),
        id: String(r.id),
        name: String(r.name ?? ''),
        parentId: str(r.parent_id),
        role: str(r.role),
        sortOrder: num(r.sort_order),
        totalEmails: num(r.total_emails),
        unreadEmails: num(r.unread_emails),
        totalThreads: num(r.total_threads),
        unreadThreads: num(r.unread_threads),
        myRightsJson: str(r.my_rights_json),
        isSubscribed: r.is_subscribed !== 0,
      }));
  }

  // ── envelopes ────────────────────────────────────────────────────────────

  /**
   * Upserts envelopes and replaces their membership rows.
   *
   * `has_body` / `body_bytes` are DELIBERATELY absent from `DO UPDATE SET`: they
   * belong to the body tier, and resetting them on an idempotent page replay
   * would look like "body missing" to the backfill job and re-download every
   * body in the page.
   *
   * `cachedAt` is a parameter rather than `Date.now()` because a reconcile must
   * stamp with its PINNED value - see `CoverageState.reconcileStampedAt`.
   */
  upsertEnvelopes(rows: readonly EnvelopeRow[], cachedAt: number): number {
    if (rows.length === 0) return 0;
    const upsert = this.db.prepare(`
      INSERT INTO replica_envelope (jmap_account_id, id, thread_id, received_at, size, subject,
        preview, from_json, to_json, cc_json, blob_id, has_attachment, keywords_json,
        has_body, body_bytes, cached_at)
      VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?)
      ON CONFLICT(jmap_account_id, id) DO UPDATE SET
        thread_id = excluded.thread_id, received_at = excluded.received_at,
        size = excluded.size, subject = excluded.subject, preview = excluded.preview,
        from_json = excluded.from_json, to_json = excluded.to_json, cc_json = excluded.cc_json,
        blob_id = excluded.blob_id, has_attachment = excluded.has_attachment,
        keywords_json = excluded.keywords_json, cached_at = excluded.cached_at
    `);
    const clearMembership = this.db.prepare(
      'DELETE FROM replica_email_mailbox WHERE jmap_account_id = ? AND email_id = ?',
    );
    const addMembership = this.db.prepare(
      'INSERT OR IGNORE INTO replica_email_mailbox (jmap_account_id, email_id, mailbox_id) VALUES (?, ?, ?)',
    );
    for (const r of rows) {
      upsert.run([
        r.jmapAccountId, r.id, r.threadId, r.receivedAt, r.size, r.subject, r.preview,
        r.fromJson, r.toJson, r.ccJson, r.blobId, r.hasAttachment ? 1 : 0, r.keywordsJson,
        cachedAt,
      ]);
      clearMembership.run([r.jmapAccountId, r.id]);
      for (const mailboxId of r.mailboxIds) addMembership.run([r.jmapAccountId, r.id, mailboxId]);
    }
    return rows.length;
  }

  /**
   * Patches the only two MUTABLE Email properties (RFC 8621 s4.1): `keywords`
   * and `mailboxIds`. Everything else - body, attachments, headers, receivedAt,
   * size, threadId, preview, subject, addresses - is immutable for the lifetime
   * of the id, which is why an `updated` id never needs a body re-fetch.
   *
   * No-ops for an id we do not hold, and only touches membership when the
   * envelope row actually existed, or we would leave membership rows for a
   * record we do not have.
   */
  patchEnvelopeMutable(
    jmapAccountId: string,
    id: string,
    patch: { keywordsJson: string; mailboxIds: string[] },
  ): boolean {
    const res = this.db
      .prepare('UPDATE replica_envelope SET keywords_json = ? WHERE jmap_account_id = ? AND id = ?')
      .run([patch.keywordsJson, jmapAccountId, id]);
    if (res.changes === 0) return false;
    this.db
      .prepare('DELETE FROM replica_email_mailbox WHERE jmap_account_id = ? AND email_id = ?')
      .run([jmapAccountId, id]);
    const add = this.db.prepare(
      'INSERT OR IGNORE INTO replica_email_mailbox (jmap_account_id, email_id, mailbox_id) VALUES (?, ?, ?)',
    );
    for (const mailboxId of patch.mailboxIds) add.run([jmapAccountId, id, mailboxId]);
    return true;
  }

  /** Bulk presence test, so the delta path can filter `updated` ids BEFORE fetching. */
  whichEnvelopesExist(jmapAccountId: string, ids: readonly string[]): Set<string> {
    if (ids.length === 0) return new Set();
    const out = new Set<string>();
    const stmt = this.db.prepare(
      'SELECT id FROM replica_envelope WHERE jmap_account_id = ? AND id = ?',
    );
    for (const id of ids) {
      if (stmt.get([jmapAccountId, id])) out.add(id);
    }
    return out;
  }

  /** Deletes an email everywhere: envelope, body, membership and any queue row. */
  deleteEmails(jmapAccountId: string, ids: readonly string[]): number {
    if (ids.length === 0) return 0;
    const statements = [
      this.db.prepare('DELETE FROM replica_body WHERE jmap_account_id = ? AND email_id = ?'),
      this.db.prepare('DELETE FROM replica_body_queue WHERE jmap_account_id = ? AND email_id = ?'),
      this.db.prepare('DELETE FROM replica_email_mailbox WHERE jmap_account_id = ? AND email_id = ?'),
    ];
    const deleteEnvelope = this.db.prepare(
      'DELETE FROM replica_envelope WHERE jmap_account_id = ? AND id = ?',
    );
    let n = 0;
    for (const id of ids) {
      for (const s of statements) s.run([jmapAccountId, id]);
      n += deleteEnvelope.run([jmapAccountId, id]).changes;
    }
    return n;
  }

  /** Retention eviction: everything strictly older than the floor. */
  evictEnvelopesBelow(jmapAccountId: string, isoFloor: string): number {
    const ids = this.db
      .prepare('SELECT id FROM replica_envelope WHERE jmap_account_id = ? AND received_at < ?')
      .all([jmapAccountId, isoFloor])
      .map((r) => String(r.id));
    return this.deleteEmails(jmapAccountId, ids);
  }

  /**
   * The reconcile sweep. Two clauses, and it REFUSES to run without a pinned
   * stamp rather than deleting unverified records.
   */
  sweep(jmapAccountId: string, sweepFloor: string, reconcileStampedAt: number | undefined): number {
    if (reconcileStampedAt === undefined) {
      throw new Error('sweep: no reconcileStampedAt pinned; refusing to delete unverified records');
    }
    const notReSeen = this.db
      .prepare(`
        SELECT id FROM replica_envelope
        WHERE jmap_account_id = ? AND received_at >= ? AND cached_at < ?
      `)
      .all([jmapAccountId, sweepFloor, reconcileStampedAt])
      .map((r) => String(r.id));
    // Records older than the pinned floor cannot be verified by an enumeration
    // that only covers the window, so they go rather than being kept on faith.
    // Normally retention has already evicted them.
    const unverifiable = this.db
      .prepare('SELECT id FROM replica_envelope WHERE jmap_account_id = ? AND received_at < ?')
      .all([jmapAccountId, sweepFloor])
      .map((r) => String(r.id));
    return this.deleteEmails(jmapAccountId, [...new Set([...notReSeen, ...unverifiable])]);
  }

  /**
   * The reconcile stamp must be derived from the DATA, not the clock:
   * `max(now, maxCachedAt + 1)`. With a frozen or coarse clock,
   * `cached_at < stamp` matches nothing and the sweep silently deletes nothing.
   */
  maxEnvelopeCachedAt(jmapAccountId: string): number {
    const row = this.db
      .prepare('SELECT MAX(cached_at) AS m FROM replica_envelope WHERE jmap_account_id = ?')
      .get([jmapAccountId]);
    return num(row?.m) ?? 0;
  }

  countEnvelopes(jmapAccountId: string): number {
    const row = this.db
      .prepare('SELECT COUNT(*) AS n FROM replica_envelope WHERE jmap_account_id = ?')
      .get([jmapAccountId]);
    return num(row?.n) ?? 0;
  }

  /** Envelopes inside the body window with no body yet - the backfill driver. */
  envelopesWithoutBody(
    jmapAccountId: string,
    receivedAfter: string,
    limit: number,
  ): Array<{ id: string; receivedAt: string; size: number }> {
    return this.db
      .prepare(`
        SELECT id, received_at, size FROM replica_envelope
        WHERE jmap_account_id = ? AND has_body = 0 AND received_at >= ?
        ORDER BY received_at DESC LIMIT ?
      `)
      .all([jmapAccountId, receivedAfter, limit])
      .map((r) => ({
        id: String(r.id),
        receivedAt: String(r.received_at),
        size: num(r.size) ?? 0,
      }));
  }

  // ── bodies ───────────────────────────────────────────────────────────────

  /**
   * Writes a body ONLY if its envelope still exists, and returns whether it did.
   *
   * Without the condition, a body fetched moments before its envelope was
   * destroyed in the same cycle lands as an orphan. This is also exactly why
   * "run bodies in parallel, it's separate state" is forbidden.
   */
  putBodyIfEnvelopeExists(jmapAccountId: string, emailId: string, json: string): boolean {
    const envelope = this.db
      .prepare('SELECT received_at FROM replica_envelope WHERE jmap_account_id = ? AND id = ?')
      .get([jmapAccountId, emailId]);
    const receivedAt = str(envelope?.received_at);
    if (receivedAt === null) return false;
    const bytes = Buffer.byteLength(json, 'utf8');
    this.db
      .prepare(`
        INSERT INTO replica_body (jmap_account_id, email_id, received_at, json, bytes)
        VALUES (?, ?, ?, ?, ?)
        ON CONFLICT(jmap_account_id, email_id) DO UPDATE SET
          received_at = excluded.received_at, json = excluded.json, bytes = excluded.bytes
      `)
      .run([jmapAccountId, emailId, receivedAt, json, bytes]);
    this.db
      .prepare('UPDATE replica_envelope SET has_body = 1, body_bytes = ? WHERE jmap_account_id = ? AND id = ?')
      .run([bytes, jmapAccountId, emailId]);
    return true;
  }

  getBody(jmapAccountId: string, emailId: string): string | null {
    const row = this.db
      .prepare('SELECT json FROM replica_body WHERE jmap_account_id = ? AND email_id = ?')
      .get([jmapAccountId, emailId]);
    return str(row?.json);
  }

  deleteBodies(jmapAccountId: string, emailIds: readonly string[]): number {
    if (emailIds.length === 0) return 0;
    const del = this.db.prepare(
      'DELETE FROM replica_body WHERE jmap_account_id = ? AND email_id = ?',
    );
    const clearFlag = this.db.prepare(
      'UPDATE replica_envelope SET has_body = 0, body_bytes = 0 WHERE jmap_account_id = ? AND id = ?',
    );
    let n = 0;
    for (const id of emailIds) {
      n += del.run([jmapAccountId, id]).changes;
      clearFlag.run([jmapAccountId, id]);
    }
    return n;
  }

  bodyBytesTotal(jmapAccountId: string): number {
    const row = this.db
      .prepare('SELECT COALESCE(SUM(bytes), 0) AS n FROM replica_body WHERE jmap_account_id = ?')
      .get([jmapAccountId]);
    return num(row?.n) ?? 0;
  }

  countBodies(jmapAccountId: string): number {
    const row = this.db
      .prepare('SELECT COUNT(*) AS n FROM replica_body WHERE jmap_account_id = ?')
      .get([jmapAccountId]);
    return num(row?.n) ?? 0;
  }

  /** Oldest bodies first - the cap-eviction order. Envelopes always survive. */
  oldestBodies(jmapAccountId: string, limit: number): Array<{ emailId: string; bytes: number }> {
    return this.db
      .prepare(`
        SELECT email_id, bytes FROM replica_body WHERE jmap_account_id = ?
        ORDER BY received_at ASC, email_id ASC LIMIT ?
      `)
      .all([jmapAccountId, limit])
      .map((r) => ({ emailId: String(r.email_id), bytes: num(r.bytes) ?? 0 }));
  }

  /** Bodies below the body-retention floor. */
  bodiesBelow(jmapAccountId: string, isoFloor: string, limit: number): string[] {
    return this.db
      .prepare(`
        SELECT email_id FROM replica_body
        WHERE jmap_account_id = ? AND received_at < ? ORDER BY received_at ASC LIMIT ?
      `)
      .all([jmapAccountId, isoFloor, limit])
      .map((r) => String(r.email_id));
  }

  /** Bodies whose envelope is gone. Invisible to cap eviction, which walks the body table. */
  orphanBodies(jmapAccountId: string, limit: number): string[] {
    return this.db
      .prepare(`
        SELECT b.email_id FROM replica_body b
        LEFT JOIN replica_envelope e
          ON e.jmap_account_id = b.jmap_account_id AND e.id = b.email_id
        WHERE b.jmap_account_id = ? AND e.id IS NULL LIMIT ?
      `)
      .all([jmapAccountId, limit])
      .map((r) => String(r.email_id));
  }

  // ── body queue ───────────────────────────────────────────────────────────

  /**
   * Insert-or-ignore. NEVER resets `attempts` on an existing row, and never
   * revives a `gave_up` row.
   *
   * Returns the number of rows ACTUALLY INSERTED. The distinction matters: the
   * caller reports this as progress, and reporting attempted-rather-than-inserted
   * made the mobile engine believe there was unfinished work on every cycle for
   * as long as any envelope lacked a body - an endless chain of cycles seconds
   * apart, changing nothing.
   */
  enqueueBodies(entries: readonly BodyQueueEntry[]): number {
    if (entries.length === 0) return 0;
    const stmt = this.db.prepare(`
      INSERT OR IGNORE INTO replica_body_queue
        (jmap_account_id, email_id, received_at, attempts, next_attempt_at, last_error, gave_up, gave_up_reason)
      VALUES (?, ?, ?, ?, ?, ?, 0, NULL)
    `);
    let inserted = 0;
    for (const e of entries) {
      inserted += stmt.run([
        e.jmapAccountId, e.emailId, e.receivedAt, e.attempts, e.nextAttemptAt ?? null,
        e.lastError ?? null,
      ]).changes;
    }
    return inserted;
  }

  /** Rows still WANTED: not given up, and past any backoff. Newest first. */
  takeBodyQueue(jmapAccountId: string, limit: number, now: number): BodyQueueEntry[] {
    return this.db
      .prepare(`
        SELECT * FROM replica_body_queue
        WHERE jmap_account_id = ? AND gave_up = 0
          AND (next_attempt_at IS NULL OR next_attempt_at <= ?)
        ORDER BY received_at DESC LIMIT ?
      `)
      .all([jmapAccountId, now, limit])
      .map((r) => ({
        emailId: String(r.email_id),
        jmapAccountId: String(r.jmap_account_id),
        receivedAt: String(r.received_at),
        attempts: num(r.attempts) ?? 0,
        lastError: str(r.last_error) ?? undefined,
        nextAttemptAt: num(r.next_attempt_at) ?? undefined,
        gaveUp: r.gave_up !== 0,
        gaveUpReason: (str(r.gave_up_reason) as BodyGiveUpReason | null) ?? undefined,
      }));
  }

  dequeueBodies(jmapAccountId: string, emailIds: readonly string[]): number {
    if (emailIds.length === 0) return 0;
    const stmt = this.db.prepare(
      'DELETE FROM replica_body_queue WHERE jmap_account_id = ? AND email_id = ?',
    );
    let n = 0;
    for (const id of emailIds) n += stmt.run([jmapAccountId, id]).changes;
    return n;
  }

  bumpBodyAttempt(
    jmapAccountId: string,
    emailId: string,
    nextAttemptAt: number,
    lastError: string,
  ): void {
    this.db
      .prepare(`
        UPDATE replica_body_queue SET attempts = attempts + 1, next_attempt_at = ?, last_error = ?
        WHERE jmap_account_id = ? AND email_id = ?
      `)
      .run([nextAttemptAt, lastError.slice(0, 400), jmapAccountId, emailId]);
  }

  /** Records a durable terminal state INSTEAD of deleting the row. */
  markBodyGaveUp(
    jmapAccountId: string,
    entries: ReadonlyArray<{ emailId: string; receivedAt: string; reason: BodyGiveUpReason; lastError?: string }>,
  ): void {
    if (entries.length === 0) return;
    // A cap-shed body may have no queue row at all (it was fetched and stored
    // successfully, then evicted), so this must be an upsert rather than an
    // update - otherwise the mark is silently dropped and the shed/re-download
    // loop stays open.
    const stmt = this.db.prepare(`
      INSERT INTO replica_body_queue
        (jmap_account_id, email_id, received_at, attempts, next_attempt_at, last_error, gave_up, gave_up_reason)
      VALUES (?, ?, ?, 0, NULL, ?, 1, ?)
      ON CONFLICT(jmap_account_id, email_id) DO UPDATE SET
        gave_up = 1, gave_up_reason = excluded.gave_up_reason,
        last_error = excluded.last_error, next_attempt_at = NULL
    `);
    for (const e of entries) {
      stmt.run([jmapAccountId, e.emailId, e.receivedAt, e.lastError?.slice(0, 400) ?? null, e.reason]);
    }
  }

  listBodyGiveUps(jmapAccountId: string, limit: number): string[] {
    return this.db
      .prepare('SELECT email_id FROM replica_body_queue WHERE jmap_account_id = ? AND gave_up = 1 LIMIT ?')
      .all([jmapAccountId, limit])
      .map((r) => String(r.email_id));
  }

  /**
   * DELETES give-up rows rather than un-flagging them, so a cleared give-up
   * looks like "never queued" and the backfill pass re-enqueues it with a clean
   * attempt count.
   *
   * Called unconditionally by a completed reconcile: 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.
   */
  clearBodyGiveUps(jmapAccountId: string, reason?: BodyGiveUpReason): number {
    if (reason) {
      return this.db
        .prepare('DELETE FROM replica_body_queue WHERE jmap_account_id = ? AND gave_up = 1 AND gave_up_reason = ?')
        .run([jmapAccountId, reason]).changes;
    }
    return this.db
      .prepare('DELETE FROM replica_body_queue WHERE jmap_account_id = ? AND gave_up = 1')
      .run([jmapAccountId]).changes;
  }

  countWantedBodies(jmapAccountId: string, now: number): number {
    const row = this.db
      .prepare(`
        SELECT COUNT(*) AS n FROM replica_body_queue
        WHERE jmap_account_id = ? AND gave_up = 0
          AND (next_attempt_at IS NULL OR next_attempt_at <= ?)
      `)
      .get([jmapAccountId, now]);
    return num(row?.n) ?? 0;
  }

  // ── purge ────────────────────────────────────────────────────────────────

  /** Wipes records AND the body queue. Leaves `replica_sync_state` (policy, cursors). */
  clearRecords(): void {
    for (const table of REPLICA_RECORD_TABLES) this.db.exec(`DELETE FROM ${table}`);
  }

  /** Everything, cursors included. The only safe pairing with a record wipe. */
  purgeAll(): void {
    for (const table of REPLICA_TABLES) this.db.exec(`DELETE FROM ${table}`);
  }

  // ── read path ────────────────────────────────────────────────────────────

  /** Envelope page for a mailbox, newest first. `mailboxId === null` = all mail. */
  listEnvelopes(
    jmapAccountId: string,
    mailboxId: string | null,
    limit: number,
    offset: number,
  ): { rows: Array<Record<string, unknown>>; total: number } {
    if (mailboxId === null) {
      const total =
        num(
          this.db
            .prepare('SELECT COUNT(*) AS n FROM replica_envelope WHERE jmap_account_id = ?')
            .get([jmapAccountId])?.n,
        ) ?? 0;
      const rows = this.db
        .prepare(`
          SELECT * FROM replica_envelope WHERE jmap_account_id = ?
          ORDER BY received_at DESC, id DESC LIMIT ? OFFSET ?
        `)
        .all([jmapAccountId, limit, offset]);
      return { rows, total };
    }
    const total =
      num(
        this.db
          .prepare('SELECT COUNT(*) AS n FROM replica_email_mailbox WHERE jmap_account_id = ? AND mailbox_id = ?')
          .get([jmapAccountId, mailboxId])?.n,
      ) ?? 0;
    const rows = this.db
      .prepare(`
        SELECT e.* FROM replica_envelope e
        JOIN replica_email_mailbox m
          ON m.jmap_account_id = e.jmap_account_id AND m.email_id = e.id
        WHERE e.jmap_account_id = ? AND m.mailbox_id = ?
        ORDER BY e.received_at DESC, e.id DESC LIMIT ? OFFSET ?
      `)
      .all([jmapAccountId, mailboxId, limit, offset]);
    return { rows, total };
  }

  getEnvelopeRaw(jmapAccountId: string, id: string): Record<string, unknown> | null {
    const row = this.db
      .prepare('SELECT * FROM replica_envelope WHERE jmap_account_id = ? AND id = ?')
      .get([jmapAccountId, id]);
    return row ?? null;
  }

  mailboxIdsFor(jmapAccountId: string, emailId: string): string[] {
    return this.db
      .prepare('SELECT mailbox_id FROM replica_email_mailbox WHERE jmap_account_id = ? AND email_id = ?')
      .all([jmapAccountId, emailId])
      .map((r) => String(r.mailbox_id));
  }

  /** Size + freshness, for the Settings surface. */
  stats(jmapAccountId: string): {
    mailboxes: number;
    envelopes: number;
    bodies: number;
    bodyBytes: number;
    wantedBodies: number;
    giveUps: number;
    newest: string | null;
    oldest: string | null;
    fileBytes: number;
  } {
    const mailboxes =
      num(
        this.db
          .prepare('SELECT COUNT(*) AS n FROM replica_mailbox WHERE jmap_account_id = ?')
          .get([jmapAccountId])?.n,
      ) ?? 0;
    const range = this.db
      .prepare('SELECT MIN(received_at) AS lo, MAX(received_at) AS hi FROM replica_envelope WHERE jmap_account_id = ?')
      .get([jmapAccountId]);
    let fileBytes = 0;
    for (const f of dbSiblings(this.dbPath)) {
      try { fileBytes += fs.statSync(f).size; } catch { /* absent sibling */ }
    }
    return {
      mailboxes,
      envelopes: this.countEnvelopes(jmapAccountId),
      bodies: this.countBodies(jmapAccountId),
      bodyBytes: this.bodyBytesTotal(jmapAccountId),
      wantedBodies: this.countWantedBodies(jmapAccountId, Date.now()),
      giveUps:
        num(
          this.db
            .prepare('SELECT COUNT(*) AS n FROM replica_body_queue WHERE jmap_account_id = ? AND gave_up = 1')
            .get([jmapAccountId])?.n,
        ) ?? 0,
      newest: str(range?.hi),
      oldest: str(range?.lo),
      fileBytes,
    };
  }

  /** Every JMAP account id with rows, so the read path can find them without a session. */
  knownJmapAccountIds(): string[] {
    const ids = new Set<string>();
    for (const table of ['replica_envelope', 'replica_mailbox'] as const) {
      for (const r of this.db.prepare(`SELECT DISTINCT jmap_account_id FROM ${table}`).all()) {
        if (typeof r.jmap_account_id === 'string') ids.add(r.jmap_account_id);
      }
    }
    return [...ids];
  }
}

function readVersion(db: SqlcipherDatabase): number | null {
  try {
    const row = db.prepare('SELECT v FROM meta WHERE k = ?').get([REPLICA_VERSION_KEY]);
    if (!row || row.v === undefined) return null;
    const n = Number(row.v);
    return Number.isFinite(n) ? n : null;
  } catch {
    // `meta` doesn't exist yet - a fresh file.
    return null;
  }
}