slave 0.12.0

//! `slave serve`: run what `slave build` compiled.
//!
//! That is the runtime: one process serving the MCP backend and every
//! embedded app, routed by `config/apps.toml`, plus one process per separate
//! app (`embed = false`), which the runtime proxies to. Output is prefixed
//! with each process's name; Ctrl+C stops them all.
//!
//! Serving never compiles: run `slave build` first and after changes. A
//! warning is printed when sources are newer than the binaries.

use std::io::{BufRead, BufReader, ErrorKind, Read};
use std::net::TcpListener;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::{Duration, Instant, SystemTime};

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

use crate::apps::{self, Apps, Env};
use crate::cli::ServeArgs;
use crate::commands::build::{self as build_cmd, LocalApp};
use crate::{project, runtime};

const DEFAULT_HOST: &str = "127.0.0.1";

const DEFAULT_PORT: u16 = 3000;

/// How long processes get to stop after Ctrl+C before they are killed.
const SHUTDOWN_GRACE: Duration = Duration::from_secs(5);

/// A process to run: a binary of the workspace and its environment.
struct Process {
    label: String,
    binary: String,
    envs: Vec<(String, String)>,
}

pub fn run(args: &ServeArgs) -> Result<()> {
    let root = project::find_root()?;
    runtime::require(&root)?;
    let env = Env::load(&root)?;
    let (host, port) = resolve_address(args, &env)?;
    let mcp_address = apps::connect_address(&host, port);
    let Some(routing) = Apps::load(&root, &env, &mcp_address)? else {
        bail!("{} is missing; the runtime routes by it", apps::CONFIG);
    };
    let local = build_cmd::local_apps(&root, &routing.config)?;

    let profile = if args.release { "release" } else { "debug" };
    let bin_dir = build_cmd::target_dir(&root)?.join(profile);
    let build_hint = if args.release {
        "slave build --release"
    } else {
        "slave build"
    };

    let mut processes = vec![Process {
        label: runtime::DIR.into(),
        binary: runtime::DIR.into(),
        envs: vec![
            ("APP_HOST".to_owned(), host.clone()),
            ("APP_PORT".to_owned(), port.to_string()),
            ("GATEWAY_LISTEN".to_owned(), routing.listen.clone()),
        ],
    }];
    for app in local.iter().filter(|app| !app.embedded) {
        processes.push(app_process(app, &mcp_address));
    }

    let missing: Vec<&str> = processes
        .iter()
        .filter(|p| !binary_path(&bin_dir, &p.binary).is_file())
        .map(|p| p.label.as_str())
        .collect();
    if !missing.is_empty() {
        bail!(
            "not built yet: {}; run `{build_hint} {}` first",
            missing.join(", "),
            missing.join(" ")
        );
    }
    warn_if_stale(&root, &bin_dir, &processes, build_hint);

    ensure_port_available(&host, port, "the MCP backend")?;
    let (listen_host, listen_port) = split_address(&routing.listen)?;
    ensure_port_available(&listen_host, listen_port, "the runtime")?;
    for app in &local {
        if app.direct || !app.embedded {
            ensure_port_available(DEFAULT_HOST, app.port, &format!("app `{}`", app.name))?;
        }
    }

    println!();
    println!("  MCP backend   http://{mcp_address}/mcp");
    println!("  Runtime       http://{}", routing.listen);
    for (_, route) in routing.routes.in_matching_order() {
        let target = if route.is_mcp {
            "MCP backend".to_owned()
        } else if local
            .iter()
            .any(|app| app.name == route.app && !app.embedded)
        {
            format!("{} (separate, {})", route.app, route.upstream)
        } else {
            route.app.clone()
        };
        println!(
            "    {:<36} → {target}",
            routing.url(route.host.as_deref(), route.path.as_deref())
        );
    }
    for app in local.iter().filter(|app| app.direct) {
        let mode = if app.embedded {
            "direct"
        } else {
            "direct, separate"
        };
        println!(
            "  {:<13} http://{DEFAULT_HOST}:{}  ({mode})",
            app.name, app.port
        );
    }
    println!("\nPress Ctrl+C to stop.\n");

    supervise(&root, &bin_dir, processes)
}

fn app_process(app: &LocalApp, mcp_address: &str) -> Process {
    let envs = vec![
        ("PORT".to_owned(), app.port.to_string()),
        (
            "SLAVE_MCP_URL".to_owned(),
            format!("http://{mcp_address}/mcp"),
        ),
        (
            "SLAVE_BASE_PATH".to_owned(),
            app.base_path.clone().unwrap_or_default(),
        ),
    ];
    Process {
        label: app.name.clone(),
        binary: app.name.clone(),
        envs,
    }
}

fn binary_path(bin_dir: &Path, name: &str) -> PathBuf {
    bin_dir.join(format!("{name}{}", std::env::consts::EXE_SUFFIX))
}

/// Warns when source files changed after the oldest binary was built.
fn warn_if_stale(root: &Path, bin_dir: &Path, processes: &[Process], build_hint: &str) {
    let built = processes
        .iter()
        .filter_map(|p| modified(&binary_path(bin_dir, &p.binary)))
        .min();
    let mut newest = None;
    // Migrations only; nothing else in `database/` is compiled in.
    for dir in ["src", "apps", "database/migrations", "Cargo.toml"] {
        newest_modified(&root.join(dir), &mut newest);
    }
    if let (Some(built), Some(newest)) = (built, newest)
        && newest > built
    {
        println!(
            "  warning: sources changed since the last build; run `{build_hint}` to include them"
        );
    }
}

