use std::path::Path;

use tracing::{info, warn};

use super::auto_trigger::read_failure_marker;
use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};

/// Walk `data_path/*/` and enqueue every pending analysis for every user.
/// Intended to run once at startup; the same primitive backs the per-user
/// self-heal triggered by web handlers.
pub async fn scan_and_enqueue(data_path: &Path, tx: &AnalyzeSender) {
    let Ok(mut users) = tokio::fs::read_dir(data_path).await else {
        return;
    };
    let mut total = 0;
    while let Ok(Some(user_entry)) = users.next_entry().await {
        total += enqueue_pending_for_user(&user_entry.path(), tx).await;
    }
    info!(count = total, "Analyze recovery scan finished");
}

/// Scan a user's case dirs and enqueue any case that has an
/// `analysis_input.json` but no `document.md`. Returns the number sent.
pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> usize {
    let Ok(mut cases) = tokio::fs::read_dir(user_root).await else {
        return 0;
    };
    let mut sent = 0;
    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;
        }
        if !case_dir.join(ANALYSIS_INPUT_FILE).exists() {
            continue;
        }
        if case_dir.join(DOCUMENT_FILE).exists() {
            continue;
        }
        // A failure marker matching the input's `last_recording_mtime`
        // means the worker has already tried this exact input and failed
        // (e.g. LLM timeout). Re-enqueueing here would block the worker
        // for another full timeout window and starve fresh user clicks.
        // `auto_trigger::evaluate_case` performs the same skip; recovery
        // must mirror it.
        if input_already_failed(&case_dir).await {
            continue;
        }
        if tx
            .send(AnalyzeJob {
                case_dir: case_dir.clone(),
            })
            .is_err()
        {
            warn!(case = %case_dir.display(), "recovery send failed (worker gone)");
            return sent;
        }
        sent += 1;
    }
    sent
}

/// Returns `true` iff a failure marker exists whose
/// `last_recording_mtime` equals the one in `analysis_input.json`. Any
/// I/O or parse error is treated as "no usable marker" — recovery then
/// proceeds with the enqueue, matching the optimistic intent of the
/// outer scan.
async fn input_already_failed(case_dir: &Path) -> bool {
    let Some(marker) = read_failure_marker(case_dir).await else {
        return false;
    };
    let Ok(bytes) = tokio::fs::read(case_dir.join(ANALYSIS_INPUT_FILE)).await else {
        return false;
    };
    let Ok(input) = serde_json::from_slice::<AnalysisInput>(&bytes) else {
        return false;
    };
    marker.last_recording_mtime == input.last_recording_mtime
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::analyze::auto_trigger::{FAILURE_MARKER_FILE, FailureMarker};
    use crate::analyze::{RecordingInput, channel as analyze_channel};
    use tempfile::TempDir;
    use tokio::fs;

    fn make_input(mtime: &str) -> AnalysisInput {
        AnalysisInput {
            last_recording_mtime: mtime.to_owned(),
            recordings: vec![RecordingInput {
                recorded_at: mtime.to_owned(),
                text: "ignored".into(),
            }],
            backend_id: String::new(),
        }
    }

    async fn write_input(case_dir: &Path, input: &AnalysisInput) {
        fs::write(
            case_dir.join(ANALYSIS_INPUT_FILE),
            serde_json::to_vec(input).unwrap(),
        )
        .await
        .unwrap();
    }

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

    /// Drain the receiver after sender drop — same pattern as
    /// `auto_trigger::tests::try_enqueue_all_sends_only_eligible_cases`.
    async fn drain(mut rx: crate::analyze::AnalyzeReceiver) -> Vec<std::path::PathBuf> {
        let mut out = Vec::new();
        while let Some(job) = rx.recv().await {
            out.push(job.case_dir);
        }
        out
    }

    #[tokio::test]
    async fn enqueues_pending_input_without_doc_or_marker() {
        let user_root = TempDir::new().unwrap();
        let case_id = "11111111-1111-1111-1111-111111111111";
        let case = user_root.path().join(case_id);
        fs::create_dir_all(&case).await.unwrap();
        write_input(&case, &make_input("2026-05-03T10:00:00Z")).await;

        let (tx, rx) = analyze_channel();
        let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
        drop(tx);

        assert_eq!(sent, 1);
        let received = drain(rx).await;
        assert_eq!(received.len(), 1);
        assert!(received[0].ends_with(case_id));
    }

    #[tokio::test]
    async fn skips_when_document_already_exists() {
        let user_root = TempDir::new().unwrap();
        let case = user_root
            .path()
            .join("22222222-2222-2222-2222-222222222222");
        fs::create_dir_all(&case).await.unwrap();
        write_input(&case, &make_input("2026-05-03T10:00:00Z")).await;
        fs::write(case.join(DOCUMENT_FILE), b"already analysed")
            .await
            .unwrap();

        let (tx, rx) = analyze_channel();
        let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
        drop(tx);

        assert_eq!(sent, 0);
        assert!(drain(rx).await.is_empty());
    }

    /// Regression: a stuck Llama job times out, leaves `analysis_input.json`
    /// behind, and writes a failure marker. Without this skip the next
    /// page reload re-enqueues the same job and blocks the worker for
    /// another full timeout — see Bug 1 (case c414cf52, 2026-05-03).
    #[tokio::test]
    async fn skips_when_failure_marker_matches_input_mtime() {
        let user_root = TempDir::new().unwrap();
        let case = user_root
            .path()
            .join("33333333-3333-3333-3333-333333333333");
        fs::create_dir_all(&case).await.unwrap();
        let mtime = "2026-05-03T10:00:00Z";
        write_input(&case, &make_input(mtime)).await;
        write_marker(&case, mtime).await;

        let (tx, rx) = analyze_channel();
        let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
        drop(tx);

        assert_eq!(sent, 0, "must not re-enqueue an already-failed input");
        assert!(drain(rx).await.is_empty());
    }

    /// A new recording bumps `last_recording_mtime`, so the old marker
    /// no longer applies and recovery should re-enqueue.
    #[tokio::test]
    async fn enqueues_when_failure_marker_is_stale() {
        let user_root = TempDir::new().unwrap();
        let case_id = "44444444-4444-4444-4444-444444444444";
        let case = user_root.path().join(case_id);
        fs::create_dir_all(&case).await.unwrap();
        write_input(&case, &make_input("2026-05-03T11:00:00Z")).await;
        write_marker(&case, "2026-05-03T10:00:00Z").await;

        let (tx, rx) = analyze_channel();
        let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
        drop(tx);

        assert_eq!(sent, 1);
        let received = drain(rx).await;
        assert_eq!(received.len(), 1);
        assert!(received[0].ends_with(case_id));
    }
}
