doctate/server/src/analyze/auto_trigger.rs

//! Opportunistic auto-trigger for LLM analysis.
//!
//! Called by the `/web/cases` and `/web/cases/{id}` handlers on every
//! request. The SSE-driven `location.reload()` in the UI turns every
//! relevant state change (upload, transcript, document) into a fresh
//! handler invocation, so no background timer is needed.
//!
//! The hot path is a pure `evaluate_case` pre-condition check; the
//! `try_enqueue` wrapper performs the actual side effects.

use std::path::Path;
use std::sync::Arc;
use std::time::SystemTime;

use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use tokio::io::AsyncWriteExt;
use tracing::warn;

use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};
use crate::config::Config;
use crate::events::{self, CaseEventKind, EventSender};
use crate::paths;
use crate::routes::case_actions::build_analysis_input;

/// Filename of the per-case failure marker. Written by the worker on
/// error, deleted on success. Presence with a matching
/// `last_recording_mtime` tells us "already tried this exact input, do
/// not retry until the input changes".
pub const FAILURE_MARKER_FILE: &str = ".analysis_failed.json";

/// Persistent failure record. JSON so it is human-readable and carries
/// a reason for post-mortem.
#[derive(Debug, Serialize, Deserialize)]
pub struct FailureMarker {
    pub last_recording_mtime: String,
    pub reason: String,
    pub failed_at: String,
}

/// Outcome of the pre-condition check. `Skip` carries a static string so
/// the common "nothing to do" path allocates nothing.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AutoDecision {
    Enqueue,
    Skip(&'static str),
}

/// Pure pre-condition check: decide whether this case is due for an
/// automatic analysis. Does file I/O but no channel sends and no
/// state mutations — safe to call redundantly.
pub async fn evaluate_case(case_dir: &Path) -> AutoDecision {
    let scan = scan_m4as(case_dir).await;

    if !scan.saw_any {
        return AutoDecision::Skip("no recordings");
    }
    if !scan.all_transcribed {
        return AutoDecision::Skip("not all transcribed");
    }
    let Some(latest_mtime) = scan.latest_mtime else {
        return AutoDecision::Skip("no recordings");
    };

    // Any existing analysis_input.json means a job is queued or running.
    // Worker removes the file on success — next reload re-evaluates.
    if tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE))
        .await
        .unwrap_or(false)
    {
        return AutoDecision::Skip("job queued");
    }

    // A failure marker matching the current input signature means we
    // already tried and the input has not changed since.
    if let Some(marker) = read_failure_marker(case_dir).await
        && marker.last_recording_mtime == rfc3339_of(latest_mtime)
    {
        return AutoDecision::Skip("previously failed");
    }

    // Fresh document older than the newest recording → stale, re-analyse.
    // Fresh document newer or equal → nothing to do.
    let doc_path = case_dir.join(DOCUMENT_FILE);
    if let Ok(meta) = tokio::fs::metadata(&doc_path).await
        && let Ok(doc_mtime) = meta.modified()
        && doc_mtime >= latest_mtime
    {
        return AutoDecision::Skip("analysis current");
    }

    AutoDecision::Enqueue
}

/// Evaluate and, on `Enqueue`, prepare and send the job. Returns `true`
/// iff a job was actually sent. Silent on routine skips; `warn!` only on
/// actual failures (build_input, write, send).
pub async fn try_enqueue(
    case_dir: &Path,
    tx: &AnalyzeSender,
    config: &Arc<Config>,
    events_tx: &EventSender,
) -> bool {
    if !config.llm_configured() {
        return false;
    }

    match evaluate_case(case_dir).await {
        AutoDecision::Skip(_) => return false,
        AutoDecision::Enqueue => {}
    }

    let input = match build_analysis_input(case_dir).await {
        Ok(i) => i,
        Err(e) => {
            warn!(case = %case_dir.display(), error = ?e, "auto-analyze: build input failed");
            return false;
        }
    };

    // Re-analyse path: delete stale document.md up front so the worker's
    // "document exists → skip" guard does not short-circuit, and the UI
    // stops showing the outdated text during the render race.
    let doc_path = case_dir.join(DOCUMENT_FILE);
    if let Err(e) = remove_if_exists(&doc_path).await {
        warn!(case = %case_dir.display(), error = %e, "auto-analyze: remove old doc failed");
        return false;
    }

    let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
    if let Err(e) = write_input_overwrite(&input_path, &input).await {
        warn!(case = %case_dir.display(), error = %e, "auto-analyze: write input failed");
        return false;
    }

    if tx
        .send(AnalyzeJob {
            case_dir: case_dir.to_path_buf(),
        })
        .is_err()
    {
        warn!(case = %case_dir.display(), "auto-analyze: send failed (worker gone)");
        return false;
    }

    events::emit(
        events_tx,
        events::user_slug_of(case_dir),
        events::case_id_of(case_dir),
        CaseEventKind::AnalysisQueued,
    );
    true
}

