//! Kubernetes implementation of the sandbox driver (cloud target): one
//! hardened pod per sandbox in a dedicated namespace with a default-deny
//! NetworkPolicy.
//!
//! Hardening parity with the Docker driver: uid 10001, cap-drop ALL,
//! no-new-privileges, read-only rootfs with emptyDir /tmp and /home/agent,
//! resource limits. Differences, stated honestly: seccomp uses the
//! runtime's `RuntimeDefault` profile (installing our strict allowlist as
//! a `Localhost` profile is per-node provisioning, documented for managed
//! clusters), and kernel-level egress enforcement of the NetworkPolicy
//! requires a NetworkPolicy-capable CNI (kind's default kindnet does not
//! enforce; production clusters with Calico/Cilium do).

use k8s_openapi::api::core::v1::{Namespace, Pod};
use k8s_openapi::api::networking::v1::NetworkPolicy;
use kube::api::{AttachParams, DeleteParams, ObjectMeta, Patch, PatchParams, PostParams};
use kube::Api;
use serde_json::json;
use tokio::io::AsyncReadExt;

use crate::spec::{ExecResult, SandboxHandle, SandboxSpec};
use crate::{SandboxDriver, SandboxError};

fn engine_err(e: impl std::fmt::Display) -> SandboxError {
    SandboxError::Engine(e.to_string())
}

pub struct K8sDriver {
    client: kube::Client,
    namespace: String,
    /// Kubelet-relative path of the strict allowlist profile when the
    /// nodes carry it (installed by the Helm DaemonSet); RuntimeDefault
    /// otherwise.
    localhost_seccomp: Option<String>,
}

impl K8sDriver {
    /// Connects via the ambient kubeconfig and ensures the sandbox
    /// namespace exists with its default-deny NetworkPolicy.
    pub async fn connect(namespace: &str) -> Result<K8sDriver, SandboxError> {
        // rustls 0.23 needs a process-level crypto provider when several
        // are linked; first caller wins, repeats are harmless.
        let _ = rustls::crypto::ring::default_provider().install_default();
        let client = kube::Client::try_default().await.map_err(engine_err)?;
        K8sDriver::with_client(client, namespace).await
    }

    /// Connects to a SPECIFIC kubeconfig context (e.g. the
    /// NetworkPolicy-enforcing test cluster) instead of the current one.
    pub async fn connect_with_context(
        namespace: &str,
        context: &str,
    ) -> Result<K8sDriver, SandboxError> {
        let _ = rustls::crypto::ring::default_provider().install_default();
        let options = kube::config::KubeConfigOptions {
            context: Some(context.to_owned()),
            ..Default::default()
        };
        let config = kube::Config::from_kubeconfig(&options)
            .await
            .map_err(engine_err)?;
        let client = kube::Client::try_from(config).map_err(engine_err)?;
        K8sDriver::with_client(client, namespace).await
    }

    async fn with_client(client: kube::Client, namespace: &str) -> Result<K8sDriver, SandboxError> {
        let driver = K8sDriver {
            client,
            namespace: namespace.to_owned(),
            localhost_seccomp: None,
        };
        driver.ensure_namespace().await?;
        Ok(driver)
    }

    async fn ensure_namespace(&self) -> Result<(), SandboxError> {
        let namespaces: Api<Namespace> = Api::all(self.client.clone());
        let ns = Namespace {
            metadata: ObjectMeta {
                name: Some(self.namespace.clone()),
                labels: Some(
                    [
                        // Pod Security Standards: refuse anything that
                        // tries to relax the hardening below.
                        (
                            "pod-security.kubernetes.io/enforce".to_owned(),
                            "restricted".to_owned(),
                        ),
                        ("app.kubernetes.io/name".to_owned(), "clawmates".to_owned()),
                    ]
                    .into(),
                ),
                ..Default::default()
            },
            ..Default::default()
        };
        namespaces
            .patch(
                &self.namespace,
                &PatchParams::apply("clawmates-sandbox").force(),
                &Patch::Apply(&ns),
            )
            .await
            .map_err(engine_err)?;

        // Default-deny everything for sandbox pods; the orchestrator talks
        // to them via the API server's exec channel, not the pod network.
        let policies: Api<NetworkPolicy> = Api::namespaced(self.client.clone(), &self.namespace);
        let deny: NetworkPolicy = serde_json::from_value(json!({
            "apiVersion": "networking.k8s.io/v1",
            "kind": "NetworkPolicy",
            "metadata": { "name": "sandbox-default-deny", "namespace": self.namespace },
            "spec": {
                "podSelector": {},
                "policyTypes": ["Ingress", "Egress"]
            }
        }))
        .expect("static policy");
        policies
            .patch(
                "sandbox-default-deny",
                &PatchParams::apply("clawmates-sandbox").force(),
                &Patch::Apply(&deny),
            )
            .await
            .map_err(engine_err)?;
        Ok(())
    }

    /// Uses the node-installed strict allowlist profile instead of the
    /// runtime default (see deploy/helm templates/seccomp-installer).
    pub fn with_localhost_seccomp(mut self, profile: &str) -> K8sDriver {
        self.localhost_seccomp = Some(profile.to_owned());
        self
    }

