doctate/server/src/transcribe/worker.rs

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

use doctate_common::oneliners::OnelinerState;
use doctate_common::{Loudness, RecordingMeta, Transcript, TranscriptState};
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::loudness;
use crate::oneliner_locks::OnelinerLocks;
use crate::paths;
use crate::settings::Settings;
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.
#[allow(clippy::too_many_arguments)]
pub async fn run(
    mut rx: TranscribeReceiver,
    config: Arc<Config>,
    settings: Arc<Settings>,
    client: reqwest::Client,
    worker_busy: WorkerBusy,
    vocab: Arc<Gazetteer>,
    events_tx: EventSender,
    oneliner_locks: OnelinerLocks,
) {
    info!("Transcription worker started");
    let timeout = Duration::from_secs(settings.whisper.timeout_seconds);

    while let Some(job) = rx.recv().await {
        let _guard = BusyGuard::new(worker_busy.clone());
        let audio_path = job.audio_path;
        let meta_path = paths::recording_meta_path(&audio_path);

        // Idempotency gate: presence of `<stem>.json` means the
        // transcriber already finished this recording (single atomic
        // write at the end of the pipeline). Skip the job, freeing
        // Whisper for the next one.
        if meta_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 user_whisper: 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;
            }
        };

        // Run audio analysis (duration + volumedetect) and Whisper
        // concurrently against the remuxed tempfile. The volumedetect
        // pass is decode-bound on a CPU core (~100-500 ms for typical
        // dictation lengths); Whisper dominates the wallclock with a
        // GPU/network round-trip in seconds, so the analysis cost
        // disappears behind the join.
        //
        // `tokio::join!` (not `try_join!`) intentionally awaits both
        // futures even on error: a failed analysis must not throw
        // away a successful transcript, and a failed transcript must
        // drop any analysis values — without a meta sidecar the
        // page-load heal re-enqueues, and that retry must run the
        // analysis fresh.
        let (analysis_result, whisper_result) = tokio::join!(
            ffmpeg::analyze_audio(remuxed.path()),
            whisper::transcribe(
                &client,
                &settings.whisper.url,
                remuxed.path(),
                timeout,
                &user_whisper,
            ),
        );

        // Whisper is the gating outcome: without a transcript we don't
        // write any meta, regardless of whether the analysis succeeded.
        let text = match whisper_result {
            Ok(t) => t,
            Err(e) => {
                error!(audio = %audio_path.display(), error = %e, "whisper call failed");
                if e.is_transient() {
                    // Leave the recording as plain `.m4a` (no `.failed` suffix)
                    // so the page-load heal (`enqueue_pending_for_user`) picks
                    // it up on the next refresh. The natural retry trigger is
                    // a human navigating the UI — no scheduler, no sidecar.
                    info!(
                        audio = %audio_path.display(),
                        error = %e,
                        reason = "transient",
                        "recording left as .m4a, page-load heal will re-enqueue"
                    );
                } else {
                    mark_failed(&audio_path, &events_tx, &job.user_slug).await;
                }
                continue;
            }
        };

        // Log analysis failures here (so the helper stays pure and
        // unit-testable) before mapping the result onto the meta
        // fields. The mapping rules — finiteness gate, silent-file
        // handling, error propagation — live in
        // `analysis_to_meta_fields` with their own tests.
        if let Err(ref e) = analysis_result {
            warn!(
                audio = %audio_path.display(),
                error = %e,
                "audio analysis failed — meta will record null duration and loudness"
            );
        }
        let (duration_seconds, loudness) = analysis_to_meta_fields(analysis_result);

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

        // Single atomic write: assemble the recording metadata
        // (transcript outcome + duration) in memory, then materialise
        // it on disk in one rename. Either readers see the previous
        // state (no sidecar = pending) or the full new sidecar — never
        // a half-written one.
        let bytes_logged = text.len();
        let transcript = if text.trim().is_empty() {
            Transcript::Silent
        } else {
            Transcript::Content { text }
        };
        let meta = RecordingMeta {
            transcript,
            duration_seconds,
            loudness,
        };
        if let Err(e) = paths::write_recording_meta_atomic(&audio_path, &meta).await {
            error!(meta = %meta_path.display(), error = %e, "writing recording meta failed");
            continue;
        }

        info!(
            audio = %audio_path.display(),
            bytes = bytes_logged,
            duration = ?duration_seconds,
            gain_db = ?meta.loudness.as_ref().map(|l| l.gain_db),
            "Recording meta 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,
                &settings,
                &vocab,
                &events_tx,
                &oneliner_locks,
            )
            .await;
        }
    }

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

