zc2 0.0.12

//! Worker discovery for the P2P broker.
//!
//! Discovery modes:
//! 1. Tailscale mode: Scans the Tailscale subnet (10.13.13.0/24 or 100.x.x.x)
//! 2. Local mode: Falls back to localhost when Tailscale unavailable
//! 3. Manual: Workers can always register via /workers endpoint

use std::net::TcpStream;
use std::sync::Arc;
use std::thread;
use std::time::Duration;

use colored::Colorize;
use super::worker::{HardwareInfo, WorkerPricing, WorkerRegistration, WorkerResources};
use super::BrokerState;

/// Discovery mode
#[derive(Debug, Clone, PartialEq)]
pub enum DiscoveryMode {
    /// Tailscale network available - scan subnet
    Tailscale { subnet: String },
    /// No Tailscale - scan localhost only
    Local,
}

/// Discovery configuration
#[derive(Debug, Clone)]
pub struct DiscoveryConfig {
    /// Tailscale subnet to scan (default: 10.13.13.0/24)
    pub subnet: String,
    /// Primary port workers listen on (default: 3960)
    pub worker_port: u16,
    /// Additional explicit ports to scan for workers
    pub extra_ports: Vec<u16>,
    /// Scan all localhost ports in this inclusive range [start, end].
    /// When set, discovery replaces the fixed extra_ports list with a full
    /// range scan so workers started on arbitrary ports are discovered.
    pub scan_port_range: Option<(u16, u16)>,
    /// Discovery interval in seconds
    pub interval_secs: u64,
    /// Enable active network scanning
    pub enable_scan: bool,
    /// Enable DNS-based discovery
    pub enable_dns: bool,
    /// Explicit peer addresses to probe (from ZAKURO_PEERS env var)
    /// Format: comma-separated "ip:port" or just "ip" (uses worker_port)
    pub peers: Vec<String>,
}

impl Default for DiscoveryConfig {
    fn default() -> Self {
        let peers = std::env::var("ZAKURO_PEERS")
            .unwrap_or_default()
            .split(',')
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
            .collect();

        // Respect ZAKURO_WORKER_PORT env var for custom port configuration
        let worker_port = std::env::var("ZAKURO_WORKER_PORT")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(3960);

        // Allow ZAKURO_SCAN_RANGE="3960-3999" for environment-driven range scan
        let scan_port_range = std::env::var("ZAKURO_SCAN_RANGE")
            .ok()
            .and_then(|v| {
                let parts: Vec<&str> = v.splitn(2, '-').collect();
                if parts.len() == 2 {
                    let start = parts[0].parse::<u16>().ok()?;
                    let end = parts[1].parse::<u16>().ok()?;
                    Some((start, end))
                } else {
                    None
                }
            });

        Self {
            subnet: "10.13.13".to_string(),
            worker_port,
            extra_ports: vec![3961, 3962],  // Common alternative ports
            scan_port_range,
            interval_secs: std::env::var("ZAKURO_SCAN_INTERVAL")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(15),  // Must be less than worker_timeout (30s)
            enable_scan: true,
            enable_dns: true,
            peers,
        }
    }
}

/// Worker discovery service
pub struct Discovery {
    config: DiscoveryConfig,
    state: Arc<BrokerState>,
    mode: DiscoveryMode,
}

impl Discovery {
    /// Create a new discovery service with automatic mode detection
    pub fn new(config: DiscoveryConfig, state: Arc<BrokerState>) -> Self {
        let mode = detect_discovery_mode(&config.subnet);
        Self { config, state, mode }
    }

    /// Get the current discovery mode
    pub fn mode(&self) -> &DiscoveryMode {
        &self.mode
    }

    /// Run discovery loop (blocking)
    pub fn run(&self, verbose: bool) {
        // Log initial mode
        if verbose {
            match &self.mode {
                DiscoveryMode::Tailscale { subnet } => {
                    println!("  {} Tailscale mode (subnet: {}.0/24)", "[DISCOVERY]".cyan(), subnet);
                }
                DiscoveryMode::Local => {
                    println!("  {} Local mode (scanning localhost)", "[DISCOVERY]".yellow());
                }
            }
            if !self.config.peers.is_empty() {
                println!("  {} Peers: {}", "[DISCOVERY]".cyan(), self.config.peers.join(", "));
            }
        }

        // Initial scan immediately
        if self.config.enable_scan {
            self.discover_workers(verbose);
        }

        // Then periodic scans
        loop {
            thread::sleep(Duration::from_secs(self.config.interval_secs));
            if self.config.enable_scan {
                self.discover_workers(verbose);
            }
        }
    }