/// Iterate over every non-deleted UUID-named case directory under
/// `user_root` and run `try_enqueue` on each. Safe to call on every
/// `/web/cases` render — skips are cheap filesystem stats.
pub async fn try_enqueue_all_for_user(
    user_root: &Path,
    tx: &AnalyzeSender,
    config: &Arc<Config>,
    events_tx: &EventSender,
) {
    let Ok(mut entries) = tokio::fs::read_dir(user_root).await else {
        return;
    };
    while let Ok(Some(entry)) = entries.next_entry().await {
        let Ok(name) = entry.file_name().into_string() else {
            continue;
        };
        if uuid::Uuid::parse_str(&name).is_err() {
            continue;
        }
        let case_path = entry.path();
        if !case_path.is_dir() {
            continue;
        }
        if crate::paths::is_closed(&case_path).await {
            continue;
        }
        try_enqueue(&case_path, tx, config, events_tx).await;
    }
}

/// Persist a failure marker atomically. Called by the worker after a
/// terminal failure. `last_recording_mtime` must equal the value read
/// from the `AnalysisInput` so `evaluate_case` can later detect an
/// unchanged input.
pub async fn write_failure_marker(
    case_dir: &Path,
    last_recording_mtime: &str,
    reason: &str,
) -> std::io::Result<()> {
    let marker = FailureMarker {
        last_recording_mtime: last_recording_mtime.to_owned(),
        reason: reason.to_owned(),
        failed_at: doctate_common::now_rfc3339(),
    };
    let bytes = serde_json::to_vec_pretty(&marker)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
    let path = case_dir.join(FAILURE_MARKER_FILE);
    let tmp = path.with_extension("json.tmp");
    let mut f = tokio::fs::File::create(&tmp).await?;
    f.write_all(&bytes).await?;
    f.sync_all().await?;
    drop(f);
    tokio::fs::rename(&tmp, &path).await
}

/// Best-effort delete. Called by the worker after a successful run so
/// the next `evaluate_case` does not mistake a resolved failure for a
/// pending one.
pub async fn remove_failure_marker(case_dir: &Path) {
    let _ = tokio::fs::remove_file(case_dir.join(FAILURE_MARKER_FILE)).await;
}

struct M4aScan {
    saw_any: bool,
    all_transcribed: bool,
    latest_mtime: Option<SystemTime>,
}

/// One-pass directory walk collecting everything `evaluate_case` needs.
/// `.m4a.failed` entries are skipped because they did not contribute a
/// transcript and will not contribute one in the future.
async fn scan_m4as(case_dir: &Path) -> M4aScan {
    let mut scan = M4aScan {
        saw_any: false,
        all_transcribed: true,
        latest_mtime: None,
    };

    let mut entries = match tokio::fs::read_dir(case_dir).await {
        Ok(r) => r,
        Err(_) => return scan,
    };

    while let Ok(Some(entry)) = entries.next_entry().await {
        let Ok(name) = entry.file_name().into_string() else {
            continue;
        };
        if !name.ends_with(".m4a") {
            continue;
        }
        scan.saw_any = true;

        if let Ok(meta) = entry.metadata().await
            && let Ok(mtime) = meta.modified()
        {
            scan.latest_mtime = Some(match scan.latest_mtime {
                Some(prev) if prev >= mtime => prev,
                _ => mtime,
            });
        }

        // Analyze gate: "all transcribed" means every m4a has passed
        // through the transcriber, regardless of whether that produced
        // `Content` or `Silent` output. Only `Pending` (no sidecar)
        // holds the trigger back.
        if paths::read_transcript_state(&entry.path())
            .await
            .is_pending()
        {
            scan.all_transcribed = false;
        }
    }

    scan
}