/// Map an [`ffmpeg::analyze_audio`] outcome onto the two
/// [`RecordingMeta`] fields it can populate.
///
/// Pure: no logging, no I/O, no Tokio. The caller is expected to log
/// failures (so the test fixture for this fn does not have to fight
/// the global tracing subscriber).
///
/// - `Ok(a)` with both `mean_db` and `max_db` finite → both fields
///   populated, `gain_db` from [`loudness::compute_gain_db`].
/// - `Ok(a)` with either non-finite (silent / zero-sample input) →
///   duration kept, loudness dropped: `serde_json` refuses non-finite
///   floats and persisting `+Inf` gain would loop the worker.
/// - `Err(_)` → both fields `None`. ffmpeg failed; the recording will
///   show up in the UI without duration or replay-gain, but the
///   transcript still gets persisted.
fn analysis_to_meta_fields(
    analysis_result: Result<ffmpeg::AudioAnalysis, ffmpeg::FfmpegError>,
) -> (Option<u32>, Option<Loudness>) {
    match analysis_result {
        Ok(a) if a.mean_db.is_finite() && a.max_db.is_finite() => {
            let gain_db = loudness::compute_gain_db(a.mean_db, a.max_db);
            (
                Some(a.duration_seconds),
                Some(Loudness {
                    mean_db: a.mean_db,
                    max_db: a.max_db,
                    gain_db,
                }),
            )
        }
        Ok(a) => (Some(a.duration_seconds), None),
        Err(_) => (None, None),
    }
}

/// 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,
            );
        }
    }
}

