//! Local OS tools for diagnostics (port of `diagnostics/processTools.js`): single-packet
//! ping probes with a bounded overlap (no root/flood needed) and traceroute/tracert.
//! Arguments are passed without a shell; output is size-limited.

use crate::core::diagnostics::common::{now_ms, stats, DiagError, Options};
use crate::core::ssh::latency::parse_ping_replies;
use once_cell::sync::Lazy;
use regex::Regex;
use serde_json::{json, Value};
use std::net::IpAddr;
use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::AsyncReadExt;
use tokio_util::sync::CancellationToken;

static CONFIG_ERROR: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)not permitted|permission denied|invalid option|unknown option|invalid argument|out of range|illegal|usage:").unwrap());
static TRANSIENT_ERROR: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)network is unreachable|no route to host|general failure|transmit failed|no buffer space|resource temporarily unavailable|host is down").unwrap());
static EXPLICIT_TIMEOUT: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)Request timed out|Request timeout|Zeitüberschreitung|Vypršel.*časov[ýy].*limit").unwrap());
const MAX_PING_IN_FLIGHT: u64 = 32;
const MAX_CONSECUTIVE_PING_ERRORS: u64 = 50;

pub struct ProcessResult {
    pub code: Option<i32>,
    pub stdout: Vec<u8>,
    pub stderr: String,
}

pub fn clean_env(command: &mut tokio::process::Command, inherit: bool) {
    if !inherit {
        command.env_clear();
        for name in ["SystemRoot", "SYSTEMROOT", "WINDIR", "TEMP", "TMP", "TMPDIR"] {
            if let Ok(value) = std::env::var(name) {
                command.env(name, value);
            }
        }
    }
    command.env("LC_ALL", "C").env("LANG", "C");
}

pub async fn run_process(
    executable: &str,
    args: &[String],
    timeout_ms: u64,
    cancel: &CancellationToken,
    max_bytes: usize,
    on_line: Option<&(dyn Fn(&str) + Send + Sync)>,
    curated_env: bool,
) -> Result<ProcessResult, DiagError> {
    if cancel.is_cancelled() {
        return Err(DiagError::new("CANCELLED"));
    }
    let mut command = tokio::process::Command::new(executable);
    command.args(args).stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped()).kill_on_drop(true);
    clean_env(&mut command, !curated_env);
    #[cfg(windows)]
    command.creation_flags(0x0800_0000);
    let mut child = command.spawn().map_err(|e| if e.kind() == std::io::ErrorKind::NotFound { DiagError::with_message("ENOENT", e.to_string()) } else { DiagError::from_io(e) })?;
    let mut stdout = child.stdout.take().expect("stdout");
    let mut stderr = child.stderr.take().expect("stderr");
    let work = async {
        let mut out = Vec::new();
        let mut err = Vec::new();
        let mut pending = String::new();
        let mut buffer = [0u8; 8192];
        let mut err_buffer = [0u8; 4096];
        let mut out_open = true;
        let mut err_open = true;
        while out_open || err_open {
            tokio::select! {
                n = stdout.read(&mut buffer), if out_open => {
                    let n = n.map_err(DiagError::from_io)?;
                    if n == 0 { out_open = false; continue; }
                    if out.len() + err.len() + n > max_bytes { return Err(DiagError::new("OUTPUT_LIMIT")); }
                    out.extend_from_slice(&buffer[..n]);
                    if let Some(callback) = on_line {
                        pending.push_str(&String::from_utf8_lossy(&buffer[..n]));
                        while let Some(index) = pending.find('\n') {
                            let line: String = pending.drain(..=index).collect();
                            callback(line.trim_end_matches(['\r', '\n']));
                        }
                    }
                }
                n = stderr.read(&mut err_buffer), if err_open => {
                    let n = n.map_err(DiagError::from_io)?;
                    if n == 0 { err_open = false; continue; }
                    if out.len() + err.len() + n > max_bytes { return Err(DiagError::new("OUTPUT_LIMIT")); }
                    err.extend_from_slice(&err_buffer[..n]);
                }
            }
        }
        if let Some(callback) = on_line {
            if !pending.is_empty() {
                callback(pending.trim_end_matches('\r'));
            }
        }
        let status = child.wait().await.map_err(DiagError::from_io)?;
        Ok(ProcessResult { code: status.code(), stdout: out, stderr: String::from_utf8_lossy(&err).to_string() })
    };
    tokio::select! {
        result = tokio::time::timeout(Duration::from_millis(timeout_ms), work) => result.map_err(|_| DiagError::new("TIMEOUT"))?,
        _ = cancel.cancelled() => Err(DiagError::new("CANCELLED")),
    }
}

