zc2 0.0.28

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

/// Pure decision: given an optional mesh IP (e.g. `Some("10.13.13.6")` when
/// this node is on the WireGuard mesh, `None` otherwise), should the
/// auto-spawned broker run with `ZAKURO_P2P=true`?
///
/// Extracted so this can be unit tested without touching real network
/// interfaces — production code feeds it the result of
/// `broker::discovery::get_mesh_ip()`.
pub fn should_enable_p2p(mesh_ip: Option<&str>) -> bool {
    mesh_ip
        .map(|ip| ip.starts_with("10.13.13."))
        .unwrap_or(false)
}

/// 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();

    let mut cmd = std::process::Command::new(&exe);
    cmd.arg("-d").arg("broker");

    if should_enable_p2p(crate::broker::discovery::get_mesh_ip().as_deref()) {
        cmd.env("ZAKURO_P2P", "true");
    }

    cmd.stdin(std::process::Stdio::null());
    cmd.stdout(std::process::Stdio::null());
    cmd.stderr(std::process::Stdio::null());

    cmd.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")
        ));
    }

    #[test]
    fn mesh_ip_present_enables_p2p() {
        assert!(should_enable_p2p(Some("10.13.13.6")));
    }

    #[test]
    fn no_mesh_ip_disables_p2p() {
        assert!(!should_enable_p2p(None));
    }

    #[test]
    fn non_mesh_ip_disables_p2p() {
        // e.g. a plain LAN IP returned by some other detection path — only
        // the 10.13.13.0/24 mesh subnet should flip P2P on.
        assert!(!should_enable_p2p(Some("192.168.1.50")));
    }

    #[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);
    }
}