zc2 0.0.14

P2P compute broker with credit-based billing, WAL, and broker mesh support
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

//! `zc://` URI resolver for broker addresses.
//!
//! The `zc://` scheme is the canonical way to address a Zakuro broker:
//!
//! - `zc://node-name`       — resolve by scanning localhost:9000-9010, match `node_name`
//! - `zc://node-name:port`  — probe a specific port and verify the node name
//! - `zc://ip:port`         — direct address (no verification); converted to `http://ip:port`
//!
//! `http://` and `https://` URLs are passed through unchanged so existing code
//! that already has a resolved URL is unaffected.
//!
//! 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;

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

    // Name-based resolution: probe brokers on localhost.
    let agent = ureq::AgentBuilder::new()
        .timeout_connect(Duration::from_millis(300))
        .timeout_read(Duration::from_millis(500))
        .build();

    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";

    for port in &ports {
        let url = format!("http://localhost:{}/health", port);
        if let Ok(resp) = agent.get(&url).call() {
            if let Ok(text) = resp.into_string() {
                if let Ok(json) = serde_json::from_str::<serde_json::Value>(&text) {
                    if wildcard || json["node_name"].as_str() == Some(name) {
                        return Ok(format!("http://localhost:{}", port));
                    }
                }
            }
        }
    }

    if let Some(p) = explicit_port {
        Err(format!("Broker '{}' not found on localhost:{}", name, p))
    } else {
        Err(format!(
            "Broker '{}' not found on localhost:{}-{} — is it running?",
            name, SCAN_START, SCAN_END
        ))
    }
}