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.
This commit is contained in:
2026-04-14 11:02:37 +02:00
parent 0ef9109876
commit fc8efc62e0
3 changed files with 23 additions and 7 deletions
+4 -1
View File
@@ -86,7 +86,10 @@ pub async fn handle_upload(
// Enqueue transcription. Failure here is non-fatal: on restart the recovery // Enqueue transcription. Failure here is non-fatal: on restart the recovery
// scan will re-enqueue any .m4a without a .transcript.txt sibling. // 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(()) => {} Ok(()) => {}
Err(TrySendError::Full(_)) => { Err(TrySendError::Full(_)) => {
warn!(file = %file_path.display(), "Transcription queue full; will be picked up on restart"); warn!(file = %file_path.display(), "Transcription queue full; will be picked up on restart");
+3
View File
@@ -13,6 +13,9 @@ pub const QUEUE_DEPTH: usize = 64;
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct TranscribeJob { pub struct TranscribeJob {
pub audio_path: PathBuf, 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<TranscribeJob>; pub type TranscribeSender = mpsc::Sender<TranscribeJob>;
+16 -6
View File
@@ -14,10 +14,11 @@ pub async fn scan_and_enqueue(data_path: &Path, tx: &TranscribeSender) {
let total = pending.len(); let total = pending.len();
info!(count = total, "Recovery scan found pending recordings"); info!(count = total, "Recovery scan found pending recordings");
for audio_path in pending { for (user_slug, audio_path) in pending {
if tx if tx
.send(TranscribeJob { .send(TranscribeJob {
audio_path: audio_path.clone(), audio_path: audio_path.clone(),
user_slug: user_slug.clone(),
}) })
.await .await
.is_err() .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<PathBuf> { /// Walks `$data_path/<slug>/open/<case>/*.m4a` and returns `(slug, audio_path)`
let mut out = Vec::new(); /// 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 { let mut users = match tokio::fs::read_dir(data_path).await {
Ok(r) => r, Ok(r) => r,
@@ -37,6 +42,9 @@ async fn collect_pending(data_path: &Path) -> Vec<PathBuf> {
}; };
while let Ok(Some(user_entry)) = users.next_entry().await { 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 open_dir = user_entry.path().join("open");
let mut cases = match tokio::fs::read_dir(&open_dir).await { let mut cases = match tokio::fs::read_dir(&open_dir).await {
Ok(r) => r, Ok(r) => r,
@@ -61,13 +69,15 @@ async fn collect_pending(data_path: &Path) -> Vec<PathBuf> {
} }
let transcript = path.with_extension("transcript.txt"); let transcript = path.with_extension("transcript.txt");
if !transcript.exists() { if !transcript.exists() {
out.push(path); out.push((slug.clone(), path));
} }
} }
} }
} }
// Oldest first (filenames are ISO timestamps). // Oldest first (filenames are ISO timestamps). Path ordering puts same-user
out.sort(); // 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 out
} }