//! `spectra serve`: the daemon (spec §7.1, §21.1). It holds the ledger, //! serves Tuner, the API and Radar on the Unix socket and on a TCP port, //! over TLS beyond loopback (§8.6), 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; /// `[server]` of spectra.toml, with the command line's overrides. struct ServerCfg { bind: String, tls: String, files: Option<(std::path::PathBuf, std::path::PathBuf)>, names: Vec, } fn server_cfg(ctx: &Ctx, bind: Option, tls: Option) -> Result { 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") { Some(toml::Value::String(m)) => mode = m.clone(), Some(toml::Value::Table(f)) => { let path = |k: &str| f.get(k).and_then(|v| v.as_str()).map(std::path::PathBuf::from); files = Some((path("cert").context("[server] tls needs cert")?, path("key").context("[server] tls needs key")?)); mode = "on".into(); } _ => {} } if let Some(m) = tls { mode = m; } if !matches!(mode.as_str(), "auto" | "on" | "off") { bail!("TLS is auto, on or off, not {mode}"); } let names = match server.get("names").and_then(|n| n.as_array()) { Some(a) => a.iter().filter_map(|v| v.as_str().map(String::from)).collect(), 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 }) } /// The host name, its fully qualified name, and localhost. fn default_names() -> Vec { 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 } type Served = std::pin::Pin> + Send>>; pub async fn serve(ctx: &Ctx, bind: Option, tls: Option, insecure: bool, open: bool) -> Result { 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() { 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 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 { Some((cert, key)) => spectra_mcp::tls::CertSource::Files { cert: cert.clone(), key: key.clone() }, 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) } else { None }; let state = AppState { daemon: daemon.clone(), origins: Arc::new(origins), sessions: Default::default() }; let app = spectra_mcp::http::router(state.clone()).merge(spectra_mcp::api::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()))); match &tls { Some(t) => { let host = if addr.ip().is_unspecified() { cfg.names.first().cloned().unwrap_or_else(|| "localhost".into()) } else { 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")); 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| async move { let _ = rx.wait_for(|s| *s).await; }; let unix_srv = axum::serve(unix, app.clone().into_make_service_with_connect_info::()).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::()).with_graceful_shutdown(stop(stop_rx)); Box::pin(async move { srv.await }) } None => { let srv = axum::serve(tcp, app.into_make_service_with_connect_info::()).with_graceful_shutdown(stop(stop_rx)); Box::pin(async move { srv.await }) } }; 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; }