pub fn ping_command(address: &str, timeout_ms: u64, options: &Options) -> Result<(String, Vec<String>), DiagError> {
    let ip: IpAddr = address.parse().map_err(|_| DiagError::new("INVALID_IP"))?;
    let v6 = ip.is_ipv6();
    let size = |flag: &str| options.packet_size.map(|s| vec![flag.to_string(), s.to_string()]).unwrap_or_default();
    let hops = |flag: &str| options.ttl.map(|t| vec![flag.to_string(), t.to_string()]).unwrap_or_default();
    let df = |flags: &[&str]| if options.dont_fragment && !v6 { flags.iter().map(|f| f.to_string()).collect() } else { Vec::new() };
    let mut args: Vec<String>;
    let executable: &str;
    if cfg!(windows) {
        executable = "ping.exe";
        args = vec![if v6 { "-6" } else { "-4" }.into(), "-n".into(), "1".into(), "-w".into(), timeout_ms.to_string()];
        args.extend(size("-l"));
        args.extend(hops("-i"));
        args.extend(df(&["-f"]));
    } else if cfg!(target_os = "macos") {
        executable = if v6 { "/sbin/ping6" } else { "/sbin/ping" };
        args = vec!["-n".into(), "-c".into(), "1".into()];
        if !v6 {
            args.extend(["-W".to_string(), timeout_ms.to_string()]);
        }
        args.extend(size("-s"));
        args.extend(hops(if v6 { "-h" } else { "-m" }));
        args.extend(df(&["-D"]));
    } else if cfg!(target_os = "linux") {
        executable = "ping";
        args = vec![if v6 { "-6" } else { "-4" }.into(), "-n".into(), "-c".into(), "1".into(), "-W".into(), timeout_ms.div_ceil(1000).to_string()];
        args.extend(size("-s"));
        args.extend(hops("-t"));
        args.extend(df(&["-M", "do"]));
    } else {
        return Err(DiagError::new("PLATFORM_UNSUPPORTED"));
    }
    args.push(address.to_string());
    Ok((executable.to_string(), args))
}

pub fn parse_ping_reply(output: &str) -> Value {
    let exact = parse_ping_replies(output).into_iter().find(|r| !r.less_than && r.ms > 0.0).map(|r| r.ms);
    let upper = Regex::new(r"(?i)<\s*(\d+(?:[.,]\d+)?)\s*ms").unwrap().captures(output).and_then(|c| c[1].replace(',', ".").parse::<f64>().ok());
    let ttl = Regex::new(r"(?i)(?:ttl|hlim|hop limit)[=:\s]+(\d+)").unwrap().captures(output).and_then(|c| c[1].parse::<u64>().ok());
    json!({ "received": exact.is_some() || upper.is_some(), "rttMs": exact, "upperBoundMs": upper, "ttl": ttl })
}

struct PingState {
    packets: Vec<Value>,
    rtts: Vec<Option<f64>>,
    sent: u64,
    received: u64,
    error_packets: u64,
    consecutive_errors: u64,
    failure: Option<Value>,
}

fn summary(state: &PingState) -> Value {
    json!({
        "sent": state.sent,
        "received": state.received,
        "errorPackets": state.error_packets,
        "lossPercent": if state.sent > 0 { json!(100.0 * (state.sent - state.received) as f64 / state.sent as f64) } else { Value::Null },
        "statistics": stats(&state.rtts)
    })
}

