//! Session latency (port of `sessions/latencyMonitor.js`). ICMP from this computer to the
//! address the SSH connection uses and the SSH transport round trip are independent: a
//! blocked ping never hides the SSH measurement and never marks SSH as broken.

use crate::core::ssh::session::{SessionManager, SessionRecord};
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;

pub const PING_COUNT: usize = 3;
const INTERVAL: Duration = Duration::from_secs(10);
const SSH_SAMPLES: usize = 3;
static REPLY_LINE: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)\b(?:ttl|hlim)\s*=\s*\d+").unwrap());
static REPLY_TIME: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)([=<])\s*(\d+(?:[.,]\d+)?)\s*ms\b").unwrap());

#[derive(Debug, Clone, PartialEq)]
pub struct Reply {
    pub ms: f64,
    pub less_than: bool,
}

pub fn median(values: &[f64]) -> Option<f64> {
    if values.is_empty() {
        return None;
    }
    let mut sorted = values.to_vec();
    sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
    let middle = sorted.len() / 2;
    Some(if sorted.len() % 2 == 1 { sorted[middle] } else { (sorted[middle - 1] + sorted[middle]) / 2.0 })
}

/// Per-reply times from macOS, Linux (iputils) and localized Windows output; "<1ms" → 1 ms, lessThan.
pub fn parse_ping_replies(stdout: &str) -> Vec<Reply> {
    let mut replies = Vec::new();
    for line in stdout.lines() {
        if !REPLY_LINE.is_match(line) {
            continue;
        }
        let Some(caps) = REPLY_TIME.captures(line) else { continue };
        let Ok(value) = caps[2].replace(',', ".").parse::<f64>() else { continue };
        if value.is_finite() && value >= 0.0 {
            let less = &caps[1] == "<";
            replies.push(Reply { ms: if less { value.max(1.0) } else { value }, less_than: less });
        }
    }
    replies
}

pub fn ping_command(address: &IpAddr) -> (String, Vec<String>) {
    let ipv6 = address.is_ipv6();
    let target = address.to_string();
    if cfg!(windows) {
        ("ping.exe".into(), vec![if ipv6 { "-6" } else { "-4" }.into(), "-n".into(), PING_COUNT.to_string(), "-w".into(), "2000".into(), target])
    } else if cfg!(target_os = "macos") {
        let mut args = vec!["-n".to_string(), "-c".into(), PING_COUNT.to_string()];
        if !ipv6 {
            args.extend(["-W".to_string(), "2000".into()]);
        }
        args.push(target);
        ((if ipv6 { "/sbin/ping6" } else { "/sbin/ping" }).into(), args)
    } else {
        ("ping".into(), vec![if ipv6 { "-6" } else { "-4" }.into(), "-n".into(), "-c".into(), PING_COUNT.to_string(), "-W".into(), "2".into(), target])
    }
}

fn classify_failure(text: &str, not_found: bool) -> (&'static str, &'static str) {
    if not_found {
        return ("unsupported", "The ping utility is not available on this computer.");
    }
    let has = |pattern: &str| Regex::new(pattern).map(|r| r.is_match(text)).unwrap_or(false);
    if has(r"(?i)Operation not permitted|Permission denied|Access is denied|not allowed") {
        return ("denied", "The operating system denied sending ICMP (for example the macOS Local Network permission or a security policy).");
    }
    if has(r"(?i)unknown host|cannot resolve|could not find host|Name or service not known") {
        return ("error", "The address could not be resolved for ping.");
    }
    if has(r"(?i)No route to host|Network is unreachable|Destination (host|net) unreachable|unreachable|nedosažiteln|nicht erreichbar") {
        return ("unreachable", "The target is unreachable for ICMP (no route or an unreachable reply).");
    }
    ("timeout", "No ICMP echo reply before the timeout; the server or a firewall may block ICMP.")
}

pub async fn measure_ping(address: &str) -> Value {
    let Ok(ip) = address.parse::<IpAddr>() else {
        return json!({ "ms": null, "status": "error", "reason": "No IP address is known for this session.", "family": null, "sent": 0, "received": 0, "loss": null });
    };
    let family = if ip.is_ipv4() { 4 } else { 6 };
    let (program, args) = ping_command(&ip);
    let mut command = tokio::process::Command::new(&program);
    command.args(&args).env("LC_ALL", "C").env("LANG", "C").stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped()).kill_on_drop(true);
    #[cfg(windows)]
    {
        command.creation_flags(0x0800_0000); // CREATE_NO_WINDOW
    }
    let output = match tokio::time::timeout(Duration::from_secs(9), command.output()).await {
        Ok(Ok(output)) => output,
        Ok(Err(error)) => {
            let (status, reason) = classify_failure(&error.to_string(), error.kind() == std::io::ErrorKind::NotFound);
            return json!({ "ms": null, "status": status, "reason": reason, "family": family, "sent": PING_COUNT, "received": 0, "loss": 1 });
        }
        Err(_) => {
            return json!({ "ms": null, "status": "timeout", "reason": classify_failure("", false).1, "family": family, "sent": PING_COUNT, "received": 0, "loss": null });
        }
    };
    let stdout = String::from_utf8_lossy(&output.stdout).to_string();
    let replies = parse_ping_replies(&stdout);
    let received = replies.len().min(PING_COUNT);
    if received > 0 {
        let times: Vec<f64> = replies.iter().map(|r| r.ms).collect();
        return json!({
            "ms": median(&times),
            "lessThan": replies.iter().all(|r| r.less_than),
            "status": "ok",
            "reason": "",
            "family": family,
            "sent": PING_COUNT,
            "received": received,
            "loss": (PING_COUNT - received) as f64 / PING_COUNT as f64
        });
    }
    let text = format!("{stdout}\n{}", String::from_utf8_lossy(&output.stderr));
    let (status, reason) = classify_failure(&text, false);
    json!({ "ms": null, "status": status, "reason": reason, "family": family, "sent": PING_COUNT, "received": 0, "loss": 1 })
}

