From fc8efc62e03ca3cb3caebb27fd65a983e8bf9a5a Mon Sep 17 00:00:00 2001 From: Brummel Date: Tue, 14 Apr 2026 11:02:37 +0200 Subject: [PATCH] Add user slug to transcription jobs The `TranscribeJob` struct now includes a `user_slug` field. This is necessary for the transcription worker to look up per-user settings such as hotwords and initial prompts. The `collect_pending` function has been updated to return user slugs along with audio paths, and the recovery scan now enqueues jobs with the correct user slug. --- server/src/routes/upload.rs | 5 ++++- server/src/transcribe/mod.rs | 3 +++ server/src/transcribe/recovery.rs | 22 ++++++++++++++++------ 3 files changed, 23 insertions(+), 7 deletions(-) diff --git a/server/src/routes/upload.rs b/server/src/routes/upload.rs index 760540a..389b457 100644 --- a/server/src/routes/upload.rs +++ b/server/src/routes/upload.rs @@ -86,7 +86,10 @@ pub async fn handle_upload( // Enqueue transcription. Failure here is non-fatal: on restart the recovery // scan will re-enqueue any .m4a without a .transcript.txt sibling. - match transcribe_tx.try_send(TranscribeJob { audio_path: file_path.clone() }) { + match transcribe_tx.try_send(TranscribeJob { + audio_path: file_path.clone(), + user_slug: user.slug.clone(), + }) { Ok(()) => {} Err(TrySendError::Full(_)) => { warn!(file = %file_path.display(), "Transcription queue full; will be picked up on restart"); diff --git a/server/src/transcribe/mod.rs b/server/src/transcribe/mod.rs index cdd2bc4..96b25f2 100644 --- a/server/src/transcribe/mod.rs +++ b/server/src/transcribe/mod.rs @@ -13,6 +13,9 @@ pub const QUEUE_DEPTH: usize = 64; #[derive(Debug, Clone)] pub struct TranscribeJob { pub audio_path: PathBuf, + /// Slug of the user who owns this recording. Used by the worker to look up + /// per-user transcription settings (hotwords, language, initial_prompt). + pub user_slug: String, } pub type TranscribeSender = mpsc::Sender; diff --git a/server/src/transcribe/recovery.rs b/server/src/transcribe/recovery.rs index c9a61e0..6d4b819 100644 --- a/server/src/transcribe/recovery.rs +++ b/server/src/transcribe/recovery.rs @@ -14,10 +14,11 @@ pub async fn scan_and_enqueue(data_path: &Path, tx: &TranscribeSender) { let total = pending.len(); info!(count = total, "Recovery scan found pending recordings"); - for audio_path in pending { + for (user_slug, audio_path) in pending { if tx .send(TranscribeJob { audio_path: audio_path.clone(), + user_slug: user_slug.clone(), }) .await .is_err() @@ -28,8 +29,12 @@ pub async fn scan_and_enqueue(data_path: &Path, tx: &TranscribeSender) { } } -async fn collect_pending(data_path: &Path) -> Vec { - let mut out = Vec::new(); +/// Walks `$data_path//open//*.m4a` and returns `(slug, audio_path)` +/// tuples for every recording without a `.transcript.txt` sibling. The slug +/// comes from the top-level user directory name — that's the only ownership +/// signal after a restart (auth context is gone). +async fn collect_pending(data_path: &Path) -> Vec<(String, PathBuf)> { + let mut out: Vec<(String, PathBuf)> = Vec::new(); let mut users = match tokio::fs::read_dir(data_path).await { Ok(r) => r, @@ -37,6 +42,9 @@ async fn collect_pending(data_path: &Path) -> Vec { }; while let Ok(Some(user_entry)) = users.next_entry().await { + let Some(slug) = user_entry.file_name().to_str().map(str::to_owned) else { + continue; + }; let open_dir = user_entry.path().join("open"); let mut cases = match tokio::fs::read_dir(&open_dir).await { Ok(r) => r, @@ -61,13 +69,15 @@ async fn collect_pending(data_path: &Path) -> Vec { } let transcript = path.with_extension("transcript.txt"); if !transcript.exists() { - out.push(path); + out.push((slug.clone(), path)); } } } } - // Oldest first (filenames are ISO timestamps). - out.sort(); + // Oldest first (filenames are ISO timestamps). Path ordering puts same-user + // entries together, which is fine — we still process strictly by timestamp + // within a user directory. + out.sort_by(|a, b| a.1.cmp(&b.1)); out }