omni-dev 0.46.0

//! CLI command for `omni-dev gmail sync`.
//!
//! Maintains a durable, incrementally-updated local archive of a mailbox as
//! `.eml` files + a JSONL manifest (#1467, Phase 2 of the Gmail
//! integration). See `docs/gmail.md`'s Sync section and
//! [ADR-0064](../../../../docs/adrs/adr-0064.md) for the archive format and
//! its watermark/reconciliation contract; the actual logic lives in
//! `engine.rs` — this file is CLI glue only.

pub(crate) mod engine;
pub(crate) mod manifest;
pub(crate) mod progress;
pub(crate) mod report;
pub(crate) mod shard;
pub(crate) mod state;

use std::io::{IsTerminal, Write};
use std::path::PathBuf;
use std::sync::Arc;

use anyhow::{Context, Result};
use chrono::SecondsFormat;
use clap::Parser;
use serde::Serialize;
use tokio::sync::mpsc;

use crate::cli::gmail::format::{output_as, write_scalar_jsonl, JsonlSerialize, OutputFormat};
use crate::gmail::client::GmailClient;

use engine::SyncOptions;
use progress::{SyncProgressBars, SyncProgressEvent};
use report::{DeferredFetch, SyncAction, SyncError, SyncReport, SyncSummary};

/// Default requested `--concurrency`, retained for configuration compatibility.
/// The engine clamps it to [`crate::gmail::messages_api::MAX_CONCURRENCY`]
/// (currently five) and separately paces requests with its token bucket.
///
/// `pub(crate)` — also `sync-all`'s fallback when neither its own
/// `--concurrency` flag nor `gmail-sync.yaml`'s `concurrency` is set
/// (ADR-0068), so the two commands share one default rather than risking
/// drift between two constants.
pub(crate) const DEFAULT_SYNC_CONCURRENCY: usize = 20;

/// Maintains a durable local archive of a Gmail mailbox (no MCP equivalent
/// — a bulk, potentially long-running filesystem operation is a poor fit
/// for a synchronous MCP tool call).
#[derive(Parser)]
pub struct SyncCommand {
    /// Directory to maintain the archive in. Created if missing.
    #[arg(long, value_name = "PATH")]
    pub output_dir: PathBuf,

    /// Restrict the archive to messages matching this Gmail search query
    /// (same syntax as `gmail search --query`). Only applied on
    /// backfill/`--full`/reconciliation passes — an incremental sync's
    /// `history.list` has no query filter, so newly-arrived mail matching
    /// the query is only picked up by a later `--full` re-run (see
    /// docs/gmail.md's Sync section).
    #[arg(long)]
    pub query: Option<String>,

    /// Excludes messages carrying this label id from the archive (#1780).
    /// Repeatable — e.g. `--exclude-label SPAM --exclude-label TRASH`.
    /// Applied fully and generally on incremental passes (no extra API
    /// calls; a `messagesAdded` history event matching this filter is never
    /// fetched, and a `labelsAdded`/`labelsRemoved` event that crosses the
    /// excluded boundary soft-deletes/undeletes the archived record). On
    /// backfill/`--full`/reconciliation passes, only entries with a known
    /// query translation take effect (currently `SPAM`/`TRASH` — folded
    /// into the listing query as `-in:spam`/`-in:trash`); anything else
    /// still filters future incremental runs, but won't be excluded
    /// retroactively on this pass unless you also pass `--query`.
    #[arg(long, value_name = "LABEL_ID")]
    pub exclude_label: Vec<String>,

    /// Forces a full backfill/reconciliation pass even if a valid watermark
    /// exists.
    #[arg(long)]
    pub full: bool,

    /// Retries every message still pending from an earlier failed fetch now,
    /// ignoring the retry backoff that would otherwise defer it (#1790). Use
    /// after fixing whatever made the fetches fail. A retry that fails again
    /// still counts toward — and restarts — that message's backoff.
    #[arg(long)]
    pub retry_pending: bool,

    /// Bounds concurrent message fetches. Clamped to
    /// `1..=gmail::messages_api::MAX_CONCURRENCY`.
    #[arg(long, default_value_t = DEFAULT_SYNC_CONCURRENCY)]
    pub concurrency: usize,

