//! Boot-time state observation for the analyze pipeline.
//!
//! Pre-refactor this module re-enqueued every case that looked pending
//! at startup. With the typed [`CaseAnalysisState`] machine the per-page
//! `try_enqueue_all_for_user` covers that responsibility lazily — boot
//! is now reduced to a telemetry pass that logs how many cases sit in
//! each observable state, without any worker channel sends.
//!
//! [`CaseAnalysisState`]: super::auto_trigger::CaseAnalysisState

use std::path::Path;
use std::time::{Duration, SystemTime};

use tracing::info;

use super::InFlight;
use super::auto_trigger::{CaseAnalysisState, evaluate_state};

/// Aggregate counts over every UUID-named, non-closed case under a user
/// root. `NoRecordings` and `Pending` are deliberately not represented:
/// they are transient pre-states that need no operator visibility.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct BootStateSummary {
    /// `analysis_input.json` present → a job is queued or in flight.
    pub queued: usize,
    /// Failure marker matches `latest_mtime` → manual retry needed.
    pub failed: usize,
    /// Idle threshold elapsed → next page render will auto-enqueue.
    pub idle: usize,
    /// Document not covering latest, idle threshold not yet reached.
    pub required: usize,
    /// Document covers `latest_mtime` → nothing to do.
    pub current: usize,
}

impl BootStateSummary {
    fn add(&mut self, other: BootStateSummary) {
        self.queued += other.queued;
        self.failed += other.failed;
        self.idle += other.idle;
        self.required += other.required;
        self.current += other.current;
    }

    fn record(&mut self, state: &CaseAnalysisState) {
        match state {
            CaseAnalysisState::Queued => self.queued += 1,
            CaseAnalysisState::Failed => self.failed += 1,
            CaseAnalysisState::Idle { .. } => self.idle += 1,
            CaseAnalysisState::Required { .. } => self.required += 1,
            CaseAnalysisState::Current => self.current += 1,
            CaseAnalysisState::NoRecordings | CaseAnalysisState::Pending => {}
        }
    }
}

/// Walk `data_path/*/` and emit a single info-level summary line. The
/// returned summary is mainly for tests; production callers ignore it.
///
/// `boot_time` is forwarded to [`evaluate_state`]. Boot-scan typically
/// passes `SystemTime::now()` here — the same instant it itself observes
/// the case directories — so old recordings count as "just seen now".
pub async fn boot_scan_state(
    data_path: &Path,
    idle_threshold: Duration,
    now: SystemTime,
    boot_time: SystemTime,
    in_flight: &InFlight,
) -> BootStateSummary {
    let mut total = BootStateSummary::default();
    let Ok(mut users) = tokio::fs::read_dir(data_path).await else {
        return total;
    };
    while let Ok(Some(user_entry)) = users.next_entry().await {
        let per_user = compute_state_for_user(
            &user_entry.path(),
            idle_threshold,
            now,
            boot_time,
            in_flight,
        )
        .await;
        total.add(per_user);
    }
    info!(
        queued = total.queued,
        failed = total.failed,
        idle = total.idle,
        required = total.required,
        current = total.current,
        "analyze boot state scan finished"
    );
    total
}

