zc2 0.0.25

//! Dashboard round-trips for the zero-trust mesh: register this node's signing
//! pubkey, pull the authorized-node roster + the dashboard's voucher-signing
//! pubkey, and obtain a signed job voucher before dispatch.
//!
//! Contract (dashboard side, `zak-dashboard`):
//!   POST /api/broker/node/register  {node_pubkey}        (X-Broker-Api-Key: zk_…)
//!   GET  /api/broker/node/roster    → [{node_pubkey, revoked}]
//!   GET  /api/broker/voucher/pubkey → {pubkey}
//!   POST /api/broker/voucher        {budget_credits}     → {signed_json, sig}

use std::time::Duration;

fn base(api_url: &str) -> &str {
    api_url.trim_end_matches('/')
}

/// Build the JSON payload for `register_node`. Omits the `mesh_endpoint` key
/// entirely when `mesh_endpoint` is `None` — absent means "unchanged"
/// server-side, so a tunnel flap must not send an explicit `null` that would
/// clear a previously-stored address. Split out from `register_node` so this
/// contract has direct test coverage.
fn register_payload(node_pubkey: &str, mesh_endpoint: Option<&str>) -> String {
    match mesh_endpoint {
        Some(ep) => serde_json::json!({ "node_pubkey": node_pubkey, "mesh_endpoint": ep }),
        None => serde_json::json!({ "node_pubkey": node_pubkey }),
    }
    .to_string()
}