    /// Reports the planned actions without writing any files.
    #[arg(long)]
    pub dry_run: bool,

    /// Also writes each message's attachment MIME parts to disk as
    /// separate files under `<eml-shard-dir>/<id>/attachments/<filename>`,
    /// alongside the existing `.eml`. Off by default: extraction is
    /// additional I/O/disk usage per message, and the `.eml` stays the
    /// lossless source of truth regardless. Only applies to messages
    /// actually fetched this run — presence-on-disk still skips an
    /// already-archived message, so turning this on does not
    /// retroactively backfill an existing archive (delete the affected
    /// `.eml` files, or the whole archive, and re-run `--full` to force
    /// re-extraction).
    #[arg(long)]
    pub extract_attachments: bool,

    /// Only shows errors/warnings, suppresses info-level output — including
    /// the live progress bars a backfill/`--full`/reconciliation pass shows
    /// on an interactive terminal (#1502).
    #[arg(long)]
    pub quiet: bool,

    /// Report format.
    #[arg(short = 'o', long, value_enum, default_value_t = OutputFormat::Table)]
    pub output: OutputFormat,
}

impl SyncCommand {
    /// Runs the command against the shared client resolved by the parent
    /// `GmailCommand::execute`.
    pub async fn execute(self, client: &GmailClient) -> Result<()> {
        run_sync_command(
            client,
            SyncOptions {
                output_dir: self.output_dir,
                query: self.query,
                exclude_labels: self.exclude_label,
                full: self.full,
                retry_pending: self.retry_pending,
                concurrency: self.concurrency,
                dry_run: self.dry_run,
                extract_attachments: self.extract_attachments,
                shared_pool: None,
            },
            self.quiet,
            &self.output,
        )
        .await
    }
}

/// Runs the sync and renders its report.
///
/// Split from [`SyncCommand::execute`] so tests can inject a wiremock
/// client without going through the credential-loading path. Mirrors
/// `src/cli/ai/claude/history/sync.rs`'s `execute`: compute the report,
/// render it, and only *then* decide the process exit condition — a
/// non-empty `errors` becomes a failing exit code after everything already
/// ran and printed, never silently.
async fn run_sync_command(
    client: &GmailClient,
    opts: SyncOptions,
    quiet: bool,
    output: &OutputFormat,
) -> Result<()> {
    let show_progress = should_show_progress(quiet, output, std::io::stderr().is_terminal());
    let report = if show_progress {
        let (tx, rx) = mpsc::unbounded_channel();
        let bars = SyncProgressBars::new();
        let render_task = tokio::spawn(bars.drain(rx));
        // Route rate-limit retry notices onto the fetch bar's own message
        // (#1651) instead of `retry_if`'s default `eprintln!`, which would
        // tear this live render.
        let notify_tx = tx.clone();
        client.set_retry_notify(Arc::new(move |status, delay_secs, attempt| {
            let _ = notify_tx.send(SyncProgressEvent::RateLimited {
                status,
                delay_secs,
                attempt,
            });
        }));
        let result = engine::run_sync_with_progress(client, &opts, Some(&tx)).await;
        // Dropping `tx` (rather than waiting for `run_sync_with_progress`'s
        // return to go out of scope) is what lets the renderer's
        // `rx.recv()` loop actually end.
        drop(tx);
        // Best-effort: a panicking/cancelled render task shouldn't fail an
        // otherwise-successful sync — it only ever formats bars.
        let _ = render_task.await;
        result?
    } else {
        engine::run_sync(client, &opts).await?
    };

    let output_view = SyncReportOutput {
        actions: &report.actions,
        errors: &report.errors,
        deferred: &report.deferred,
        summary: report.summary(),
    };
    if !output_as(&output_view, output)? {
        let stdout = std::io::stdout();
        let mut handle = stdout.lock();
        // Bars already rendered every fetch/delete live, and `--quiet`
        // asked for exactly this suppression explicitly — see
        // `render_report_text`'s doc comment for why `quiet` is checked
        // again here rather than just reusing `show_progress` (a non-tty
        // stderr makes both false, but that case still wants full detail).
        render_report_text(&report, &mut handle, !quiet && !show_progress)?;
    }

    if !report.errors.is_empty() {
        anyhow::bail!(
            "{} message(s) failed to sync; see errors above",
            report.errors.len()
        );
    }
    Ok(())
}