    /// Discover workers based on current mode
    fn discover_workers(&self, verbose: bool) {
        // Always scan localhost first (finds local worker in any mode)
        self.scan_localhost(verbose);

        // Probe explicit peers early (before slow subnet scan)
        self.scan_peers(verbose);

        // Then mode-specific subnet scanning (skip for CGNAT 100.x.x.x — IPs aren't contiguous)
        if let DiscoveryMode::Tailscale { subnet } = &self.mode {
            if subnet != "peers" && !subnet.starts_with("100.") {
                self.scan_subnet(subnet, verbose);
            }
        }
    }

    /// Probe explicit peer addresses (workers + peer brokers)
    fn scan_peers(&self, verbose: bool) {
        for peer in &self.config.peers {
            let (host, port) = if let Some((h, p)) = peer.rsplit_once(':') {
                (h.to_string(), p.parse().unwrap_or(self.config.worker_port))
            } else {
                (peer.clone(), self.config.worker_port)
            };
            if host.is_empty() {
                continue;
            }

            let broker_port = self.state.config.port;
            if port == broker_port {
                // Peer specified as broker address — fetch workers via broker-to-broker API
                self.try_register_peer_broker(&host, broker_port, verbose);
                self.fetch_workers_from_peer_broker(&host, broker_port, verbose);
            } else {
                // Peer specified as worker port — direct probe (same-network workers)
                self.try_register_worker(&host, port, verbose);
                // Also register the peer broker for credit operations
                self.try_register_peer_broker(&host, broker_port, verbose);
                self.fetch_workers_from_peer_broker(&host, broker_port, verbose);
            }
        }
    }

    /// Fetch workers from a peer broker via /peer/workers and register them locally.
    /// Worker URIs are rewritten from 127.0.0.1:PORT to PEER_IP:PORT so they are
    /// treated as remote (non-local) workers and billed accordingly.
    fn fetch_workers_from_peer_broker(&self, host: &str, broker_port: u16, verbose: bool) {
        let peer_key = self.state.peer_manager.peer_key();
        let url = format!("http://{}:{}/peer/workers", host, broker_port);

        let result = ureq::get(&url)
            .set("X-Peer-Key", peer_key)
            .timeout(Duration::from_secs(3))
            .call();

        let body = match result {
            Ok(resp) => match resp.into_string() {
                Ok(s) => s,
                Err(_) => return,
            },
            Err(_) => return,
        };

        let json: serde_json::Value = match serde_json::from_str(&body) {
            Ok(v) => v,
            Err(_) => return,
        };

        let workers = match json["workers"].as_array() {
            Some(arr) => arr,
            None => return,
        };

        for w in workers {
            let name = w["name"].as_str().unwrap_or("").to_string();
            let uri = w["uri"].as_str().unwrap_or("").to_string();
            let worker_type = w["worker_type"].as_str().unwrap_or("zakuro").to_string();

            if name.is_empty() || uri.is_empty() {
                continue;
            }

            // Rewrite uri: http://127.0.0.1:PORT  →  http://PEER_IP:PORT
            let rewritten_uri = if let Some(rest) = uri.strip_prefix("http://127.0.0.1:") {
                format!("http://{}:{}", host, rest.split('/').next().unwrap_or("3960"))
            } else if let Some(rest) = uri.strip_prefix("http://localhost:") {
                format!("http://{}:{}", host, rest.split('/').next().unwrap_or("3960"))
            } else {
                uri.clone() // already has correct host
            };

            // If already registered, refresh heartbeat to keep it alive, then skip
            let existing = self.state.workers.list();
            if let Some(known) = existing.iter().find(|e| e.name == name || e.uri == rewritten_uri) {
                self.state.workers.refresh_heartbeat(&known.id);
                continue;
            }

            let registration = WorkerRegistration {
                name: name.clone(),
                uri: rewritten_uri.clone(),
                worker_type,
                resources: WorkerResources {
                    cpus_available: w["cpus_available"].as_f64().unwrap_or(1.0),
                    cpus_total: w["cpus_total"].as_f64().unwrap_or(1.0),
                    memory_available: (w["memory_available_gib"].as_f64().unwrap_or(1.0) * 1024.0 * 1024.0 * 1024.0) as u64,
                    memory_total: (w["memory_total_gib"].as_f64().unwrap_or(1.0) * 1024.0 * 1024.0 * 1024.0) as u64,
                    gpus_available: w["gpus_available"].as_u64().unwrap_or(0) as u32,
                    gpus_total: w["gpus_total"].as_u64().unwrap_or(0) as u32,
                },
                pricing: WorkerPricing {
                    price_per_hour: w["price_per_hour"].as_f64().unwrap_or(3.6),
                    min_charge: w["min_charge"].as_f64().unwrap_or(0.001),
                },
                tags: vec![],
                max_timeout_secs: 0.0,
                hardware: HardwareInfo {
                    cpu_model: w["cpu_model"].as_str().map(|s| s.to_string()),
                    gpu_model: w["gpu_model"].as_str().map(|s| s.to_string()),
                    gpu_vram_gb: w["gpu_vram_gb"].as_u64().map(|v| v as u32),
                    storage_gb: w["storage_gb"].as_u64().map(|v| v as u32),
                },
                tailscale_ip: Some(host.to_string()),
                is_docker: w["is_docker"].as_bool(),
            };

            let worker = self.state.workers.register(registration);
            if verbose {
                println!(
                    "  {} Discovered peer worker {} at {} (via broker {}:{})",
                    "[DISCOVERY]".cyan(),
                    worker.name,
                    rewritten_uri,
                    host,
                    broker_port,
                );
            }
        }
    }

