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

use tokio::io::AsyncWriteExt;
use tracing::{error, info, warn};

use super::backend::{default_backend, find_backend};
use super::{
    ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, auto_trigger, llm, prompt,
};
use crate::events::{self, CaseEventKind, EventSender};
use crate::gazetteer::Gazetteer;
use crate::{BusyGuard, WorkerBusy};

/// Consume analyze jobs sequentially. The external LLM tolerates parallelism,
/// but sequential processing keeps the architecture simple and shields the
/// provider from bursts. Trivial to lift later via `tokio::spawn(process(..))`.
pub async fn run(
    mut rx: AnalyzeReceiver,
    client: reqwest::Client,
    worker_busy: WorkerBusy,
    vocab: Arc<Gazetteer>,
    events_tx: EventSender,
) {
    info!(vocab_entries = vocab.len(), "Analyze worker started");

    while let Some(job) = rx.recv().await {
        let _guard = BusyGuard::new(worker_busy.clone());
        process(&job.case_dir, &client, &vocab, &events_tx).await;
    }

    warn!("Analyze worker stopped (channel closed)");
}

async fn process(
    case_dir: &Path,
    client: &reqwest::Client,
    vocab: &Gazetteer,
    events_tx: &EventSender,
) {
    let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
    let document_path = case_dir.join(DOCUMENT_FILE);

    // If the document already exists, the job is obsolete (e.g. recovery
    // re-enqueued a finished case during a race). Skip silently.
    if document_path.exists() {
        return;
    }

    let input: AnalysisInput = match tokio::fs::read(&input_path).await {
        Ok(bytes) => match serde_json::from_slice(&bytes) {
            Ok(v) => v,
            Err(e) => {
                error!(path = %input_path.display(), error = %e, "analysis input parse failed");
                return;
            }
        },
        Err(e) => {
            error!(path = %input_path.display(), error = %e, "analysis input read failed");
            return;
        }
    };

    // Silent-only case: no usable transcripts. Skip LLM, write a stub.
    if input.recordings.is_empty() {
        let stub = b"_Keine verwertbaren Aufnahmen (alle Aufnahmen still)._\n";
        if let Err(e) = write_atomic(&document_path, stub).await {
            error!(path = %document_path.display(), error = %e, "writing stub document failed");
            let _ = auto_trigger::write_failure_marker(
                case_dir,
                &input.last_recording_mtime,
                &format!("stub write: {e}"),
                None,
            )
            .await;
            return;
        }
        if let Err(e) = tokio::fs::remove_file(&input_path).await {
            warn!(path = %input_path.display(), error = %e, "removing analysis input failed");
        }
        auto_trigger::remove_failure_marker(case_dir).await;
        info!(case = %case_dir.display(), "analysis done (stub, no recordings)");
        events::emit(
            events_tx,
            events::user_slug_of(case_dir),
            events::case_id_of(case_dir),
            CaseEventKind::DocumentReady,
        );
        return;
    }

    let user_content = prompt::render_prompt(&input);
    let total_chars = user_content.chars().count();
    let backend = find_backend(&input.backend_id).unwrap_or_else(default_backend);
    info!(
        case = %case_dir.display(),
        recording_count = input.recordings.len(),
        total_chars,
        backend_id = %backend.id,
        model = %backend.model_id,
        "sending to llm"
    );

    // NOTE: If the server crashes between a successful LLM response and the
    // document write below, the recovery scan will re-enqueue this job and
    // the call will be made again. Accepted cost — rare event, small
    // per-call price. Response-caching in a `.tmp` file would prevent it.
    let document = match llm::chat_once(client, backend, &user_content).await {
        Ok(text) => text,
        Err(e) => {
            // Do not log the response body — LlmError::Display is already
            // redacted, but reinforce the rule here for future readers.
            error!(case = %case_dir.display(), backend_id = %backend.id, error = %e, "llm call failed");
            let _ = auto_trigger::write_failure_marker(
                case_dir,
                &input.last_recording_mtime,
                &format!("llm: {e}"),
                Some(&backend.id),
            )
            .await;
            return;
        }
    };

    // Post-AI terminology normalization: enforce vocab-canonical forms
    // over the LLM's training priors (e.g. rewrite anglicized INNs like
    // `Amiodarone` back to `Amiodaron`).
    let document = vocab.replace(&document);

    // Silent-deletion guard: schema-aware backends can return formally valid
    // but empty envelopes (`{"document": ""}`) when the prompt is incompatible
    // with the format constraint. Without this guard the worker would write
    // an empty document.md, drop the failure marker, and the user would see
    // a blank note with no diagnostic. Observed on Llama 3.1 405B FP8
    // (test T4, 2026-05-03) and previously on Mistral 24B.
    if document.trim().is_empty() {
        error!(
            case = %case_dir.display(),
            backend_id = %backend.id,
            "llm returned empty document — silent deletion guard triggered"
        );
        let _ = auto_trigger::write_failure_marker(
            case_dir,
            &input.last_recording_mtime,
            "llm returned empty document",
            Some(&backend.id),
        )
        .await;
        return;
    }

    if let Err(e) = write_atomic(&document_path, document.as_bytes()).await {
        error!(path = %document_path.display(), error = %e, "writing document failed");
        let _ = auto_trigger::write_failure_marker(
            case_dir,
            &input.last_recording_mtime,
            &format!("write document: {e}"),
            Some(&backend.id),
        )
        .await;
        return;
    }

    // Job done — drop the queue marker so the live "analyzing" flag
    // becomes accurate immediately and recovery on next boot does not
    // re-enqueue this version.
    if let Err(e) = tokio::fs::remove_file(&input_path).await {
        warn!(path = %input_path.display(), error = %e, "removing analysis input failed");
    }
    auto_trigger::remove_failure_marker(case_dir).await;

    info!(
        case = %case_dir.display(),
        bytes = document.len(),
        "analysis done"
    );

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

/// Write `bytes` to `path` atomically: write to `<path>.tmp`, fsync, rename.
/// A crash between write and rename leaves no half-written target file — the
/// state machine simply sees the job as still pending and retries.
async fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
    let tmp = path.with_extension({
        let mut ext = path
            .extension()
            .map(|s| s.to_os_string())
            .unwrap_or_default();
        ext.push(".tmp");
        ext
    });

    let mut file = tokio::fs::File::create(&tmp).await?;
    file.write_all(bytes).await?;
    file.sync_all().await?;
    drop(file);
    tokio::fs::rename(&tmp, path).await?;
    Ok(())
}
