AkurAI-Mail / tools/akurai-mail-index.ts

Latest change 7a284f40b2637792d0599c38c4cac0342da676d7 - Merge live EC2 ops copies into scripts/ and tools/ by Ólafur Búi Ólafsson

#!/usr/bin/env bun
const _Set: SetConstructor = Set;

import { SQL } from "bun";
import { createHash } from "node:crypto";
import { readdir, readFile, stat } from "node:fs/promises";
import { homedir } from "node:os";
import { join } from "node:path";
import { ImapFlow, type FetchMessageObject, type MessageAddressObject } from "imapflow";
import PostalMime, { type Address } from "postal-mime";

// Optional env file (KEY=VALUE lines); variables already present in the environment win.
// Loaded before the config constants so a sudo re-exec (maildir reads) needs no preserved env.
{
  const envFile = process.env.AKURAI_MAIL_ENV_FILE ?? "/home/ubuntu/.config/akurai-mail-index/env";
  const text = await readFile(envFile, "utf8").catch(() => "");
  for (const line of text.split("\n")) {
    const match = line.match(/^([A-Z0-9_]+)=(.*)$/);
    if (match && !(match[1]! in process.env)) process.env[match[1]!] = match[2]!;
  }
}

const HOME = homedir();
// Under the sudo re-exec (maildir sync) HOME is /root; keep defaults on the service user's config dir.
const CONFIG_DIR = process.getuid?.() === 0 ? "/home/ubuntu/.config/akurai-mail-index" : join(HOME, ".config/akurai-mail-index");
const ACCOUNTS_DIR = process.env.AKURAI_MAIL_ACCOUNTS_DIR ?? join(HOME, ".omp/agent/skills/akurai-mail/accounts");
const DB_PASSWORD_FILE = process.env.AKURAI_MAIL_DB_PASSWORD_FILE ?? join(CONFIG_DIR, "db.pass");
const DB_HOST = process.env.AKURAI_MAIL_DB_HOST ?? "127.0.0.1";
const DB_PORT = Number(process.env.AKURAI_MAIL_DB_PORT ?? 3306);
const EMBED_URL = process.env.AKURAI_MAIL_EMBED_URL ?? "https://openrouter.ai/api/v1/embeddings";
const RERANK_URL = process.env.AKURAI_MAIL_RERANK_URL ?? "";
const EMBED_MODEL = process.env.AKURAI_MAIL_EMBED_MODEL ?? "voyageai/voyage-4";
const EMBED_API_KEY = process.env.AKURAI_MAIL_EMBED_API_KEY ?? "";
const EMBED_DIMENSIONS = 1024;
const SEARCH_TEXT_LIMIT = 1_500;

interface AccountConfig {
  alias: string;
  address: string;
  displayName: string;
  host: string;
  port: number;
  secure: boolean;
  login: string;
  passwordFile: string;
  maildirPath?: string;
}

interface IndexedMessage {
  id: string;
  account: string;
  address: string;
  folder: string;
  uid: number;
  subject: string;
  senderName: string;
  senderAddress: string;
  sentAt: string | null;
  body: string;
  contentSha: Buffer;
}

interface SearchRow {
  id: string | number | bigint;
  account: string;
  address: string;
  folder: string;
  uid: number;
  subject: string;
  sender_name: string | null;
  sender_address: string | null;
  sent_at: string | Date | null;
  body_text: string | null;
  body_html: string | null;
  distance: number;
}

type MailCategory = "direct-personal" | "personal-administrative" | "security" | "likely-phishing" | "marketing" | "other";

interface RecentRow extends Omit<SearchRow, "distance"> {
  date: string | Date | null;
}

interface CliOptions {
  account?: string;
  folder?: string;
  from?: string;
  after?: string;
  before?: string;
  since?: string;
  category?: string;
  sync: boolean;
  skipEmbeddings: boolean;
  help: boolean;
  limit: number;
  limitProvided: boolean;
  used: Set<string>;
  json: boolean;
}

interface SyncReport {
  status: "succeeded" | "skipped" | "deferred";
  reason?: string;
  accountsScanned?: number;
  foldersScanned?: number;
  messagesAdded?: number;
  embeddingsUpdated?: number;
  embeddingsSkipped?: boolean;
  messagesDeleted?: number;
  newMessages?: Array<Omit<IndexedMessage, "contentSha" | "body"> & { preview: string }>;
}