fn merge(mut base: Value, extra: Value) -> Value {
    if let (Some(target), Value::Object(fields)) = (base.as_object_mut(), extra) {
        target.extend(fields);
    }
    base
}

pub async fn ping(address: &str, options: &Options, cancel: &CancellationToken, progress: Arc<dyn Fn(Value) + Send + Sync>) -> Result<Value, DiagError> {
    let interval = options.interval_ms.max(1);
    let timeout = options.ping_timeout_ms.unwrap_or(options.timeout_ms);
    let total = if options.continuous { u64::MAX } else { options.count * options.repetitions };
    let max_in_flight = (timeout + 500).div_ceil(interval).saturating_add(1).clamp(1, MAX_PING_IN_FLIGHT);
    let state = Arc::new(parking_lot::Mutex::new(PingState { packets: Vec::new(), rtts: Vec::new(), sent: 0, received: 0, error_packets: 0, consecutive_errors: 0, failure: None }));
    let (executable, _) = ping_command(address, timeout, options)?;
    let _ = executable;
    let mut in_flight = tokio::task::JoinSet::new();
    let mut launched = 0u64;
    let mut throttled = 0u64;
    let started = now_ms();
    let mut next_at = started;
    let mut last_launch = started;
    let base_options = options.clone();
    while launched < total && state.lock().failure.is_none() {
        if cancel.is_cancelled() {
            in_flight.abort_all();
            return Err(DiagError::new("CANCELLED"));
        }
        if in_flight.len() as u64 >= max_in_flight {
            throttled += 1;
            tokio::select! {
                _ = in_flight.join_next() => {},
                _ = cancel.cancelled() => { in_flight.abort_all(); return Err(DiagError::new("CANCELLED")); }
            }
            continue;
        }
        launched += 1;
        let sequence = launched;
        let address = address.to_string();
        let state = state.clone();
        let progress = progress.clone();
        let cancel_probe = cancel.clone();
        let probe_options = base_options.clone();
        in_flight.spawn(async move {
            let (exe, args) = match ping_command(&address, timeout, &probe_options) {
                Ok(c) => c,
                Err(e) => {
                    state.lock().failure = Some(json!({ "sequence": sequence, "address": address, "received": false, "rttMs": null, "status": "error", "code": e.code, "raw": e.message }));
                    return;
                }
            };
            let mut packet = match run_process(&exe, &args, timeout + 500, &cancel_probe, 256 * 1024, None, false).await {
                Ok(result) => {
                    let raw = String::from_utf8_lossy(&result.stdout).to_string();
                    let reply = parse_ping_reply(&raw);
                    let received = reply["received"] == true;
                    let status = if received { "success" } else if EXPLICIT_TIMEOUT.is_match(&raw) { "timeout" } else { "no-reply" };
                    let combined = format!("{raw}{}", result.stderr);
                    let mut packet = merge(json!({ "sequence": sequence, "address": address }), reply);
                    packet["status"] = json!(status);
                    packet["raw"] = json!(combined.chars().take(if received { 300 } else { 4096 }).collect::<String>());
                    let threshold = if cfg!(target_os = "macos") { 2 } else { 1 };
                    if !received && (result.code.map(|c| c > threshold).unwrap_or(false) || CONFIG_ERROR.is_match(&result.stderr) || TRANSIENT_ERROR.is_match(&result.stderr)) {
                        packet["status"] = json!("error");
                        packet["fatal"] = json!(CONFIG_ERROR.is_match(&result.stderr));
                    }
                    packet
                }
                Err(error) => {
                    if cancel_probe.is_cancelled() {
                        return;
                    }
                    let status = match error.code.as_str() { "ENOENT" => "unavailable", "TIMEOUT" => "timeout", _ => "error" };
                    json!({ "sequence": sequence, "address": address, "received": false, "rttMs": null, "status": status, "code": error.code, "raw": error.message.chars().take(400).collect::<String>() })
                }
            };
            if cancel_probe.is_cancelled() {
                return;
            }
            let mut guard = state.lock();
            if guard.failure.is_some() {
                return;
            }
            packet["measuredAt"] = json!(chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true));
            let is_error = packet["status"] == "error";
            let fatal = packet["status"] == "unavailable" || (is_error && (packet["fatal"] == true || guard.sent == 0 || guard.consecutive_errors + 1 >= MAX_CONSECUTIVE_PING_ERRORS));
            if let Some(object) = packet.as_object_mut() {
                object.remove("fatal");
            }
            if fatal {
                if is_error && guard.sent > 0 && packet.get("code").is_none() {
                    packet["code"] = json!("PING_REPEATED_ERRORS");
                }
                guard.packets.push(packet.clone());
                guard.failure = Some(packet);
                return;
            }
            guard.consecutive_errors = if is_error { guard.consecutive_errors + 1 } else { 0 };
            if is_error {
                guard.error_packets += 1;
            }
            guard.sent += 1;
            if packet["received"] == true {
                guard.received += 1;
                let index = (sequence - 1) as usize;
                if guard.rtts.len() <= index {
                    guard.rtts.resize(index + 1, None);
                }
                guard.rtts[index] = packet["rttMs"].as_f64();
            }
            guard.packets.push(packet.clone());
            if guard.packets.len() > 500 {
                guard.packets.remove(0);
            }
            let event = merge(merge(json!({ "kind": "ping-packet" }), packet), summary(&guard));
            drop(guard);
            progress(event);
        });
        last_launch = now_ms();
        next_at = (next_at + interval as f64).max(last_launch - interval as f64 / 2.0);
        let wait = next_at - now_ms();
        if launched < total && wait > 0.0 {
            tokio::select! {
                _ = tokio::time::sleep(Duration::from_secs_f64(wait / 1000.0)) => {},
                _ = cancel.cancelled() => { in_flight.abort_all(); return Err(DiagError::new("CANCELLED")); }
            }
        }
    }
    while in_flight.join_next().await.is_some() {}
    if cancel.is_cancelled() {
        return Err(DiagError::new("CANCELLED"));
    }
    let mut guard = state.lock();
    guard.packets.sort_by_key(|p| p["sequence"].as_u64().unwrap_or(0));
    let timing = json!({
        "intervalMs": interval,
        "effectiveIntervalMs": if launched > 1 { json!((last_launch - started) / (launched - 1) as f64) } else { Value::Null },
        "timeoutMs": timeout,
        "maxInFlight": max_in_flight,
        "throttled": throttled,
        "packetSize": options.packet_size,
        "ttl": options.ttl,
        "dontFragment": options.dont_fragment
    });
    let sum = summary(&guard);
    if let Some(failure) = guard.failure.clone() {
        progress(merge(merge(json!({ "kind": "ping-packet" }), failure.clone()), sum.clone()));
        return Ok(merge(
            json!({ "status": failure["status"], "code": failure.get("code").cloned().unwrap_or(json!("PING_TOOL_ERROR")), "reason": failure["raw"], "address": address, "packets": guard.packets }),
            merge(sum, json!({ "timing": timing, "note": "ICMP failure does not establish server unavailability or prove a firewall." })),
        ));
    }
    Ok(merge(
        json!({ "status": if guard.received > 0 { "success" } else { "warning" }, "address": address, "packets": std::mem::take(&mut guard.packets) }),
        merge(sum, json!({ "timing": timing, "note": "No ICMP reply may mean filtering, ICMP rate limiting or loss; it does not establish server unavailability." })),
    ))
}

