doctate/server/src/main.rs

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::{AsrBackend, Config};
use doctate_server::events;
use doctate_server::gazetteer::{Gazetteer, SpellbookDict};
use doctate_server::settings::Settings;
use doctate_server::transcribe;

#[tokio::main]
async fn main() {
    dotenvy::dotenv().ok();
    let config = Arc::new(Config::from_env());

    let settings_path = std::env::var("SETTINGS_FILE").unwrap_or_else(|_| "settings.toml".into());
    let settings = Arc::new(Settings::load_or_default(&settings_path));

    // 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.
    let backends = doctate_server::analyze::backend::backends();
    let available = doctate_server::analyze::backend::available_backends();
    info!(
        total = backends.len(),
        available = available.len(),
        "LLM backends loaded"
    );
    for b in backends {
        info!(
            id = %b.id,
            label = %b.label,
            model = %b.model_id,
            available = b.is_available(),
            requires_api_key = b.requires_api_key,
            "LLM backend"
        );
    }
    if available.is_empty() {
        warn!(
            "No LLM backend available — analyze button will be hidden. \
             Set IONOS_API_KEY (or another provider's env var) to enable."
        );
    }

    info!(backend = ?config.asr_backend, "ASR backend");
    if config.asr_backend == AsrBackend::Canary {
        warn!(
            "ASR_BACKEND=canary — Whisper container on the GPU host \
             must be stopped (12 GB VRAM budget shared with Ollama; \
             see docs/canary.md §2.6)."
        );
    }

    // 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 `/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(),
        settings.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 settings = settings.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, &settings, &vocab, &events, &locks,
            )
            .await;
        });
    }

    // Analyze pipeline: channel + worker. Boot-scan is observation-only —
    // the per-render `try_enqueue_all_for_user` covers auto-trigger via
    // the typed `CaseAnalysisState` machine.
    let (analyze_tx, analyze_rx) = analyze::channel();
    let analyze_busy: doctate_server::WorkerBusy =
        std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let analyze_in_flight = analyze::InFlight::new();
    tokio::spawn(analyze::worker::run(
        analyze_rx,
        http_client.clone(),
        analyze_busy.clone(),
        analyze_in_flight.clone(),
        vocab.clone(),
        events_tx.clone(),
    ));
    let server_boot_time = std::time::SystemTime::now();
    {
        let data_path = config.data_path.clone();
        let idle_threshold = std::time::Duration::from_secs(config.analysis_idle_threshold_secs);
        let in_flight = analyze_in_flight.clone();
        tokio::spawn(async move {
            analyze::recovery::boot_scan_state(
                &data_path,
                idle_threshold,
                std::time::SystemTime::now(),
                server_boot_time,
                &in_flight,
            )
            .await;
        });
    }

    let state = AppState {
        config: config.clone(),
        settings: settings.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,
        boot_time: doctate_server::BootTime(server_boot_time),
        analyze_in_flight,
    };

    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);
    }
}