fn modified(path: &Path) -> Option<SystemTime> {
    path.metadata().and_then(|m| m.modified()).ok()
}

/// The latest modification time of `path` or any file below it, skipping
/// build output.
fn newest_modified(path: &Path, newest: &mut Option<SystemTime>) {
    let Ok(metadata) = path.metadata() else {
        return;
    };
    if metadata.is_dir() {
        if path.file_name().is_some_and(|name| name == "target") {
            return;
        }
        if let Ok(entries) = path.read_dir() {
            for entry in entries.flatten() {
                newest_modified(&entry.path(), newest);
            }
        }
    } else if let Ok(time) = metadata.modified()
        && newest.is_none_or(|current| time > current)
    {
        *newest = Some(time);
    }
}

/// Runs every process until Ctrl+C or until one of them exits.
fn supervise(root: &Path, bin_dir: &Path, processes: Vec<Process>) -> Result<()> {
    let stop = Arc::new(AtomicBool::new(false));
    {
        let stop = Arc::clone(&stop);
        ctrlc::set_handler(move || stop.store(true, Ordering::SeqCst))
            .context("failed to install the Ctrl+C handler")?;
    }

    let width = processes.iter().map(|p| p.label.len()).max().unwrap_or(0);
    let mut children: Vec<(String, Child)> = Vec::new();
    for process in processes {
        let binary = bin_dir.join(format!(
            "{}{}",
            process.binary,
            std::env::consts::EXE_SUFFIX
        ));
        let spawned = Command::new(&binary)
            .current_dir(root)
            .envs(process.envs.iter().map(|(k, v)| (k.as_str(), v.as_str())))
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn();
        let mut child = match spawned {
            Ok(child) => child,
            Err(err) => {
                stop_all(&mut children, Duration::ZERO);
                return Err(err).with_context(|| format!("failed to start {}", binary.display()));
            }
        };
        let prefix = format!("[{:<width$}]", process.label);
        forward(child.stdout.take(), prefix.clone(), false);
        forward(child.stderr.take(), prefix, true);
        children.push((process.label, child));
    }

    loop {
        if stop.load(Ordering::SeqCst) {
            stop_all(&mut children, SHUTDOWN_GRACE);
            return Ok(());
        }
        for (label, child) in &mut children {
            if let Some(status) = child.try_wait()? {
                let label = label.clone();
                stop_all(&mut children, SHUTDOWN_GRACE);
                bail!("`{label}` exited ({status}); stopped the other processes");
            }
        }
        thread::sleep(Duration::from_millis(100));
    }
}

/// Copies a child's output to ours, line by line, behind `prefix`.
fn forward(stream: Option<impl Read + Send + 'static>, prefix: String, to_stderr: bool) {
    let Some(stream) = stream else { return };
    thread::spawn(move || {
        for line in BufReader::new(stream).lines().map_while(Result::ok) {
            if to_stderr {
                eprintln!("{prefix} {line}");
            } else {
                println!("{prefix} {line}");
            }
        }
    });
}

/// Asks children to stop (SIGINT on Unix, as Ctrl+C in a terminal would; the
/// gateway treats SIGTERM as a slow graceful drain), waits up to `grace`, then
/// kills whatever is left.
fn stop_all(children: &mut [(String, Child)], grace: Duration) {
    #[cfg(unix)]
    for (_, child) in children.iter_mut() {
        if matches!(child.try_wait(), Ok(None)) {
            let _ = Command::new("kill")
                .args(["-INT", &child.id().to_string()])
                .status();
        }
    }
    let deadline = Instant::now() + grace;
    for (_, child) in children.iter_mut() {
        while matches!(child.try_wait(), Ok(None)) && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(50));
        }
        let _ = child.kill();
        let _ = child.wait();
    }
}

/// Resolves host and port with the precedence: flag > process environment > `.env` > default.
fn resolve_address(args: &ServeArgs, env: &Env) -> Result<(String, u16)> {
    let host = args
        .host
        .clone()
        .or_else(|| env.get("APP_HOST"))
        .unwrap_or_else(|| DEFAULT_HOST.to_owned());

    let port = match (args.port, env.get("APP_PORT")) {
        (Some(port), _) => port,
        (None, Some(value)) => value
            .trim()
            .parse()
            .with_context(|| format!("APP_PORT={value:?} is not a valid port"))?,
        (None, None) => DEFAULT_PORT,
    };
    Ok((host, port))
}

fn split_address(address: &str) -> Result<(String, u16)> {
    let (host, port) = address
        .rsplit_once(':')
        .with_context(|| format!("`{address}` must be host:port"))?;
    let port = port
        .parse()
        .with_context(|| format!("`{address}` has an invalid port"))?;
    Ok((host.trim_matches(['[', ']']).to_owned(), port))
}

/// Fails fast with a clear message instead of waiting for a full build to hit the same error.
fn ensure_port_available(host: &str, port: u16, what: &str) -> Result<()> {
    match TcpListener::bind((host, port)) {
        Ok(_) => Ok(()),
        Err(err) if err.kind() == ErrorKind::AddrInUse => {
            bail!("port {port} on {host} (for {what}) is already in use")
        }
        Err(err) => Err(err).with_context(|| format!("cannot bind to {host}:{port} for {what}")),
    }
}