Add analyze channel to AppState

Introduces a new channel for the analyze pipeline, including its sender
in the `AppState`.
Also updates the test router to create both transcribe and analyze
channels.

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<Config>,
     pub transcribe_tx: TranscribeSender,
+    pub analyze_tx: AnalyzeSender,
     pub session_store: SessionStore,
 }
 
@@ -37,20 +39,28 @@ impl FromRef<AppState> for TranscribeSender {
     }
 }
 
+impl FromRef<AppState> for AnalyzeSender {
+    fn from_ref(state: &AppState) -> Self {
+        state.analyze_tx.clone()
+    }
+}
+
 impl FromRef<AppState> for SessionStore {
     fn from_ref(state: &AppState) -> Self {
         state.session_store.clone()
     }
 }
 
-/// Test/simple entrypoint: jobs pushed into the transcribe channel are dropped
+/// Test/simple entrypoint: jobs pushed into either channel are dropped
-/// because the receiver is not retained. Use [`create_router_with_state`] from
+/// because the receivers are not retained. Use [`create_router_with_state`]
-/// `main.rs` where a real worker owns the receiver.
+/// from `main.rs` where real workers own the receivers.
 pub fn create_router(config: Arc<Config>) -> 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(),
     })
 }

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.
+    // Shared HTTP client for both downstream pipelines.
-    let (transcribe_tx, transcribe_rx) = transcribe::channel();
     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(),
     };