//! Windows-only Job Object implementation of [`CellBackend`].
//!
//! Lifecycle (mirrors the gVisor backend's "owns execution" posture):
//!
//! - `create(spec)`:
//!   1. Build a per-cell [`JobObject`] with kill-on-close + the spec's memory /
//!      CPU caps.
//!   2. When the spec declares `run.argv`, spawn the workload with a cleared
//!      environment + fixed `PATH`, then `AssignProcessToJobObject` it so the
//!      whole tree is contained. The child is tracked so `wait_for_in_vm_exit`
//!      can join it.
//! - `wait_for_in_vm_exit(cell_id)`:
//!   - `Some(Ok(code))` once the tracked workload exits; `None` when this
//!      backend has no workload child for the id (the supervisor's
//!      host-subprocess path takes over).
//! - `destroy(handle)`:
//!   - `TerminateJobObject` (kills the tree), then drop the handle. Idempotent:
//!      a destroy after a completed `wait` finds nothing and still reports
//!      `destroyed: true`.

use std::collections::HashMap;
use std::process::Stdio;
use std::sync::Arc;

use async_trait::async_trait;
use tokio::process::{Child, Command};
use tokio::sync::Mutex;
use tracing::{instrument, warn};
use windows::Win32::Foundation::HANDLE;

use cellos_core::ports::{CellBackend, CellHandle, TeardownReport};
use cellos_core::{CellosError, ExecutionCellDocument, RunCpuMax, SecretDeliveryMode};

use crate::job::JobObject;

/// Fixed child `PATH` — never inherit the host `PATH` (search-order hijack);
/// matches the supervisor's host-subprocess shim (`run_cell_command_tokio`).
const FIXED_PATH: &str = r"C:\Windows\System32;C:\Windows";

/// cgroup-style CPU period used when a `cpuMax` omits its own period, in micros
/// (100 ms — the cgroup-v2 default).
const DEFAULT_CPU_PERIOD_MICROS: u64 = 100_000;

struct TrackedCell {
    /// The workload child, when the spec declared `run.argv`. `None` for cells
    /// with no workload (the job still provides containment + teardown).
    child: Option<Child>,
    /// The cell's job. Dropping it closes the handle → kill-on-close.
    job: JobObject,
}

/// Job-Object-backed [`CellBackend`]. Holds only an in-memory id → cell map; no
/// persistent state.
pub struct WindowsCellBackend {
    tracked: Arc<Mutex<HashMap<String, TrackedCell>>>,
}