async fn measure_ssh(manager: &SessionManager, record: &Arc<SessionRecord>) -> (Option<f64>, Option<&'static str>, usize, Option<String>) {
    let mut samples = Vec::new();
    let mut method = None;
    let mut error = None;
    for _ in 0..SSH_SAMPLES {
        match manager.ssh_round_trip(record).await {
            Ok((Some(ms), used)) if ms > 0.0 => {
                samples.push(ms);
                method = Some(used);
            }
            Ok((_, used)) => method = Some(used),
            Err(cause) => {
                error = Some(cause.message);
                break;
            }
        }
    }
    let error = if samples.is_empty() { Some(error.unwrap_or_else(|| "SSH round trip failed".into())) } else { None };
    (median(&samples), method, samples.len(), error)
}

pub(crate) async fn measure_route_ping(address: &str, jump: bool) -> Value {
    if jump {
        json!({ "ms": null, "status": "jump", "reason": "Target reached through SSH forwarding.", "family": null })
    } else { measure_ping(address).await }
}

pub fn start(manager: Arc<SessionManager>, record: Arc<SessionRecord>) {
    let task_record = record.clone();
    let task = tokio::spawn(async move {
        let record = task_record;
        loop {
            if !record.is_connected() {
                break;
            }
            let address = record.address.lock().clone();
            let ping = measure_route_ping(&address, record.has_jump());
            let (ping, ssh) = tokio::join!(ping, measure_ssh(&manager, &record));
            if !record.is_connected() {
                break;
            }
            let valid = |v: Option<f64>| v.filter(|v| v.is_finite() && *v > 0.0);
            let family = ping.get("family").cloned().filter(|v| !v.is_null()).unwrap_or_else(|| {
                address.parse::<IpAddr>().map(|ip| json!(if ip.is_ipv4() { 4 } else { 6 })).unwrap_or(Value::Null)
            });
            let latency = json!({
                "pingMs": valid(ping["ms"].as_f64()),
                "pingStatus": ping["status"].as_str().unwrap_or("error"),
                "pingReason": ping["reason"].as_str().unwrap_or(""),
                "pingLoss": ping["loss"].as_f64(),
                "pingFamily": family,
                "pingLessThan": ping["lessThan"].as_bool() == Some(true),
                "sshRttMs": valid(ssh.0),
                "sshRttMethod": ssh.1,
                "sshRttSamples": ssh.2,
                "sshRttError": ssh.3,
                "measuredAt": chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
            });
            *record.latency.lock() = Some(latency);
            manager.publish_state(&record);
            tokio::time::sleep(INTERVAL).await;
        }
    });
    *record.stop_latency.lock() = Some(task);
}

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

    #[test]
    fn parses_platform_outputs() {
        let mac = "64 bytes from 1.1.1.1: icmp_seq=0 ttl=57 time=12.345 ms\n64 bytes from 1.1.1.1: icmp_seq=1 ttl=57 time=10.1 ms\nround-trip min/avg/max = 1/2/3 ms";
        assert_eq!(parse_ping_replies(mac).len(), 2);
        let windows_cs = "Odpověď od 10.0.0.1: bajty=32 čas<1ms TTL=128\nOdpověď od 10.0.0.1: bajty=32 čas=2ms TTL=128\nMinimum = 0ms";
        let replies = parse_ping_replies(windows_cs);
        assert_eq!(replies, vec![Reply { ms: 1.0, less_than: true }, Reply { ms: 2.0, less_than: false }]);
        assert_eq!(median(&[3.0, 1.0, 2.0]), Some(2.0));
        assert_eq!(median(&[1.0, 2.0]), Some(1.5));
    }

    #[test]
    fn argv_has_no_shell() {
        let (program, args) = ping_command(&"::1".parse().unwrap());
        assert!(program.contains("ping"));
        assert!(args.iter().all(|a| !a.contains(';')));
    }
}