zc2 0.0.29

//! `zc://` URI resolver for broker addresses.
//!
//! The `zc://` scheme is the canonical way to address a Zakuro broker:
//!
//! - `zc://node-<fp>`       — a broker by key-derived identity. Resolved first
//!   against brokers on localhost (matching `/health`'s `node_id`), then via
//!   the mesh directory (hub roster → the broker's advertised endpoint), so a
//!   fleet broker is reachable by id from any machine on the mesh.
//! - `zc://<name>`          — legacy: resolve by scanning localhost:9000-9010
//!   and matching `/health`'s `node_name`.
//! - `zc://<name>:port`     — probe a specific local port and verify the name.
//! - `zc://ip:port`         — direct address (no verification); converted to
//!   `http://ip:port`.
//! - `zc://localhost`       — the first broker found on localhost:9000-9010.
//!
//! A port only counts as a broker when its `/health` body says
//! `"service":"zakuro-broker"` — any other 200 on those ports (a MinIO console
//! was the real case) is skipped.
//!
//! `http://` and `https://` URLs are rejected: callers hold `zc://` handles, and
//! the rule keeps raw addresses out of user-facing flows.
//!
//! Returns `Err` with a human-readable message when resolution fails so callers
//! can surface it before making any network calls.

use std::time::Duration;

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

/// Minimal view of a `/health` body, enough to decide "is this a broker, and
/// which one".
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub(crate) struct HealthIdentity {
    pub is_broker: bool,
    pub node_name: Option<String>,
    /// Key-derived `zc://node-<fp>` (brokers ≥ 0.0.24); `None` on older builds.
    pub node_id: Option<String>,
}

pub(crate) fn parse_health(text: &str) -> HealthIdentity {
    let Ok(json) = serde_json::from_str::<serde_json::Value>(text) else {
        return HealthIdentity::default();
    };
    HealthIdentity {
        is_broker: json.get("service").and_then(|s| s.as_str()) == Some("zakuro-broker"),
        node_name: json
            .get("node_name")
            .and_then(|s| s.as_str())
            .map(str::to_string),
        node_id: json
            .get("node_id")
            .and_then(|s| s.as_str())
            .map(str::to_string),
    }
}

/// Does a local broker's health identity satisfy the user's `name` argument?
/// `name` may be a fingerprint spelling (`node-<fp>` / `<fp>`) or a legacy
/// hostname label. `wildcard` accepts any broker.
pub(crate) fn health_matches(h: &HealthIdentity, name: &str, wildcard: bool) -> bool {
    if !h.is_broker {
        return false;
    }
    if wildcard {
        return true;
    }
    let want_fp = crate::broker::node_identity::strip_node_arg(name);
    if let Some(id) = &h.node_id {
        if crate::broker::node_identity::strip_node_arg(id) == want_fp {
            return true;
        }
    }
    h.node_name.as_deref() == Some(name)
}

fn local_agent() -> ureq::Agent {
    ureq::Agent::new_with_config(
        ureq::Agent::config_builder()
            .timeout_connect(Some(Duration::from_millis(300)))
            .timeout_recv_response(Some(Duration::from_millis(500)))
            .build(),
    )
}

/// `/health` identity of whatever listens on `localhost:port`, if it is a broker.
pub(crate) fn probe_local(port: u16) -> Option<HealthIdentity> {
    let url = format!("http://localhost:{}/health", port);
    let text = local_agent()
        .get(&url)
        .call()
        .ok()?
        .into_body()
        .read_to_string()
        .ok()?;
    let h = parse_health(&text);
    if h.is_broker {
        Some(h)
    } else {
        None
    }
}