pub fn traceroute_command(address: &str, options: &Options) -> Result<(String, Vec<String>), DiagError> {
    let ip: IpAddr = address.parse().map_err(|_| DiagError::new("INVALID_IP"))?;
    let v6 = ip.is_ipv6();
    let hops = options.max_hops.to_string();
    if cfg!(windows) {
        Ok(("tracert.exe".into(), vec![if v6 { "-6" } else { "-4" }.into(), "-d".into(), "-h".into(), hops, "-w".into(), options.timeout_ms.to_string(), address.into()]))
    } else if cfg!(target_os = "macos") {
        Ok(((if v6 { "/usr/sbin/traceroute6" } else { "/usr/sbin/traceroute" }).into(), vec!["-n".into(), "-m".into(), hops, "-w".into(), options.timeout_ms.div_ceil(1000).to_string(), address.into()]))
    } else if cfg!(target_os = "linux") {
        let wait = options.timeout_ms as f64 / 1000.0;
        Ok(("traceroute".into(), vec![if v6 { "-6" } else { "-4" }.into(), "-n".into(), "-m".into(), hops, "-w".into(), if wait.fract() == 0.0 { format!("{}", wait as u64) } else { wait.to_string() }, address.into()]))
    } else {
        Err(DiagError::new("PLATFORM_UNSUPPORTED"))
    }
}