    fn pod_spec(&self, spec: &SandboxSpec) -> Pod {
        let seccomp = match &self.localhost_seccomp {
            Some(profile) => json!({ "type": "Localhost", "localhostProfile": profile }),
            None => json!({ "type": "RuntimeDefault" }),
        };
        serde_json::from_value(json!({
            "apiVersion": "v1",
            "kind": "Pod",
            "metadata": {
                "name": spec.name,
                "namespace": self.namespace,
                "labels": { "app.kubernetes.io/name": "clawmates-sandbox" }
            },
            "spec": {
                "restartPolicy": "Never",
                "automountServiceAccountToken": false,
                "securityContext": {
                    "runAsNonRoot": true,
                    "runAsUser": 10001,
                    "runAsGroup": 10001,
                    "seccompProfile": seccomp
                },
                "containers": [{
                    "name": "sandbox",
                    "image": spec.image,
                    "imagePullPolicy": "Never",
                    "command": ["sleep", "infinity"],
                    "securityContext": {
                        "allowPrivilegeEscalation": false,
                        "capabilities": { "drop": ["ALL"] },
                        "readOnlyRootFilesystem": true
                    },
                    "resources": {
                        "limits": {
                            "memory": format!("{}", spec.memory_bytes),
                            "cpu": format!("{}m", spec.nano_cpus / 1_000_000)
                        }
                    },
                    "volumeMounts": [
                        { "name": "tmp", "mountPath": "/tmp" },
                        { "name": "home", "mountPath": "/home/agent" }
                    ]
                }],
                "volumes": [
                    { "name": "tmp", "emptyDir": { "sizeLimit": "64Mi" } },
                    { "name": "home", "emptyDir": { "sizeLimit": "256Mi" } }
                ]
            }
        }))
        .expect("static pod spec")
    }

    fn pods(&self) -> Api<Pod> {
        Api::namespaced(self.client.clone(), &self.namespace)
    }
}

#[async_trait::async_trait]
impl SandboxDriver for K8sDriver {
    async fn provision(&self, spec: &SandboxSpec) -> Result<SandboxHandle, SandboxError> {
        let pods = self.pods();
        pods.create(&PostParams::default(), &self.pod_spec(spec))
            .await
            .map_err(engine_err)?;
        // Wait until Running (image is preloaded, so this is fast).
        for _ in 0..120 {
            let pod = pods.get(&spec.name).await.map_err(engine_err)?;
            let phase = pod.status.as_ref().and_then(|s| s.phase.clone());
            match phase.as_deref() {
                Some("Running") => {
                    return Ok(SandboxHandle {
                        id: spec.name.clone(),
                        name: spec.name.clone(),
                    })
                }
                Some("Failed") => return Err(engine_err("sandbox pod failed to start")),
                _ => tokio::time::sleep(std::time::Duration::from_millis(500)).await,
            }
        }
        Err(engine_err("sandbox pod never reached Running"))
    }

    async fn exec(&self, handle: &SandboxHandle, cmd: &[&str]) -> Result<ExecResult, SandboxError> {
        let pods = self.pods();
        let params = AttachParams::default().stdout(true).stderr(true);
        let mut attached = pods
            .exec(&handle.name, cmd.to_vec(), &params)
            .await
            .map_err(engine_err)?;

        let mut stdout = String::new();
        let mut stderr = String::new();
        if let Some(mut out) = attached.stdout() {
            out.read_to_string(&mut stdout).await.map_err(engine_err)?;
        }
        if let Some(mut err) = attached.stderr() {
            err.read_to_string(&mut stderr).await.map_err(engine_err)?;
        }
        let status = attached.take_status();
        attached.join().await.map_err(engine_err)?;

        // Exit code travels in the v1.Status the API server sends on close.
        let exit_code = match status {
            Some(rx) => match rx.await {
                Some(s) if s.status.as_deref() == Some("Success") => 0,
                Some(s) => s
                    .details
                    .and_then(|d| {
                        d.causes.unwrap_or_default().into_iter().find_map(|c| {
                            (c.reason.as_deref() == Some("ExitCode"))
                                .then(|| c.message.and_then(|m| m.parse().ok()))
                                .flatten()
                        })
                    })
                    .unwrap_or(1),
                None => 0,
            },
            None => 0,
        };
        Ok(ExecResult {
            exit_code,
            stdout,
            stderr,
        })
    }

    async fn destroy(&self, handle: &SandboxHandle) -> Result<(), SandboxError> {
        self.pods()
            .delete(&handle.name, &DeleteParams::default().grace_period(0))
            .await
            .map_err(engine_err)?;
        Ok(())
    }

    async fn health(&self, handle: &SandboxHandle) -> Result<bool, SandboxError> {
        match self
            .pods()
            .get_opt(&handle.name)
            .await
            .map_err(engine_err)?
        {
            Some(pod) => Ok(pod
                .status
                .and_then(|s| s.phase)
                .is_some_and(|phase| phase == "Running")),
            None => Ok(false),
        }
    }
}