Coverage / src/cli/serve.rs 82.7%

fn server_cfg(ctx: &Ctx, bind: Option<String>, tls: Option<String>) -> Result<ServerCfg> {

    let text = std::fs::read_to_string(ctx.data.join("spectra.toml")).unwrap_or_default();

    let t: toml::Table = text.parse().with_context(|| format!("reading {}", ctx.data.join("spectra.toml").display()))?;

    let server = t.get("server").and_then(|s| s.as_table()).cloned().unwrap_or_default();

    let (mut mode, mut files) = (String::from("auto"), None);

    match server.get("tls") {

        _ => {}

    if let Some(m) = tls {

        mode = m;

    }

    if !matches!(mode.as_str(), "auto" | "on" | "off") {

    }

    let names = match server.get("names").and_then(|n| n.as_array()) {

        None => default_names(),

    let bind = bind.or_else(|| server.get("bind").and_then(|b| b.as_str()).map(String::from)).unwrap_or_else(|| "127.0.0.1:7350".into());

    Ok(ServerCfg { bind, tls: mode, files, names })

}

fn default_names() -> Vec<String> {

    let mut names = Vec::new();

    let short = std::fs::read_to_string("/etc/hostname").ok().map(|h| h.trim().to_string()).filter(|h| !h.is_empty());

    let fqdn = std::process::Command::new("hostname")

        .arg("--fqdn")

        .output()

        .ok()

        .filter(|o| o.status.success())

        .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())

        .filter(|h| !h.is_empty());

    for n in [short, fqdn, Some("localhost".to_string())].into_iter().flatten() {

        if !names.contains(&n) {

            names.push(n);

        }

    names

}

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

    let cfg = server_cfg(ctx, bind, tls)?;

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

    let use_tls = match cfg.tls.as_str() {

        "on" => true,

        "off" => false,

        _ => !addr.ip().is_loopback(),

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

        bail!("Spectra serves other machines over TLS only (spec §21.1); drop --tls off, or pass --insecure to bind {addr} in plain HTTP anyway");

    }

    let socket = ctx.socket();

    if socket.exists() {

    }

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

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

        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 mut origins = vec![format!("http://127.0.0.1:{port}"), format!("http://localhost:{port}"), format!("http://{addr}")];

    let tls = if use_tls {

        let source = match &cfg.files {

            None => spectra_mcp::tls::CertSource::Own { names: cfg.names.clone() },

        let t = spectra_mcp::tls::ServerTls::load(&ctx.data, &source)?;

        let instance = spectra_core::keys::KeyPair::instance(&ctx.data, "spectra")?;

        spectra_mcp::instance::record_tls(&daemon.ledger, &instance, &t.spki_sha256, &t.names).await?;

        for n in &cfg.names {

            origins.push(format!("https://{n}:{port}"));

        }

        origins.push(format!("https://{addr}"));

        Some(t)

        None

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

    let app = spectra_mcp::http::router(state.clone()).merge(spectra_mcp::api::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())));

    match &tls {

        Some(t) => {

            let host = if addr.ip().is_unspecified() {

                addr.ip().to_string()

            println!("{}", ui::kv("https", format!("https://{host}:{port}")));

            println!(

                ui::kv(

                    "certificate",

                    format!("{}{}", ui::hash(hex::encode(&t.spki_sha256[..8])), ui::dim(" (signed by the instance key)"))

        None => println!("{}", ui::kv("http", format!("http://{}:{port}", addr.ip()))),

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

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

    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: Served = match &tls {

        Some(t) => {

            let listener = spectra_mcp::tls::TlsListener::new(tcp, t.config.clone())?;

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

            Box::pin(async move { srv.await })

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

            Box::pin(async move { srv.await })

    let signal = async move {

        shutdown_signal().await;

        let _ = stop_tx.send(true);

    };

    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;

    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() {

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

        tokio::select! {

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

            _ = term.recv() => {}

}