use std::process::{Command, Output, Stdio};
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

use crate::core::engine::command::{
    isolate_process_tree, wait_with_bounded_output, wait_with_bounded_output_until_cancelled,
    CommandCaptureMetadata, DEFAULT_CAPTURE_LIMIT_BYTES,
};
use crate::core::error::{Error, Result};

const SAMPLE_INTERVAL: Duration = Duration::from_millis(1000);
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
#[cfg(target_os = "linux")]
const DEFAULT_RSS_LIMIT_BYTES: u64 = 16 * 1024 * 1024 * 1024;
#[cfg(target_os = "linux")]
const DEFAULT_PROCESS_COUNT_LIMIT: u64 = 128;
#[cfg(target_os = "linux")]
const RSS_LIMIT_ENV: &str = "HOMEBOY_RUNNER_RESOURCE_GUARD_RSS_BYTES";
#[cfg(target_os = "linux")]
const PROCESS_COUNT_LIMIT_ENV: &str = "HOMEBOY_RUNNER_RESOURCE_GUARD_PROCESS_COUNT";

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RunnerResourceMetrics {
    pub duration_ms: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cpu_user_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cpu_system_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub peak_rss_bytes: Option<u64>,
    pub sample_count: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub child_process_count_peak: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub guard_violation: Option<RunnerResourceGuardViolation>,
    pub source: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RunnerResourceGuardViolation {
    pub reason: String,
    pub message: String,
    pub rss_bytes: u64,
    pub rss_limit_bytes: u64,
    pub process_count: u64,
    pub process_count_limit: u64,
}

#[derive(Debug)]
pub(crate) struct MeasuredOutput {
    pub output: Output,
    pub capture: CommandCaptureMetadata,
    pub metrics: RunnerResourceMetrics,
}

pub(crate) type RunnerCommandProgressSink = Arc<dyn Fn(Value) + Send + Sync + 'static>;

#[derive(Debug, Default)]
struct MetricsState {
    sample_count: u64,
    peak_rss_bytes: u64,
    child_process_count_peak: u64,
    cpu_user_ms: u64,
    cpu_system_ms: u64,
    guard_violation: Option<RunnerResourceGuardViolation>,
}

pub(crate) fn measured_command_output(command: &mut Command) -> Result<MeasuredOutput> {
    command.stdout(Stdio::piped()).stderr(Stdio::piped());
    let started = Instant::now();
    let child = command.spawn().map_err(|err| {
        Error::internal_io(err.to_string(), Some("execute runner command".to_string()))
    })?;
    let pid = child.id();
    let collector = ResourceMetricsCollector::start(pid, started, None, None);
    let bounded_output =
        wait_with_bounded_output(child, DEFAULT_CAPTURE_LIMIT_BYTES).map_err(|err| {
            Error::internal_io(err.to_string(), Some("wait for runner command".to_string()))
        })?;
    let capture = bounded_output.capture.clone();
    let output = bounded_output.into_output();
    let metrics = collector.finish(started.elapsed());
    Ok(MeasuredOutput {
        output,
        capture,
        metrics,
    })
}

pub(crate) fn measured_command_output_until_cancelled_with_progress(
    command: &mut Command,
    mut is_cancelled: impl FnMut() -> bool,
    progress_sink: Option<RunnerCommandProgressSink>,
) -> Result<MeasuredOutput> {
    command.stdout(Stdio::piped()).stderr(Stdio::piped());
    isolate_process_tree(command);
    let started = Instant::now();
    let mut child = command.spawn().map_err(|err| {
        Error::internal_io(err.to_string(), Some("execute runner command".to_string()))
    })?;
    let pid = child.id();
    let guard_violation = Arc::new(Mutex::new(None));
    let collector = ResourceMetricsCollector::start(
        pid,
        started,
        progress_sink,
        Some(Arc::clone(&guard_violation)),
    );
    let bounded_output =
        wait_with_bounded_output_until_cancelled(&mut child, DEFAULT_CAPTURE_LIMIT_BYTES, || {
            is_cancelled()
                || guard_violation
                    .lock()
                    .expect("resource guard mutex poisoned")
                    .is_some()
        })
        .map_err(|err| {
            Error::internal_io(err.to_string(), Some("wait for runner command".to_string()))
        })?;
    let capture = bounded_output.capture.clone();
    let output = bounded_output.into_output();
    let metrics = collector.finish(started.elapsed());
    Ok(MeasuredOutput {
        output,
        capture,
        metrics,
    })
}

struct ResourceMetricsCollector {
    supported: bool,
    stop: Option<mpsc::Sender<()>>,
    state: Arc<Mutex<MetricsState>>,
    handle: Option<thread::JoinHandle<()>>,
}

impl ResourceMetricsCollector {
    fn start(
        root_pid: u32,
        started: Instant,
        progress_sink: Option<RunnerCommandProgressSink>,
        guard_violation: Option<Arc<Mutex<Option<RunnerResourceGuardViolation>>>>,
    ) -> Self {
        let supported = cfg!(target_os = "linux") && std::path::Path::new("/proc").exists();
        let state = Arc::new(Mutex::new(MetricsState::default()));
        if !supported && progress_sink.is_none() {
            return Self {
                supported,
                stop: None,
                state,
                handle: None,
            };
        }

        let (stop, stop_rx) = mpsc::channel();
        sample(root_pid, &state, guard_violation.as_ref());
        let thread_state = Arc::clone(&state);
        let mut last_heartbeat = Instant::now();
        let handle = thread::spawn(move || loop {
            sample(root_pid, &thread_state, guard_violation.as_ref());
            if let Some(progress) = progress_sink.as_ref() {
                if last_heartbeat.elapsed() >= HEARTBEAT_INTERVAL {
                    progress(runner_command_heartbeat_data(
                        started.elapsed(),
                        root_pid,
                        process_tree_resource_summary(root_pid),
                    ));
                    last_heartbeat = Instant::now();
                }
            }
            if stop_rx.recv_timeout(SAMPLE_INTERVAL).is_ok() {
                break;
            }
        });

        Self {
            supported,
            stop: Some(stop),
            state,
            handle: Some(handle),
        }
    }

    fn finish(mut self, duration: Duration) -> RunnerResourceMetrics {
        if let Some(stop) = self.stop.take() {
            let _ = stop.send(());
        }
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
        let state = self.state.lock().expect("resource metrics mutex poisoned");
        RunnerResourceMetrics {
            duration_ms: duration.as_millis().try_into().unwrap_or(u64::MAX),
            cpu_user_ms: (state.sample_count > 0).then_some(state.cpu_user_ms),
            cpu_system_ms: (state.sample_count > 0).then_some(state.cpu_system_ms),
            peak_rss_bytes: (state.sample_count > 0).then_some(state.peak_rss_bytes),
            sample_count: state.sample_count,
            child_process_count_peak: (state.sample_count > 0)
                .then_some(state.child_process_count_peak),
            guard_violation: state.guard_violation.clone(),
            source: if self.supported {
                "linux_procfs_process_tree".to_string()
            } else {
                "duration_only".to_string()
            },
        }
    }
}

pub(crate) fn runner_command_heartbeat_data(
    elapsed: Duration,
    root_pid: u32,
    resources: Option<Value>,
) -> Value {
    json!({
        "phase": "heartbeat",
        "elapsed_ms": elapsed.as_millis(),
        "process": {
            "root_pid": root_pid,
            "resources": resources,
        },
    })
}

#[cfg(target_os = "linux")]
fn process_tree_resource_summary(root_pid: u32) -> Option<Value> {
    let snapshot = process_tree_snapshot(root_pid)?;
    Some(json!({
        "source": "linux_procfs_process_tree",
        "process_count": snapshot.process_count,
        "child_process_count": snapshot.process_count.saturating_sub(1),
        "rss_bytes": snapshot.rss_bytes,
        "cpu_user_ms": snapshot.cpu_user_ms,
        "cpu_system_ms": snapshot.cpu_system_ms,
    }))
}

#[cfg(not(target_os = "linux"))]
fn process_tree_resource_summary(_root_pid: u32) -> Option<Value> {
    None
}

#[cfg(target_os = "linux")]
fn sample(
    root_pid: u32,
    state: &Arc<Mutex<MetricsState>>,
    guard_violation: Option<&Arc<Mutex<Option<RunnerResourceGuardViolation>>>>,
) {
    if let Some(snapshot) = process_tree_snapshot(root_pid) {
        let mut state = state.lock().expect("resource metrics mutex poisoned");
        state.sample_count += 1;
        state.peak_rss_bytes = state.peak_rss_bytes.max(snapshot.rss_bytes);
        state.child_process_count_peak = state
            .child_process_count_peak
            .max(snapshot.process_count.saturating_sub(1));
        state.cpu_user_ms = state.cpu_user_ms.max(snapshot.cpu_user_ms);
        state.cpu_system_ms = state.cpu_system_ms.max(snapshot.cpu_system_ms);
        if state.guard_violation.is_none() {
            if let Some(violation) = resource_guard_violation(&snapshot) {
                if let Some(shared_violation) = guard_violation {
                    *shared_violation
                        .lock()
                        .expect("resource guard mutex poisoned") = Some(violation.clone());
                }
                state.guard_violation = Some(violation);
            }
        }
    }
}

#[cfg(not(target_os = "linux"))]
fn sample(
    _root_pid: u32,
    _state: &Arc<Mutex<MetricsState>>,
    _guard_violation: Option<&Arc<Mutex<Option<RunnerResourceGuardViolation>>>>,
) {
}

#[cfg(target_os = "linux")]
#[derive(Debug)]
struct ProcessSnapshot {
    process_count: u64,
    rss_bytes: u64,
    cpu_user_ms: u64,
    cpu_system_ms: u64,
}

#[cfg(target_os = "linux")]
fn process_tree_snapshot(root_pid: u32) -> Option<ProcessSnapshot> {
    let stats = read_process_stats();
    let root_pid = root_pid as i32;
    stats.get(&root_pid)?;

    let mut children: std::collections::HashMap<i32, Vec<i32>> = std::collections::HashMap::new();
    for stat in stats.values() {
        children.entry(stat.ppid).or_default().push(stat.pid);
    }

    let mut queue = std::collections::VecDeque::from([root_pid]);
    let mut seen = std::collections::HashSet::new();
    let mut rss_bytes = 0_u64;
    let mut user_ticks = 0_u64;
    let mut system_ticks = 0_u64;

    while let Some(pid) = queue.pop_front() {
        if !seen.insert(pid) {
            continue;
        }
        if let Some(stat) = stats.get(&pid) {
            rss_bytes = rss_bytes.saturating_add(stat.rss_bytes);
            user_ticks = user_ticks.saturating_add(stat.user_ticks);
            system_ticks = system_ticks.saturating_add(stat.system_ticks);
            if let Some(child_pids) = children.get(&pid) {
                queue.extend(child_pids);
            }
        }
    }

    let clock_ticks = clock_ticks_per_second().max(1);
    Some(ProcessSnapshot {
        process_count: seen.len().try_into().unwrap_or(u64::MAX),
        rss_bytes,
        cpu_user_ms: user_ticks.saturating_mul(1000) / clock_ticks,
        cpu_system_ms: system_ticks.saturating_mul(1000) / clock_ticks,
    })
}

#[cfg(target_os = "linux")]
fn resource_guard_violation(snapshot: &ProcessSnapshot) -> Option<RunnerResourceGuardViolation> {
    let rss_limit_bytes = resource_guard_limit(RSS_LIMIT_ENV, DEFAULT_RSS_LIMIT_BYTES);
    let process_count_limit =
        resource_guard_limit(PROCESS_COUNT_LIMIT_ENV, DEFAULT_PROCESS_COUNT_LIMIT);
    classify_resource_guard_violation(
        snapshot.rss_bytes,
        snapshot.process_count,
        rss_limit_bytes,
        process_count_limit,
    )
}

#[cfg(any(target_os = "linux", test))]
fn classify_resource_guard_violation(
    rss_bytes: u64,
    process_count: u64,
    rss_limit_bytes: u64,
    process_count_limit: u64,
) -> Option<RunnerResourceGuardViolation> {
    let reason = if rss_limit_bytes > 0 && rss_bytes >= rss_limit_bytes {
        "rss_limit_exceeded"
    } else if process_count_limit > 0 && process_count >= process_count_limit {
        "process_count_limit_exceeded"
    } else {
        return None;
    };
    Some(RunnerResourceGuardViolation {
        reason: reason.to_string(),
        message: format!(
            "runner job resource guard stopped process tree after rss_bytes={rss_bytes}, process_count={process_count}; limits rss_bytes={rss_limit_bytes}, process_count={process_count_limit}"
        ),
        rss_bytes,
        rss_limit_bytes,
        process_count,
        process_count_limit,
    })
}

#[cfg(target_os = "linux")]
fn resource_guard_limit(env_name: &str, default_value: u64) -> u64 {
    std::env::var(env_name)
        .ok()
        .and_then(|value| value.trim().parse::<u64>().ok())
        .unwrap_or(default_value)
}

#[cfg(target_os = "linux")]
#[derive(Debug)]
struct ProcessStat {
    pid: i32,
    ppid: i32,
    user_ticks: u64,
    system_ticks: u64,
    rss_bytes: u64,
}

#[cfg(target_os = "linux")]
fn read_process_stats() -> std::collections::HashMap<i32, ProcessStat> {
    let mut stats = std::collections::HashMap::new();
    let Ok(entries) = std::fs::read_dir("/proc") else {
        return stats;
    };
    for entry in entries.flatten() {
        let file_name = entry.file_name();
        let Some(pid_text) = file_name.to_str() else {
            continue;
        };
        let Ok(pid) = pid_text.parse::<i32>() else {
            continue;
        };
        if let Some(stat) = read_process_stat(pid) {
            stats.insert(pid, stat);
        }
    }
    stats
}

#[cfg(target_os = "linux")]
fn read_process_stat(pid: i32) -> Option<ProcessStat> {
    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
    let end_command = stat.rfind(')')?;
    let fields: Vec<&str> = stat[end_command + 2..].split_whitespace().collect();
    let ppid = fields.get(1)?.parse().ok()?;
    let user_ticks = fields.get(11).and_then(|v| v.parse().ok()).unwrap_or(0)
        + fields.get(13).and_then(|v| v.parse().ok()).unwrap_or(0);
    let system_ticks = fields.get(12).and_then(|v| v.parse().ok()).unwrap_or(0)
        + fields.get(14).and_then(|v| v.parse().ok()).unwrap_or(0);
    let rss_pages: u64 = fields.get(21).and_then(|v| v.parse().ok()).unwrap_or(0);
    Some(ProcessStat {
        pid,
        ppid,
        user_ticks,
        system_ticks,
        rss_bytes: rss_pages.saturating_mul(page_size_bytes()),
    })
}

#[cfg(target_os = "linux")]
fn clock_ticks_per_second() -> u64 {
    static CLOCK_TICKS_PER_SECOND: std::sync::OnceLock<u64> = std::sync::OnceLock::new();

    *CLOCK_TICKS_PER_SECOND.get_or_init(|| command_u64("getconf", &["CLK_TCK"]).unwrap_or(100))
}

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

    #[test]
    fn heartbeat_payload_includes_elapsed_pid_and_optional_resources() {
        let payload = runner_command_heartbeat_data(
            Duration::from_millis(42_000),
            1234,
            Some(json!({
                "source": "fixture",
                "process_count": 3,
                "child_process_count": 2,
            })),
        );

        assert_eq!(payload["phase"], "heartbeat");
        assert_eq!(payload["elapsed_ms"], 42_000);
        assert_eq!(payload["process"]["root_pid"], 1234);
        assert_eq!(payload["process"]["resources"]["process_count"], 3);
        assert_eq!(payload["process"]["resources"]["child_process_count"], 2);
    }

    #[test]
    fn heartbeat_payload_allows_missing_resource_summary() {
        let payload = runner_command_heartbeat_data(Duration::from_millis(5), 9, None);

        assert_eq!(payload["phase"], "heartbeat");
        assert_eq!(payload["elapsed_ms"], 5);
        assert_eq!(payload["process"]["root_pid"], 9);
        assert!(payload["process"]["resources"].is_null());
    }

    #[test]
    fn resource_guard_classifies_rss_limit_exceeded() {
        let violation = classify_resource_guard_violation(17, 3, 16, 128).expect("rss violation");

        assert_eq!(violation.reason, "rss_limit_exceeded");
        assert_eq!(violation.rss_bytes, 17);
        assert_eq!(violation.rss_limit_bytes, 16);
        assert_eq!(violation.process_count, 3);
    }

    #[test]
    fn resource_guard_classifies_process_count_limit_exceeded() {
        let violation =
            classify_resource_guard_violation(8, 129, 16, 128).expect("process count violation");

        assert_eq!(violation.reason, "process_count_limit_exceeded");
        assert_eq!(violation.process_count, 129);
        assert_eq!(violation.process_count_limit, 128);
    }

    #[test]
    fn resource_guard_allows_zero_limits_to_disable_guard() {
        assert!(classify_resource_guard_violation(u64::MAX, u64::MAX, 0, 0).is_none());
    }
}

#[cfg(target_os = "linux")]
fn page_size_bytes() -> u64 {
    static PAGE_SIZE_BYTES: std::sync::OnceLock<u64> = std::sync::OnceLock::new();

    *PAGE_SIZE_BYTES.get_or_init(|| command_u64("getconf", &["PAGESIZE"]).unwrap_or(4096))
}

#[cfg(target_os = "linux")]
fn command_u64(command: &str, args: &[&str]) -> Option<u64> {
    let output = Command::new(command).args(args).output().ok()?;
    String::from_utf8_lossy(&output.stdout).trim().parse().ok()
}