/// Regenerate `case_dir/oneliner.json` from **all** `Content` 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.
///
/// Three terminal shapes:
/// - At least one `Content` transcript → call the LLM; persist `Ready`,
///   `Empty`, or `Error` based on the response.
/// - No `Content` but all other recordings are `Silent` (nothing more
///   coming from the transcriber) → persist `Empty` directly and skip
///   the LLM call. Without this, silent-only cases would leave the UI
///   stuck on "Generating" forever.
/// - Still-`Pending` recordings present → no-op; the next transcript
///   write re-enters this function with more information.
pub async fn update_oneliner(
    case_dir: &Path,
    user_slug: &str,
    client: &reqwest::Client,
    settings: &Settings,
    vocab: &Gazetteer,
    events_tx: &EventSender,
    locks: &OnelinerLocks,
) {
    // Early-Latch: a manual override is sticky. Skip the LLM call
    // entirely so the doctor's edit cannot be overwritten and Ollama
    // does not waste a 60s round-trip on a result that will be dropped.
    // Read-only check, no lock needed.
    let (existing, _) = paths::read_oneliner_state(case_dir).await;
    if matches!(existing, Some(OnelinerState::Manual { .. })) {
        info!(
            user = %user_slug,
            case = %case_dir.display(),
            "oneliner manual override is sticky — skip auto-regen"
        );
        return;
    }

    let transcript = match all_transcripts_joined(case_dir).await {
        Some(t) => t,
        None => {
            // No `Content` transcripts in the case. Two distinct reasons:
            //   a) Some recording is still `Pending` — transcriber will
            //      trigger us again when the last one lands. Skip for now.
            //   b) Every recording is `Silent` — transcriber is done and
            //      will never produce text. Without a persisted state
            //      the UI would stick on `Generating` forever (bug that
            //      motivated this code path). Persist `Empty` so the UI
            //      settles and recovery treats it as terminal.
            if !has_pending_recordings(case_dir).await {
                let generated_at = OffsetDateTime::now_utc()
                    .format(&Rfc3339)
                    .unwrap_or_default();
                let state = OnelinerState::Empty { generated_at };
                // Re-check under lock: a PUT may have arrived between
                // the early-latch read and now (no LLM call here, but
                // the silent-empty resolution still races with PUTs).
                let _guard = locks.lock_for(case_dir).await;
                let (current, _) = paths::read_oneliner_state(case_dir).await;
                if matches!(current, Some(OnelinerState::Manual { .. })) {
                    info!(
                        user = %user_slug,
                        case = %case_dir.display(),
                        "manual override appeared during silent-empty resolution — drop"
                    );
                    return;
                }
                if let Err(e) = paths::write_oneliner_state(case_dir, &state).await {
                    error!(
                        case = %case_dir.display(),
                        error = %e,
                        "writing oneliner.json (all-silent empty) failed"
                    );
                    return;
                }
                info!(
                    user = %user_slug,
                    case = %case_dir.display(),
                    "oneliner empty — all recordings silent (no content transcripts)"
                );
                events::emit(
                    events_tx,
                    user_slug,
                    events::case_id_of(case_dir),
                    CaseEventKind::OnelinerUpdated,
                );
            }
            return;
        }
    };

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

    // LLM call runs WITHOUT the lock — holding the per-case mutex for
    // 60s would block any concurrent PUT for the same case. The
    // post-LLM re-check (below) catches manual overrides that arrived
    // during the call.
    let state = match ollama::generate_oneliner(
        client,
        &settings.ollama.url,
        &settings.ollama.model,
        settings.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 }
        }
    };

    // Re-check under the per-case lock: if a manual override landed
    // while the LLM was thinking, drop the auto result so the doctor's
    // edit always wins. The PUT handler holds the same lock around its
    // write, so the read-and-write below is atomic vs. that path.
    let _guard = locks.lock_for(case_dir).await;
    let (current, _) = paths::read_oneliner_state(case_dir).await;
    if matches!(current, Some(OnelinerState::Manual { .. })) {
        info!(
            user = %user_slug,
            case = %case_dir.display(),
            "manual override appeared during regen — dropping LLM result"
        );
        return;
    }

    if let Err(e) = paths::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,
    );
}

/// Return true if `case_dir` contains at least one `.m4a` recording
/// whose transcript is still `Pending` (no sidecar file yet). 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.
/// `Silent` recordings count as finished, not pending — that's the
/// whole point of the three-state model: the transcriber already ran,
/// it just didn't produce text.
/// 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 paths::read_transcript_state(&path).await.is_pending() {
            return true;
        }
    }
    false
}