/// Live progress bars are only worth constructing when a human is actually
/// watching an interactive terminal render `-o table` text: `--quiet`
/// suppresses them explicitly, a machine `-o` format would otherwise mix
/// indicatif's escape sequences into output a script parses, and a
/// non-terminal stderr (redirected to a file, CI) is exactly what
/// `indicatif` itself would no-op on anyway — checking it here means that
/// no-op is a decision this function makes rather than one left implicit.
///
/// Takes `stderr_is_terminal` as a value rather than calling
/// [`std::io::IsTerminal`] itself, so tests can exercise every combination
/// without depending on the test runner's own stderr (which is typically
/// captured, i.e. never a terminal) — see STYLE-0028.
///
/// `pub(crate)` — also `sync-all`'s gate for its own shared-`MultiProgress`
/// bars (ADR-0068's Decision 4 follow-up, #1504), so the two commands agree
/// on exactly when live progress rendering makes sense rather than risking
/// the conditions drifting apart.
pub(crate) fn should_show_progress(
    quiet: bool,
    output: &OutputFormat,
    stderr_is_terminal: bool,
) -> bool {
    !quiet && matches!(output, OutputFormat::Table) && stderr_is_terminal
}

/// `-o json`/`-o yaml`/`-o yamls`/`-o jsonl` view of a [`SyncReport`]: the
/// same `actions`/`errors` plus a computed `summary` field, so a machine
/// consumer gets the same at-a-glance total the text output gains (#1488).
/// Mirrors `SyncOutput` in `src/cli/ai/claude/history.rs`, which adds a
/// `dry_run` field the same way — a borrowing wrapper rather than a field on
/// `SyncReport` itself, since `summary` is only ever valid once a run has
/// finished pushing to `actions`/`errors`.
#[derive(Serialize)]
struct SyncReportOutput<'a> {
    actions: &'a [SyncAction],
    errors: &'a [SyncError],
    deferred: &'a [DeferredFetch],
    summary: SyncSummary,
}

impl JsonlSerialize for SyncReportOutput<'_> {
    fn write_jsonl(&self, out: &mut dyn Write) -> Result<()> {
        write_scalar_jsonl(self, out)
    }
}

/// Renders a report as one line per action, then one line per error.
///
/// `show_action_detail` gates the per-item lines (`Fetched`/`Deleted`/...,
/// but never `Note`, `Vanished`, `Error`, or the trailing summary — see the
/// call site):
/// when live progress bars already rendered every fetch/delete as it
/// happened, or `--quiet` asked for exactly this, repeating the whole batch
/// as text on top would just be a second, redundant dump — the "silent,
/// then dump everything at once" experience #1502 introduced bars to fix
/// in the first place. `false` is only for the cases neither of those
/// covers (a non-interactive `stderr`, e.g. a redirected/cron log): there,
/// this text *is* the only record of what happened, so it still gets the
/// full detail — see ADR-0064's authoritative-record framing.
fn render_report_text(
    report: &SyncReport,
    out: &mut dyn Write,
    show_action_detail: bool,
) -> Result<()> {
    if report.actions.is_empty() && report.errors.is_empty() && report.deferred.is_empty() {
        writeln!(out, "Nothing to do.").context("Failed to write sync report")?;
        return Ok(());
    }
    for action in &report.actions {
        if !show_action_detail
            && !matches!(
                action,
                SyncAction::Note { .. } | SyncAction::Vanished { .. }
            )
        {
            continue;
        }
        let line = match action {
            SyncAction::Fetched { id, path, bytes } => {
                format!("Fetched {id} -> {} ({bytes} bytes)", path.display())
            }
            SyncAction::WouldFetch { id } => format!("Would fetch {id}"),
            SyncAction::LabelsUpdated { id, added, removed } => {
                let mut parts = Vec::new();
                if !added.is_empty() {
                    parts.push(format!("+{}", added.join(",")));
                }
                if !removed.is_empty() {
                    parts.push(format!("-{}", removed.join(",")));
                }
                format!("Labels updated on {id}: {}", parts.join(" "))
            }
            SyncAction::Deleted { id } => format!("Deleted {id}"),
            SyncAction::Undeleted { id } => format!("Undeleted {id}"),
            SyncAction::WouldDelete { id } => format!("Would delete {id}"),
            SyncAction::WouldUndelete { id } => format!("Would undelete {id}"),
            SyncAction::Vanished { id } => {
                format!("Vanished {id} (message no longer existed on the server; skipped, not an error)")
            }
            SyncAction::Excluded { id } => {
                format!("Excluded {id} (matches an excluded label; not archived)")
            }
            SyncAction::Note { message } => format!("Note: {message}"),
        };
        writeln!(out, "{line}").context("Failed to write sync report")?;
    }
    for error in &report.errors {
        writeln!(out, "Error: {} failed: {}", error.id, error.reason)
            .context("Failed to write sync report")?;
    }
    for line in deferred_warning_lines(&report.deferred) {
        writeln!(out, "{line}").context("Failed to write sync report")?;
    }
    writeln!(out, "{}", format_summary_line(&report.summary()))
        .context("Failed to write sync report")?;
    Ok(())
}

