diff --git a/server/src/lib.rs b/server/src/lib.rs index 5dc0203..a07d03c 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -12,6 +12,7 @@ use std::sync::Arc; use axum::extract::FromRef; use axum::Router; +use analyze::AnalyzeSender; use config::Config; use transcribe::TranscribeSender; use web_session::SessionStore; @@ -22,6 +23,7 @@ use web_session::SessionStore; pub struct AppState { pub config: Arc, pub transcribe_tx: TranscribeSender, + pub analyze_tx: AnalyzeSender, pub session_store: SessionStore, } @@ -37,20 +39,28 @@ impl FromRef for TranscribeSender { } } +impl FromRef for AnalyzeSender { + fn from_ref(state: &AppState) -> Self { + state.analyze_tx.clone() + } +} + impl FromRef for SessionStore { fn from_ref(state: &AppState) -> Self { state.session_store.clone() } } -/// Test/simple entrypoint: jobs pushed into the transcribe channel are dropped -/// because the receiver is not retained. Use [`create_router_with_state`] from -/// `main.rs` where a real worker owns the receiver. +/// 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. pub fn create_router(config: Arc) -> Router { - let (tx, _rx) = transcribe::channel(); + let (transcribe_tx, _transcribe_rx) = transcribe::channel(); + let (analyze_tx, _analyze_rx) = analyze::channel(); create_router_with_state(AppState { config, - transcribe_tx: tx, + transcribe_tx, + analyze_tx, session_store: web_session::new_store(), }) } diff --git a/server/src/main.rs b/server/src/main.rs index 057e11d..8c40e5b 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -8,6 +8,7 @@ use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; use tracing_subscriber::EnvFilter; +use doctate_server::analyze; use doctate_server::config::Config; use doctate_server::transcribe; use doctate_server::AppState; @@ -39,18 +40,18 @@ async fn main() { ) .init(); - // Transcription pipeline: shared HTTP client + worker task. - let (transcribe_tx, transcribe_rx) = transcribe::channel(); + // Shared HTTP client for both downstream pipelines. let http_client = reqwest::Client::builder() .build() .expect("reqwest client build failed"); + + // Transcription pipeline: channel + worker + recovery scan. + let (transcribe_tx, transcribe_rx) = transcribe::channel(); tokio::spawn(transcribe::worker::run( transcribe_rx, config.clone(), - http_client, + http_client.clone(), )); - - // Re-enqueue pending recordings left over from a previous run. { let tx = transcribe_tx.clone(); let data_path = config.data_path.clone(); @@ -59,9 +60,25 @@ async fn main() { }); } + // Analyze pipeline: channel + worker + recovery scan. + let (analyze_tx, analyze_rx) = analyze::channel(); + tokio::spawn(analyze::worker::run( + analyze_rx, + config.clone(), + http_client.clone(), + )); + { + let tx = analyze_tx.clone(); + let data_path = config.data_path.clone(); + tokio::spawn(async move { + analyze::recovery::scan_and_enqueue(&data_path, &tx).await; + }); + } + let state = AppState { config: config.clone(), transcribe_tx, + analyze_tx, session_store: doctate_server::web_session::new_store(), };