slave 0.5.0

//! `slave serve`: build the workspace and run everything it needs.
//!
//! That is the MCP backend, and — when the project has `config/apps.toml` —
//! the domain gateway and every web app in `apps/`. Output is prefixed with
//! each process's name; Ctrl+C stops them all.

use std::collections::BTreeMap;
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};

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

use crate::apps::{self, Apps, Env};
use crate::cli::ServeArgs;
use crate::{cargo, dioxus, project};

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)>,
    /// Path prefix the app is served under (`SLAVE_BASE_PATH`), if any.
    base_path: Option<String>,
    /// A Dioxus app whose single-page bundle `dx` must build first.
    spa: bool,
}

pub fn run(args: &ServeArgs) -> Result<()> {
    let root = project::find_root()?;
    let env = Env::load(&root)?;
    let (host, port) = resolve_address(args, &env)?;
    ensure_port_available(&host, port, "the MCP backend")?;

    let mcp_address = apps::connect_address(&host, port);
    let backend_env = vec![
        ("APP_HOST".to_owned(), host.clone()),
        ("APP_PORT".to_owned(), port.to_string()),
    ];
    let mut processes = vec![Process {
        label: "mcp".into(),
        binary: project::package_name(&root)?,
        envs: backend_env.clone(),
        base_path: None,
        spa: false,
    }];

    let routing = Apps::load(&root, &env, &mcp_address)?;
    if let Some(routing) = &routing {
        let (listen_host, listen_port) = split_address(&routing.listen)?;
        ensure_port_available(&listen_host, listen_port, "the gateway")?;
        processes.push(Process {
            label: "gateway".into(),
            binary: "gateway".into(),
            envs: backend_env.clone(),
            base_path: None,
            spa: false,
        });
        for (app, (port, base_path)) in local_apps(&root, routing)? {
            ensure_port_available(DEFAULT_HOST, port, &format!("app `{app}`"))?;
            processes.push(Process {
                label: app.clone(),
                spa: dioxus::is_dioxus_app(&root.join("apps").join(&app)),
                binary: app,
                envs: vec![
                    ("PORT".to_owned(), port.to_string()),
                    (
                        "SLAVE_MCP_URL".to_owned(),
                        format!("http://{mcp_address}/mcp"),
                    ),
                    (
                        "SLAVE_BASE_PATH".to_owned(),
                        base_path.clone().unwrap_or_default(),
                    ),
                ],
                base_path,
            });
        }
    }

    if processes.iter().any(|p| p.spa) && !dioxus::dx_installed() {
        bail!(
            "this project has Dioxus apps, but `dx` was not found; {}",
            dioxus::INSTALL_HINT
        );
    }
    build(&root, args.release)?;
    let target = target_dir(&root)?;
    for process in processes.iter_mut().filter(|p| p.spa) {
        let spa_dir = dioxus::build_spa(
            &root,
            &target,
            &process.binary,
            process.base_path.as_deref(),
            args.release,
        )?;
        process
            .envs
            .push(("SLAVE_SPA_DIR".to_owned(), spa_dir.display().to_string()));
    }
    let bin_dir = target.join(if args.release { "release" } else { "debug" });

    println!();
    println!("  MCP backend   http://{mcp_address}/mcp");
    if let Some(routing) = &routing {
        println!("  Gateway       http://{}", routing.listen);
        for route in routing.routes.iter() {
            let target = if route.is_mcp {
                "MCP backend"
            } else {
                &route.app
            };
            println!(
                "    {:<36} → {target}",
                routing.url(route.host.as_deref(), route.path.as_deref())
            );
        }
    }
    println!("\nPress Ctrl+C to stop.\n");

    supervise(&root, &bin_dir, processes)
}

/// Web apps from `apps.toml` that live in `apps/<name>`, with their ports and
/// base paths. Apps routed to an explicit `upstream` run elsewhere and are not
/// started.
fn local_apps(root: &Path, routing: &Apps) -> Result<BTreeMap<String, (u16, Option<String>)>> {
    let mut found = BTreeMap::new();
    for domain in &routing.config.domains {
        if domain.is_mcp() || domain.upstream.is_some() {
            continue;
        }
        let Some(port) = domain.port else { continue };
        if !root
            .join("apps")
            .join(&domain.app)
            .join("Cargo.toml")
            .is_file()
        {
            println!(
                "  note: apps/{} not found; start `{}` on port {port} yourself",
                domain.app, domain.app
            );
            continue;
        }
        let entry = (port, domain.path.clone());
        if let Some(previous) = found.insert(domain.app.clone(), entry.clone())
            && previous != entry
        {
            bail!(
                "app `{}` is listed with different ports or paths in {}; one app serves one port and base path",
                domain.app,
                apps::CONFIG
            );
        }
    }
    Ok(found)
}

fn build(root: &Path, release: bool) -> Result<()> {
    let mut args = vec!["build", "--workspace", "--bins"];
    if release {
        args.push("--release");
    }
    let status = cargo::run(root, &args, &[])?;
    if !status.success() {
        bail!("build failed");
    }
    Ok(())
}

/// The workspace's target directory, as Cargo reports it.
fn target_dir(root: &Path) -> Result<PathBuf> {
    let output = cargo::command()
        .args(["metadata", "--format-version", "1", "--no-deps"])
        .current_dir(root)
        .stderr(Stdio::inherit())
        .output()
        .context("failed to run `cargo metadata`")?;
    if !output.status.success() {
        bail!("`cargo metadata` failed");
    }
    let metadata: serde_json::Value =
        serde_json::from_slice(&output.stdout).context("unexpected `cargo metadata` output")?;
    metadata["target_directory"]
        .as_str()
        .map(PathBuf::from)
        .context("`cargo metadata` did not report a target directory")
}

/// 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}")),
    }
}