zc2 0.0.30

//! "Model A" — `zc` auto-runs a local broker on demand.
//!
//! Several commands (`workers`, `discovery`) target `ZAKURO_BROKER` which
//! defaults to `zc://localhost` (a scan of localhost:9000-9010, see
//! `broker::uri::resolve`). Historically, if nothing was listening there the
//! command just failed with "no local broker...". This module makes the
//! *default* case self-service: when the user didn't ask for a specific
//! broker, we make sure one exists — reusing it if already running, quietly
//! spawning one in the background otherwise — and only fail hard when the
//! user explicitly named a broker target that isn't reachable.

use std::time::Duration;

const SCAN_START: u16 = 9000;
const SCAN_END: u16 = 9010;
const DEFAULT_PORT: u16 = 9000;

/// Should we auto-spawn a local broker, or respect an explicit target?
///
/// Pure decision function: auto-spawn only applies to the *default* case —
/// no explicit `zc://...` argument on the command line, and no
/// `ZAKURO_BROKER` env var override. If the user named a target explicitly
/// (either way), we never auto-spawn — we use their target and let it fail
/// with the normal error if unreachable.
pub fn should_auto_spawn(explicit_cli: Option<&str>, env_broker: Option<&str>) -> bool {
    explicit_cli.is_none() && env_broker.is_none()
}

/// Probe `http://localhost:PORT/health` for PORT in `SCAN_START..=SCAN_END`.
/// Returns the URL of the first broker that responds.
///
/// Generic over the probe function so this is testable without opening real
/// sockets: production code passes a real HTTP GET, tests pass a fake.
fn find_running_broker<F>(mut probe: F) -> Option<String>
where
    F: FnMut(u16) -> bool,
{
    for port in SCAN_START..=SCAN_END {
        if probe(port) {
            return Some(format!("http://localhost:{}", port));
        }
    }
    None
}

/// A port is "a running broker" only when its `/health` body identifies as
/// `zakuro-broker`. Any other 2xx on localhost:9000-9010 (a MinIO console on
/// 9001 was the real case) used to be adopted as the broker, and every
/// following request then died parsing HTML as JSON.
fn real_probe(port: u16) -> bool {
    crate::broker::uri::probe_local(port).is_some()
}

/// The command that auto-spawns the local broker. It sets no `ZAKURO_P2P`:
/// the child inherits the caller's, and an explicit value always wins. Unset,
/// the spawned `zc broker` turns P2P on for any mesh route itself
/// (`broker::p2p_default`).
fn broker_command(exe: &std::path::Path) -> std::process::Command {
    let mut cmd = std::process::Command::new(exe);
    cmd.arg("-d").arg("broker");
    cmd.stdin(std::process::Stdio::null());
    cmd.stdout(std::process::Stdio::null());
    cmd.stderr(std::process::Stdio::null());
    cmd
}

/// Ensure a local broker is reachable, spawning one in the background if
/// needed. Returns the broker's base URL (e.g. `http://localhost:9000`) on
/// success.
///
/// `verbose` mirrors the convention used elsewhere in this crate (see
/// `BrokerConfig.verbose` in `main.rs`) — only chatter when actually doing
/// something (spawning), not when silently reusing an existing broker.
pub fn ensure_local_broker(verbose: bool) -> Result<String, String> {
    // 1. Reuse an already-running broker if there is one.
    if let Some(url) = find_running_broker(real_probe) {
        return Ok(url);
    }

    // 2. Nothing listening — spawn one in the background.
    if verbose {
        println!("  starting local broker…");
    }

    let exe = std::env::current_exe()
        .map_err(|e| format!("could not resolve current executable: {}", e))?;

    // Make sure ZAKURO_API_KEY (if any) is available to the child — it
    // inherits the parent's env by default, but load_into_env() may have
    // populated the *parent's* env from disk credentials just before this
    // runs, so re-run it defensively in case ensure_local_broker() is ever
    // called before that happens.
    crate::credentials::load_into_env();
    crate::broker::apply_user_broker_defaults();

    broker_command(&exe)
        .spawn()
        .map_err(|e| format!("could not spawn local broker: {}", e))?;

    // 3. Poll until it comes up, or give up after ~10s.
    let deadline = std::time::Instant::now() + Duration::from_secs(10);
    while std::time::Instant::now() < deadline {
        if real_probe(DEFAULT_PORT) {
            return Ok(format!("http://localhost:{}", DEFAULT_PORT));
        }
        std::thread::sleep(Duration::from_millis(400));
    }

    Err("could not start a local broker — run `zc broker` manually to see errors".to_string())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn should_auto_spawn_when_nothing_explicit() {
        assert!(should_auto_spawn(None, None));
    }

    #[test]
    fn should_not_auto_spawn_with_explicit_cli_arg() {
        assert!(!should_auto_spawn(Some("zc://node-foo"), None));
    }

    #[test]
    fn should_not_auto_spawn_with_env_broker_set() {
        assert!(!should_auto_spawn(None, Some("zc://node-foo")));
    }

    #[test]
    fn should_not_auto_spawn_with_both_set() {
        assert!(!should_auto_spawn(
            Some("zc://node-foo"),
            Some("zc://node-bar")
        ));
    }

    /// The spawn sets no `ZAKURO_P2P`, so the caller's explicit value reaches
    /// the child untouched: `ZAKURO_P2P=false` stays off on the host route.
    /// Unset, the child still peers there and on the proxy route.
    #[test]
    fn an_explicit_zakuro_p2p_false_stays_false_on_the_host_route() {
        use crate::broker::p2p_default_from;
        use crate::vpn::fixtures::{host_route, proxy_route};
        let cmd = broker_command(std::path::Path::new("zc"));
        assert_eq!(cmd.get_args().collect::<Vec<_>>(), ["-d", "broker"]);
        assert!(
            cmd.get_envs().all(|(k, _)| k != "ZAKURO_P2P"),
            "the auto-spawn must not override the caller's ZAKURO_P2P"
        );
        assert!(!p2p_default_from(Some("false"), &host_route()));
        assert!(p2p_default_from(None, &host_route()));
        assert!(p2p_default_from(None, &proxy_route()));
    }

    #[test]
    fn find_running_broker_reuses_first_hit() {
        let seen = std::cell::RefCell::new(Vec::new());
        let url = find_running_broker(|port| {
            seen.borrow_mut().push(port);
            port == 9003
        });
        assert_eq!(url, Some("http://localhost:9003".to_string()));
        // Probed in order and stopped at the first hit.
        assert_eq!(*seen.borrow(), vec![9000, 9001, 9002, 9003]);
    }

    #[test]
    fn find_running_broker_none_when_all_fail() {
        let url = find_running_broker(|_port| false);
        assert_eq!(url, None);
    }
}