    /// Try to register a peer broker at the given address.
    /// Probes /peer/health and registers in PeerManager if alive.
    fn try_register_peer_broker(&self, host: &str, port: u16, verbose: bool) {
        let addr = format!("{}:{}", host, port);

        // Resolve hostname to socket address (supports both IPs and hostnames)
        use std::net::ToSocketAddrs;
        let sock_addr: std::net::SocketAddr = match addr.to_socket_addrs() {
            Ok(mut addrs) => match addrs.next() {
                Some(a) => a,
                None => return,
            },
            Err(_) => return,
        };

        if TcpStream::connect_timeout(&sock_addr, Duration::from_millis(200)).is_err() {
            return;
        }

        let base_url = format!("http://{}:{}", host, port);
        self.state.peer_manager.register_peer(base_url.clone());

        // Check health
        if let Some(client) = self.state.peer_manager.get_client(&base_url) {
            if client.check_health() && verbose {
                println!(
                    "  {} Peer broker alive at {}:{}",
                    "[DISCOVERY]".cyan(),
                    host, port
                );
            }
        }
    }

    /// Scan localhost for workers — parallel TCP probing for fast discovery.
    ///
    /// All ports are probed concurrently: TCP SYN is sent to every port
    /// simultaneously, so the total time is bounded by a single connect
    /// timeout rather than N × timeout for sequential scanning.
    fn scan_localhost(&self, verbose: bool) {
        // Build the port list to scan
        let ports: Vec<u16> = if let Some((start, end)) = self.config.scan_port_range {
            (start..=end).collect()
        } else {
            let mut p = vec![self.config.worker_port];
            p.extend(&self.config.extra_ports);
            p
        };

        // Phase 1 — parallel TCP connect (50ms timeout, fast reject for closed ports)
        let open_ports: Vec<u16> = {
            let (tx, rx) = std::sync::mpsc::channel();
            let mut handles = Vec::with_capacity(ports.len());

            for port in &ports {
                let port = *port;
                let tx = tx.clone();
                handles.push(thread::spawn(move || {
                    let addr: std::net::SocketAddr =
                        format!("127.0.0.1:{}", port).parse().unwrap();
                    if TcpStream::connect_timeout(&addr, Duration::from_millis(50)).is_ok() {
                        let _ = tx.send(port);
                    }
                }));
            }
            drop(tx); // close sender so rx.iter() terminates

            let mut open: Vec<u16> = rx.iter().collect();
            for h in handles {
                let _ = h.join();
            }
            open.sort_unstable();
            open
        };

        // Phase 2 — probe only the open ports (health + info) in parallel.
        // We use a thread per open port; on loopback each call is < 5ms.
        // Arc the state so each thread can register workers independently.
        let state = Arc::clone(&self.state);
        let config = self.config.clone();
        let mut probe_handles = Vec::with_capacity(open_ports.len());

        for port in open_ports {
            let state = Arc::clone(&state);
            let config = config.clone();
            probe_handles.push(thread::spawn(move || {
                let tmp = Discovery { config, state: Arc::clone(&state), mode: DiscoveryMode::Local };
                tmp.try_register_worker("127.0.0.1", port, verbose);
            }));
        }
        for h in probe_handles {
            let _ = h.join();
        }
    }