pub fn parse_hop(line: &str) -> Option<Value> {
    let caps = Regex::new(r"^\s*(\d{1,2})\s+(.+)").unwrap().captures(line)?;
    let rest = caps[2].to_string();
    let mut addresses: Vec<String> = Vec::new();
    for token in rest.split_whitespace() {
        let cleaned: String = token.chars().filter(|c| !"()[],".contains(*c)).collect();
        if cleaned.parse::<IpAddr>().is_ok() && !addresses.contains(&cleaned) {
            addresses.push(cleaned);
        }
    }
    let probe_re = Regex::new(r"(?i)(<\s*)?(\d+(?:[.,]\d+)?)\s*ms|\*").unwrap();
    let probes: Vec<Value> = probe_re
        .captures_iter(&rest)
        .map(|m| {
            if &m[0] == "*" {
                json!({ "status": "timeout", "rttMs": null })
            } else {
                let value = m[2].replace(',', ".").parse::<f64>().unwrap_or(0.0);
                if m.get(1).is_some() { json!({ "status": "success", "rttMs": null, "upperBoundMs": value }) } else { json!({ "status": "success", "rttMs": value, "upperBoundMs": null }) }
            }
        })
        .collect();
    if probes.is_empty() && addresses.is_empty() {
        return None;
    }
    let rtts: Vec<Option<f64>> = probes.iter().map(|p| p["rttMs"].as_f64()).collect();
    let timeouts = probes.iter().filter(|p| p["status"] == "timeout").count();
    Some(json!({
        "hop": caps[1].parse::<u64>().unwrap_or(0),
        "addresses": addresses,
        "probes": probes,
        "averageMs": stats(&rtts)["average"],
        "lossPercent": if probes.is_empty() { Value::Null } else { json!(100.0 * timeouts as f64 / probes.len() as f64) },
        "raw": line.chars().take(1024).collect::<String>()
    }))
}

fn same_ip(a: &str, b: &str) -> bool {
    match (a.parse::<IpAddr>(), b.parse::<IpAddr>()) {
        (Ok(x), Ok(y)) => x == y,
        _ => a == b,
    }
}

