doctate/server/src/transcribe/worker.rs

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

use doctate_common::oneliners::{ONELINER_FILENAME, OnelinerState};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use tracing::{error, info, warn};

use super::{TranscribeReceiver, ffmpeg, ollama, whisper};
use crate::config::{Config, WhisperUserSettings};
use crate::events::{self, CaseEventKind, EventSender};
use crate::gazetteer::Gazetteer;
use crate::{BusyGuard, WorkerBusy};

const ONELINER_TIMEOUT: Duration = Duration::from_secs(60);

/// Consume transcription jobs sequentially. Whisper holds the GPU, so parallelism
/// here would only cause contention. One job in flight at a time.
pub async fn run(
    mut rx: TranscribeReceiver,
    config: Arc<Config>,
    client: reqwest::Client,
    worker_busy: WorkerBusy,
    vocab: Arc<Gazetteer>,
    events_tx: EventSender,
) {
    info!("Transcription worker started");
    let timeout = Duration::from_secs(config.whisper_timeout_seconds);

    while let Some(job) = rx.recv().await {
        let _guard = BusyGuard::new(worker_busy.clone());
        let audio_path = job.audio_path;
        let transcript_path = audio_path.with_extension("transcript.txt");

        if transcript_path.exists() {
            continue;
        }

        // Look up per-user Whisper settings live from the shared config so that
        // edits to users.toml (on next restart) reach in-flight jobs naturally.
        // Unknown slug → empty settings (service falls back to language=de).
        let settings: WhisperUserSettings = config
            .users
            .iter()
            .find(|u| u.slug == job.user_slug)
            .map(|u| u.whisper.clone())
            .unwrap_or_default();

        info!(audio = %audio_path.display(), user = %job.user_slug, "Transcribing");

        let remuxed = match ffmpeg::remux_faststart(&audio_path).await {
            Ok(f) => f,
            Err(e) => {
                error!(audio = %audio_path.display(), error = %e, "ffmpeg remux failed");
                mark_failed(&audio_path, &events_tx, &job.user_slug).await;
                continue;
            }
        };

        // Persist the container duration as a sidecar so the UI can render it
        // without the browser having to HEAD every audio file. Non-fatal — the
        // recordings scan has a lazy-backfill path, and transcription is the
        // critical workload here.
        write_duration_sidecar(&audio_path, remuxed.path()).await;

        let text = match whisper::transcribe(
            &client,
            &config.whisper_url,
            remuxed.path(),
            timeout,
            &settings,
        )
        .await
        {
            Ok(t) => t,
            Err(e) => {
                error!(audio = %audio_path.display(), error = %e, "whisper call failed");
                mark_failed(&audio_path, &events_tx, &job.user_slug).await;
                continue;
            }
        };

        // Normalize terminology against the curated vocab before
        // persisting. Downstream consumers (oneliner, analyze) read
        // the canonical path.
        let text = vocab.replace(&text);

        if let Err(e) = tokio::fs::write(&transcript_path, &text).await {
            error!(transcript = %transcript_path.display(), error = %e, "writing transcript failed");
            continue;
        }

        info!(
            audio = %audio_path.display(),
            bytes = text.len(),
            "Transcript written"
        );

        // Notify the live-update bus so subscribed browsers can refresh the
        // recordings view. case_id = parent dir of the audio file.
        if let Some(case_dir) = audio_path.parent() {
            events::emit(
                &events_tx,
                &job.user_slug,
                events::case_id_of(case_dir),
                CaseEventKind::TranscriptReady,
            );
        }

        // Regenerate the oneliner, but only after the *last* pending
        // recording in the batch — otherwise we'd burn LLM calls on
        // incomplete inputs that the next job would redo. Later
        // recordings still override earlier ones (symmetric to the
        // analyze-LLM "Spätere Aufnahmen haben Vorrang" rule). Silent /
        // empty cases are no-ops; LLM errors leave any existing
        // oneliner untouched.
        if let Some(case_dir) = audio_path.parent()
            && !has_pending_recordings(case_dir).await
        {
            update_oneliner(
                case_dir,
                &job.user_slug,
                &client,
                &config,
                &vocab,
                &events_tx,
            )
            .await;
        }
    }

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

/// Atomically rename a failed recording from `<ts>.m4a` to `<ts>.m4a.failed`
/// so the recovery scan no longer picks it up. The UI still surfaces these
/// files (with a "failed" flag) so the user can listen to the audio and
/// manually decide whether to retry (by renaming back) or to delete.
async fn mark_failed(audio_path: &Path, events_tx: &EventSender, user_slug: &str) {
    let failed_path = audio_path.with_extension("m4a.failed");
    if let Err(e) = tokio::fs::rename(audio_path, &failed_path).await {
        // Nothing to roll back — if rename fails, the worst case is that we
        // retry the same file on next restart. Log and move on.
        error!(
            audio = %audio_path.display(),
            error = %e,
            "Failed to mark recording as failed; may be retried on restart",
        );
    } else {
        warn!(
            audio = %audio_path.display(),
            failed = %failed_path.display(),
            "Recording marked as failed",
        );
        if let Some(case_dir) = audio_path.parent() {
            events::emit(
                events_tx,
                user_slug,
                events::case_id_of(case_dir),
                CaseEventKind::TranscriptFailed,
            );
        }
    }
}

/// Probe the (remuxed) audio and persist the duration next to the original file
/// as `<ts>.duration.txt`. Probes on the remuxed copy because its `moov` atom
/// sits at the start — ffprobe returns without seeking to EOF.
async fn write_duration_sidecar(audio_path: &Path, probe_path: &Path) {
    let secs = match ffmpeg::probe_duration_seconds(probe_path).await {
        Ok(s) => s,
        Err(e) => {
            warn!(audio = %audio_path.display(), error = %e, "ffprobe duration failed — UI will lazy-backfill");
            return;
        }
    };
    let sidecar = audio_path.with_extension("duration.txt");
    if let Err(e) = tokio::fs::write(&sidecar, secs.to_string()).await {
        warn!(sidecar = %sidecar.display(), error = %e, "writing duration sidecar failed");
    }
}

/// Regenerate `case_dir/oneliner.json` from **all** non-empty transcripts
/// in the case, joined chronologically. Called after every successful
/// transcript write so later recordings can correct earlier ones (mirror
/// of the analyze-LLM rule "later recordings override"). Every outcome —
/// generated text, deliberate empty ("no medical content"), or call error
/// — is persisted as an [`OnelinerState`] variant, overwriting any
/// previous state. Silent case (no non-empty transcript) is a no-op; the
/// next transcript write retries.
pub(crate) async fn update_oneliner(
    case_dir: &Path,
    user_slug: &str,
    client: &reqwest::Client,
    config: &Config,
    vocab: &Gazetteer,
    events_tx: &EventSender,
) {
    let transcript = match all_transcripts_joined(case_dir).await {
        Some(t) => t,
        None => return,
    };

    let generated_at = OffsetDateTime::now_utc()
        .format(&Rfc3339)
        .unwrap_or_default();

    let state = match ollama::generate_oneliner(
        client,
        &config.ollama_url,
        &config.ollama_model,
        config.ollama_keep_alive,
        &transcript,
        ONELINER_TIMEOUT,
    )
    .await
    {
        Ok(line) => {
            let text = vocab.replace(&line);
            info!(
                user = %user_slug,
                case = %case_dir.display(),
                chars = text.chars().count(),
                "Oneliner ready"
            );
            OnelinerState::Ready { text, generated_at }
        }
        Err(ollama::OllamaError::EmptyResponse) => {
            warn!(
                case = %case_dir.display(),
                "oneliner empty — model returned nothing (no medical keyword in transcript)"
            );
            OnelinerState::Empty { generated_at }
        }
        Err(e) => {
            error!(case = %case_dir.display(), error = %e, "oneliner generation failed");
            OnelinerState::Error { generated_at }
        }
    };

    if let Err(e) = write_oneliner_state(case_dir, &state).await {
        error!(case = %case_dir.display(), error = %e, "writing oneliner.json failed");
        return;
    }
    events::emit(
        events_tx,
        user_slug,
        events::case_id_of(case_dir),
        CaseEventKind::OnelinerUpdated,
    );
}

/// Write `state` atomically to `case_dir/oneliner.json` via
/// `write(tmp) + rename(tmp, final)`. Rename is atomic within a single
/// filesystem, so concurrent readers never see a half-serialized JSON
/// document.
async fn write_oneliner_state(case_dir: &Path, state: &OnelinerState) -> std::io::Result<()> {
    let final_path = case_dir.join(ONELINER_FILENAME);
    let tmp_path = case_dir.join(format!("{ONELINER_FILENAME}.tmp"));
    let payload = serde_json::to_vec(state).expect("OnelinerState is infallibly serializable");
    tokio::fs::write(&tmp_path, &payload).await?;
    tokio::fs::rename(&tmp_path, &final_path).await
}

/// Return true if `case_dir` contains at least one `.m4a` recording
/// that has no corresponding `.transcript.txt`. Used to gate the
/// oneliner-regenerate call: we only run it after the *last* pending
/// transcription in a batch, not after every single one. `.m4a.failed`
/// entries are skipped — they don't block the oneliner. Fail-open on
/// read errors: if we can't scan, return false so the oneliner runs
/// anyway (one redundant call beats never).
pub(crate) async fn has_pending_recordings(case_dir: &Path) -> bool {
    let Ok(mut entries) = tokio::fs::read_dir(case_dir).await else {
        return false;
    };
    while let Ok(Some(entry)) = entries.next_entry().await {
        let path = entry.path();
        if path.extension().and_then(|s| s.to_str()) != Some("m4a") {
            continue;
        }
        if !path.with_extension("transcript.txt").exists() {
            return true;
        }
    }
    false
}

/// Read every non-empty `*.transcript.txt` in `case_dir`, ordered
/// chronologically (ISO-8601 filename sort equals chronological order), and
/// join them with `\n\n---\n\n`. Empty / whitespace-only transcripts (silent
/// recordings) are skipped. Returns None if no non-empty transcript exists.
pub(crate) async fn all_transcripts_joined(case_dir: &Path) -> Option<String> {
    let mut entries = tokio::fs::read_dir(case_dir).await.ok()?;
    let mut files: Vec<std::path::PathBuf> = Vec::new();
    while let Ok(Some(entry)) = entries.next_entry().await {
        let name = entry.file_name();
        let Some(s) = name.to_str() else { continue };
        if s.ends_with(".transcript.txt") {
            files.push(entry.path());
        }
    }
    files.sort();

    let mut parts: Vec<String> = Vec::with_capacity(files.len());
    for path in files {
        if let Ok(text) = tokio::fs::read_to_string(&path).await {
            let trimmed = text.trim();
            if !trimmed.is_empty() {
                parts.push(trimmed.to_string());
            }
        }
    }
    if parts.is_empty() {
        return None;
    }
    Some(parts.join("\n\n---\n\n"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    /// Regression guard for the Womax bug: the joined transcript must
    /// include ALL recordings in chronological order, so the LLM sees
    /// both the initial term and any later correction.
    #[tokio::test]
    async fn all_transcripts_joined_includes_all_in_order() {
        let dir = tempdir().unwrap();
        tokio::fs::write(
            dir.path().join("2026-04-16T10-05-00Z.transcript.txt"),
            "Korrektur. Das Mittel heißt Vomex.",
        )
        .await
        .unwrap();
        tokio::fs::write(
            dir.path().join("2026-04-16T10-00-00Z.transcript.txt"),
            "Jetzt weiß ich es wieder. Es heißt Womax.",
        )
        .await
        .unwrap();

        let got = all_transcripts_joined(dir.path()).await.expect("Some");
        assert_eq!(
            got,
            "Jetzt weiß ich es wieder. Es heißt Womax.\n\n---\n\nKorrektur. Das Mittel heißt Vomex."
        );
    }

    /// Silent recordings are skipped, not treated as end-of-stream — a
    /// later non-empty recording still contributes.
    #[tokio::test]
    async fn all_transcripts_joined_skips_empty_recordings() {
        let dir = tempdir().unwrap();
        tokio::fs::write(dir.path().join("2026-04-16T10-00-00Z.transcript.txt"), "")
            .await
            .unwrap();
        tokio::fs::write(
            dir.path().join("2026-04-16T10-05-00Z.transcript.txt"),
            "non-empty content",
        )
        .await
        .unwrap();

        let got = all_transcripts_joined(dir.path()).await.expect("Some");
        assert_eq!(got, "non-empty content");
    }

    #[tokio::test]
    async fn all_transcripts_joined_is_none_when_all_empty() {
        let dir = tempdir().unwrap();
        tokio::fs::write(
            dir.path().join("2026-04-16T10-00-00Z.transcript.txt"),
            "   \n  ",
        )
        .await
        .unwrap();
        assert!(all_transcripts_joined(dir.path()).await.is_none());
    }

    #[tokio::test]
    async fn all_transcripts_joined_is_none_when_no_transcripts() {
        let dir = tempdir().unwrap();
        tokio::fs::write(dir.path().join("2026-04-16T10-00-00Z.m4a"), b"audio")
            .await
            .unwrap();
        assert!(all_transcripts_joined(dir.path()).await.is_none());
    }

    #[tokio::test]
    async fn has_pending_recordings_true_when_m4a_without_transcript() {
        let dir = tempdir().unwrap();
        tokio::fs::write(dir.path().join("2026-04-16T10-00-00Z.m4a"), b"a")
            .await
            .unwrap();
        tokio::fs::write(
            dir.path().join("2026-04-16T10-00-00Z.transcript.txt"),
            "text",
        )
        .await
        .unwrap();
        tokio::fs::write(dir.path().join("2026-04-16T10-05-00Z.m4a"), b"b")
            .await
            .unwrap();
        assert!(has_pending_recordings(dir.path()).await);
    }

    #[tokio::test]
    async fn has_pending_recordings_false_when_all_transcribed() {
        let dir = tempdir().unwrap();
        tokio::fs::write(dir.path().join("2026-04-16T10-00-00Z.m4a"), b"a")
            .await
            .unwrap();
        tokio::fs::write(
            dir.path().join("2026-04-16T10-00-00Z.transcript.txt"),
            "text1",
        )
        .await
        .unwrap();
        tokio::fs::write(dir.path().join("2026-04-16T10-05-00Z.m4a"), b"b")
            .await
            .unwrap();
        tokio::fs::write(
            dir.path().join("2026-04-16T10-05-00Z.transcript.txt"),
            "text2",
        )
        .await
        .unwrap();
        assert!(!has_pending_recordings(dir.path()).await);
    }

    #[tokio::test]
    async fn has_pending_recordings_ignores_failed() {
        let dir = tempdir().unwrap();
        tokio::fs::write(dir.path().join("2026-04-16T10-00-00Z.m4a.failed"), b"x")
            .await
            .unwrap();
        assert!(!has_pending_recordings(dir.path()).await);
    }

    #[tokio::test]
    async fn has_pending_recordings_false_for_empty_dir() {
        let dir = tempdir().unwrap();
        assert!(!has_pending_recordings(dir.path()).await);
    }
}