    /// Scan a subnet for workers (Tailscale mode).
    ///
    /// Phase 1: parallel TCP SYN to all 253 hosts (50ms timeout) to find open ports fast.
    /// Phase 2: probe only reachable hosts for worker info (in parallel).
    /// Total time ≈ 50ms + per-host probe time, not 253 × 50ms sequential.
    fn scan_subnet(&self, subnet: &str, verbose: bool) {
        let port = self.config.worker_port;

        // Phase 1 — parallel TCP connect to all hosts in the /24 subnet (skip .0 and .1)
        let open_hosts: Vec<String> = {
            let (tx, rx) = std::sync::mpsc::channel();
            let mut handles = Vec::with_capacity(253);

            for i in 2u8..=254 {
                let ip = format!("{}.{}", subnet, i);
                let tx = tx.clone();
                handles.push(thread::spawn(move || {
                    let addr_str = format!("{}:{}", ip, port);
                    if let Ok(addr) = addr_str.parse::<std::net::SocketAddr>() {
                        if TcpStream::connect_timeout(&addr, Duration::from_millis(50)).is_ok() {
                            let _ = tx.send(ip);
                        }
                    }
                }));
            }
            drop(tx);

            let mut hosts: Vec<String> = rx.iter().collect();
            for h in handles {
                let _ = h.join();
            }
            hosts.sort();
            hosts
        };

        // Phase 2 — probe only the open hosts (health + info) in parallel
        let state = Arc::clone(&self.state);
        let config = self.config.clone();
        let mut probe_handles = Vec::with_capacity(open_hosts.len());

        for host in open_hosts {
            let state = Arc::clone(&state);
            let config = config.clone();
            probe_handles.push(thread::spawn(move || {
                let tmp = Discovery { config, state: Arc::clone(&state), mode: DiscoveryMode::Local };
                tmp.try_register_worker(&host, port, verbose);
            }));
        }
        for h in probe_handles {
            let _ = h.join();
        }
    }