impl Default for WindowsCellBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl WindowsCellBackend {
    pub fn new() -> Self {
        Self {
            tracked: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// How many cells this host still tracks (tests + operators; not an auth
    /// decision). Parity with the proprietary backend's helper.
    pub async fn tracked_cell_count(&self) -> usize {
        self.tracked.lock().await.len()
    }

    /// True if any host-tracked state remains for this `cell_id`.
    pub async fn has_tracked_state(&self, cell_id: &str) -> bool {
        self.tracked.lock().await.contains_key(cell_id)
    }
}

/// Translate a cgroup-style `cpuMax` quota/period into the hard CPU-rate cap
/// Windows wants: 1/100 of a percent of total CPU, clamped to `(0, 10000]`.
fn cpu_rate_hundredths_percent(cpu: &RunCpuMax) -> Option<u32> {
    let period = cpu.period_micros.unwrap_or(DEFAULT_CPU_PERIOD_MICROS);
    if period == 0 {
        return None;
    }
    // fraction = quota / period; the Windows unit is fraction * 10_000.
    let hundredths = (cpu.quota_micros as u128 * 10_000) / period as u128;
    Some(hundredths.clamp(1, 10_000) as u32)
}

#[async_trait]
impl CellBackend for WindowsCellBackend {
    #[instrument(skip(self, spec), fields(cell_id = %spec.spec.id))]
    async fn create(&self, spec: &ExecutionCellDocument) -> Result<CellHandle, CellosError> {
        let cell_id = spec.spec.id.clone();
        if cell_id.is_empty() {
            return Err(CellosError::InvalidSpec("spec.id must be non-empty".into()));
        }

        let run = spec.spec.run.as_ref();

        // Fail-closed: this backend owns execution, so the supervisor's env
        // secret-injection path is bypassed. Refuse `secretDelivery: env` with
        // declared secrets rather than run the workload without its secrets.
        if let Some(run) = run {
            let has_secret_refs = spec
                .spec
                .authority
                .secret_refs
                .as_ref()
                .is_some_and(|r| !r.is_empty());
            if matches!(run.secret_delivery, SecretDeliveryMode::Env) && has_secret_refs {
                return Err(CellosError::InvalidSpec(
                    "windows backend: secretDelivery=env with secretRefs is unsupported \
                     (this backend owns execution; use a broker-socket delivery mode such as \
                     the default runtimeLeasedBroker)"
                        .into(),
                ));
            }
        }

        // Network egress is not enforceable on Windows in v1 — warn loudly so
        // operators don't mistake "no error" for "enforced".
        if spec
            .spec
            .authority
            .egress_rules
            .as_ref()
            .is_some_and(|r| !r.is_empty())
        {
            warn!(
                target: "cellos.host.windows",
                cell_id = %cell_id,
                "spec declares authority.egressRules but the Windows backend does NOT enforce \
                 network egress (no nft equivalent); rules are advisory only on this host"
            );
        }

        let mut tracked = self.tracked.lock().await;
        if tracked.contains_key(&cell_id) {
            return Err(CellosError::Host(format!(
                "cell id {cell_id:?} already active on host (no duplicate live cells)"
            )));
        }

        // Resolve resource caps from the spec.
        let limits = run.and_then(|r| r.limits.as_ref());
        let memory_max_bytes = limits.and_then(|l| l.memory_max_bytes);
        let cpu_rate = limits
            .and_then(|l| l.cpu_max.as_ref())
            .and_then(cpu_rate_hundredths_percent);

        let job = JobObject::create(memory_max_bytes, cpu_rate)?;

        // Spawn the workload into the job when the spec declares one.
        let child = match run {
            Some(r) if !r.argv.is_empty() => {
                let mut cmd = Command::new(&r.argv[0]);
                if r.argv.len() > 1 {
                    cmd.args(&r.argv[1..]);
                }
                cmd.env_clear();
                cmd.env("PATH", FIXED_PATH);
                if let Some(wd) = &r.working_directory {
                    cmd.current_dir(wd);
                }
                cmd.stdin(Stdio::null())
                    .stdout(Stdio::null())
                    .stderr(Stdio::null());

                let child = cmd.spawn().map_err(|e| {
                    CellosError::Host(format!("windows: spawn {:?} failed: {e}", r.argv[0]))
                })?;

                // Assign immediately after spawn. There is a sub-millisecond
                // window between spawn and assignment in which the child could
                // fork a grandchild outside the job; `TerminateJobObject` +
                // kill-on-close still bound the tree, and the assigned child's
                // own descendants are contained. Documented in README.
                match child.raw_handle() {
                    Some(raw) => job.assign(HANDLE(raw))?,
                    None => {
                        return Err(CellosError::Host(
                            "windows: child raw handle unavailable; cannot assign to job".into(),
                        ));
                    }
                }
                Some(child)
            }
            _ => None,
        };

        tracked.insert(cell_id.clone(), TrackedCell { child, job });
        drop(tracked);

        Ok(CellHandle {
            cell_id,
            // No cgroup on Windows.
            cgroup_path: None,
            // This backend does not enforce host network policy.
            nft_rules_applied: None,
            // No boot-artifact manifest (not a VM).
            kernel_digest_sha256: None,
            rootfs_digest_sha256: None,
            firecracker_digest_sha256: None,
        })
    }

    #[instrument(skip(self))]
    async fn wait_for_in_vm_exit(&self, cell_id: &str) -> Option<Result<i32, CellosError>> {
        let mut tracked = self.tracked.lock().await;
        // Only own execution when we actually spawned a workload child for this
        // id; otherwise defer to the supervisor's host-subprocess path.
        match tracked.get(cell_id) {
            Some(t) if t.child.is_some() => {}
            _ => return None,
        }
        // Pull the entry out so we can await without holding the map lock. When
        // the entry (and its `JobObject`) drops at the end of this scope, the
        // handle closes and kill-on-close reaps any lingering descendants.
        let mut entry = tracked.remove(cell_id).expect("entry present above");
        drop(tracked);

        let mut child = entry.child.take().expect("child present above");
        let status = match child.wait().await {
            Ok(s) => s,
            Err(e) => {
                return Some(Err(CellosError::Host(format!(
                    "windows: wait for workload of {cell_id}: {e}"
                ))));
            }
        };
        Some(Ok(status.code().unwrap_or(-1)))
    }

    #[instrument(skip(self, handle), fields(cell_id = %handle.cell_id))]
    async fn destroy(&self, handle: &CellHandle) -> Result<TeardownReport, CellosError> {
        let mut tracked = self.tracked.lock().await;
        let entry = tracked.remove(&handle.cell_id);
        let peers_tracked_after = tracked.len();
        drop(tracked);

        if let Some(t) = entry {
            // Deterministically kill the tree before the handle drops. We never
            // fail the teardown report on this — the job handle's Drop is the
            // backstop, and an error here usually means the job already emptied.
            if let Err(e) = t.job.terminate(1) {
                warn!(
                    target: "cellos.host.windows",
                    cell_id = %handle.cell_id,
                    error = %e,
                    "TerminateJobObject failed (continuing; kill-on-close will still fire on drop)"
                );
            }
            // `t` (child handle + job) drops here.
        }

        Ok(TeardownReport {
            cell_id: handle.cell_id.clone(),
            destroyed: true,
            peers_tracked_after,
        })
    }
}

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

    fn doc_with_run(id: &str, argv: &[&str]) -> ExecutionCellDocument {
        serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": id,
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 },
                "run": { "argv": argv }
            }
        }))
        .unwrap()
    }

    fn doc_no_run(id: &str) -> ExecutionCellDocument {
        serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": id,
                "authority": { "secretRefs": [] },
                "lifetime": { "ttlSeconds": 60 }
            }
        }))
        .unwrap()
    }

    #[test]
    fn cpu_rate_translation() {
        // 50% (50_000 / 100_000 micros) → 5000 hundredths-of-a-percent.
        assert_eq!(
            cpu_rate_hundredths_percent(&RunCpuMax {
                quota_micros: 50_000,
                period_micros: Some(100_000)
            }),
            Some(5_000)
        );
        // Default period when omitted.
        assert_eq!(
            cpu_rate_hundredths_percent(&RunCpuMax {
                quota_micros: 100_000,
                period_micros: None
            }),
            Some(10_000)
        );
        // Over-100% is clamped to the 10_000 ceiling.
        assert_eq!(
            cpu_rate_hundredths_percent(&RunCpuMax {
                quota_micros: 1_000_000,
                period_micros: Some(100_000)
            }),
            Some(10_000)
        );
        // Zero period is rejected (no divide-by-zero).
        assert_eq!(
            cpu_rate_hundredths_percent(&RunCpuMax {
                quota_micros: 10,
                period_micros: Some(0)
            }),
            None
        );
    }

    #[tokio::test]
    async fn empty_id_rejected() {
        let host = WindowsCellBackend::new();
        let err = host.create(&doc_no_run("")).await.unwrap_err();
        assert!(matches!(err, CellosError::InvalidSpec(_)));
    }

    #[tokio::test]
    async fn duplicate_id_rejected() {
        let host = WindowsCellBackend::new();
        // A long-lived workload so the first cell stays tracked.
        let doc = doc_with_run("dup", &["cmd.exe", "/c", "ping -n 30 127.0.0.1 >NUL"]);
        let h = host.create(&doc).await.unwrap();
        let err = host.create(&doc).await.unwrap_err();
        assert!(matches!(err, CellosError::Host(_)));
        host.destroy(&h).await.unwrap();
    }

    #[tokio::test]
    async fn env_secret_delivery_with_refs_refused() {
        let host = WindowsCellBackend::new();
        let doc: ExecutionCellDocument = serde_json::from_value(serde_json::json!({
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {
                "id": "secret-env",
                "authority": { "secretRefs": ["db/password"] },
                "lifetime": { "ttlSeconds": 60 },
                "run": { "argv": ["cmd.exe", "/c", "exit 0"], "secretDelivery": "env" }
            }
        }))
        .unwrap();
        let err = host.create(&doc).await.unwrap_err();
        assert!(matches!(err, CellosError::InvalidSpec(_)));
    }

    #[tokio::test]
    async fn workload_runs_and_exit_code_is_surfaced() {
        let host = WindowsCellBackend::new();
        let doc = doc_with_run("rc", &["cmd.exe", "/c", "exit 7"]);
        let h = host.create(&doc).await.unwrap();
        assert_eq!(host.tracked_cell_count().await, 1);
        let code = host.wait_for_in_vm_exit("rc").await;
        assert_eq!(code.unwrap().unwrap(), 7);
        // wait consumed the tracked entry (job dropped, kill-on-close fired).
        assert_eq!(host.tracked_cell_count().await, 0);
        // destroy after wait is idempotent and still reports destroyed.
        let report = host.destroy(&h).await.unwrap();
        assert!(report.destroyed);
        assert_eq!(report.peers_tracked_after, 0);
    }

    #[tokio::test]
    async fn destroy_before_exit_kills_tree_and_clears_residue() {
        let host = WindowsCellBackend::new();
        // Long sleeper — destroy must terminate it without waiting.
        let doc = doc_with_run("kill-me", &["cmd.exe", "/c", "ping -n 60 127.0.0.1 >NUL"]);
        let h = host.create(&doc).await.unwrap();
        assert_eq!(host.tracked_cell_count().await, 1);
        let report = host.destroy(&h).await.unwrap();
        assert!(report.destroyed);
        assert_eq!(report.peers_tracked_after, 0);
        assert!(!host.has_tracked_state("kill-me").await);
    }

    #[tokio::test]
    async fn peers_tracked_after_counts_remaining() {
        let host = WindowsCellBackend::new();
        let a = host
            .create(&doc_with_run(
                "a",
                &["cmd.exe", "/c", "ping -n 60 127.0.0.1 >NUL"],
            ))
            .await
            .unwrap();
        let _b = host
            .create(&doc_with_run(
                "b",
                &["cmd.exe", "/c", "ping -n 60 127.0.0.1 >NUL"],
            ))
            .await
            .unwrap();
        let r = host.destroy(&a).await.unwrap();
        assert_eq!(r.peers_tracked_after, 1);
        assert!(host.has_tracked_state("b").await);
    }

    #[tokio::test]
    async fn no_workload_cell_defers_execution_to_supervisor() {
        let host = WindowsCellBackend::new();
        let h = host.create(&doc_no_run("bare")).await.unwrap();
        // No child → backend does not own execution for this id.
        assert!(host.wait_for_in_vm_exit("bare").await.is_none());
        // But the job is tracked and torn down on destroy.
        assert_eq!(host.tracked_cell_count().await, 1);
        host.destroy(&h).await.unwrap();
        assert_eq!(host.tracked_cell_count().await, 0);
    }
}