/// Join every `Content` transcript in `case_dir` chronologically with
/// `\n\n---\n\n` separators. `Pending` (no sidecar) and `Silent` (empty
/// sidecar) recordings are skipped. Returns None if no `Content`
/// transcript exists — the caller decides whether that means "wait for
/// more transcripts" or "this case is terminal silent" by consulting
/// [`has_pending_recordings`] afterwards.
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 m4as: Vec<std::path::PathBuf> = Vec::new();
    while let Ok(Some(entry)) = entries.next_entry().await {
        let path = entry.path();
        if path.extension().and_then(|s| s.to_str()) == Some("m4a") {
            m4as.push(path);
        }
    }
    m4as.sort();

    let mut parts: Vec<String> = Vec::with_capacity(m4as.len());
    for path in m4as {
        if let TranscriptState::Content { text } = paths::read_transcript_state(&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;

    /// Helper: seed an m4a plus an optional metadata sidecar, matching
    /// how the worker lays files out on disk. `all_transcripts_joined`
    /// iterates `.m4a` files and reads the sidecar through the shared
    /// reader, so bare meta files without m4as would be ignored.
    ///
    /// `transcript`: `None` → no sidecar (Pending); `Some(s)` with
    /// whitespace-only `s` → `Transcript::Silent`; otherwise
    /// `Transcript::Content`.
    async fn seed(dir: &Path, stem: &str, transcript: Option<&str>) {
        tokio::fs::write(dir.join(format!("{stem}.m4a")), b"audio")
            .await
            .unwrap();
        if let Some(text) = transcript {
            let variant = if text.trim().is_empty() {
                Transcript::Silent
            } else {
                Transcript::Content {
                    text: text.to_owned(),
                }
            };
            paths::write_recording_meta_sync(dir, stem, variant, None);
        }
    }

    /// 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();
        seed(
            dir.path(),
            "2026-04-16T10-05-00Z",
            Some("Korrektur. Das Mittel heißt Vomex."),
        )
        .await;
        seed(
            dir.path(),
            "2026-04-16T10-00-00Z",
            Some("Jetzt weiß ich es wieder. Es heißt Womax."),
        )
        .await;

        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();
        seed(dir.path(), "2026-04-16T10-00-00Z", Some("")).await;
        seed(
            dir.path(),
            "2026-04-16T10-05-00Z",
            Some("non-empty content"),
        )
        .await;

        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();
        seed(dir.path(), "2026-04-16T10-00-00Z", Some("   \n  ")).await;
        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();
        seed(dir.path(), "2026-04-16T10-00-00Z", Some("text")).await;
        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();
        seed(dir.path(), "2026-04-16T10-00-00Z", Some("text1")).await;
        seed(dir.path(), "2026-04-16T10-05-00Z", Some("text2")).await;
        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);
    }

    /// Successful analysis with finite mean and max — both meta fields
    /// are populated and `gain_db` matches the loudness formula
    /// (mean=-22, max=-15 → target-limited at +6 dB; same case as
    /// `loudness::tests::target_limits_gain`).
    #[test]
    fn analysis_to_meta_fields_finite_populates_both() {
        let analysis = ffmpeg::AudioAnalysis {
            duration_seconds: 30,
            mean_db: -22.0,
            max_db: -15.0,
        };
        let (duration, loudness) = analysis_to_meta_fields(Ok(analysis));
        assert_eq!(duration, Some(30));
        assert_eq!(
            loudness,
            Some(Loudness {
                mean_db: -22.0,
                max_db: -15.0,
                gain_db: 6.0,
            })
        );
    }

    /// Silent file: volumedetect returns `-inf dB` for both. Duration
    /// is still valid (the decode pass measured it), but loudness must
    /// be `None` — `serde_json` would refuse to serialize an
    /// `Infinity` and the worker would loop forever on retry.
    #[test]
    fn analysis_to_meta_fields_silent_drops_loudness() {
        let analysis = ffmpeg::AudioAnalysis {
            duration_seconds: 5,
            mean_db: f64::NEG_INFINITY,
            max_db: f64::NEG_INFINITY,
        };
        let (duration, loudness) = analysis_to_meta_fields(Ok(analysis));
        assert_eq!(duration, Some(5));
        assert_eq!(loudness, None);
    }

    /// Mixed case: only one of the two values is non-finite. The gate
    /// is `&&` so a single non-finite value still drops loudness — we
    /// never feed a partially-defined measurement into the gain
    /// formula.
    #[test]
    fn analysis_to_meta_fields_partial_inf_drops_loudness() {
        let analysis = ffmpeg::AudioAnalysis {
            duration_seconds: 10,
            mean_db: -27.3,
            max_db: f64::NEG_INFINITY,
        };
        let (duration, loudness) = analysis_to_meta_fields(Ok(analysis));
        assert_eq!(duration, Some(10));
        assert_eq!(loudness, None);
    }

    /// ffmpeg failed: both fields are `None`, the transcript still
    /// gets persisted by the caller. The worker logs the error before
    /// calling this helper, so the helper itself stays log-free.
    #[test]
    fn analysis_to_meta_fields_err_yields_none_none() {
        let err = ffmpeg::FfmpegError::Parse("synthetic test error".into());
        let (duration, loudness) = analysis_to_meta_fields(Err(err));
        assert_eq!(duration, None);
        assert_eq!(loudness, None);
    }
}