    /// Try to register a worker at the given address
    fn try_register_worker(&self, host: &str, port: u16, verbose: bool) {
        let addr = format!("{}:{}", host, port);

        // Resolve hostname to socket address (supports both IPs and hostnames)
        use std::net::ToSocketAddrs;
        let sock_addr = match addr.to_socket_addrs() {
            Ok(mut addrs) => match addrs.next() {
                Some(a) => a,
                None => return,
            },
            Err(_) => return,
        };

        // Quick connection check (short timeout — loopback is nearly instant)
        if TcpStream::connect_timeout(&sock_addr, Duration::from_millis(50)).is_err() {
            return;
        }

        // Build canonical URI for dedup
        let uri = format!("http://{}:{}", host, port);

        // Check if already registered (by URI or localhost variants)
        let existing = self.state.workers.list();
        let existing_worker = existing.iter().find(|w| {
            w.uri == uri ||
            (host == "127.0.0.1" && w.uri.contains("localhost")) ||
            (host == "localhost" && w.uri.contains("127.0.0.1"))
        });

        if let Some(worker) = existing_worker {
            // Worker exists and TCP reachable — re-probe /info to get fresh resources
            if let Some(info) = self.probe_worker(host, port) {
                if let Some(resources) = info.resources {
                    self.state.workers.update_resources(
                        &worker.id,
                        resources,
                        info.hardware.unwrap_or_default(),
                    );
                } else {
                    self.state.workers.refresh_heartbeat(&worker.id);
                }
            } else {
                self.state.workers.refresh_heartbeat(&worker.id);
            }
            return;
        }

        // New endpoint — do full probe to get worker info
        if let Some(worker_info) = self.probe_worker(host, port) {
            // Dedup by worker name (same worker at different IPs)
            let worker_name = worker_info.name.as_deref().unwrap_or("");
            let name_match = existing.iter().find(|w| {
                !worker_name.is_empty() && w.name == worker_name
            });

            if let Some(worker) = name_match {
                // Same worker name at different IP — update resources
                if let Some(resources) = worker_info.resources.clone() {
                    self.state.workers.update_resources(
                        &worker.id,
                        resources,
                        worker_info.hardware.clone().unwrap_or_default(),
                    );
                } else {
                    self.state.workers.refresh_heartbeat(&worker.id);
                }
            } else {
                let registration = WorkerRegistration {
                    name: worker_info.name.unwrap_or_else(|| format!("worker-{}", host)),
                    uri,
                    worker_type: worker_info.worker_type.unwrap_or_else(|| "zakuro".to_string()),
                    resources: worker_info.resources.unwrap_or_default(),
                    pricing: worker_info.pricing.unwrap_or_default(),
                    tags: worker_info.tags.unwrap_or_default(),
                    max_timeout_secs: 0.0,
                    hardware: worker_info.hardware.unwrap_or_default(),
                    tailscale_ip: None,
                    is_docker: None,
                };

                let worker = self.state.workers.register(registration);
                if verbose {
                    println!(
                        "  {} Discovered worker {} at {}",
                        "[DISCOVERY]".cyan(),
                        worker.name,
                        worker.uri
                    );
                }

                // Sync discovered worker immediately to dashboard
                if let Some(ref owner_id) = self.state.config.owner_user_id {
                    let node_name = self.state.config.node_name.as_deref();

                    // Prefer API sync if configured
                    if let (Some(ref api_url), Some(ref api_key)) =
                        (&self.state.config.api_url, &self.state.config.api_key)
                    {
                        match crate::broker::ledger::Ledger::sync_workers_via_api(
                            owner_id,
                            &vec![worker.clone()],
                            api_url,
                            api_key,
                            node_name,
                            self.state.own_tailscale_ip.as_deref(),
                        ) {
                            Ok(()) => {
                                if verbose {
                                    println!("  [WORKER_SYNC] Worker {} synced to dashboard", worker.name);
                                }
                            }
                            Err(e) => {
                                eprintln!("  [WORKER_SYNC] Failed to sync {}: {}", worker.name, e);
                            }
                        }
                    }
                }
            }
        }
    }

    /// Probe a potential worker for its info
    /// Only returns Some if the worker has a valid /info endpoint with worker_type
    fn probe_worker(&self, ip: &str, port: u16) -> Option<WorkerProbeResult> {
        let health_url = format!("http://{}:{}/health", ip, port);

        // First check health
        match ureq::get(&health_url)
            .timeout(Duration::from_secs(2))
            .call()
        {
            Ok(response) if response.status() == 200 => {
                // Must have a valid /info endpoint to be a zakuro worker
                let info_url = format!("http://{}:{}/info", ip, port);
                match ureq::get(&info_url)
                    .timeout(Duration::from_secs(2))
                    .call()
                {
                    Ok(info_response) if info_response.status() == 200 => {
                        if let Ok(body) = info_response.into_string() {
                            if let Ok(info) = serde_json::from_str::<WorkerProbeResult>(&body) {
                                // Only accept workers with a recognized worker_type
                                if info.worker_type.is_some() {
                                    return Some(info);
                                }
                            }
                        }
                        None
                    }
                    _ => None, // No /info endpoint = not a zakuro worker
                }
            }
            _ => None,
        }
    }
}

/// Result from probing a worker
#[derive(Debug, Clone, serde::Deserialize)]
struct WorkerProbeResult {
    name: Option<String>,
    worker_type: Option<String>,
    resources: Option<WorkerResources>,
    pricing: Option<WorkerPricing>,
    tags: Option<Vec<String>>,
    /// Hardware details reported by worker
    #[serde(default)]
    hardware: Option<HardwareInfo>,
}

/// Trait for colored output
trait ColorExt {
    fn cyan(&self) -> String;
    fn yellow(&self) -> String;
    fn green(&self) -> String;
}

impl ColorExt for &str {
    fn cyan(&self) -> String {
        format!("\x1b[36m{}\x1b[0m", self)
    }
    fn yellow(&self) -> String {
        format!("\x1b[33m{}\x1b[0m", self)
    }
    fn green(&self) -> String {
        format!("\x1b[32m{}\x1b[0m", self)
    }
}