/// Most deferred ids listed individually in a warning; the rest are counted.
/// A sustained rate limit can defer thousands, and `-o json` carries them all.
const DEFERRED_LINES_SHOWN: usize = 10;

/// The warning for ids skipped by retry backoff (#1790): a header line, up to
/// [`DEFERRED_LINES_SHOWN`] indented per-id lines, and an overflow count.
/// Empty when nothing was deferred. A warning, not an error — it never
/// contributes to the exit code.
///
/// `pub(crate)` so `sync-all` can print the same text under its own
/// `<account>:` prefix.
pub(crate) fn deferred_warning_lines(deferred: &[DeferredFetch]) -> Vec<String> {
    if deferred.is_empty() {
        return Vec::new();
    }
    let mut lines = vec![format!(
        "Warning: {} message(s) deferred by retry backoff after repeated failures; \
         pass --retry-pending to retry them now",
        deferred.len()
    )];
    for entry in deferred.iter().take(DEFERRED_LINES_SHOWN) {
        let next = entry.next_retry_at.map_or_else(
            || "unscheduled".to_string(),
            |at| at.to_rfc3339_opts(SecondsFormat::Secs, true),
        );
        let reason = entry
            .last_error
            .as_ref()
            .map_or_else(String::new, |reason| format!("; last error: {reason}"));
        lines.push(format!(
            "  {}: {} consecutive failure(s), next retry {next}{reason}",
            entry.id, entry.failures
        ));
    }
    if deferred.len() > DEFERRED_LINES_SHOWN {
        lines.push(format!(
            "  ... and {} more",
            deferred.len() - DEFERRED_LINES_SHOWN
        ));
    }
    lines
}

