use std::sync::Arc; use tokio::net::TcpListener; use tower_http::limit::RequestBodyLimitLayer; use tower_http::trace::TraceLayer; use tracing::{info, warn}; use tracing_subscriber::EnvFilter; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; use doctate_server::AppState; use doctate_server::analyze; use doctate_server::config::Config; use doctate_server::events; use doctate_server::gazetteer::{Gazetteer, SpellbookDict}; use doctate_server::transcribe; #[tokio::main] async fn main() { dotenvy::dotenv().ok(); let config = Arc::new(Config::from_env()); // Logging: stdout + daily rotating log files. let env_filter = EnvFilter::try_new(&config.log_level).unwrap_or_else(|_| EnvFilter::new("info")); let file_appender = tracing_appender::rolling::daily(&config.log_path, "recorder.log"); let (file_writer, _guard) = tracing_appender::non_blocking(file_appender); tracing_subscriber::registry() .with(env_filter) .with( tracing_subscriber::fmt::layer() .with_target(false) .with_writer(std::io::stdout), ) .with( tracing_subscriber::fmt::layer() .with_target(false) .with_ansi(false) .with_writer(file_writer), ) .init(); // Surface configuration gaps that silently disable features. if !config.llm_configured() { warn!( "LLM not configured — 'Fall abschließen' is disabled. \ Set LLM_URL, LLM_API_KEY and LLM_MODEL in .env to enable." ); } { let custom = std::env::var("LLM_SYSTEM_PROMPT") .map(|v| !v.is_empty()) .unwrap_or(false); info!( prompt_source = if custom { "env" } else { "default" }, prompt_chars = config.llm_system_prompt.chars().count(), "system prompt loaded" ); } // Shared HTTP client for both downstream pipelines. let http_client = reqwest::Client::builder() .build() .expect("reqwest client build failed"); // Gazetteer: post-AI terminology normalization filter applied to // every AI output (Whisper, oneliner, analyze). Missing / unreadable // vocab directory is non-fatal — the pipeline runs without // normalization. let base_gazetteer = match Gazetteer::load_dir(&config.vocab_dir).await { Ok(g) => { info!( dir = %config.vocab_dir.display(), entries = g.len(), "Gazetteer loaded" ); g } Err(e) => { warn!( dir = %config.vocab_dir.display(), error = %e, "Gazetteer not available; running without proper-name correction" ); Gazetteer::empty() } }; // Dict-veto: preserve tokens that are real German words, even if // they sit at DL ≤ 2 from a vocab entry (e.g. `Kaktus` stays // `Kaktus`, does not become `Lantus`). Missing / unreadable // hunspell files are non-fatal — pipeline runs without the veto. let vocab = Arc::new(match SpellbookDict::load(&config.hunspell_dict_path) { Ok(dict) => { info!( stem = %config.hunspell_dict_path.display(), "Hunspell dict loaded for gazetteer veto" ); base_gazetteer.with_dict(Box::new(dict)) } Err(e) => { warn!( stem = %config.hunspell_dict_path.display(), error = %e, "Hunspell dict unavailable; gazetteer running without dict veto" ); base_gazetteer } }); // Live-update bus: background workers and handlers publish case // events here; the SSE route at `/web/events` fans them out to // subscribed browsers. let events_tx = events::channel(); // Dedup flag shared by the startup oneliner-regen task and every // request-driven `heal_orphans_if_idle` spawn. One-at-a-time semantics // keep Ollama from seeing overlapping regen loops. let oneliner_heal_busy: doctate_server::WorkerBusy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); // Per-`case_dir` mutex registry. Shared by the PUT handler (manual // override write) and the transcribe worker (post-LLM re-check) // so a doctor's manual edit always wins over a concurrent // auto-regen, no matter the timing. let oneliner_locks = doctate_server::oneliner_locks::OnelinerLocks::new(); // Transcription pipeline: channel + worker + recovery scan. let (transcribe_tx, transcribe_rx) = transcribe::channel(); let transcribe_busy: doctate_server::WorkerBusy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); tokio::spawn(transcribe::worker::run( transcribe_rx, config.clone(), http_client.clone(), transcribe_busy.clone(), vocab.clone(), events_tx.clone(), oneliner_locks.clone(), )); { let tx = transcribe_tx.clone(); let data_path = config.data_path.clone(); let client = http_client.clone(); let cfg = config.clone(); let vocab = vocab.clone(); let events = events_tx.clone(); let heal_busy = oneliner_heal_busy.clone(); let locks = oneliner_locks.clone(); tokio::spawn(async move { transcribe::recovery::scan_and_enqueue(&data_path, &tx).await; // Then catch up oneliners for cases that crashed between // transcript-write and oneliner-write in a previous run. // Hold the shared heal-busy flag while running so a concurrent // page-load does not spawn a second parallel regen loop. let _guard = doctate_server::BusyGuard::new(heal_busy); transcribe::recovery::regenerate_missing_oneliners( &data_path, &client, &cfg, &vocab, &events, &locks, ) .await; }); } // Analyze pipeline: channel + worker + recovery scan. let (analyze_tx, analyze_rx) = analyze::channel(); let analyze_busy: doctate_server::WorkerBusy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); tokio::spawn(analyze::worker::run( analyze_rx, config.clone(), http_client.clone(), analyze_busy.clone(), vocab.clone(), events_tx.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(), magic_link_store: doctate_server::magic_link::new_store(), analyze_busy: doctate_server::AnalyzeBusy(analyze_busy), transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy), oneliner_heal_busy: doctate_server::OnelinerHealBusy(oneliner_heal_busy), oneliner_locks, events_tx, http_client, vocab, }; let addr = format!("0.0.0.0:{}", config.server_port); info!(port = config.server_port, "Server starting"); let app = doctate_server::create_router_with_state(state) .layer(TraceLayer::new_for_http()) .layer(RequestBodyLimitLayer::new(50 * 1024 * 1024)); // 50 MB let listener = match TcpListener::bind(&addr).await { Ok(l) => l, Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => { eprintln!( "Port {} is already in use. Is another doctate-server instance running?", config.server_port ); std::process::exit(1); } Err(e) => { eprintln!("Failed to bind to {addr}: {e}"); std::process::exit(1); } }; if let Err(e) = axum::serve(listener, app).await { eprintln!("Server runtime error: {e}"); std::process::exit(1); } }