/// Register this node's Ed25519 public key with the dashboard (idempotent).
pub fn register_node(
    api_url: &str,
    api_key: &str,
    node_pubkey: &str,
    mesh_endpoint: Option<&str>,
) -> Result<(), String> {
    let endpoint = format!("{}/api/broker/node/register", base(api_url));
    let payload = register_payload(node_pubkey, mesh_endpoint);
    let resp = ureq::post(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .header("Content-Type", "application/json")
        .send(payload.as_str())
        .map_err(|e| e.to_string())?;
    let code = resp.status().as_u16();
    if (200..300).contains(&code) {
        Ok(())
    } else {
        Err(format!("register_node HTTP {code}"))
    }
}

/// Parse a roster response body: `[{"node_pubkey": "...", "revoked": bool}, …]`
/// → `(pubkey, revoked)` pairs. Unknown/malformed entries are skipped.
pub fn parse_roster(body: &str) -> Vec<(String, bool)> {
    let v: serde_json::Value = match serde_json::from_str(body) {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };
    v.as_array()
        .map(|arr| {
            arr.iter()
                .filter_map(|e| {
                    let pk = e.get("node_pubkey")?.as_str()?.to_string();
                    let revoked = e.get("revoked").and_then(|r| r.as_bool()).unwrap_or(false);
                    Some((pk, revoked))
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Parse `mesh_endpoint`s from a roster body → `(pubkey, "ip:port")` pairs.
/// Entries with a null or absent endpoint are skipped: they are authorized
/// nodes with no tunnel up, which is normal, not malformed.
pub fn parse_endpoints(body: &str) -> Vec<(String, String)> {
    let v: serde_json::Value = match serde_json::from_str(body) {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };
    v.as_array()
        .map(|arr| {
            arr.iter()
                .filter_map(|e| {
                    let pk = e.get("node_pubkey")?.as_str()?.to_string();
                    let ep = e.get("mesh_endpoint")?.as_str()?.to_string();
                    Some((pk, ep))
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Request the roster and read the response body (shared by `fetch_roster`
/// and `fetch_roster_full` so the HTTP round-trip is implemented once).
fn fetch_roster_body(api_url: &str, api_key: &str) -> Result<String, String> {
    let endpoint = format!("{}/api/broker/node/roster", base(api_url));
    let resp = ureq::get(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .call()
        .map_err(|e| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!("fetch_roster HTTP {}", resp.status().as_u16()));
    }
    resp.into_body().read_to_string().map_err(|e| e.to_string())
}

/// Fetch the authorized-node roster from the dashboard.
pub fn fetch_roster(api_url: &str, api_key: &str) -> Result<Vec<(String, bool)>, String> {
    Ok(parse_roster(&fetch_roster_body(api_url, api_key)?))
}

/// `(pubkey_b64, revoked)` authorization pairs, as `parse_roster` returns them.
pub type RosterEntries = Vec<(String, bool)>;

/// `(pubkey_b64, "ip:port")` mesh endpoints, as `parse_endpoints` returns them.
pub type RosterEndpoints = Vec<(String, String)>;

/// Fetch the roster once and return both authorization pairs and endpoints.
///
/// `fetch_roster` reads and discards the body, so a caller that also wants
/// endpoints would have to poll again. This is the tick-loop path, polled by
/// every broker, so it stays one request.
pub fn fetch_roster_full(
    api_url: &str,
    api_key: &str,
) -> Result<(RosterEntries, RosterEndpoints), String> {
    let body = fetch_roster_body(api_url, api_key)?;
    Ok((parse_roster(&body), parse_endpoints(&body)))
}

/// Fetch the dashboard's voucher-signing public key (b64).
pub fn fetch_voucher_pubkey(api_url: &str) -> Result<String, String> {
    let endpoint = format!("{}/api/broker/voucher/pubkey", base(api_url));
    let resp = ureq::get(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .call()
        .map_err(|e| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!(
            "fetch_voucher_pubkey HTTP {}",
            resp.status().as_u16()
        ));
    }
    let text = resp
        .into_body()
        .read_to_string()
        .map_err(|e| e.to_string())?;
    let v: serde_json::Value = serde_json::from_str(&text).map_err(|e| e.to_string())?;
    v.get("pubkey")
        .and_then(|p| p.as_str())
        .map(|s| s.to_string())
        .ok_or_else(|| "voucher pubkey response missing `pubkey`".into())
}

/// Obtain a dashboard-signed job voucher. Returns `(signed_json, sig)` — the
/// exact bytes the dashboard signed plus its signature (see `voucher.rs`).
pub fn obtain_voucher(
    api_url: &str,
    api_key: &str,
    budget_credits: f64,
) -> Result<(String, String), String> {
    let endpoint = format!("{}/api/broker/voucher", base(api_url));
    let payload = serde_json::json!({ "budget_credits": budget_credits }).to_string();
    let resp = ureq::post(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .header("Content-Type", "application/json")
        .send(payload.as_str())
        .map_err(|e| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!("obtain_voucher HTTP {}", resp.status().as_u16()));
    }
    let text = resp
        .into_body()
        .read_to_string()
        .map_err(|e| e.to_string())?;
    let v: serde_json::Value = serde_json::from_str(&text).map_err(|e| e.to_string())?;
    let signed_json = v
        .get("signed_json")
        .and_then(|s| s.as_str())
        .ok_or("voucher response missing `signed_json`")?
        .to_string();
    let sig = v
        .get("sig")
        .and_then(|s| s.as_str())
        .ok_or("voucher response missing `sig`")?
        .to_string();
    Ok((signed_json, sig))
}

/// Redeem a voucher's nonce at the dashboard (double-spend guard). Returns
/// `Ok(true)` on first redemption, `Ok(false)` if already spent. Best-effort:
/// the caller logs failures and continues (settlement rides the existing path).
pub fn redeem_voucher(
    api_url: &str,
    api_key: &str,
    task_nonce: &str,
    actual_cost: f64,
) -> Result<bool, String> {
    let endpoint = format!("{}/api/broker/voucher/redeem", base(api_url));
    let payload =
        serde_json::json!({ "task_nonce": task_nonce, "actual_cost": actual_cost }).to_string();
    let resp = ureq::post(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .header("Content-Type", "application/json")
        .send(payload.as_str())
        .map_err(|e| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!("redeem_voucher HTTP {}", resp.status().as_u16()));
    }
    let text = resp
        .into_body()
        .read_to_string()
        .map_err(|e| e.to_string())?;
    let v: serde_json::Value = serde_json::from_str(&text).map_err(|e| e.to_string())?;
    Ok(v.get("redeemed").and_then(|r| r.as_bool()).unwrap_or(false))
}

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

    #[test]
    fn register_payload_omits_mesh_endpoint_key_when_none() {
        let payload = super::register_payload("PUBKEY", None);
        let v: serde_json::Value = serde_json::from_str(&payload).unwrap();
        assert_eq!(v["node_pubkey"], "PUBKEY");
        assert!(
            v.get("mesh_endpoint").is_none(),
            "mesh_endpoint key must be absent, not null: {payload}"
        );
        assert!(
            !payload.contains("mesh_endpoint"),
            "payload must not mention mesh_endpoint at all: {payload}"
        );
    }

    #[test]
    fn register_payload_includes_exact_mesh_endpoint_when_some() {
        let payload = super::register_payload("PUBKEY", Some("10.13.13.6:9000"));
        let v: serde_json::Value = serde_json::from_str(&payload).unwrap();
        assert_eq!(v["node_pubkey"], "PUBKEY");
        assert_eq!(v["mesh_endpoint"], "10.13.13.6:9000");
    }

    #[test]
    fn parse_roster_reads_pairs_and_defaults_revoked() {
        let body = r#"[
            {"node_pubkey":"AAA","revoked":true},
            {"node_pubkey":"BBB"},
            {"nope":1}
        ]"#;
        let r = parse_roster(body);
        assert_eq!(r.len(), 2);
        assert_eq!(r[0], ("AAA".to_string(), true));
        assert_eq!(r[1], ("BBB".to_string(), false));
    }

    #[test]
    fn parse_roster_handles_garbage() {
        assert!(parse_roster("not json").is_empty());
        assert!(parse_roster("{}").is_empty());
    }

    #[test]
    fn parses_endpoints_skipping_absent_and_null() {
        let body = r#"[
            {"node_pubkey":"A","revoked":false,"mesh_endpoint":"10.13.13.6:9000"},
            {"node_pubkey":"B","revoked":false,"mesh_endpoint":null},
            {"node_pubkey":"C","revoked":false}
        ]"#;
        assert_eq!(
            super::parse_endpoints(body),
            vec![("A".to_string(), "10.13.13.6:9000".to_string())]
        );
    }

    #[test]
    fn parse_roster_still_reads_bodies_without_endpoints() {
        // An older dashboard omits the field entirely; authorization must
        // keep working.
        let body = r#"[{"node_pubkey":"A","revoked":false}]"#;
        assert_eq!(super::parse_roster(body), vec![("A".to_string(), false)]);
        assert!(super::parse_endpoints(body).is_empty());
    }

    use std::io::{Read, Write};
    use std::net::TcpListener;

    /// Minimal mock: serve a fixed JSON body for one request, capture the request line.
    fn mock_once(status: u16, body: &'static str) -> (std::thread::JoinHandle<String>, u16) {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let h = std::thread::spawn(move || {
            let (mut s, _) = listener.accept().unwrap();
            // read until headers + full Content-Length body are in (a single read
            // can return before the body bytes arrive)
            let mut raw = Vec::new();
            let mut buf = [0u8; 2048];
            loop {
                let n = s.read(&mut buf).unwrap_or(0);
                if n == 0 {
                    break;
                }
                raw.extend_from_slice(&buf[..n]);
                let text = String::from_utf8_lossy(&raw);
                if let Some(hdr_end) = text.find("\r\n\r\n") {
                    let want: usize = text
                        .lines()
                        .find_map(|l| {
                            l.to_ascii_lowercase()
                                .strip_prefix("content-length:")
                                .map(|v| v.trim().parse().unwrap_or(0))
                        })
                        .unwrap_or(0);
                    if raw.len() >= hdr_end + 4 + want {
                        break;
                    }
                }
            }
            let req = String::from_utf8_lossy(&raw).to_string();
            let resp = format!(
                "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                body.len()
            );
            let _ = s.write_all(resp.as_bytes());
            req
        });
        (h, port)
    }

    // Network round-trip tests: reliable in isolation, but the blocking mock
    // server can starve under the full suite's fd/CPU pressure. Run explicitly
    // with `cargo test -- --ignored`. The parse logic above is always covered.
    #[test]
    #[ignore = "network mock; run with --ignored"]
    fn fetch_roster_parses_live_response() {
        let (h, port) = mock_once(
            200,
            r#"[{"node_pubkey":"AAA","revoked":false},{"node_pubkey":"BBB","revoked":true}]"#,
        );
        let url = format!("http://127.0.0.1:{port}");
        let roster = fetch_roster(&url, "zk_1_x").unwrap();
        let req = h.join().unwrap();
        assert!(req.starts_with("GET /api/broker/node/roster"));
        assert!(req
            .to_ascii_lowercase()
            .contains("x-broker-api-key: zk_1_x"));
        assert_eq!(roster.len(), 2);
        assert_eq!(roster[1], ("BBB".to_string(), true));
    }

    #[test]
    #[ignore = "network mock; run with --ignored"]
    fn obtain_voucher_extracts_signed_json_and_sig() {
        let (h, port) = mock_once(200, r#"{"signed_json":"{\"v\":1}","sig":"deadbeef"}"#);
        let url = format!("http://127.0.0.1:{port}");
        let (signed, sig) = obtain_voucher(&url, "zk_1_x", 5.0).unwrap();
        let req = h.join().unwrap();
        assert!(req.starts_with("POST /api/broker/voucher"));
        assert!(req.contains("budget_credits"));
        assert_eq!(signed, "{\"v\":1}");
        assert_eq!(sig, "deadbeef");
    }
}