/// Formats `summary` as a trailing comma-separated line, e.g.
/// `"3 fetched, 1 deleted, 0 errors"`. Zero counts are omitted except
/// `errors`, which is always shown so a clean run is visible at a glance.
///
/// `pub(crate)` so `sync-all` (ADR-0068) can reuse the exact same
/// formatting for its per-account and combined-total lines.
pub(crate) fn format_summary_line(summary: &SyncSummary) -> String {
    let mut parts = Vec::new();
    let mut push = |count: usize, label: &str| {
        if count > 0 {
            parts.push(format!("{count} {label}"));
        }
    };
    push(summary.fetched, "fetched");
    push(summary.would_fetch, "would fetch");
    push(summary.vanished, "vanished");
    push(summary.excluded, "excluded");
    push(summary.labels_updated, "labels updated");
    push(summary.deleted, "deleted");
    push(summary.undeleted, "undeleted");
    push(summary.would_delete, "would delete");
    push(summary.would_undelete, "would undelete");
    push(summary.deferred, "deferred");
    parts.push(format!("{} errors", summary.errors));
    parts.join(", ")
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use crate::gmail::auth::{GmailCredentials, GmailScope};
    use crate::utils::secret::Secret;
    use base64::Engine as _;

    fn test_credentials() -> GmailCredentials {
        GmailCredentials {
            client_id: "client-1".to_string(),
            client_secret: Secret::new("secret-1"),
            refresh_token: Secret::new("refresh-1"),
            scope: GmailScope::ReadOnly,
        }
    }

    async fn client_with_bootstrapped_token(server: &wiremock::MockServer) -> GmailClient {
        wiremock::Mock::given(wiremock::matchers::method("POST"))
            .and(wiremock::matchers::path("/token"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "access_token": "test-token",
                    "expires_in": 3600,
                })),
            )
            .mount(server)
            .await;

        let mut client = GmailClient::new(&server.uri(), &test_credentials()).unwrap();
        crate::gmail::client::test_support::replace_session(
            &mut client,
            &test_credentials(),
            &format!("{}/token", server.uri()),
        );
        client
    }

    // ── should_show_progress (#1502) ─────────────────────────────────────

    #[test]
    fn should_show_progress_gate() {
        let cases = [
            (
                "quiet suppresses even on a tty",
                true,
                OutputFormat::Table,
                true,
                false,
            ),
            (
                "machine format suppresses even on a tty",
                false,
                OutputFormat::Json,
                true,
                false,
            ),
            (
                "non-tty stderr suppresses",
                false,
                OutputFormat::Table,
                false,
                false,
            ),
            (
                "table + not quiet + tty shows bars",
                false,
                OutputFormat::Table,
                true,
                true,
            ),
        ];
        for (case, quiet, output, stderr_is_terminal, expected) in cases {
            assert_eq!(
                should_show_progress(quiet, &output, stderr_is_terminal),
                expected,
                "case: {case}"
            );
        }
    }

    // ── render_report_text ─────────────────────────────────────────────

    #[test]
    fn render_report_text_reports_nothing_to_do_when_empty() {
        let report = SyncReport::default();
        let mut buf = Vec::new();
        render_report_text(&report, &mut buf, true).unwrap();
        assert_eq!(String::from_utf8(buf).unwrap(), "Nothing to do.\n");
    }

    #[test]
    fn render_report_text_writes_one_line_per_action_and_error() {
        let report = SyncReport {
            actions: vec![
                SyncAction::Fetched {
                    id: "m1".to_string(),
                    path: PathBuf::from("messages/m1/m1.eml"),
                    bytes: 100,
                },
                SyncAction::Deleted {
                    id: "m2".to_string(),
                },
            ],
            errors: vec![SyncError {
                id: "m3".to_string(),
                reason: "boom".to_string(),
            }],
            ..SyncReport::default()
        };
        let mut buf = Vec::new();
        render_report_text(&report, &mut buf, true).unwrap();
        let text = String::from_utf8(buf).unwrap();
        assert!(text.contains("Fetched m1"));
        assert!(text.contains("Deleted m2"));
        assert!(text.contains("Error: m3 failed: boom"));
        assert!(text.contains("1 fetched, 1 deleted, 1 errors"));
        assert_eq!(text.lines().count(), 4);
    }

    #[test]
    fn render_report_text_formats_label_updates() {
        let report = SyncReport {
            actions: vec![SyncAction::LabelsUpdated {
                id: "m1".to_string(),
                added: vec!["IMPORTANT".to_string()],
                removed: vec!["UNREAD".to_string()],
            }],
            errors: vec![],
            ..SyncReport::default()
        };
        let mut buf = Vec::new();
        render_report_text(&report, &mut buf, true).unwrap();
        let text = String::from_utf8(buf).unwrap();
        assert!(text.contains("+IMPORTANT"));
        assert!(text.contains("-UNREAD"));
    }

    #[test]
    fn render_report_text_formats_excluded_and_tallies_it_in_the_summary() {
        let report = SyncReport {
            actions: vec![SyncAction::Excluded {
                id: "spam1".to_string(),
            }],
            errors: vec![],
            ..SyncReport::default()
        };
        let mut buf = Vec::new();
        render_report_text(&report, &mut buf, true).unwrap();
        let text = String::from_utf8(buf).unwrap();
        assert!(text.contains("Excluded spam1"));
        assert!(text.contains("1 excluded, 0 errors"));
    }

    #[test]
    fn render_report_text_summary_line_omits_zero_counts_under_dry_run() {
        let report = SyncReport {
            actions: vec![
                SyncAction::WouldFetch {
                    id: "m1".to_string(),
                },
                SyncAction::WouldDelete {
                    id: "m2".to_string(),
                },
                SyncAction::WouldUndelete {
                    id: "m3".to_string(),
                },
            ],
            errors: vec![],
            ..SyncReport::default()
        };
        let mut buf = Vec::new();
        render_report_text(&report, &mut buf, true).unwrap();
        let text = String::from_utf8(buf).unwrap();
        assert!(text.contains("1 would fetch, 1 would delete, 1 would undelete, 0 errors"));
        assert!(!text.contains("fetched,"));
        assert!(!text.contains("deleted,"));
    }

    #[test]
    fn render_report_text_suppresses_per_item_actions_but_keeps_notes_errors_and_summary() {
        let report = SyncReport {
            actions: vec![
                SyncAction::Fetched {
                    id: "m1".to_string(),
                    path: PathBuf::from("messages/m1/m1.eml"),
                    bytes: 100,
                },
                SyncAction::Deleted {
                    id: "m2".to_string(),
                },
                SyncAction::Note {
                    message: "watermark expired (404 on startHistoryId); reconciling".to_string(),
                },
            ],
            errors: vec![SyncError {
                id: "m3".to_string(),
                reason: "boom".to_string(),
            }],
            ..SyncReport::default()
        };
        let mut buf = Vec::new();
        render_report_text(&report, &mut buf, false).unwrap();
        let text = String::from_utf8(buf).unwrap();
        assert!(
            !text.contains("Fetched m1"),
            "flood actions must be suppressed"
        );
        assert!(
            !text.contains("Deleted m2"),
            "flood actions must be suppressed"
        );
        assert!(
            text.contains("Note: watermark expired"),
            "a rare contextual Note must survive suppression"
        );
        assert!(
            text.contains("Error: m3 failed: boom"),
            "errors must survive suppression"
        );
        assert!(
            text.contains("1 fetched, 1 deleted, 1 errors"),
            "the summary must survive suppression"
        );
    }

    // ── SyncReportOutput (`-o json`/`-o yaml`/`-o yamls`/`-o jsonl`) ─────

    #[test]
    fn sync_report_output_yaml_includes_summary_field() {
        let report = SyncReport {
            actions: vec![SyncAction::Fetched {
                id: "m1".to_string(),
                path: PathBuf::from("m1.eml"),
                bytes: 1,
            }],
            errors: vec![],
            ..SyncReport::default()
        };
        let output_view = SyncReportOutput {
            actions: &report.actions,
            errors: &report.errors,
            deferred: &report.deferred,
            summary: report.summary(),
        };
        let yaml = serde_yaml::to_string(&output_view).unwrap();
        assert!(yaml.contains("summary:"));
        assert!(yaml.contains("fetched: 1"));
    }

    // ── run_sync_command / SyncCommand::execute glue ────────────────────

    #[tokio::test]
    async fn run_sync_command_dry_run_reports_would_fetch_and_touches_no_files() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/profile"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "emailAddress": "user@example.com", "messagesTotal": 1, "threadsTotal": 1, "historyId": "1"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "messages": [{"id": "m1", "threadId": "t1"}]
                })),
            )
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        run_sync_command(
            &client,
            SyncOptions {
                output_dir: output_dir.clone(),
                query: None,
                exclude_labels: Vec::new(),
                full: false,
                retry_pending: false,
                concurrency: 4,
                dry_run: true,
                extract_attachments: false,
                shared_pool: None,
            },
            false,
            &OutputFormat::Table,
        )
        .await
        .unwrap();

        assert!(!output_dir.exists());
    }

    fn deferred(id: &str, failures: u32) -> DeferredFetch {
        DeferredFetch {
            id: id.to_string(),
            failures,
            next_retry_at: Some(
                chrono::DateTime::parse_from_rfc3339("2026-01-01T00:40:00Z")
                    .unwrap()
                    .with_timezone(&chrono::Utc),
            ),
            last_error: Some("rate limited".to_string()),
        }
    }

    #[test]
    fn deferred_warning_lines_is_empty_when_nothing_was_deferred() {
        assert!(deferred_warning_lines(&[]).is_empty());
    }

    #[test]
    fn deferred_warning_lines_names_each_id_and_the_override() {
        let lines = deferred_warning_lines(&[deferred("m1", 4)]);
        assert_eq!(lines.len(), 2);
        assert!(lines[0].starts_with("Warning: 1 message(s) deferred"));
        assert!(lines[0].contains("--retry-pending"));
        assert_eq!(
            lines[1],
            "  m1: 4 consecutive failure(s), next retry 2026-01-01T00:40:00Z; \
             last error: rate limited"
        );
    }

    #[test]
    fn deferred_warning_lines_reports_an_unscheduled_entry() {
        let entry = DeferredFetch {
            next_retry_at: None,
            last_error: None,
            ..deferred("m1", 2)
        };
        let lines = deferred_warning_lines(&[entry]);
        assert!(lines[1].contains("next retry unscheduled"), "{}", lines[1]);
    }

    #[test]
    fn deferred_warning_lines_caps_the_per_id_lines_and_counts_the_rest() {
        let many: Vec<DeferredFetch> = (0..13).map(|n| deferred(&format!("m{n}"), 2)).collect();
        let lines = deferred_warning_lines(&many);
        // Header, ten ids, overflow line.
        assert_eq!(lines.len(), 12);
        assert!(lines[0].contains("13 message(s)"));
        assert_eq!(lines[11], "  ... and 3 more");
    }

    #[test]
    fn render_report_text_shows_deferred_warnings_even_without_action_detail() {
        let report = SyncReport {
            deferred: vec![deferred("m1", 3)],
            ..SyncReport::default()
        };
        let mut buf = Vec::new();
        render_report_text(&report, &mut buf, false).unwrap();
        let text = String::from_utf8(buf).unwrap();
        assert!(!text.contains("Nothing to do"));
        assert!(text.contains("Warning: 1 message(s) deferred"));
        assert!(text.contains("m1: 3 consecutive failure(s)"));
        assert!(text.contains("1 deferred, 0 errors"));
    }

    #[tokio::test]
    async fn run_sync_command_exits_zero_when_the_only_problem_is_a_deferred_id() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/profile"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "emailAddress": "user@example.com", "messagesTotal": 1, "threadsTotal": 1, "historyId": "1"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/history"))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"historyId": "2"})),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .respond_with(wiremock::ResponseTemplate::new(500))
            .expect(0)
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        std::fs::create_dir_all(&output_dir).unwrap();
        state::save(
            &state::ArchiveState {
                history_id: "1".to_string(),
                email_address: "user@example.com".to_string(),
                last_sync: chrono::Utc::now(),
                query: None,
                pending_fetch: vec![state::PendingFetch::failed(
                    None,
                    "m1",
                    "rate limited",
                    chrono::Utc::now(),
                )],
            },
            &output_dir.join("state.json"),
        )
        .unwrap();

        run_sync_command(
            &client,
            SyncOptions {
                output_dir,
                query: None,
                exclude_labels: Vec::new(),
                full: false,
                retry_pending: false,
                concurrency: 4,
                dry_run: false,
                extract_attachments: false,
                shared_pool: None,
            },
            true,
            &OutputFormat::Table,
        )
        .await
        .expect("a deferred id is a warning, not a failure");
    }

    #[tokio::test]
    async fn run_sync_command_surfaces_a_non_zero_exit_on_per_item_errors() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/profile"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "emailAddress": "user@example.com", "messagesTotal": 1, "threadsTotal": 1, "historyId": "1"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "messages": [{"id": "m1", "threadId": "t1"}]
                })),
            )
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .respond_with(wiremock::ResponseTemplate::new(500).set_body_string("boom"))
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let err = run_sync_command(
            &client,
            SyncOptions {
                output_dir: dir.path().join("archive"),
                query: None,
                exclude_labels: Vec::new(),
                full: false,
                retry_pending: false,
                concurrency: 4,
                dry_run: false,
                extract_attachments: false,
                shared_pool: None,
            },
            false,
            &OutputFormat::Table,
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("1 message(s) failed"));
    }

    #[tokio::test]
    async fn execute_passes_flags_through() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/profile"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "emailAddress": "user@example.com", "messagesTotal": 0, "threadsTotal": 0, "historyId": "1"
            })))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .respond_with(
                wiremock::ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"messages": []})),
            )
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let cmd = SyncCommand {
            output_dir: dir.path().join("archive"),
            query: None,
            exclude_label: Vec::new(),
            full: false,
            retry_pending: false,
            concurrency: DEFAULT_SYNC_CONCURRENCY,
            dry_run: false,
            extract_attachments: false,
            quiet: false,
            output: OutputFormat::Json,
        };
        cmd.execute(&client).await.unwrap();
    }

    // ── --extract-attachments ────────────────────────────────────────

    fn multipart_with_base64_attachment() -> String {
        let encoded_attachment = base64::engine::general_purpose::STANDARD.encode(b"PDF-CONTENT");
        format!(
            "Subject: Report\r\n\
From: a@example.com\r\n\
MIME-Version: 1.0\r\n\
Content-Type: multipart/mixed; boundary=\"BOUNDARY\"\r\n\
\r\n\
--BOUNDARY\r\n\
Content-Type: text/plain\r\n\
\r\n\
Hello\r\n\
--BOUNDARY\r\n\
Content-Type: application/pdf\r\n\
Content-Transfer-Encoding: base64\r\n\
Content-Disposition: attachment; filename=\"report.pdf\"\r\n\
\r\n\
{encoded_attachment}\r\n\
--BOUNDARY--\r\n"
        )
    }

    /// Mounts a single-message mailbox (`m1`, a multipart message with one
    /// base64-encoded `application/pdf` attachment) for the
    /// `--extract-attachments` tests below.
    async fn mount_single_message_with_attachment(server: &wiremock::MockServer) {
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/profile"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "emailAddress": "user@example.com", "messagesTotal": 1, "threadsTotal": 1, "historyId": "1"
            })))
            .mount(server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "messages": [{"id": "m1", "threadId": "t1"}]
                })),
            )
            .mount(server)
            .await;
        let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .encode(multipart_with_base64_attachment());
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
            .and(wiremock::matchers::query_param("format", "raw"))
            .respond_with(
                wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "m1",
                    "threadId": "t1",
                    "labelIds": ["INBOX"],
                    "internalDate": "1700000000000",
                    "historyId": "500",
                    "raw": encoded,
                })),
            )
            .mount(server)
            .await;
    }

    fn expected_attachment_path(output_dir: &std::path::Path) -> PathBuf {
        let date = chrono::DateTime::from_timestamp_millis(1_700_000_000_000).unwrap();
        shard::attachments_dir(output_dir, "m1", Some(date)).join("report.pdf")
    }

    #[tokio::test]
    async fn run_sync_command_extract_attachments_writes_attachment_files() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_single_message_with_attachment(&server).await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        run_sync_command(
            &client,
            SyncOptions {
                output_dir: output_dir.clone(),
                query: None,
                exclude_labels: Vec::new(),
                full: false,
                retry_pending: false,
                concurrency: 4,
                dry_run: false,
                extract_attachments: true,
                shared_pool: None,
            },
            false,
            &OutputFormat::Table,
        )
        .await
        .unwrap();

        let contents = std::fs::read(expected_attachment_path(&output_dir)).unwrap();
        assert_eq!(contents, b"PDF-CONTENT");
    }

    #[tokio::test]
    async fn run_sync_command_without_extract_attachments_writes_no_attachment_files() {
        let server = wiremock::MockServer::start().await;
        let client = client_with_bootstrapped_token(&server).await;
        mount_single_message_with_attachment(&server).await;

        let dir = tempfile::tempdir().unwrap();
        let output_dir = dir.path().join("archive");
        run_sync_command(
            &client,
            SyncOptions {
                output_dir: output_dir.clone(),
                query: None,
                exclude_labels: Vec::new(),
                full: false,
                retry_pending: false,
                concurrency: 4,
                dry_run: false,
                extract_attachments: false,
                shared_pool: None,
            },
            false,
            &OutputFormat::Table,
        )
        .await
        .unwrap();

        assert!(!expected_attachment_path(&output_dir).exists());
        assert!(!expected_attachment_path(&output_dir)
            .parent()
            .unwrap()
            .exists());
    }
}