pub async fn traceroute(address: &str, options: &Options, cancel: &CancellationToken, progress: Arc<dyn Fn(Value) + Send + Sync>) -> Result<Value, DiagError> {
    let mut routes: Vec<Value> = Vec::new();
    for iteration in 0..options.repetitions {
        if cancel.is_cancelled() {
            return Err(DiagError::new("CANCELLED"));
        }
        let (exe, args) = traceroute_command(address, options)?;
        let hops = Arc::new(parking_lot::Mutex::new(Vec::<Value>::new()));
        let hops_cb = hops.clone();
        let progress_cb = progress.clone();
        let address_owned = address.to_string();
        let on_line = move |line: &str| {
            if let Some(hop) = parse_hop(line) {
                hops_cb.lock().push(hop.clone());
                progress_cb(merge(json!({ "kind": "trace-hop", "iteration": iteration, "address": address_owned }), hop));
            }
        };
        let timeout = (options.max_hops * options.timeout_ms.div_ceil(1000) * 1000 * 3 + 2000).min(180_000);
        match run_process(&exe, &args, timeout, cancel, 256 * 1024, Some(&on_line), false).await {
            Ok(result) => {
                let hops = hops.lock().clone();
                routes.push(json!({ "iteration": iteration, "hops": hops, "exitCode": result.code, "raw": String::from_utf8_lossy(&result.stdout).chars().take(32000).collect::<String>(), "reason": result.stderr.chars().take(1000).collect::<String>() }));
            }
            Err(error) => {
                if cancel.is_cancelled() {
                    return Err(error);
                }
                let hops = hops.lock().clone();
                if !hops.is_empty() {
                    routes.push(json!({ "iteration": iteration, "hops": hops, "incomplete": true, "reason": error.message }));
                }
                return Ok(json!({ "status": if error.code == "ENOENT" { "unavailable" } else { "error" }, "code": error.code, "reason": error.message, "address": address, "routes": routes }));
            }
        }
    }
    let reached = routes.iter().any(|r| r["hops"].as_array().map(|h| h.iter().any(|hop| hop["addresses"].as_array().map(|a| a.iter().any(|ip| same_ip(ip.as_str().unwrap_or(""), address))).unwrap_or(false))).unwrap_or(false));
    let all_failed = routes.iter().all(|r| r["exitCode"] != 0 && r["hops"].as_array().map(|h| h.is_empty()).unwrap_or(true));
    let addresses_of = |route: &Value| route["hops"].as_array().map(|h| h.iter().map(|hop| hop["addresses"].clone()).collect::<Vec<_>>()).unwrap_or_default();
    let changed = if routes.len() > 1 { json!(routes.iter().any(|r| addresses_of(r) != addresses_of(&routes[0]))) } else { Value::Null };
    Ok(json!({
        "status": if reached { "success" } else if all_failed { "error" } else { "warning" },
        "address": address,
        "routes": routes,
        "effectiveProbeTimeoutMs": if cfg!(target_os = "macos") { options.timeout_ms.div_ceil(1000) * 1000 } else { options.timeout_ms },
        "routeChanged": changed,
        "method": if cfg!(windows) { "ICMP tracert" } else { "System UDP traceroute" },
        "note": "Intermediate timeouts do not prove a broken route. Loss is the fraction of unanswered displayed probes, not end-to-end packet loss."
    }))
}

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

    #[test]
    fn hop_parsing() {
        let hop = parse_hop(" 3  10.0.0.1  1.234 ms  * <1 ms").unwrap();
        assert_eq!(hop["hop"], 3);
        assert_eq!(hop["addresses"][0], "10.0.0.1");
        assert_eq!(hop["probes"].as_array().unwrap().len(), 3);
        assert_eq!(hop["probes"][2]["upperBoundMs"], 1.0);
        assert!(parse_hop("traceroute to x").is_none());
    }

    #[test]
    fn ping_reply_parsing() {
        let reply = parse_ping_reply("64 bytes from 1.1.1.1: icmp_seq=0 ttl=57 time=12.5 ms");
        assert_eq!(reply["rttMs"], 12.5);
        assert_eq!(reply["ttl"], 57);
        let windows = parse_ping_reply("Reply from 10.0.0.1: bytes=32 time<1ms TTL=128");
        assert_eq!(windows["received"], true);
        assert_eq!(windows["upperBoundMs"], 1.0);
    }

    #[tokio::test]
    async fn process_limits_and_cancel() {
        if cfg!(windows) {
            return;
        }
        let cancel = CancellationToken::new();
        let result = run_process("/bin/sh", &["-c".into(), "echo hi; echo err >&2".into()], 5000, &cancel, 1024, None, false).await.unwrap();
        assert_eq!(String::from_utf8_lossy(&result.stdout), "hi\n");
        assert_eq!(result.stderr, "err\n");
        let limited = run_process("/bin/sh", &["-c".into(), "yes".into()], 5000, &cancel, 1024, None, false).await;
        assert_eq!(limited.err().unwrap().code, "OUTPUT_LIMIT");
        let timed = run_process("/bin/sh", &["-c".into(), "sleep 5".into()], 200, &cancel, 1024, None, false).await;
        assert_eq!(timed.err().unwrap().code, "TIMEOUT");
    }
}