//! `spectra serve`: the daemon (spec §7.1, §21.1). It holds the ledger,

//! serves Tuner (and, as they land, the API and Radar) on the Unix socket

//! and on a loopback port, and writes a checkpoint when it stops.

use super::{ui, Ctx};

use anyhow::{bail, Context, Result};

use spectra_core::daemon::Daemon;

use spectra_mcp::client::Client;

use spectra_mcp::http::{AppState, PeerInfo};

use std::net::SocketAddr;

use std::sync::Arc;

pub async fn serve(ctx: &Ctx, bind: &str, insecure: bool, open: bool) -> Result<i32> {

    let addr: SocketAddr = bind.parse().with_context(|| format!("--bind {bind} is not an address and port"))?;

    if !addr.ip().is_loopback() && !insecure {

        bail!("Spectra serves loopback only unless it has TLS (spec §21.1); pass --insecure to bind {addr} in plain HTTP anyway");

    }

    let socket = ctx.socket();

    if socket.exists() {

        if Client::alive(&socket).await {

            bail!("Spectra is already serving {} (its socket answers)", ctx.data.display());

        }

        std::fs::remove_file(&socket).ok();

    }

    let daemon = Daemon::start(&ctx.data).await?;

    let unix = tokio::net::UnixListener::bind(&socket).with_context(|| format!("binding {}", socket.display()))?;

    #[cfg(unix)]

    {

        use std::os::unix::fs::PermissionsExt;

        std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o660))?;

    }

    let tcp = tokio::net::TcpListener::bind(addr).await.with_context(|| format!("binding {addr}"))?;

    let port = tcp.local_addr()?.port();

    let origins = vec![format!("http://127.0.0.1:{port}"), format!("http://localhost:{port}"), format!("http://{addr}")];

    let state = AppState { daemon: daemon.clone(), origins: Arc::new(origins) };

    let app = spectra_mcp::http::router(state.clone()).merge(radar::router(state));

    let (head, _) = daemon.ledger.head().await;

    let agents = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM agents WHERE disabled_ms IS NULL")

        .fetch_one(daemon.ledger.pool())

        .await

        .unwrap_or(0);

    println!("{} {}", ui::done("Serving"), ui::subject("Spectra"));

    println!("{}", ui::kv("socket", ui::path(socket.display())));

    println!("{}", ui::kv("http", format!("http://127.0.0.1:{port}")));

    println!("{}", ui::kv("ledger", format!("{head} events{}{agents} agent(s)", ui::sep())));

    println!("  {} {}", ui::note(), ui::dim("agents connect with: spectra mcp --agent NAME"));

    println!("  {} {}", ui::note(), ui::dim("open Radar with: spectra open"));

    if open {

        let ctx2 = Ctx { data: ctx.data.clone() };

        tokio::spawn(async move {

            // Once the socket answers, sign the link as the owner and open it.

            tokio::time::sleep(std::time::Duration::from_millis(200)).await;

            if let Err(e) = super::open::open(&ctx2, None, true).await {

                eprintln!("{} {e:#}", ui::caution());

            }

        });

    }

    let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);

    let stop = |mut rx: tokio::sync::watch::Receiver<bool>| async move {

        let _ = rx.wait_for(|s| *s).await;

    };

    let unix_srv =

        axum::serve(unix, app.clone().into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx.clone()));

    let tcp_srv = axum::serve(tcp, app.into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx));

    let signal = async move {

        shutdown_signal().await;

        let _ = stop_tx.send(true);

    };

    // After the signal, open connections get five seconds to finish: a

    // `subscriptions/listen` stream never does by itself.

    let (stopping_tx, mut stopping) = tokio::sync::watch::channel(false);

    tokio::spawn(async move {

        signal.await;

        let _ = stopping_tx.send(true);

    });

    let servers = async { tokio::join!(unix_srv, tcp_srv) };

    tokio::select! {

        (u, t) = servers => {

            u?;

            t?;

        }

        _ = async {

            let _ = stopping.wait_for(|s| *s).await;

            tokio::time::sleep(std::time::Duration::from_secs(5)).await;

        } => {

            tracing::info!("connections still open five seconds after the signal; stopping anyway");

        }

    }

    daemon.shutdown().await;

    std::fs::remove_file(&socket).ok();

    println!("{} {}", ui::done("Stopped"), ui::dim(format!("ledger checkpointed at {} events", daemon.ledger.head().await.0)));

    Ok(0)

}

async fn shutdown_signal() {

    #[cfg(unix)]

    {

        let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("a SIGTERM handler");

        tokio::select! {

            _ = tokio::signal::ctrl_c() => {}

            _ = term.recv() => {}

        }

    }

    #[cfg(not(unix))]

    let _ = tokio::signal::ctrl_c().await;

}