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

    // 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(),
    ));
    {
        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();
        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,
            )
            .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),
        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);
    }
}