From 13375b8297754c0af5b28cf8801f5db80e2aa4da Mon Sep 17 00:00:00 2001 From: Brummel Date: Mon, 13 Apr 2026 16:47:38 +0200 Subject: [PATCH] feat: Add recovery scan for pending recordings This commit introduces a new module `transcribe::recovery` and a function `scan_and_enqueue`. This function is called at startup to find audio files that have not yet been transcribed and re-enqueues them for processing. This ensures that recordings are not lost if the server restarts before transcription is complete. A new integration test `recovery_enqueues_only_pending_recordings` has been added to verify the functionality. --- server/src/main.rs | 9 ++++ server/src/transcribe/mod.rs | 1 + server/src/transcribe/recovery.rs | 73 +++++++++++++++++++++++++++++++ server/tests/transcribe_test.rs | 39 +++++++++++++++++ 4 files changed, 122 insertions(+) create mode 100644 server/src/transcribe/recovery.rs diff --git a/server/src/main.rs b/server/src/main.rs index e151cef..5708e7a 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -50,6 +50,15 @@ async fn main() { http_client, )); + // Re-enqueue pending recordings left over from a previous run. + { + let tx = transcribe_tx.clone(); + let data_path = config.data_path.clone(); + tokio::spawn(async move { + transcribe::recovery::scan_and_enqueue(&data_path, &tx).await; + }); + } + let state = AppState { config: config.clone(), transcribe_tx, diff --git a/server/src/transcribe/mod.rs b/server/src/transcribe/mod.rs index e4de5dc..45ad2e5 100644 --- a/server/src/transcribe/mod.rs +++ b/server/src/transcribe/mod.rs @@ -3,6 +3,7 @@ use std::path::PathBuf; use tokio::sync::mpsc; pub mod ffmpeg; +pub mod recovery; pub mod whisper; pub mod worker; diff --git a/server/src/transcribe/recovery.rs b/server/src/transcribe/recovery.rs new file mode 100644 index 0000000..c9a61e0 --- /dev/null +++ b/server/src/transcribe/recovery.rs @@ -0,0 +1,73 @@ +use std::path::{Path, PathBuf}; + +use tracing::{info, warn}; + +use super::{TranscribeJob, TranscribeSender}; + +/// Walk `data_path/*/open/*/*.m4a` and enqueue every recording that does not +/// yet have a `.transcript.txt` sibling. +/// +/// Intended to run once at startup (spawn as a task — sends may back-pressure +/// if many recordings are pending and the queue fills). +pub async fn scan_and_enqueue(data_path: &Path, tx: &TranscribeSender) { + let pending = collect_pending(data_path).await; + let total = pending.len(); + info!(count = total, "Recovery scan found pending recordings"); + + for audio_path in pending { + if tx + .send(TranscribeJob { + audio_path: audio_path.clone(), + }) + .await + .is_err() + { + warn!(file = %audio_path.display(), "Recovery send failed (worker gone)"); + break; + } + } +} + +async fn collect_pending(data_path: &Path) -> Vec { + let mut out = Vec::new(); + + let mut users = match tokio::fs::read_dir(data_path).await { + Ok(r) => r, + Err(_) => return out, + }; + + while let Ok(Some(user_entry)) = users.next_entry().await { + let open_dir = user_entry.path().join("open"); + let mut cases = match tokio::fs::read_dir(&open_dir).await { + Ok(r) => r, + Err(_) => continue, + }; + + while let Ok(Some(case_entry)) = cases.next_entry().await { + let case_dir = case_entry.path(); + if !case_dir.is_dir() { + continue; + } + + let mut files = match tokio::fs::read_dir(&case_dir).await { + Ok(r) => r, + Err(_) => continue, + }; + + while let Ok(Some(file)) = files.next_entry().await { + let path = file.path(); + if path.extension().and_then(|s| s.to_str()) != Some("m4a") { + continue; + } + let transcript = path.with_extension("transcript.txt"); + if !transcript.exists() { + out.push(path); + } + } + } + } + + // Oldest first (filenames are ISO timestamps). + out.sort(); + out +} diff --git a/server/tests/transcribe_test.rs b/server/tests/transcribe_test.rs index 1988709..ee63bf6 100644 --- a/server/tests/transcribe_test.rs +++ b/server/tests/transcribe_test.rs @@ -1,7 +1,9 @@ use std::path::PathBuf; use std::time::Duration; +use doctate_server::transcribe; use doctate_server::transcribe::ffmpeg::remux_faststart; +use doctate_server::transcribe::recovery::scan_and_enqueue; use doctate_server::transcribe::whisper::{transcribe, WhisperError}; use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -116,3 +118,40 @@ async fn whisper_client_times_out() { assert!(matches!(err, WhisperError::Http(_)), "expected Http error, got {err:?}"); } + +#[tokio::test] +async fn recovery_enqueues_only_pending_recordings() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + + // User 1, open case with two .m4a — one pending, one already transcribed. + let case_a = root.join("dr_a/open/aaaa"); + std::fs::create_dir_all(&case_a).unwrap(); + std::fs::write(case_a.join("2026-04-10T10-00-00Z.m4a"), b"x").unwrap(); + std::fs::write(case_a.join("2026-04-11T10-00-00Z.m4a"), b"x").unwrap(); + std::fs::write(case_a.join("2026-04-11T10-00-00Z.transcript.txt"), b"done").unwrap(); + + // User 2, open case with one pending .m4a. + let case_b = root.join("dr_b/open/bbbb"); + std::fs::create_dir_all(&case_b).unwrap(); + std::fs::write(case_b.join("2026-04-12T10-00-00Z.m4a"), b"x").unwrap(); + + // done/ cases must be ignored (not re-transcribed). + let case_c = root.join("dr_a/done/cccc"); + std::fs::create_dir_all(&case_c).unwrap(); + std::fs::write(case_c.join("2026-04-09T10-00-00Z.m4a"), b"x").unwrap(); + + let (tx, mut rx) = transcribe::channel(); + scan_and_enqueue(root, &tx).await; + drop(tx); // close channel so the loop below terminates + + let mut received = Vec::new(); + while let Some(job) = rx.recv().await { + received.push(job.audio_path); + } + + assert_eq!(received.len(), 2, "got: {received:?}"); + // Oldest first: + assert!(received[0].ends_with("2026-04-10T10-00-00Z.m4a")); + assert!(received[1].ends_with("2026-04-12T10-00-00Z.m4a")); +}