async fn read_failure_marker(case_dir: &Path) -> Option<FailureMarker> {
    let bytes = tokio::fs::read(case_dir.join(FAILURE_MARKER_FILE))
        .await
        .ok()?;
    serde_json::from_slice(&bytes).ok()
}

async fn remove_if_exists(path: &Path) -> std::io::Result<()> {
    match tokio::fs::remove_file(path).await {
        Ok(_) => Ok(()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(e),
    }
}

/// Overwrite-capable sibling of `case_actions::write_input_create_new`.
/// Auto-trigger re-analyses may legitimately replace an existing input
/// (e.g. a new recording arrived between two reloads while the previous
/// job was still queued). `.tmp` + fsync + rename keeps it atomic.
async fn write_input_overwrite(path: &Path, input: &AnalysisInput) -> std::io::Result<()> {
    let bytes = serde_json::to_vec_pretty(input)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
    let tmp = path.with_extension("json.tmp");
    let mut f = tokio::fs::File::create(&tmp).await?;
    f.write_all(&bytes).await?;
    f.sync_all().await?;
    drop(f);
    tokio::fs::rename(&tmp, path).await
}

/// Format an mtime in the same RFC3339 shape that
/// `case_actions::build_analysis_input` writes, so equality comparison
/// against a stored marker is string-exact.
fn rfc3339_of(t: SystemTime) -> String {
    OffsetDateTime::from(t).format(&Rfc3339).unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use super::*;
    use filetime::{FileTime, set_file_mtime};
    use std::time::Duration;
    use tempfile::TempDir;
    use tokio::fs;

    async fn touch(path: &std::path::Path) {
        fs::write(path, b"").await.unwrap();
    }

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

    #[tokio::test]
    async fn empty_dir_skips() {
        let dir = TempDir::new().unwrap();
        assert_eq!(
            evaluate_case(dir.path()).await,
            AutoDecision::Skip("no recordings")
        );
    }

    #[tokio::test]
    async fn missing_transcript_skips() {
        let dir = TempDir::new().unwrap();
        touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
        touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
        touch(&dir.path().join("2026-01-01T11-00-00Z.m4a")).await;
        assert_eq!(
            evaluate_case(dir.path()).await,
            AutoDecision::Skip("not all transcribed")
        );
    }

    #[tokio::test]
    async fn all_transcribed_no_document_enqueues() {
        let dir = TempDir::new().unwrap();
        touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
        touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
        assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
    }

    #[tokio::test]
    async fn document_newer_than_recordings_skips() {
        let dir = TempDir::new().unwrap();
        let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
        touch(&m4a).await;
        touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
        let doc = dir.path().join(DOCUMENT_FILE);
        touch(&doc).await;

        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        set_mtime(&m4a, base);
        set_mtime(&doc, base + Duration::from_secs(60));

        assert_eq!(
            evaluate_case(dir.path()).await,
            AutoDecision::Skip("analysis current")
        );
    }

    #[tokio::test]
    async fn document_older_than_recordings_enqueues() {
        let dir = TempDir::new().unwrap();
        let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
        touch(&m4a).await;
        touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
        let doc = dir.path().join(DOCUMENT_FILE);
        touch(&doc).await;

        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        set_mtime(&doc, base);
        set_mtime(&m4a, base + Duration::from_secs(60));

        assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
    }

    #[tokio::test]
    async fn analysis_input_present_skips() {
        let dir = TempDir::new().unwrap();
        touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
        touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
        touch(&dir.path().join(ANALYSIS_INPUT_FILE)).await;
        assert_eq!(
            evaluate_case(dir.path()).await,
            AutoDecision::Skip("job queued")
        );
    }

    #[tokio::test]
    async fn failure_marker_matching_mtime_skips() {
        let dir = TempDir::new().unwrap();
        let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
        touch(&m4a).await;
        touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;

        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        set_mtime(&m4a, base);

        let marker = FailureMarker {
            last_recording_mtime: rfc3339_of(base),
            reason: "test".into(),
            failed_at: "2026-01-01T10-05-00Z".into(),
        };
        fs::write(
            dir.path().join(FAILURE_MARKER_FILE),
            serde_json::to_vec(&marker).unwrap(),
        )
        .await
        .unwrap();

        assert_eq!(
            evaluate_case(dir.path()).await,
            AutoDecision::Skip("previously failed")
        );
    }

    #[tokio::test]
    async fn failure_marker_stale_mtime_enqueues() {
        let dir = TempDir::new().unwrap();
        let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
        touch(&m4a).await;
        touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;

        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        set_mtime(&m4a, base);

        // Marker records an older mtime → input has since changed.
        let marker = FailureMarker {
            last_recording_mtime: rfc3339_of(base - Duration::from_secs(3600)),
            reason: "test".into(),
            failed_at: "2025-12-31T23-00-00Z".into(),
        };
        fs::write(
            dir.path().join(FAILURE_MARKER_FILE),
            serde_json::to_vec(&marker).unwrap(),
        )
        .await
        .unwrap();

        assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
    }

    #[tokio::test]
    async fn m4a_failed_is_ignored() {
        let dir = TempDir::new().unwrap();
        // Only .m4a.failed present — counts as "no usable recordings".
        touch(&dir.path().join("2026-01-01T10-00-00Z.m4a.failed")).await;
        assert_eq!(
            evaluate_case(dir.path()).await,
            AutoDecision::Skip("no recordings")
        );
    }

    #[tokio::test]
    async fn try_enqueue_all_sends_only_eligible_cases() {
        use crate::analyze::channel as analyze_channel;
        use crate::config::Config;
        use crate::events::channel as events_channel;

        let user_root = TempDir::new().unwrap();

        // Eligible: all transcribed, no document, plausible content.
        let eligible_id = "11111111-1111-1111-1111-111111111111";
        let eligible = user_root.path().join(eligible_id);
        fs::create_dir_all(&eligible).await.unwrap();
        fs::write(eligible.join("2026-01-01T10-00-00Z.m4a"), b"fake")
            .await
            .unwrap();
        fs::write(
            eligible.join("2026-01-01T10-00-00Z.transcript.txt"),
            b"patient mit brustschmerz",
        )
        .await
        .unwrap();

        // Not eligible: has a fresh document.
        let current_id = "22222222-2222-2222-2222-222222222222";
        let current = user_root.path().join(current_id);
        fs::create_dir_all(&current).await.unwrap();
        let m4a = current.join("2026-01-01T10-00-00Z.m4a");
        fs::write(&m4a, b"fake").await.unwrap();
        fs::write(
            current.join("2026-01-01T10-00-00Z.transcript.txt"),
            b"stable",
        )
        .await
        .unwrap();
        let doc = current.join(DOCUMENT_FILE);
        fs::write(&doc, b"doc").await.unwrap();
        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        set_mtime(&m4a, base);
        set_mtime(&doc, base + Duration::from_secs(60));

        // Ignored: non-UUID directory name.
        let junk = user_root.path().join("not-a-uuid");
        fs::create_dir_all(&junk).await.unwrap();
        fs::write(junk.join("2026-01-01T10-00-00Z.m4a"), b"fake")
            .await
            .unwrap();
        fs::write(junk.join("2026-01-01T10-00-00Z.transcript.txt"), b"x")
            .await
            .unwrap();

        let (tx, mut rx) = analyze_channel();
        let events_tx = events_channel();
        let mut cfg = Config::test_default();
        cfg.llm_url = "http://localhost:9999".into();
        cfg.llm_model = "test".into();
        let cfg = Arc::new(cfg);

        try_enqueue_all_for_user(user_root.path(), &tx, &cfg, &events_tx).await;

        // Drop the sender half so recv closes after draining.
        drop(tx);
        let mut received = Vec::new();
        while let Some(job) = rx.recv().await {
            received.push(job.case_dir);
        }
        assert_eq!(received.len(), 1, "expected exactly one enqueued job");
        assert!(
            received[0].ends_with(eligible_id),
            "expected eligible case, got {}",
            received[0].display()
        );
    }
}