/// Detect the appropriate discovery mode based on network availability
pub fn detect_discovery_mode(preferred_subnet: &str) -> DiscoveryMode {
    // First, check for Tailscale IP (env var or interface detection)
    if let Some(tailscale_ip) = get_tailscale_ip() {
        // Extract subnet from Tailscale IP (e.g., "100.64.0.5" -> "100.64.0")
        let parts: Vec<&str> = tailscale_ip.split('.').collect();
        if parts.len() == 4 {
            let subnet = format!("{}.{}.{}", parts[0], parts[1], parts[2]);
            return DiscoveryMode::Tailscale { subnet };
        }
    }

    // Check if preferred subnet is reachable (might be on VPN/custom network)
    let test_ip = format!("{}.1", preferred_subnet);
    if let Ok(addr) = format!("{}:1", test_ip).parse() {
        if TcpStream::connect_timeout(&addr, Duration::from_millis(50)).is_ok() {
            return DiscoveryMode::Tailscale {
                subnet: preferred_subnet.to_string(),
            };
        }
    }

    // If peers are configured, use Tailscale mode with a dummy subnet
    // (actual discovery happens via peer probing, not subnet scan)
    if std::env::var("ZAKURO_PEERS").map(|v| !v.is_empty()).unwrap_or(false) {
        return DiscoveryMode::Tailscale {
            subnet: "peers".to_string(),
        };
    }

    // Fall back to local mode
    DiscoveryMode::Local
}

/// Get the best available IP for this node to advertise to peers.
/// Prefers Tailscale (100.x.x.x) over the primary LAN IP.
/// Returns None only if no non-loopback IP can be determined.
pub fn get_effective_node_ip() -> Option<String> {
    if let Some(ip) = get_tailscale_ip() {
        return Some(ip);
    }
    // Fall back to the primary LAN IP (for brokers not on Tailscale)
    if let Ok(socket) = std::net::UdpSocket::bind("0.0.0.0:0") {
        if socket.connect("8.8.8.8:80").is_ok() {
            if let Ok(addr) = socket.local_addr() {
                let ip = addr.ip().to_string();
                if ip != "127.0.0.1" && ip != "::1" {
                    return Some(ip);
                }
            }
        }
    }
    None
}

/// Discover other broker instances on localhost by probing /peer/health.
/// Used when ZAKURO_PEERS is empty so brokers on the same machine can find each other
/// without manual config. Excludes `self_port` from the result.
/// Returns URLs like "http://127.0.0.1:9001".
pub fn discover_broker_peers_on_localhost(
    self_port: u16,
    port_start: u16,
    port_end: u16,
) -> Vec<String> {
    let mut out = Vec::new();
    let agent = ureq::AgentBuilder::new()
        .timeout(Duration::from_millis(500))
        .build();
    for port in port_start..=port_end {
        if port == self_port {
            continue;
        }
        let url = format!("http://127.0.0.1:{}/peer/health", port);
        if agent.get(&url).call().map(|r| r.status() == 200).unwrap_or(false) {
            out.push(format!("http://127.0.0.1:{}", port));
        }
    }
    out
}

/// Get the local Tailscale IP
pub fn get_tailscale_ip() -> Option<String> {
    // Check env var first (for userspace Tailscale / Docker sidecar mode)
    if let Ok(ip) = std::env::var("ZAKURO_TAILSCALE_IP") {
        if !ip.is_empty() {
            return Some(ip);
        }
    }

    #[cfg(unix)]
    {
        for iface in ifaces::Interface::get_all().ok()?.into_iter() {
            // Only check the Tailscale kernel interface; wg0 is a generic WireGuard name
            // that is NOT specific to Tailscale and can false-positive on other VPNs.
            if iface.name == "tailscale0" || iface.name.starts_with("ts") {
                if let Some(addr) = iface.addr {
                    let addr_str = addr.to_string();
                    // Remove port suffix if present
                    let ip = addr_str.trim_end_matches(":0");
                    // Tailscale always uses 100.64.0.0/10 (CGNAT range)
                    if ip.starts_with("100.") {
                        return Some(ip.to_string());
                    }
                }
            }
        }
    }
    None
}