/// Resolve a `zc://` URI to a plain `http://` URL.
///
/// Panics if `uri` does not start with `zc://`, `http://`, or `https://`.
/// Use [`try_resolve`] when the input is untrusted.
pub fn resolve(uri: &str) -> Result<String, String> {
    if uri.starts_with("http://") || uri.starts_with("https://") {
        return Err(format!(
            "Plain HTTP URL '{}' is not allowed — use zc://node-name instead.",
            uri
        ));
    }

    let rest = match uri.strip_prefix("zc://") {
        Some(r) => r.trim_end_matches('/'),
        None => {
            return Err(format!(
                "Invalid broker URI '{}'. Expected zc://node-name.",
                uri
            ))
        }
    };

    if rest.is_empty() {
        return Err("Empty broker name in zc:// URI.".to_string());
    }

    // Split into (host_or_name, optional_port)
    let (name, explicit_port): (&str, Option<u16>) = if let Some(colon) = rest.rfind(':') {
        let port_str = &rest[colon + 1..];
        match port_str.parse::<u16>() {
            Ok(p) => (&rest[..colon], Some(p)),
            Err(_) => (rest, None),
        }
    } else {
        (rest, None)
    };

    // If name looks like an IP address (all digits and dots), convert directly.
    let looks_like_ip =
        name.split('.').all(|octet| octet.parse::<u8>().is_ok()) && name.contains('.');

    if looks_like_ip {
        let port = explicit_port.unwrap_or(9000);
        return Ok(format!("http://{}:{}", name, port));
    }

    let ports: Vec<u16> = if let Some(p) = explicit_port {
        vec![p]
    } else {
        (SCAN_START..=SCAN_END).collect()
    };

    // "localhost" is a wildcard — return the first broker found on any scanned port.
    let wildcard = name == "localhost" || name == "127.0.0.1";

    // 1. Local brokers: match by key-derived id (or legacy hostname label).
    for port in &ports {
        if let Some(h) = probe_local(*port) {
            if health_matches(&h, name, wildcard) {
                return Ok(format!("http://localhost:{}", port));
            }
        }
    }

    // 2. Mesh directory: a fingerprint resolves to the broker's advertised
    //    endpoint. Only for fingerprint spellings — hostnames are never looked
    //    up remotely, so `zc://my-laptop` cannot silently dial the wrong box.
    let bare = crate::broker::node_identity::strip_node_arg(name);
    if !wildcard && crate::mesh_dir::looks_like_fingerprint(bare) {
        if let Some(ep) = crate::mesh_dir::endpoint_for(bare) {
            let url = match explicit_port {
                // An explicit port overrides the advertised one (same host).
                Some(p) => {
                    let host = ep.rsplit_once(':').map(|(h, _)| h).unwrap_or(&ep);
                    format!("http://{}:{}", host, p)
                }
                None => format!("http://{}", ep),
            };
            return Ok(url);
        }
        return Err(format!(
            "broker zc://node-{bare} is not reachable: it is not advertising a mesh endpoint \
             (offline, or not on the mesh). List live brokers with `zc brokers`."
        ));
    }

    if let Some(p) = explicit_port {
        Err(format!("Broker '{}' not found on localhost:{}", name, p))
    } else {
        Err(format!(
            "no local broker on localhost:{}-{}.\n  \
             Start one with `zc broker` (or `zc share` for a broker + worker), or target a\n  \
             mesh broker with `ZAKURO_BROKER=zc://node-<id> zc ls` (list them: `zc brokers`).",
            SCAN_START, SCAN_END
        ))
    }
}

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

    #[test]
    fn rejects_plain_http() {
        assert!(resolve("http://1.2.3.4:9000").is_err());
        assert!(resolve("https://example.com").is_err());
    }

    #[test]
    fn rejects_malformed() {
        assert!(resolve("node-abc").is_err());
        assert!(resolve("zc://").is_err());
    }

    #[test]
    fn ip_forms_are_direct() {
        assert_eq!(
            resolve("zc://10.13.13.33:9000").unwrap(),
            "http://10.13.13.33:9000"
        );
        assert_eq!(
            resolve("zc://10.13.13.33").unwrap(),
            "http://10.13.13.33:9000"
        );
        assert_eq!(
            resolve("zc://10.13.13.33/").unwrap(),
            "http://10.13.13.33:9000"
        );
    }

    #[test]
    fn health_parsing_requires_the_broker_service_tag() {
        let h = parse_health(
            r#"{"status":"healthy","service":"zakuro-broker","node_name":"lxd","node_id":"zc://node-13a9551e881f7f8c"}"#,
        );
        assert!(h.is_broker);
        assert_eq!(h.node_name.as_deref(), Some("lxd"));
        assert_eq!(h.node_id.as_deref(), Some("zc://node-13a9551e881f7f8c"));

        // MinIO console: HTML, not JSON
        assert!(!parse_health("<!doctype html>").is_broker);
        // JSON but not a broker
        assert!(!parse_health(r#"{"status":"ok"}"#).is_broker);
        // Older broker without node_id still counts as a broker
        let old =
            parse_health(r#"{"status":"healthy","service":"zakuro-broker","node_name":"lxd"}"#);
        assert!(old.is_broker);
        assert!(old.node_id.is_none());
    }

    #[test]
    fn matching_by_fingerprint_name_and_wildcard() {
        let h = parse_health(
            r#"{"service":"zakuro-broker","node_name":"lxd","node_id":"zc://node-13a9551e881f7f8c"}"#,
        );
        assert!(health_matches(&h, "node-13a9551e881f7f8c", false));
        assert!(health_matches(&h, "13a9551e881f7f8c", false));
        assert!(health_matches(&h, "zc://node-13a9551e881f7f8c", false));
        assert!(health_matches(&h, "lxd", false));
        assert!(!health_matches(&h, "node-0000000000000000", false));
        assert!(health_matches(&h, "anything", true));
        let not_broker = parse_health("<html>");
        assert!(!health_matches(&not_broker, "anything", true));
    }
}