From e995cdd7c47fb64bb94d4266e6a3fb2ef746f667 Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 19 Apr 2026 15:14:22 +0200 Subject: [PATCH] refactor(server): bundle worker pipeline into PipelineState Group the four worker handles (analyze/transcribe x busy/tx) into a new PipelineState struct with a FromRef impl. Convert heal_orphans_if_idle from a free function with 6 parameters into a method on PipelineState. The two web handlers (handle_my_cases, handle_case_detail) now take one State instead of four separate State extractors. The original FromRef impls for AnalyzeBusy/Sender and TranscribeBusy/ Sender remain intact so handlers that need only one of them keep their narrower signature. --- server/src/lib.rs | 25 ++++++++++++++ server/src/routes/user_web.rs | 64 +++++++++++------------------------ 2 files changed, 44 insertions(+), 45 deletions(-) diff --git a/server/src/lib.rs b/server/src/lib.rs index d597f13..d7ffa91 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -51,6 +51,20 @@ impl Drop for BusyGuard { } } +/// Bundle of the four worker-pipeline handles (analyze + transcribe, +/// each with a busy-flag and a job sender). Extracted so handlers that +/// need the whole pipeline (self-heal, status probes) can take a single +/// `State` instead of four separate `State(...)` params. +/// The individual `FromRef` impls below remain valid, so +/// handlers that only need one sender keep their narrower signature. +#[derive(Clone)] +pub struct PipelineState { + pub analyze_busy: AnalyzeBusy, + pub analyze_tx: AnalyzeSender, + pub transcribe_busy: TranscribeBusy, + pub transcribe_tx: TranscribeSender, +} + /// Shared application state. Clonable — all fields are cheap to clone /// (`Arc` and `mpsc::Sender`). #[derive(Clone)] @@ -99,6 +113,17 @@ impl FromRef for TranscribeBusy { } } +impl FromRef for PipelineState { + fn from_ref(state: &AppState) -> Self { + PipelineState { + analyze_busy: state.analyze_busy.clone(), + analyze_tx: state.analyze_tx.clone(), + transcribe_busy: state.transcribe_busy.clone(), + transcribe_tx: state.transcribe_tx.clone(), + } + } +} + /// Test/simple entrypoint: jobs pushed into either channel are dropped /// because the receivers are not retained. Use [`create_router_with_state`] /// from `main.rs` where real workers own the receivers. diff --git a/server/src/routes/user_web.rs b/server/src/routes/user_web.rs index e136674..e7d7336 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -9,14 +9,14 @@ use tracing::{info, warn}; use time::{format_description::well_known::Rfc3339, Date, OffsetDateTime}; -use crate::analyze::{recovery as analyze_recovery, AnalyzeSender, ANALYSIS_INPUT_FILE, DOCUMENT_FILE}; +use crate::analyze::{recovery as analyze_recovery, ANALYSIS_INPUT_FILE, DOCUMENT_FILE}; use crate::auth::AuthenticatedWebUser; use crate::case_id::{CaseId, CaseIdPath}; use crate::config::Config; use crate::error::AppError; use crate::routes::web::{scan_recordings, RecordingView}; -use crate::transcribe::{recovery as transcribe_recovery, TranscribeSender}; -use crate::{AnalyzeBusy, TranscribeBusy}; +use crate::transcribe::recovery as transcribe_recovery; +use crate::PipelineState; struct UserCaseView { case_id: String, @@ -201,42 +201,27 @@ pub(crate) async fn any_document_exists(case_dir: &Path) -> bool { /// Self-heal: if a worker is idle but orphans exist on disk for this user, /// re-enqueue them. Page-load is the trigger; no cron, no periodic task. /// Runs for both pipelines so a page-load on either view cleans both. -async fn heal_orphans_if_idle( - user_root: &Path, - slug: &str, - analyze_busy: &AnalyzeBusy, - analyze_tx: &AnalyzeSender, - transcribe_busy: &TranscribeBusy, - transcribe_tx: &TranscribeSender, -) { - if !analyze_busy.0.load(Ordering::Acquire) { - analyze_recovery::enqueue_pending_for_user(user_root, analyze_tx).await; - } - if !transcribe_busy.0.load(Ordering::Acquire) { - transcribe_recovery::enqueue_pending_for_user(user_root, slug, transcribe_tx).await; +impl PipelineState { + pub async fn heal_orphans_if_idle(&self, user_root: &Path, slug: &str) { + if !self.analyze_busy.0.load(Ordering::Acquire) { + analyze_recovery::enqueue_pending_for_user(user_root, &self.analyze_tx).await; + } + if !self.transcribe_busy.0.load(Ordering::Acquire) { + transcribe_recovery::enqueue_pending_for_user(user_root, slug, &self.transcribe_tx) + .await; + } } } pub async fn handle_my_cases( user: AuthenticatedWebUser, State(config): State>, - State(analyze_busy): State, - State(analyze_tx): State, - State(transcribe_busy): State, - State(transcribe_tx): State, + State(pipeline): State, ) -> Result, AppError> { let user_root = config.data_path.join(&user.slug); - heal_orphans_if_idle( - &user_root, - &user.slug, - &analyze_busy, - &analyze_tx, - &transcribe_busy, - &transcribe_tx, - ) - .await; + pipeline.heal_orphans_if_idle(&user_root, &user.slug).await; - let busy = analyze_busy.0.load(Ordering::Acquire); + let busy = pipeline.analyze_busy.0.load(Ordering::Acquire); let cases = scan_user_cases(&config, &user.slug, busy).await; let total = cases.len(); let groups = group_by_utc_date(cases); @@ -260,23 +245,12 @@ pub async fn handle_my_cases( pub async fn handle_case_detail( user: AuthenticatedWebUser, State(config): State>, - State(analyze_busy): State, - State(analyze_tx): State, - State(transcribe_busy): State, - State(transcribe_tx): State, + State(pipeline): State, CaseIdPath(case_id): CaseIdPath, ) -> Result, AppError> { // IDOR guard: case_dir must live under the session's user_slug. let user_root = config.data_path.join(&user.slug); - heal_orphans_if_idle( - &user_root, - &user.slug, - &analyze_busy, - &analyze_tx, - &transcribe_busy, - &transcribe_tx, - ) - .await; + pipeline.heal_orphans_if_idle(&user_root, &user.slug).await; let case_dir = locate_case_or_404( &user_root, @@ -295,8 +269,8 @@ pub async fn handle_case_detail( .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()); - let a_busy = analyze_busy.0.load(Ordering::Acquire); - let t_busy = transcribe_busy.0.load(Ordering::Acquire); + let a_busy = pipeline.analyze_busy.0.load(Ordering::Acquire); + let t_busy = pipeline.transcribe_busy.0.load(Ordering::Acquire); let flags = compute_flags(&case_dir, &recordings, config.llm_configured(), a_busy).await; let case_id_short = case_id_str.chars().take(8).collect(); let is_admin = user.is_admin();