const schema = [
  `CREATE TABLE IF NOT EXISTS accounts (
    id INT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    alias VARCHAR(191) NOT NULL UNIQUE,
    address VARCHAR(320) NOT NULL,
    display_name VARCHAR(320) NOT NULL DEFAULT '',
    last_sync_at DATETIME(6) NULL,
    last_error TEXT NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
    INDEX accounts_address (address)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
  `CREATE TABLE IF NOT EXISTS folders (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    account_id INT UNSIGNED NOT NULL,
    path VARCHAR(512) NOT NULL,
    uid_validity BIGINT UNSIGNED NOT NULL DEFAULT 0,
    last_uid BIGINT UNSIGNED NOT NULL DEFAULT 0,
    message_count INT UNSIGNED NOT NULL DEFAULT 0,
    last_sync_at DATETIME(6) NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
    UNIQUE KEY folders_account_path (account_id, path),
    CONSTRAINT folders_account_fk FOREIGN KEY (account_id) REFERENCES accounts(id) ON DELETE CASCADE
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
  `CREATE TABLE IF NOT EXISTS messages (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    account_id INT UNSIGNED NOT NULL,
    folder_id BIGINT UNSIGNED NOT NULL,
    uid BIGINT UNSIGNED NOT NULL,
    message_id VARCHAR(998) NULL,
    in_reply_to VARCHAR(998) NULL,
    subject TEXT NOT NULL,
    sender_name VARCHAR(512) NULL,
    sender_address VARCHAR(320) NULL,
    recipients JSON NOT NULL,
    cc JSON NOT NULL,
    sent_at DATETIME(6) NULL,
    internal_date DATETIME(6) NULL,
    flags JSON NOT NULL,
    size_bytes BIGINT UNSIGNED NOT NULL DEFAULT 0,
    body_text MEDIUMTEXT NOT NULL,
    body_html MEDIUMTEXT NOT NULL,
    attachments JSON NOT NULL,
    raw_message LONGBLOB NOT NULL,
    content_sha BINARY(32) NOT NULL,
    deleted_at DATETIME(6) NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
    UNIQUE KEY messages_folder_uid (folder_id, uid),
    INDEX messages_account_date (account_id, sent_at),
    INDEX messages_folder_date (folder_id, sent_at),
    INDEX messages_sender (sender_address),
    INDEX messages_message_id (message_id(191)),
    FULLTEXT INDEX messages_text (subject, sender_name, sender_address, body_text),
    CONSTRAINT messages_account_fk FOREIGN KEY (account_id) REFERENCES accounts(id) ON DELETE CASCADE,
    CONSTRAINT messages_folder_fk FOREIGN KEY (folder_id) REFERENCES folders(id) ON DELETE CASCADE
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
  `CREATE TABLE IF NOT EXISTS embeddings (
    message_id BIGINT UNSIGNED NOT NULL PRIMARY KEY,
    content_sha BINARY(32) NOT NULL,
    model VARCHAR(191) NOT NULL,
    embedding VECTOR(1024) NOT NULL,
    embedded_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    VECTOR INDEX embedding_vector (embedding) M=8 DISTANCE=cosine,
    CONSTRAINT embeddings_message_fk FOREIGN KEY (message_id) REFERENCES messages(id) ON DELETE CASCADE
  ) ENGINE=InnoDB`,
  `CREATE TABLE IF NOT EXISTS sync_runs (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    started_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    finished_at DATETIME(6) NULL,
    status ENUM('running','succeeded','failed','skipped') NOT NULL DEFAULT 'running',
    accounts_scanned INT UNSIGNED NOT NULL DEFAULT 0,
    folders_scanned INT UNSIGNED NOT NULL DEFAULT 0,
    messages_added INT UNSIGNED NOT NULL DEFAULT 0,
    messages_updated INT UNSIGNED NOT NULL DEFAULT 0,
    messages_deleted INT UNSIGNED NOT NULL DEFAULT 0,
    error_text TEXT NULL,
    INDEX sync_runs_started (started_at)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`
];

async function database(): Promise<SQL> {
  const password = (await readFile(DB_PASSWORD_FILE, "utf8")).trim();
  if (!password) throw new Error(`empty database password file: ${DB_PASSWORD_FILE}`);
  return new SQL({ adapter: "mysql", hostname: DB_HOST, port: DB_PORT, database: "akurai_mail", username: "mail_index", password, max: 4, idleTimeout: 30 });
}

async function ensureSchema(db: SQL): Promise<void> {
  for (const statement of schema) await db.unsafe(statement);
}

async function loadAccounts(selected?: string): Promise<AccountConfig[]> {
  const files = (await readdir(ACCOUNTS_DIR)).filter((name) => name.endsWith(".conf")).sort();
  const accounts: AccountConfig[] = [];
  for (const file of files) {
    const parsed = Bun.TOML.parse(await readFile(join(ACCOUNTS_DIR, file), "utf8")) as Record<string, unknown>;
    const root = parsed.accounts as Record<string, Record<string, unknown>> | undefined;
    if (!root) continue;
    for (const [alias, value] of Object.entries(root)) {
      if (selected && alias !== selected) continue;
      const maildir = value.maildir as Record<string, unknown> | undefined;
      const backend = value.backend as Record<string, unknown> | undefined;
      if (maildir) {
        accounts.push({
          alias,
          address: String(value.email ?? ""),
          displayName: String(value["display-name"] ?? ""),
          host: "",
          port: 0,
          secure: false,
          login: String(value.email ?? ""),
          passwordFile: "",
          maildirPath: String(maildir.path ?? "")
        });
        if (!accounts[accounts.length - 1]!.maildirPath) throw new Error(`${alias}: maildir.path is required`);
        continue;
      }
      if (!backend) throw new Error(`${alias}: account needs either backend (IMAP) or maildir section`);
      const auth = backend.auth as Record<string, unknown>;
      const encryption = backend.encryption as Record<string, unknown>;
      const command = String(auth.cmd ?? "");
      const match = command.match(/^cat\s+(.+)$/);
      if (!match) throw new Error(`${alias}: backend.auth.cmd must be a simple cat command`);
      accounts.push({
        alias,
        address: String(value.email ?? backend.login ?? ""),
        displayName: String(value["display-name"] ?? ""),
        host: String(backend.host ?? ""),
        port: Number(backend.port ?? 993),
        secure: encryption.type === "tls",
        login: String(backend.login ?? value.email ?? ""),
        passwordFile: match[1]!.trim()
      });
    }
  }
  if (selected && accounts.length === 0) throw new Error(`unknown account: ${selected}`);
  if (accounts.length === 0) throw new Error(`no mail accounts found in ${ACCOUNTS_DIR}`);
  return accounts;
}

function mysqlDate(value: Date | string | undefined): string | null {
  if (!value) return null;
  const date = value instanceof Date ? value : new Date(value);
  return Number.isNaN(date.getTime()) ? null : date.toISOString().slice(0, 23).replace("T", " ");
}

function addressList(values: MessageAddressObject[] | undefined): Array<{ name: string; address: string }> {
  return (values ?? []).map((item) => ({ name: item.name ?? "", address: item.address ?? "" }));
}

function postalAddresses(values: Address[] | undefined): Array<{ name: string; address: string }> {
  const result: Array<{ name: string; address: string }> = [];
  for (const value of values ?? []) {
    const mailboxes = "group" in value && value.group ? value.group : [value];
    for (const mailbox of mailboxes) {
      if (mailbox.address) result.push({ name: mailbox.name, address: mailbox.address });
    }
  }
  return result;
}

function searchableText(message: Pick<IndexedMessage, "subject" | "senderName" | "senderAddress" | "body">): string {
  return [`Subject: ${message.subject}`, `From: ${message.senderName} <${message.senderAddress}>`, message.body].join("\n").slice(0, SEARCH_TEXT_LIMIT);
}

function htmlToText(html: string): string {
  return html
    .replace(/<(script|style)\b[^>]*>[\s\S]*?<\/\1>/gi, " ")
    .replace(/<br\s*\/?>/gi, "\n")
    .replace(/<\/(?:div|h[1-6]|li|p|tr)>/gi, "\n")
    .replace(/<[^>]+>/g, " ")
    .replace(/&nbsp;/gi, " ")
    .replace(/&amp;/gi, "&")
    .replace(/&lt;/gi, "<")
    .replace(/&gt;/gi, ">")
    .replace(/&quot;/gi, "\"")
    .replace(/&#39;|&apos;/gi, "'")
    .replace(/&#(?:x[0-9a-f]+|\d+);/gi, " ")
    .replace(/&(?:zwnj|zwj|lrm|rlm);/gi, " ")
    .replace(/[ \t]+\n/g, "\n")
    .replace(/\n[ \t]+/g, "\n")
    .replace(/[ \t]{2,}/g, " ")
    .replace(/\n{3,}/g, "\n\n")
    .trim();
}

function plainBody(text: unknown, html: unknown): string {
  const body = typeof text === "string" ? text.trim() : "";
  return body || htmlToText(typeof html === "string" ? html : "");
}

type ImapCommandError = Error & { code?: string; responseStatus?: string; responseText?: string };

class TemporaryImapError extends Error {}

function imapErrorDetail(error: unknown): string {
  if (!(error instanceof Error)) return String(error);
  const commandError = error as ImapCommandError;
  const response = commandError.responseText?.trim();
  return response ? `${error.message}: ${response}` : error.message;
}

function isTemporaryImapError(error: unknown): boolean {
  if (!(error instanceof Error)) return false;
  const commandError = error as ImapCommandError;
  return /temporar|unavailable|try again|rate.?limit|timeout|timed out/i.test(`${commandError.responseText ?? ""} ${error.message}`)
    || ["ECONNRESET", "ECONNREFUSED", "EHOSTUNREACH", "ENETUNREACH", "ETIMEDOUT"].includes(commandError.code ?? "");
}

async function connectImap(account: AccountConfig, password: string): Promise<ImapFlow> {
  const delays = [0, 1_000, 3_000];
  let lastError: unknown;
  for (const delay of delays) {
    if (delay) await Bun.sleep(delay);
    const client = new ImapFlow({ host: account.host, port: account.port, secure: account.secure, auth: { user: account.login, pass: password }, logger: false });
    try {
      await client.connect();
      return client;
    } catch (error) {
      client.close();
      if (!isTemporaryImapError(error)) throw new Error(`${account.alias}: ${imapErrorDetail(error)}`);
      lastError = error;
    }
  }
  throw new TemporaryImapError(`${account.alias}: IMAP temporarily unavailable after ${delays.length} attempts: ${imapErrorDetail(lastError)}`);
}

async function embeddingBatch(texts: string[]): Promise<number[][]> {
  if (texts.length === 0) return [];
  const headers: Record<string, string> = { "content-type": "application/json" };
  if (EMBED_API_KEY) headers.Authorization = `Bearer ${EMBED_API_KEY}`;
  const response = await fetch(EMBED_URL, { method: "POST", headers, body: JSON.stringify({ model: EMBED_MODEL, input: texts }), signal: AbortSignal.timeout(120_000) });
  const body = await response.text();
  if (!response.ok) throw new Error(`embedding service returned ${response.status}: ${body.slice(0, 300)}`);
  const payload: unknown = JSON.parse(body);
  if (!payload || typeof payload !== "object" || !("data" in payload) || !Array.isArray(payload.data)) throw new Error("embedding service returned an invalid response");
  const ordered = payload.data.map((item, index) => {
    if (!item || typeof item !== "object" || !(("embedding" in item) && Array.isArray(item.embedding))) throw new Error(`embedding ${index} is invalid`);
    return item.embedding.map(Number);
  });
  if (ordered.length !== texts.length || ordered.some((vector) => vector.length !== EMBED_DIMENSIONS)) throw new Error("embedding service returned an invalid vector batch");
  return ordered;
}

async function rerank(query: string, texts: string[]): Promise<Array<{ index: number; score: number }>> {
  if (!RERANK_URL) return texts.map((_text, index) => ({ index, score: 1 }));
  try {
    const response = await fetch(RERANK_URL, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ query, texts, return_text: false }), signal: AbortSignal.timeout(120_000) });
    if (!response.ok) throw new Error(`reranker returned ${response.status}`);
    return await response.json() as Array<{ index: number; score: number }>;
  } catch {
    return texts.map((_text, index) => ({ index, score: 1 }));
  }
}

async function upsertAccount(db: SQL, account: AccountConfig): Promise<number> {
  await db`INSERT INTO accounts (alias, address, display_name) VALUES (${account.alias}, ${account.address}, ${account.displayName}) ON DUPLICATE KEY UPDATE address=VALUES(address), display_name=VALUES(display_name), last_error=NULL`;
  const rows = await db`SELECT id FROM accounts WHERE alias=${account.alias}` as Array<{ id: number }>;
  return Number(rows[0]!.id);
}

async function upsertFolder(db: SQL, accountId: number, path: string, uidValidity: bigint | number): Promise<{ id: number; uidValidity: bigint }> {
  await db`INSERT INTO folders (account_id, path, uid_validity) VALUES (${accountId}, ${path}, ${String(uidValidity)}) ON DUPLICATE KEY UPDATE path=VALUES(path)`;
  const rows = await db`SELECT id, uid_validity FROM folders WHERE account_id=${accountId} AND path=${path}` as Array<{ id: number; uid_validity: bigint }>;
  return { id: Number(rows[0]!.id), uidValidity: BigInt(rows[0]!.uid_validity) };
}

async function storeMessage(db: SQL, accountId: number, folderId: number, account: AccountConfig, folder: string, item: FetchMessageObject): Promise<IndexedMessage> {
  if (!item.source) throw new Error(`${account.alias}/${folder}/${item.uid}: source missing`);
  const parsed = await PostalMime.parse(item.source, { attachmentEncoding: "arraybuffer" });
  const envelope = item.envelope;
  const from = postalAddresses(parsed.from ? [parsed.from] : undefined)[0] ?? addressList(envelope?.from)[0] ?? { name: "", address: "" };
  const body = plainBody(parsed.text, parsed.html);
  const html = parsed.html ?? "";
  const subject = parsed.subject ?? envelope?.subject ?? "";
  const contentSha = createHash("sha256").update(item.source).digest();
  const recipients = postalAddresses(parsed.to).length ? postalAddresses(parsed.to) : addressList(envelope?.to);
  const cc = postalAddresses(parsed.cc).length ? postalAddresses(parsed.cc) : addressList(envelope?.cc);
  const attachments = parsed.attachments.map((attachment) => ({ filename: attachment.filename, mimeType: attachment.mimeType, disposition: attachment.disposition, contentId: attachment.contentId, size: typeof attachment.content === "string" ? Buffer.byteLength(attachment.content) : attachment.content.byteLength }));
  await db`INSERT INTO messages (account_id, folder_id, uid, message_id, in_reply_to, subject, sender_name, sender_address, recipients, cc, sent_at, internal_date, flags, size_bytes, body_text, body_html, attachments, raw_message, content_sha, deleted_at)
    VALUES (${accountId}, ${folderId}, ${item.uid}, ${parsed.messageId ?? envelope?.messageId ?? null}, ${parsed.inReplyTo ?? envelope?.inReplyTo ?? null}, ${subject}, ${from.name}, ${from.address}, ${JSON.stringify(recipients)}, ${JSON.stringify(cc)}, ${mysqlDate(parsed.date ?? envelope?.date)}, ${mysqlDate(item.internalDate)}, ${JSON.stringify([...item.flags ?? []])}, ${item.size ?? item.source.byteLength}, ${body}, ${html}, ${JSON.stringify(attachments)}, ${item.source}, ${contentSha}, NULL)
    ON DUPLICATE KEY UPDATE message_id=VALUES(message_id), in_reply_to=VALUES(in_reply_to), subject=VALUES(subject), sender_name=VALUES(sender_name), sender_address=VALUES(sender_address), recipients=VALUES(recipients), cc=VALUES(cc), sent_at=VALUES(sent_at), internal_date=VALUES(internal_date), flags=VALUES(flags), size_bytes=VALUES(size_bytes), body_text=VALUES(body_text), body_html=VALUES(body_html), attachments=VALUES(attachments), raw_message=VALUES(raw_message), content_sha=VALUES(content_sha), deleted_at=NULL`;
  const rows = await db`SELECT id FROM messages WHERE folder_id=${folderId} AND uid=${item.uid}` as Array<{ id: string | number | bigint }>;
  return { id: String(rows[0]!.id), account: account.alias, address: account.address, folder, uid: item.uid, subject, senderName: from.name, senderAddress: from.address, sentAt: mysqlDate(parsed.date ?? envelope?.date), body, contentSha };
}

async function storeEmbeddings(db: SQL, messages: IndexedMessage[]): Promise<void> {
  for (let offset = 0; offset < messages.length; offset += 4) {
    const batch = messages.slice(offset, offset + 4);
    const vectors = await embeddingBatch(batch.map(searchableText));
    for (let index = 0; index < batch.length; index++) {
      const message = batch[index]!;
      await db`INSERT INTO embeddings (message_id, content_sha, model, embedding) VALUES (${message.id}, ${message.contentSha}, ${EMBED_MODEL}, VEC_FromText(${JSON.stringify(vectors[index])})) ON DUPLICATE KEY UPDATE content_sha=VALUES(content_sha), model=VALUES(model), embedding=VALUES(embedding), embedded_at=CURRENT_TIMESTAMP(6)`;
    }
  }
}

async function embedPending(db: SQL): Promise<number> {
  let embedded = 0;
  while (true) {
    const rows = await db`SELECT m.id, a.alias AS account, a.address, f.path AS folder, m.uid, m.subject, m.sender_name, m.sender_address, m.sent_at, m.body_text, m.content_sha
      FROM messages m JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id LEFT JOIN embeddings e ON e.message_id=m.id
      WHERE m.deleted_at IS NULL AND (e.message_id IS NULL OR e.content_sha<>m.content_sha OR e.model<>${EMBED_MODEL})
      ORDER BY m.id LIMIT 100` as Array<{ id: string | number | bigint; account: string; address: string; folder: string; uid: number; subject: string; sender_name: string | null; sender_address: string | null; sent_at: string | null; body_text: string | null; content_sha: Uint8Array }>;
    if (rows.length === 0) return embedded;
    const messages = rows.map((row) => ({ id: String(row.id), account: row.account, address: row.address, folder: row.folder, uid: Number(row.uid), subject: row.subject, senderName: row.sender_name ?? "", senderAddress: row.sender_address ?? "", sentAt: row.sent_at, body: row.body_text ?? "", contentSha: Buffer.from(row.content_sha) }));
    await storeEmbeddings(db, messages);
    embedded += messages.length;
  }
}

async function syncAccount(db: SQL, account: AccountConfig): Promise<{ folders: number; added: IndexedMessage[]; deleted: number }> {
  const accountId = await upsertAccount(db, account);
  const password = (await readFile(account.passwordFile, "utf8")).trim();
  let client: ImapFlow | null = null;
  const added: IndexedMessage[] = [];
  let deleted = 0;
  let scannedFolders = 0;
  try {
    client = await connectImap(account, password);
    const mailboxes = (await client.list()).filter((mailbox) => !mailbox.flags.has("\\Noselect"));
    for (const mailboxInfo of mailboxes) {
      const mailbox = await client.mailboxOpen(mailboxInfo.path, { readOnly: true });
      scannedFolders++;
      const folder = await upsertFolder(db, accountId, mailboxInfo.path, mailbox.uidValidity);
      if (folder.uidValidity !== 0n && folder.uidValidity !== BigInt(mailbox.uidValidity)) {
        await db`UPDATE messages SET deleted_at=CURRENT_TIMESTAMP(6) WHERE folder_id=${folder.id} AND deleted_at IS NULL`;
      }
      const existingRows = await db`SELECT uid, HEX(content_sha) AS sha FROM messages WHERE folder_id=${folder.id} AND deleted_at IS NULL` as Array<{ uid: number; sha: string }>;
      const existing: Record<number, string> = Object.fromEntries(existingRows.map((row) => [Number(row.uid), row.sha]));
      const current: Set<number> = new _Set<number>();
      const newUids: number[] = [];
      if (mailbox.exists > 0) {
        for await (const message of client.fetch("1:*", { uid: true })) {
          current.add(message.uid);
          if (!(String(message.uid) in existing)) newUids.push(message.uid);
        }
      }
      for (let offset = 0; offset < newUids.length; offset += 25) {
        const chunk = newUids.slice(offset, offset + 25);
        const fetched: FetchMessageObject[] = [];
        for await (const message of client.fetch(chunk, { uid: true, flags: true, envelope: true, internalDate: true, size: true, source: true }, { uid: true })) fetched.push(message);
        for (const message of fetched) added.push(await storeMessage(db, accountId, folder.id, account, mailboxInfo.path, message));
      }
      const removed = Object.keys(existing).map(Number).filter((uid) => !current.has(uid));
      for (const uid of removed) await db`UPDATE messages SET deleted_at=CURRENT_TIMESTAMP(6) WHERE folder_id=${folder.id} AND uid=${uid} AND deleted_at IS NULL`;
      deleted += removed.length;
      const lastUid = current.size ? Math.max(...current) : 0;
      await db`UPDATE folders SET uid_validity=${String(mailbox.uidValidity)}, last_uid=${lastUid}, message_count=${mailbox.exists}, last_sync_at=CURRENT_TIMESTAMP(6) WHERE id=${folder.id}`;
    }
    await db`UPDATE accounts SET last_sync_at=CURRENT_TIMESTAMP(6), last_error=NULL WHERE id=${accountId}`;
    return { folders: scannedFolders, added, deleted };
  } catch (error) {
    await db`UPDATE accounts SET last_error=${imapErrorDetail(error)} WHERE id=${accountId}`;
    throw error;
  } finally {
    if (client) await client.logout().catch(() => undefined);
  }
}

const MAILDIR_FLAGS: Record<string, string> = { R: "\\Seen", T: "\\Deleted", S: "\\Answered", F: "\\Flagged", P: "\\Draft" };


async function readMaildirUidlist(dir: string): Promise<{ uidValidity: bigint; uidByBase: Map<string, number> }> {
  const text = await readFile(join(dir, "dovecot-uidlist"), "utf8").catch(() => "");
  const uidByBase = new Map<string, number>();
  let uidValidity = 0n;
  for (const line of text.split("\n")) {
    const header = line.match(/^\d+ V(\S+)/);
    if (header) {
      uidValidity = BigInt(header[1]!);
      continue;
    }
    const entry = line.match(/^(\d+)\s+:?(.+)$/);
    if (entry) uidByBase.set(entry[2]!.trim().replace(/,[SW]=\d+/g, ""), Number(entry[1]!));
  }
  return { uidValidity, uidByBase };
}


function maildirFlags(fileName: string): Set<string> {
  const flags = new Set<string>();
  const info = fileName.match(/:2,(.*)$/);
  if (info) for (const character of info[1]!) {
    const flag = MAILDIR_FLAGS[character];
    if (flag) flags.add(flag);
  }
  return flags;
}

async function syncAccountMaildir(db: SQL, account: AccountConfig): Promise<{ folders: number; added: IndexedMessage[]; deleted: number }> {
  const accountId = await upsertAccount(db, account);
  const root = account.maildirPath!;
  const added: IndexedMessage[] = [];
  let deleted = 0;
  let scannedFolders = 0;
  try {
    const entries = await readdir(root, { withFileTypes: true });
    const folderDirs: Array<{ name: string; path: string }> = [{ name: "INBOX", path: root }];
    for (const entry of entries) {
      if (entry.isDirectory() && entry.name.startsWith(".")) folderDirs.push({ name: entry.name.slice(1).replace(/\./g, "/"), path: join(root, entry.name) });
    }
    for (const folderDir of folderDirs) {
      const { uidValidity, uidByBase } = await readMaildirUidlist(folderDir.path);
      const folder = await upsertFolder(db, accountId, folderDir.name, uidValidity);
      if (folder.uidValidity !== 0n && folder.uidValidity !== uidValidity) {
        await db`UPDATE messages SET deleted_at=CURRENT_TIMESTAMP(6) WHERE folder_id=${folder.id} AND deleted_at IS NULL`;
      }
      const existingRows = await db`SELECT uid, HEX(content_sha) AS sha FROM messages WHERE folder_id=${folder.id} AND deleted_at IS NULL` as Array<{ uid: number; sha: string }>;
      const existing: Record<number, string> = Object.fromEntries(existingRows.map((row) => [Number(row.uid), row.sha]));
      const current: Set<number> = new _Set<number>();
      const pending: Array<{ uid: number; file: string; dir: string }> = [];
      for (const sub of ["new", "cur"] as const) {
        const subDir = join(folderDir.path, sub);
        const files = await readdir(subDir).catch(() => [] as string[]);
        for (const file of files) {
          if (file.startsWith(".")) continue;
          const uid = uidByBase.get(file.replace(/:2,.*$/, "").replace(/,[SW]=\d+/g, ""));
          if (uid === undefined || current.has(uid)) continue;
          current.add(uid);
          if (!(String(uid) in existing)) pending.push({ uid, file, dir: subDir });
        }
      }
      for (const item of pending) {
        const fullPath = join(item.dir, item.file);
        const source = await readFile(fullPath);
        const message: FetchMessageObject = { seq: item.uid, uid: item.uid, flags: maildirFlags(item.file), internalDate: (await stat(fullPath)).mtime, size: source.byteLength, source };
        added.push(await storeMessage(db, accountId, folder.id, account, folderDir.name, message));
      }
      const removed = Object.keys(existing).map(Number).filter((uid) => !current.has(uid));
      for (const uid of removed) await db`UPDATE messages SET deleted_at=CURRENT_TIMESTAMP(6) WHERE folder_id=${folder.id} AND uid=${uid} AND deleted_at IS NULL`;
      deleted += removed.length;
      const lastUid = current.size ? Math.max(...current) : 0;
      await db`UPDATE folders SET uid_validity=${String(uidValidity)}, last_uid=${lastUid}, message_count=${current.size}, last_sync_at=CURRENT_TIMESTAMP(6) WHERE id=${folder.id}`;
      scannedFolders++;
    }
    await db`UPDATE accounts SET last_sync_at=CURRENT_TIMESTAMP(6), last_error=NULL WHERE id=${accountId}`;
    return { folders: scannedFolders, added, deleted };
  } catch (error) {
    await db`UPDATE accounts SET last_error=${imapErrorDetail(error)} WHERE id=${accountId}`;
    throw error;
  }
}

async function sync(options: CliOptions, emit = true, updateEmbeddings = true): Promise<SyncReport> {
  const db = await database();
  await ensureSchema(db);
  const lockRows = await db`SELECT GET_LOCK('akurai_mail_sync', 0) AS acquired` as Array<{ acquired: number }>;
  if (Number(lockRows[0]?.acquired) !== 1) {
    const report: SyncReport = { status: "skipped", reason: "sync already running" };
    if (emit) console.log(options.json ? JSON.stringify(report) : "Mail sync already running; skipped.");
    await db.close();
    return report;
  }
  await db`INSERT INTO sync_runs () VALUES ()`;
  const runRows = await db`SELECT LAST_INSERT_ID() AS id` as Array<{ id: string | number | bigint }>;
  const runId = String(runRows[0]!.id);
  const allAdded: IndexedMessage[] = [];
  let accountsScanned = 0;
  let foldersScanned = 0;
  let deleted = 0;
  try {
    for (const account of await loadAccounts(options.account)) {
      const result = account.maildirPath ? await syncAccountMaildir(db, account) : await syncAccount(db, account);
      accountsScanned++;
      foldersScanned += result.folders;
      deleted += result.deleted;
      allAdded.push(...result.added);
    }
    const embeddingsUpdated = updateEmbeddings ? await embedPending(db) : 0;
    await db`UPDATE sync_runs SET finished_at=CURRENT_TIMESTAMP(6), status='succeeded', accounts_scanned=${accountsScanned}, folders_scanned=${foldersScanned}, messages_added=${allAdded.length}, messages_updated=${embeddingsUpdated}, messages_deleted=${deleted} WHERE id=${runId}`;
    const report: SyncReport = { status: "succeeded", accountsScanned, foldersScanned, messagesAdded: allAdded.length, embeddingsUpdated, embeddingsSkipped: !updateEmbeddings, messagesDeleted: deleted, newMessages: allAdded.map(({ contentSha: _contentSha, body, ...message }) => ({ ...message, preview: body.replace(/\s+/g, " ").trim().slice(0, 240) })) };
    if (emit) console.log(options.json ? JSON.stringify(report) : `Synced ${accountsScanned} account(s), ${foldersScanned} folder(s); added ${allAdded.length}, embedded ${embeddingsUpdated}, removed ${deleted}.`);
    return report;
  } catch (error) {
    const detail = imapErrorDetail(error);
    if (error instanceof TemporaryImapError) {
      await db`UPDATE sync_runs SET finished_at=CURRENT_TIMESTAMP(6), status='skipped', accounts_scanned=${accountsScanned}, folders_scanned=${foldersScanned}, messages_added=${allAdded.length}, messages_deleted=${deleted}, error_text=${detail} WHERE id=${runId}`;
      const report: SyncReport = { status: "deferred", reason: detail };
      if (emit) console.log(options.json ? JSON.stringify(report) : `Mail sync deferred; the existing index remains available. ${detail}`);
      return report;
    }
    await db`UPDATE sync_runs SET finished_at=CURRENT_TIMESTAMP(6), status='failed', accounts_scanned=${accountsScanned}, folders_scanned=${foldersScanned}, messages_added=${allAdded.length}, messages_deleted=${deleted}, error_text=${detail} WHERE id=${runId}`;
    throw error;
  } finally {
    await db`SELECT RELEASE_LOCK('akurai_mail_sync')`;
    await db.close();
  }
}

function senderFromNaturalLanguage(query: string): string | null {
  const match = query.match(/\b(?:from|frá)\s+["“]?(.+?)["”]?\s*$/iu);
  if (!match?.[1]) return null;
  return match[1].replace(/[?.!,;:]+$/, "").trim() || null;
}

async function searchMail(query: string, options: CliOptions): Promise<void> {
  const db = await database();
  await ensureSchema(db);
  const [vector] = await embeddingBatch([query]);
  const account = options.account ?? null;
  const folder = options.folder ?? null;
  const senderName = options.from ?? senderFromNaturalLanguage(query);
  const sender = senderName ? `%${senderName}%` : null;
  const after = options.after ? calendarDate(options.after, "--after") : null;
  const before = options.before ? calendarDate(options.before, "--before") : null;
  if (after && before && before <= after) throw new Error("--before must be later than --after");
  const candidateLimit = Math.max(options.limit * 5, 30);
  const rows = await db`SELECT m.id, a.alias AS account, a.address, f.path AS folder, m.uid, m.subject, m.sender_name, m.sender_address, m.sent_at, m.body_text, m.body_html, VEC_DISTANCE_COSINE(e.embedding, VEC_FromText(${JSON.stringify(vector)})) AS distance
    FROM embeddings e JOIN messages m ON m.id=e.message_id JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id
    WHERE m.deleted_at IS NULL
      AND (${account} IS NULL OR a.alias=${account})
      AND (${folder} IS NULL OR f.path=${folder})
      AND (${sender} IS NULL OR m.sender_address LIKE ${sender} OR m.sender_name LIKE ${sender})
      AND (${after} IS NULL OR COALESCE(m.sent_at,m.internal_date)>=${after})
      AND (${before} IS NULL OR COALESCE(m.sent_at,m.internal_date)<${before})
    ORDER BY VEC_DISTANCE_COSINE(e.embedding, VEC_FromText(${JSON.stringify(vector)})) ASC LIMIT ${candidateLimit}` as SearchRow[];
  const ranked = rows.length > 1 ? await rerank(query, rows.map((row) => searchableText({ subject: row.subject, senderName: row.sender_name ?? "", senderAddress: row.sender_address ?? "", body: plainBody(row.body_text, row.body_html).slice(0, 4000) }))) : rows.map((_row, index) => ({ index, score: 1 }));
  const results = ranked.slice(0, options.limit).map((rank) => {
    const row = rows[rank.index]!;
    return { id: String(row.id), account: row.account, address: row.address, folder: row.folder, uid: row.uid, subject: row.subject, from: { name: row.sender_name ?? "", address: row.sender_address ?? "" }, date: row.sent_at, score: rank.score, semanticDistance: Number(row.distance), preview: plainBody(row.body_text, row.body_html).replace(/\s+/g, " ").slice(0, 240) };
  });
  if (options.json) console.log(JSON.stringify({ query, results }));
  else if (results.length === 0) console.log("No matching mail.");
  else for (const [index, result] of results.entries()) console.log(`${index + 1}. [${result.id}] ${result.subject || "(no subject)"}\n   ${result.from.name || result.from.address} <${result.from.address}> · ${result.account}/${result.folder} · ${result.date ?? "unknown date"}\n   ${result.preview}`);
  await db.close();
}

async function topSenders(options: CliOptions): Promise<void> {
  const db = await database();
  await ensureSchema(db);
  const account = options.account ?? null;
  const folder = options.folder ?? null;
  const after = options.after ? calendarDate(options.after, "--after") : null;
  const before = options.before ? calendarDate(options.before, "--before") : null;
  if (after && before && before <= after) throw new Error("--before must be later than --after");
  const rows = await db`SELECT COALESCE(MAX(NULLIF(m.sender_name, '')), MIN(m.sender_address)) AS name, MIN(m.sender_address) AS address, COUNT(*) AS count
    FROM messages m JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id
    WHERE m.deleted_at IS NULL AND COALESCE(m.sender_address, '')<>''
      AND LOWER(m.sender_address)<>LOWER(a.address)
      AND LOWER(f.path) NOT IN ('sent', 'drafts')
      AND (${account} IS NULL OR a.alias=${account})
      AND (${folder} IS NULL OR f.path=${folder})
      AND (${after} IS NULL OR COALESCE(m.sent_at,m.internal_date)>=${after})
      AND (${before} IS NULL OR COALESCE(m.sent_at,m.internal_date)<${before})
    GROUP BY LOWER(m.sender_address)
    ORDER BY count DESC, address ASC LIMIT ${options.limit}` as Array<{ name: string; address: string; count: number }>;
  const senders = rows.map((row) => ({ name: row.name ?? "", address: row.address, count: Number(row.count) }));
  if (options.json) console.log(JSON.stringify({ senders }));
  else if (senders.length === 0) console.log("No incoming senders found.");
  else for (const [index, sender] of senders.entries()) console.log(`${index + 1}. ${sender.name || sender.address}${sender.name ? ` <${sender.address}>` : ""} — ${sender.count}`);
  await db.close();
}

async function showMessage(id: string, json: boolean): Promise<void> {
  const db = await database();
  const rows = await db`SELECT m.id, a.alias AS account, a.address, f.path AS folder, m.uid, m.subject, m.sender_name, m.sender_address, m.recipients, m.cc, m.sent_at, m.internal_date, m.flags, m.body_text, m.body_html, m.attachments, m.deleted_at FROM messages m JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id WHERE m.id=${id}` as Array<Record<string, unknown>>;
  if (!rows[0]) throw new Error(`message ${id} not found`);
  const row = rows[0];
  const bodyText = plainBody(row.body_text, row.body_html);
  console.log(json ? JSON.stringify({ ...row, body_text: bodyText }) : [`[${row.id}] ${row.subject}`, `From: ${row.sender_name ?? ""} <${row.sender_address ?? ""}>`, `Account: ${row.account} (${row.address})`, `Folder: ${row.folder} · UID ${row.uid}`, `Date: ${row.sent_at ?? row.internal_date ?? "unknown"}`, "", bodyText].join("\n"));
  await db.close();
}

function calendarDate(value: string, option: string): string {
  if (!/^\d{4}-\d{2}-\d{2}$/.test(value)) throw new Error(`${option} requires YYYY-MM-DD`);
  const date = new Date(`${value}T00:00:00Z`);
  if (Number.isNaN(date.getTime()) || date.toISOString().slice(0, 10) !== value) throw new Error(`${option} has an invalid calendar date: ${value}`);
  return `${value} 00:00:00`;
}

function sinceDate(value: string, now = new Date()): string {
  const match = value.match(/^(\d+)([hdw])$/i);
  const amount = Number(match?.[1]);
  const unit = match?.[2]?.toLowerCase() as "h" | "d" | "w" | undefined;
  const maximum = unit ? { h: 87_600, d: 3_650, w: 520 }[unit] : 0;
  if (!unit || !Number.isSafeInteger(amount) || amount < 1 || amount > maximum) throw new Error(`invalid --since value "${value}"; use a bounded positive duration such as 24h or 7d`);
  const unitMs = { h: 3_600_000, d: 86_400_000, w: 604_800_000 }[unit];
  const result = mysqlDate(new Date(now.getTime() - amount * unitMs));
  if (!result) throw new Error(`invalid --since value "${value}"`);
  return result;
}

function categoryMatches(filter: string | undefined, category: MailCategory): boolean {
  if (!filter || filter === "all") return true;
  if (filter === "personal") return category === "direct-personal" || category === "personal-administrative";
  if (filter === "actionable") return category === "direct-personal" || category === "personal-administrative" || category === "security";
  return filter === category;
}

function classifyMail(row: RecentRow): { category: MailCategory; confidence: number; reason: string } {
  const subject = row.subject.toLowerCase();
  const body = plainBody(row.body_text, row.body_html).toLowerCase();
  const address = (row.sender_address ?? "").toLowerCase();
  const local = address.split("@")[0] ?? "";
  const domain = address.split("@")[1] ?? "";
  const brandDomains: Array<[RegExp, string[]]> = [
    [/\bnetflix\b/i, ["netflix.com"]],
    [/\btrezor\b/i, ["trezor.io"]],
    [/\bledger\b/i, ["ledger.com"]],
    [/\bchase\b/i, ["chase.com"]]
  ];
  for (const [brand, domains] of brandDomains) {
    if (brand.test(`${row.sender_name ?? ""} ${row.subject}`) && !domains.some((allowed) => domain === allowed || domain.endsWith(`.${allowed}`))) {
      return { category: "likely-phishing", confidence: 0.8, reason: "brand name does not match a known official sender domain; verify independently" };
    }
  }
  if (/\b(new sign-in|unrecognized device|security alert|secure link|login details|password reset|two-factor|2fa)\b/i.test(subject)) {
    return { category: "security", confidence: 0.95, reason: "subject reports an account-security event" };
  }
  if (/\b(skuld|vanskil|debt|collection|nýtt skjal|new document|boð um vistun|samþykkt af forráðamanni|ný skráning|application|additional information|appointment|booking violation|áminning)\b/i.test(`${subject} ${body.slice(0, 800)}`)) {
    return { category: "personal-administrative", confidence: 0.9, reason: "message concerns personal administration, family, finance, or an appointment" };
  }
  if (/\b(unsubscribe|afþakka|newsletter|discount|offer|sale|tilboð|reward|livestream|webinar)\b/i.test(`${subject} ${body.slice(-1200)}`)) {
    return { category: "marketing", confidence: 0.9, reason: "message contains promotional or mailing-list language" };
  }
  const senderName = (row.sender_name ?? "").trim();
  const humanName = /^[\p{L}.'’-]+(?:\s+[\p{L}.'’-]+)+$/u.test(senderName) && !/\b(?:team|support|customer|service|studio|official)\b/i.test(senderName);
  const personalGreeting = /^(?:sæll|sæl|hæ|hi|hello|dear)(?=\s|[,!])/i.test(body.trim());
  if (address && humanName && personalGreeting && !/^(?:no-?reply|do-?not-?reply|notifications?|marketing|news|newsletter|mailer-daemon)$/i.test(local)) {
    return { category: "direct-personal", confidence: 0.65, reason: "human-named reply address and personal greeting; sender identity remains unverified" };
  }
  return { category: "other", confidence: 0.6, reason: "message does not match a higher-confidence category" };
}

function safePreview(row: RecentRow, limit = 480): string {
  return plainBody(row.body_text, row.body_html)
    .replace(/\b\d{6}[- ]?\d{4}\b/g, "[redacted identifier]")
    .replace(/\b(?:\d[ -]*?){13,19}\b/g, "[redacted payment number]")
    .replace(/\s+/g, " ")
    .trim()
    .slice(0, limit);
}

async function recentRows(options: CliOptions, scanLimit: number): Promise<{ rows: RecentRow[]; total: number; after: string; before: string | null }> {
  const db = await database();
  await ensureSchema(db);
  const account = options.account ?? null;
  const folder = options.folder ?? null;
  const sender = options.from ? `%${options.from}%` : null;
  const after = options.since ? sinceDate(options.since) : options.after ? calendarDate(options.after, "--after") : sinceDate("7d");
  const before = options.before ? calendarDate(options.before, "--before") : null;
  if (before && before <= after) throw new Error("--before must be later than the lower date bound");
  const countRows = await db`SELECT COUNT(*) AS count FROM messages m JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id
    WHERE m.deleted_at IS NULL
      AND (${account} IS NULL OR a.alias=${account})
      AND (${folder} IS NULL OR f.path=${folder})
      AND (${sender} IS NULL OR m.sender_address LIKE ${sender} OR m.sender_name LIKE ${sender})
      AND COALESCE(m.sent_at,m.internal_date)>=${after}
      AND (${before} IS NULL OR COALESCE(m.sent_at,m.internal_date)<${before})` as Array<{ count: string | number | bigint }>;
  const rows = await db`SELECT m.id, a.alias AS account, a.address, f.path AS folder, m.uid, m.subject, m.sender_name, m.sender_address, m.sent_at, m.internal_date AS date, m.body_text, m.body_html
    FROM messages m JOIN accounts a ON a.id=m.account_id JOIN folders f ON f.id=m.folder_id
    WHERE m.deleted_at IS NULL
      AND (${account} IS NULL OR a.alias=${account})
      AND (${folder} IS NULL OR f.path=${folder})
      AND (${sender} IS NULL OR m.sender_address LIKE ${sender} OR m.sender_name LIKE ${sender})
      AND COALESCE(m.sent_at,m.internal_date)>=${after}
      AND (${before} IS NULL OR COALESCE(m.sent_at,m.internal_date)<${before})
    ORDER BY COALESCE(m.sent_at,m.internal_date) DESC LIMIT ${scanLimit}` as RecentRow[];
  await db.close();
  return { rows, total: Number(countRows[0]?.count ?? 0), after, before };
}

function digestMessage(row: RecentRow): Record<string, unknown> {
  return {
    id: String(row.id),
    account: row.account,
    folder: row.folder,
    subject: row.subject,
    from: { name: row.sender_name ?? "", address: row.sender_address ?? "" },
    date: row.sent_at ?? row.date,
    ...classifyMail(row),
    preview: safePreview(row)
  };
}

async function recentMail(options: CliOptions): Promise<void> {
  const limit = options.limitProvided ? options.limit : 50;
  const result = await recentRows(options, options.category ? 1_000 : Math.min(Math.max(limit * 5, 100), 1_000));
  if (options.category && result.total > result.rows.length) throw new Error("category window exceeds the 1000-message classification bound; narrow --since or use --after/--before");
  const messages = result.rows.map(digestMessage).filter((message) => categoryMatches(options.category, message.category as MailCategory)).slice(0, limit);
  const output = { window: { after: result.after, before: result.before }, scanned: result.total, classified: result.rows.length, truncated: result.total > result.rows.length, messages };
  if (options.json) console.log(JSON.stringify(output));
  else if (messages.length === 0) console.log("No matching recent mail.");
  else for (const message of messages) {
    const from = message.from as { name: string; address: string };
    console.log(`[${message.id}] ${message.subject}\n  ${from.name || from.address} · ${message.date} · ${message.category}\n  ${message.preview}`);
  }
}

async function digestMail(options: CliOptions): Promise<void> {
  const fullSyncReport = options.sync ? await sync(options, false, false) : null;
  if (fullSyncReport && fullSyncReport.status !== "succeeded") throw new Error(`digest sync did not complete: ${fullSyncReport.reason ?? fullSyncReport.status}`);
  const syncReport = fullSyncReport ? (({ newMessages: _messages, ...summary }) => summary)(fullSyncReport) : null;
  const limit = options.limitProvided ? options.limit : 50;
  const result = await recentRows(options, 1_000);
  if (options.category && result.total > result.rows.length) throw new Error("category window exceeds the 1000-message classification bound; narrow --since or use --after/--before");
  const classified = result.rows.map(digestMessage);
  const categories: MailCategory[] = ["direct-personal", "personal-administrative", "security", "likely-phishing", "marketing", "other"];
  const classifiedCounts = Object.fromEntries(categories.map((category) => [category, classified.filter((message) => message.category === category).length]));
  const messages = classified.filter((message) => categoryMatches(options.category, message.category as MailCategory)).slice(0, limit);
  const output = { window: { after: result.after, before: result.before }, sync: syncReport, scanned: result.total, classified: result.rows.length, truncated: result.total > result.rows.length, classifiedCounts, messages };
  if (options.json) console.log(JSON.stringify(output));
  else {
    console.log(`Mail digest: ${result.total} message(s) since ${result.after}${result.before ? ` before ${result.before}` : ""}.`);
    console.log(`Classified categories: ${Object.entries(classifiedCounts).map(([category, count]) => `${category} ${count}`).join(", ")}.`);
    if (messages.length === 0) console.log("No messages match the requested category.");
    else for (const message of messages) {
      const from = message.from as { name: string; address: string };
      console.log(`\n[${message.id}] ${message.subject}\n  ${from.name || from.address} · ${message.date} · ${message.category} (${message.confidence})\n  ${message.preview}`);
    }
  }
}

async function status(json: boolean): Promise<void> {
  const db = await database();
  await ensureSchema(db);
  const accounts = await db`SELECT a.alias, a.address, a.last_sync_at, a.last_error, COUNT(DISTINCT f.id) AS folders, COUNT(m.id) AS messages, COALESCE(SUM(m.id IS NOT NULL AND m.deleted_at IS NULL), 0) AS active_messages FROM accounts a LEFT JOIN folders f ON f.account_id=a.id LEFT JOIN messages m ON m.folder_id=f.id GROUP BY a.id ORDER BY a.alias` as Array<Record<string, unknown>>;
  const pending = await db`SELECT COUNT(*) AS count FROM messages m LEFT JOIN embeddings e ON e.message_id=m.id WHERE m.deleted_at IS NULL AND (e.message_id IS NULL OR e.content_sha<>m.content_sha OR e.model<>${EMBED_MODEL})` as Array<{ count: number }>;
  const runs = await db`SELECT id, started_at, finished_at, status, messages_added, messages_deleted, error_text FROM sync_runs ORDER BY id DESC LIMIT 5` as Array<Record<string, unknown>>;
  const normalizedAccounts: Array<Record<string, unknown>> = accounts.map((account) => ({ ...account, folders: Number(account.folders ?? 0), messages: Number(account.messages ?? 0), active_messages: Number(account.active_messages ?? 0) }));
  const pendingEmbeddings = Number(pending[0]?.count ?? 0);
  const result = { accounts: normalizedAccounts, pendingEmbeddings, embeddingIndexCurrent: pendingEmbeddings === 0, recentRuns: runs };
  if (json) console.log(JSON.stringify(result));
  else {
    for (const account of normalizedAccounts) console.log(`${account.alias} (${account.address}): ${account.active_messages} active mail in ${account.folders} folders; last sync ${account.last_sync_at ?? "never"}${account.last_error ? `; ERROR ${account.last_error}` : ""}`);
    console.log(`Pending embeddings: ${result.pendingEmbeddings}`);
  }
  await db.close();
}

async function doctor(): Promise<void> {
  const db = await database();
  await ensureSchema(db);
  await db`SELECT 1`;
  const vector = await embeddingBatch(["mail index health check"]);
  await rerank("important mail", ["A security alert needs attention.", "A weekly newsletter."]);
  for (const account of await loadAccounts()) {
    if (account.maildirPath) {
      const info = await stat(account.maildirPath).catch(() => null);
      if (!info?.isDirectory()) throw new Error(`${account.alias}: maildir ${account.maildirPath} not readable`);
      console.log(`${account.alias}: Maildir ready (${account.maildirPath})`);
      continue;
    }
    const password = (await readFile(account.passwordFile, "utf8")).trim();
    const client = await connectImap(account, password);
    await client.logout();
    console.log(`${account.alias}: IMAP ready`);
  }
  console.log(`MariaDB ${DB_HOST}:${DB_PORT} ready; embeddings ${EMBED_MODEL} ${vector[0]!.length}d; reranker ${RERANK_URL ? "ready" : "disabled (vector+lexical ranking)"}`);
  await db.close();
}

function parseOptions(args: string[]): { positional: string[]; options: CliOptions } {
  const positional: string[] = [];
  const options: CliOptions = { limit: 10, limitProvided: false, used: new Set(), json: false, sync: false, skipEmbeddings: false, help: false };
  const value = (index: number, option: string): string => {
    const result = args[index];
    if (!result || result.startsWith("--")) throw new Error(`${option} requires a value`);
    return result;
  };
  for (let index = 0; index < args.length; index++) {
    const arg = args[index]!;
    if (arg.startsWith("-")) options.used.add(arg === "-h" ? "--help" : arg);
    if (arg === "--json") options.json = true;
    else if (arg === "--sync") options.sync = true;
    else if (arg === "--no-embeddings") options.skipEmbeddings = true;
    else if (arg === "--help" || arg === "-h") options.help = true;
    else if (arg === "--account") options.account = value(++index, arg);
    else if (arg === "--folder") options.folder = value(++index, arg);
    else if (arg === "--from") options.from = value(++index, arg);
    else if (arg === "--after") options.after = value(++index, arg);
    else if (arg === "--before") options.before = value(++index, arg);
    else if (arg === "--since") options.since = value(++index, arg);
    else if (arg === "--category") options.category = value(++index, arg);
    else if (arg === "--limit") {
      const limit = Number(value(++index, arg));
      if (!Number.isSafeInteger(limit) || limit < 1) throw new Error("--limit requires a positive integer");
      options.limit = Math.min(200, limit);
      options.limitProvided = true;
    } else if (arg.startsWith("-")) throw new Error(`unknown option: ${arg}`);
    else positional.push(arg);
  }
  if (options.after && options.since) throw new Error("use either --after or --since, not both");
  if (options.category && !["all", "personal", "actionable", "direct-personal", "personal-administrative", "security", "likely-phishing", "marketing", "other"].includes(options.category)) throw new Error(`unknown category: ${options.category}`);
  return { positional, options };
}

function validateCommandOptions(command: string | undefined, positional: string[], options: CliOptions): void {
  const common = ["--json", "--help"];
  const allowed: Record<string, string[]> = {
    sync: [...common, "--account", "--no-embeddings"],
    search: [...common, "--account", "--folder", "--from", "--after", "--before", "--limit"],
    recent: [...common, "--account", "--folder", "--from", "--since", "--after", "--before", "--category", "--limit"],
    digest: [...common, "--account", "--folder", "--from", "--since", "--after", "--before", "--category", "--limit", "--sync"],
    "top-senders": [...common, "--account", "--folder", "--after", "--before", "--limit"],
    show: common,
    status: common,
    accounts: [...common, "--account"],
    doctor: ["--help"],
    "self-test": [],
    help: []
  };
  if (!command) return;
  if (!allowed[command]) throw new Error(`unknown command: ${command}`);
  for (const option of options.used) if (!allowed[command]!.includes(option)) throw new Error(`${option} is not valid for ${command}`);
  if (command === "show" && positional.length !== 1) throw new Error("show requires exactly one MESSAGE_ID");
  if (!["search", "show"].includes(command) && positional.length > 0) throw new Error(`${command} does not accept positional arguments`);
}

function usage(exitCode = 2): never {
  const output = `AkurAI Mail semantic index

Indexes every configured AkurAI mailbox (local Maildir or IMAP) into MariaDB and
retrieves hybrid lexical+vector candidates using OpenRouter ${EMBED_MODEL} embeddings.
Reranking is optional (AKURAI_MAIL_RERANK_URL); without it results keep vector distance order.

Usage:
  akurai-mail-index sync [--account ALIAS] [--no-embeddings] [--json]
  akurai-mail-index search <natural language query> [--account ALIAS] [--folder NAME] [--from TEXT] [--after YYYY-MM-DD] [--before YYYY-MM-DD] [--limit N] [--json]
  akurai-mail-index recent [--account ALIAS] [--folder NAME] [--since 7d | --after YYYY-MM-DD] [--before YYYY-MM-DD] [--category CATEGORY] [--limit N] [--json]
  akurai-mail-index digest [--account ALIAS] [--folder NAME] [--since 7d | --after YYYY-MM-DD] [--before YYYY-MM-DD] [--category CATEGORY] [--limit N] [--sync] [--json]
  akurai-mail-index top-senders [--account ALIAS] [--folder NAME] [--after YYYY-MM-DD] [--before YYYY-MM-DD] [--limit N] [--json]
  akurai-mail-index show <MESSAGE_ID> [--json]
  akurai-mail-index status [--json]
  akurai-mail-index accounts [--json]
  akurai-mail-index doctor
  akurai-mail-index help

Search spans all accounts unless --account is supplied. Natural-language
"from …" and "frá …" clauses become sender filters; explicit --from wins.
Date filters are half-open: --after includes its date; --before excludes its date.
Recent and digest default to --since 7d. Categories: personal, actionable,
direct-personal, personal-administrative, security, likely-phishing, marketing,
other, or all. Digest sync skips embeddings. Standalone sync updates embeddings unless
--no-embeddings is explicit. JSON reports embeddingsSkipped so callers cannot confuse
zero updates with completed embedding work. Digest previews redact common 10-digit identity and 13–19 digit payment-number patterns.

Workbench protocol:
  Search result:  workbench:mail-search:<JSON from search --json>
  Sender ranking: workbench:mail-senders:<JSON from top-senders --json>
  New mail:       the AkurAI-Mail delivery hook calls pi-bun /api/ingest-mail,
                  which runs "sync --json" and posts workbench:mail-received cards.

Operations and architecture: skill://akurai-mail-semantic
Raw IMAP/SMTP:                 skill://akurai-mail`;
  console[exitCode === 0 ? "log" : "error"](output);
  process.exit(exitCode);
}

function selfTest(): void {
  const check = (condition: unknown, message: string): void => {
    if (!condition) throw new Error(`self-test failed: ${message}`);
  };
  const row = (subject: string, senderAddress: string, bodyText = "", senderName = "Sender"): RecentRow => ({ id: 1, account: "test", address: "test@example.com", folder: "INBOX", uid: 1, subject, sender_name: senderName, sender_address: senderAddress, sent_at: "2026-08-26 12:00:00", date: "2026-08-26 12:00:00", body_text: bodyText, body_html: "" });
  const fails = (operation: () => void): boolean => {
    try { operation(); return false; } catch { return true; }
  };
  check(sinceDate("7d", new Date("2026-08-26T12:00:00Z")) === "2026-08-19 12:00:00.000", "relative date");
  check(classifyMail(row("Netflix security alert", "notice@example.net")).category === "likely-phishing", "brand-domain phishing detection");
  check(classifyMail(row("New sign-in to your account", "noreply@example.com")).category === "security", "security classification");
  check(classifyMail(row("Nýtt skjal", "noreply@example.is")).category === "personal-administrative", "administrative classification");
  check(classifyMail(row("Weekly newsletter", "news@example.com", "unsubscribe")).category === "marketing", "marketing classification");
  check(classifyMail(row("Tee time", "jon@example.is", "Sæll Óli, þetta er skilaboð.", "Jón Heiðar")).category === "direct-personal", "direct personal classification");
  check(classifyMail(row("Hello", "attacker@example.net", "Open this link")).category === "other", "unknown reply-capable sender remains neutral");
  check(categoryMatches("personal", "direct-personal") && categoryMatches("personal", "personal-administrative") && !categoryMatches("personal", "marketing"), "personal category alias");
  check(fails(() => calendarDate("2026-02-30", "--after")), "invalid calendar date rejection");
  check(fails(() => sinceDate("999999999999d")), "oversized relative date rejection");
  const unsupported = parseOptions(["recent", "--sync"]);
  check(fails(() => validateCommandOptions(unsupported.positional.shift(), unsupported.positional, unsupported.options)), "unsupported option rejection");
  const noEmbeddings = parseOptions(["sync", "--no-embeddings"]);
  check(noEmbeddings.options.skipEmbeddings, "embedding-free sync parsing");
  validateCommandOptions(noEmbeddings.positional.shift(), noEmbeddings.positional, noEmbeddings.options);
  check(!safePreview(row("Family", "person@example.com", "Kennitala 2307203040 card 4111 1111 1111 1111")).includes("2307203040"), "identifier redaction");
  check(parseOptions(["search", "--help"]).options.help, "subcommand help parsing");
  console.log("akurai-mail-index self-test passed");
}

async function main(): Promise<void> {
  const { positional, options } = parseOptions(process.argv.slice(2));
  const command = positional.shift();
  validateCommandOptions(command, positional, options);
  if (options.help) usage(0);
  // Maildir reads need root (/var/vmail is vmail:mail 0770). Re-exec under sudo; the
  // env file inside the script restores configuration after sudo's env reset.
  if ((command === "sync" || command === "doctor" || (command === "digest" && options.sync)) && process.env.AKURAI_MAIL_SOURCE === "maildir" && typeof process.getuid === "function" && process.getuid() !== 0) {
    const rerun = Bun.spawnSync(["sudo", "-n", "bun", "run", import.meta.path, ...process.argv.slice(2)], { stdin: "inherit", stdout: "inherit", stderr: "inherit" });
    process.exit(rerun.exitCode ?? 1);
  }
  if (command === "sync") await sync(options, true, !options.skipEmbeddings);
  else if (command === "search") {
    const query = positional.join(" ").trim();
    if (!query) usage();
    await searchMail(query, options);
  } else if (command === "recent") await recentMail(options);
  else if (command === "digest") await digestMail(options);
  else if (command === "top-senders") await topSenders(options);
  else if (command === "show") {
    if (!positional[0]) usage();
    await showMessage(positional[0], options.json);
  } else if (command === "status") await status(options.json);
  else if (command === "accounts") {
    const accounts = await loadAccounts(options.account);
    console.log(options.json ? JSON.stringify(accounts.map(({ passwordFile: _secret, ...account }) => account)) : accounts.map((account) => account.maildirPath ? `${account.alias}\t${account.address}\t${account.maildirPath}` : `${account.alias}\t${account.address}\t${account.host}:${account.port}`).join("\n"));
  } else if (command === "doctor") await doctor();
  else if (command === "self-test") selfTest();
  else if (command === "help" || command === "--help" || command === "-h") usage(0);
  else usage();
}

try {
  await main();
} catch (error) {
  console.error(`akurai-mail-index: ${error instanceof Error ? error.message : String(error)}`);
  process.exit(1);
}