/// Aggregate the per-case state for one user root. Skips closed cases
/// and non-UUID directory names. Pure observation — no side effects.
pub async fn compute_state_for_user(
    user_root: &Path,
    idle_threshold: Duration,
    now: SystemTime,
    boot_time: SystemTime,
    in_flight: &InFlight,
) -> BootStateSummary {
    let mut summary = BootStateSummary::default();
    let Ok(mut cases) = tokio::fs::read_dir(user_root).await else {
        return summary;
    };
    while let Ok(Some(case_entry)) = cases.next_entry().await {
        let case_dir = case_entry.path();
        if !case_dir.is_dir() || crate::paths::is_closed(&case_dir).await {
            continue;
        }
        let Ok(name) = case_entry.file_name().into_string() else {
            continue;
        };
        if uuid::Uuid::parse_str(&name).is_err() {
            continue;
        }
        let state = evaluate_state(&case_dir, idle_threshold, now, boot_time, in_flight).await;
        summary.record(&state);
    }
    summary
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::analyze::auto_trigger::{
        FAILURE_MARKER_FILE, FailureMarker, rfc3339_of as auto_rfc3339,
    };
    use crate::analyze::{DOCUMENT_FILE, Document};
    use filetime::{FileTime, set_file_mtime};
    use tempfile::TempDir;
    use tokio::fs;

    fn case_id(n: u8) -> String {
        let h = format!("{n:08x}");
        format!("{h}-1111-1111-1111-111111111111")
    }

    fn set_mtime(path: &Path, t: SystemTime) {
        set_file_mtime(path, FileTime::from_system_time(t)).unwrap();
    }

    /// Seed a case with a single transcribed recording at `mtime`.
    /// Returns the case_dir for further fixture operations.
    async fn seed_transcribed_case(
        user_root: &Path,
        n: u8,
        mtime: SystemTime,
    ) -> std::path::PathBuf {
        let case = user_root.join(case_id(n));
        fs::create_dir_all(&case).await.unwrap();
        let stem = "2026-01-01T10-00-00Z";
        let m4a = case.join(format!("{stem}.m4a"));
        fs::write(&m4a, b"fake").await.unwrap();
        crate::paths::write_recording_meta_sync(
            &case,
            stem,
            doctate_common::Transcript::Content {
                text: "transcript".into(),
            },
            None,
        );
        set_mtime(&m4a, mtime);
        case
    }

    async fn write_document_covering(case_dir: &Path, covered_mtime: SystemTime) {
        let rfc = auto_rfc3339(covered_mtime);
        let doc = Document {
            covered_mtime: rfc.clone(),
            written_at: rfc,
            backend_id: "test".into(),
            content_md: "doc".into(),
        };
        fs::write(
            case_dir.join(DOCUMENT_FILE),
            serde_json::to_vec(&doc).unwrap(),
        )
        .await
        .unwrap();
    }

    async fn write_marker(case_dir: &Path, mtime: SystemTime) {
        let marker = FailureMarker {
            last_recording_mtime: auto_rfc3339(mtime),
            reason: "test".into(),
            failed_at: "2026-05-03T00:00:00Z".into(),
            backend_id: None,
            details: None,
        };
        fs::write(
            case_dir.join(FAILURE_MARKER_FILE),
            serde_json::to_vec(&marker).unwrap(),
        )
        .await
        .unwrap();
    }

    fn base_time() -> SystemTime {
        SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000)
    }

    /// `Queued` is now driven by the in-process [`InFlight`] set, not
    /// by a file on disk. A claimed slot for a transcribed case must
    /// surface as `queued`.
    #[tokio::test]
    async fn in_flight_claim_counts_as_queued() {
        let user_root = TempDir::new().unwrap();
        let case = seed_transcribed_case(user_root.path(), 1, base_time()).await;

        let in_flight = InFlight::new();
        assert!(in_flight.try_claim(&case));

        let summary = compute_state_for_user(
            user_root.path(),
            Duration::ZERO,
            base_time(),
            SystemTime::UNIX_EPOCH,
            &in_flight,
        )
        .await;

        assert_eq!(summary.queued, 1);
        assert_eq!(summary.idle, 0);
        assert_eq!(summary.required, 0);
    }

    /// Pre-refactor leftover: a stale `analysis_input.json` on disk (no
    /// claim in [`InFlight`]) must NOT register as `Queued` any more —
    /// that was the whole reason for the refactor. The case still
    /// produces a transcribed-but-no-document state, so it counts as
    /// `idle` (zero idle threshold collapses Required→Idle).
    #[tokio::test]
    async fn orphan_input_file_does_not_count_as_queued() {
        let user_root = TempDir::new().unwrap();
        let case = seed_transcribed_case(user_root.path(), 7, base_time()).await;
        fs::write(case.join("analysis_input.json"), b"{}")
            .await
            .unwrap();

        let summary = compute_state_for_user(
            user_root.path(),
            Duration::ZERO,
            base_time(),
            SystemTime::UNIX_EPOCH,
            &InFlight::new(),
        )
        .await;

        assert_eq!(summary.queued, 0);
        assert_eq!(summary.idle, 1);
    }

    #[tokio::test]
    async fn document_covering_latest_counts_as_current() {
        let user_root = TempDir::new().unwrap();
        let case = seed_transcribed_case(user_root.path(), 2, base_time()).await;
        write_document_covering(&case, base_time()).await;

        let summary = compute_state_for_user(
            user_root.path(),
            Duration::ZERO,
            base_time(),
            SystemTime::UNIX_EPOCH,
            &InFlight::new(),
        )
        .await;

        assert_eq!(summary.current, 1);
        assert_eq!(summary.queued, 0);
    }

    /// Failure marker matching the input → operator must intervene; not
    /// retried automatically.
    #[tokio::test]
    async fn failure_marker_matching_mtime_counts_as_failed() {
        let user_root = TempDir::new().unwrap();
        let case = seed_transcribed_case(user_root.path(), 3, base_time()).await;
        write_marker(&case, base_time()).await;

        let summary = compute_state_for_user(
            user_root.path(),
            Duration::ZERO,
            base_time(),
            SystemTime::UNIX_EPOCH,
            &InFlight::new(),
        )
        .await;

        assert_eq!(summary.failed, 1);
    }

    /// Stale failure marker (older than current input) → recoverable on
    /// next idle tick.
    #[tokio::test]
    async fn stale_failure_marker_falls_through_to_idle() {
        let user_root = TempDir::new().unwrap();
        let case = seed_transcribed_case(user_root.path(), 4, base_time()).await;
        write_marker(&case, base_time() - Duration::from_secs(3600)).await;

        let summary = compute_state_for_user(
            user_root.path(),
            Duration::ZERO,
            base_time(),
            SystemTime::UNIX_EPOCH,
            &InFlight::new(),
        )
        .await;

        assert_eq!(summary.idle, 1);
        assert_eq!(summary.failed, 0);
    }

    #[tokio::test]
    async fn closed_cases_are_skipped() {
        let user_root = TempDir::new().unwrap();
        let case = seed_transcribed_case(user_root.path(), 5, base_time()).await;
        // Mark closed; the closed-cases scan must skip this entirely.
        fs::write(case.join(crate::paths::CLOSE_MARKER), "{}")
            .await
            .unwrap();

        let summary = compute_state_for_user(
            user_root.path(),
            Duration::ZERO,
            base_time(),
            SystemTime::UNIX_EPOCH,
            &InFlight::new(),
        )
        .await;

        assert_eq!(summary, BootStateSummary::default());
    }

    /// Idle threshold not yet reached → state is `Required`, not `Idle`.
    #[tokio::test]
    async fn idle_threshold_not_reached_counts_as_required() {
        let user_root = TempDir::new().unwrap();
        let _ = seed_transcribed_case(user_root.path(), 6, base_time()).await;

        // `now` is only 100s after the recording — below a 900s threshold.
        let summary = compute_state_for_user(
            user_root.path(),
            Duration::from_secs(900),
            base_time() + Duration::from_secs(100),
            SystemTime::UNIX_EPOCH,
            &InFlight::new(),
        )
        .await;

        assert_eq!(summary.required, 1);
        assert_eq!(summary.idle, 0);
    }
}
