use async_trait::async_trait;
use at_run_core::{
    ArchiveLimits, ArchiveSummary, ArchiveValidationError, PinnedImageRef, RuntimeTarget,
    validate_archive,
};
use bollard::{
    Docker, body_full,
    models::{ContainerCreateBody, HostConfig, Mount, MountType, VolumeCreateRequest},
    query_parameters::{
        CreateImageOptionsBuilder, DownloadFromContainerOptionsBuilder,
        RemoveContainerOptionsBuilder, RemoveVolumeOptionsBuilder, UploadToContainerOptionsBuilder,
    },
};
use futures_util::StreamExt;
use std::{collections::HashMap, path::Path};
use thiserror::Error;

const MATERIALIZATION_TARGET: &str = "/at-run-output";

/// Small portability boundary around the container backend.
///
/// Bollard selects Unix sockets, Windows named pipes, or HTTP transports from
/// the local Docker environment. Runner orchestration should depend on this
/// trait rather than on Bollard's generated request models directly.
#[async_trait]
pub trait ContainerEngine: Send + Sync {
    fn backend_name(&self) -> &'static str;

    async fn ping(&self) -> Result<(), EngineError>;

    /// Return the OCI platform used by containers created by this daemon.
    async fn runtime_target(&self) -> Result<RuntimeTarget, EngineError>;

    /// Pull and safely materialize one exported directory from a public image
    /// into a named Docker volume without starting that image.
    async fn materialize_image_export(
        &self,
        image: &PinnedImageRef,
        export_path: &str,
        volume_name: &str,
        limits: ArchiveLimits,
    ) -> Result<MaterializedVolume, EngineError>;

    /// Copy a validated tar into a volume through a stopped carrier
    /// container. The carrier image is never started by this operation.
    async fn materialize_archive(
        &self,
        carrier_image: &PinnedImageRef,
        archive: &[u8],
        volume_name: &str,
        limits: ArchiveLimits,
    ) -> Result<MaterializedVolume, EngineError>;

    /// Create, but do not start, the runner-owned sandbox container.
    async fn create_sandbox(
        &self,
        specification: &SandboxSpec,
    ) -> Result<StoppedSandbox, EngineError>;
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MaterializedVolume {
    pub name: String,
    pub archive: ArchiveSummary,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SandboxMount {
    pub volume_name: String,
    pub target: String,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SandboxSpec {
    pub image: PinnedImageRef,
    pub user: String,
    pub supervisor: String,
    pub command: Vec<String>,
    pub environment: Vec<String>,
    pub mounts: Vec<SandboxMount>,
    pub memory_bytes: i64,
    pub pids: i64,
    pub work_bytes: u64,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StoppedSandbox {
    pub id: String,
    pub warnings: Vec<String>,
}

#[derive(Clone)]
pub struct DockerEngine {
    client: Docker,
}

impl DockerEngine {
    /// Connect using the platform's local Docker defaults.
    pub fn connect_local() -> Result<Self, EngineError> {
        if std::env::var_os("DOCKER_HOST").is_some() {
            return Ok(Self {
                client: Docker::connect_with_defaults()?,
            });
        }

        match Docker::connect_with_local_defaults() {
            Ok(client) => Ok(Self { client }),
            Err(default_error) => {
                #[cfg(unix)]
                if let Some(client) = docker_desktop_client() {
                    return Ok(Self { client });
                }

                Err(default_error.into())
            }
        }
    }

    pub fn client(&self) -> &Docker {
        &self.client
    }

    async fn ensure_image(&self, image: &str) -> Result<(), EngineError> {
        match self.client.inspect_image(image).await {
            Ok(_) => return Ok(()),
            Err(bollard::errors::Error::DockerResponseServerError {
                status_code: 404, ..
            }) => {}
            Err(error) => return Err(error.into()),
        }

        let mut progress = self.client.create_image(
            Some(
                CreateImageOptionsBuilder::default()
                    .from_image(image)
                    .build(),
            ),
            None,
            None,
        );
        while let Some(result) = progress.next().await {
            result?;
        }
        Ok(())
    }

    async fn materialize_existing_image_export(
        &self,
        image: &str,
        export_path: &str,
        volume_name: &str,
        limits: ArchiveLimits,
    ) -> Result<MaterializedVolume, EngineError> {
        validate_export_path(export_path)?;

        let mut labels = HashMap::new();
        labels.insert("dev.at-run.managed".to_owned(), "true".to_owned());
        labels.insert("dev.at-run.source.image".to_owned(), image.to_owned());
        labels.insert("dev.at-run.source.path".to_owned(), export_path.to_owned());

        let volume = self
            .client
            .create_volume(VolumeCreateRequest {
                name: Some(volume_name.to_owned()),
                labels: Some(labels),
                ..Default::default()
            })
            .await?;

        let result = self
            .populate_volume(image, export_path, &volume.name, limits)
            .await;

        if result.is_err() {
            let _ = self
                .client
                .remove_volume(
                    &volume.name,
                    Some(RemoveVolumeOptionsBuilder::default().force(true).build()),
                )
                .await;
        }

        result.map(|archive| MaterializedVolume {
            name: volume.name,
            archive,
        })
    }

    async fn populate_volume(
        &self,
        image: &str,
        export_path: &str,
        volume_name: &str,
        limits: ArchiveLimits,
    ) -> Result<ArchiveSummary, EngineError> {
        let created = self
            .client
            .create_container(
                None,
                ContainerCreateBody {
                    image: Some(image.to_owned()),
                    host_config: Some(HostConfig {
                        mounts: Some(vec![Mount {
                            target: Some(MATERIALIZATION_TARGET.to_owned()),
                            source: Some(volume_name.to_owned()),
                            typ: Some(MountType::VOLUME),
                            read_only: Some(false),
                            ..Default::default()
                        }]),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
            )
            .await?;

        let operation = async {
            let archive = self
                .archive_from_container(&created.id, export_path)
                .await?;
            let summary = validate_archive(&archive, limits)?;

            self.client
                .upload_to_container(
                    &created.id,
                    Some(
                        UploadToContainerOptionsBuilder::default()
                            .path(MATERIALIZATION_TARGET)
                            .build(),
                    ),
                    body_full(archive.into()),
                )
                .await?;

            Ok::<_, EngineError>(summary)
        }
        .await;

        let cleanup = self
            .client
            .remove_container(
                &created.id,
                Some(RemoveContainerOptionsBuilder::default().force(true).build()),
            )
            .await;

        match (operation, cleanup) {
            (Ok(summary), Ok(())) => Ok(summary),
            (Err(error), _) => Err(error),
            (Ok(_), Err(error)) => Err(error.into()),
        }
    }

    async fn populate_volume_from_archive(
        &self,
        carrier_image: &str,
        archive: Vec<u8>,
        volume_name: &str,
    ) -> Result<(), EngineError> {
        let created = self
            .client
            .create_container(
                None,
                ContainerCreateBody {
                    image: Some(carrier_image.to_owned()),
                    host_config: Some(HostConfig {
                        mounts: Some(vec![Mount {
                            target: Some(MATERIALIZATION_TARGET.to_owned()),
                            source: Some(volume_name.to_owned()),
                            typ: Some(MountType::VOLUME),
                            read_only: Some(false),
                            ..Default::default()
                        }]),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
            )
            .await?;

        let operation = self
            .client
            .upload_to_container(
                &created.id,
                Some(
                    UploadToContainerOptionsBuilder::default()
                        .path(MATERIALIZATION_TARGET)
                        .build(),
                ),
                body_full(archive.into()),
            )
            .await;
        let cleanup = self
            .client
            .remove_container(
                &created.id,
                Some(RemoveContainerOptionsBuilder::default().force(true).build()),
            )
            .await;

        match (operation, cleanup) {
            (Ok(()), Ok(())) => Ok(()),
            (Err(error), _) | (Ok(()), Err(error)) => Err(error.into()),
        }
    }

    async fn archive_from_container(
        &self,
        container_id: &str,
        path: &str,
    ) -> Result<Vec<u8>, EngineError> {
        let mut stream = self.client.download_from_container(
            container_id,
            Some(
                DownloadFromContainerOptionsBuilder::default()
                    .path(path)
                    .build(),
            ),
        );
        let mut archive = Vec::new();
        while let Some(chunk) = stream.next().await {
            archive.extend_from_slice(&chunk?);
        }
        Ok(archive)
    }
}

#[cfg(unix)]
fn docker_desktop_client() -> Option<Docker> {
    let socket = std::env::var_os("HOME")
        .map(std::path::PathBuf::from)?
        .join(".docker/run/docker.sock");
    if !socket.exists() {
        return None;
    }

    let host = format!("unix://{}", socket.display());
    Docker::connect_with_host(&host).ok()
}

#[async_trait]
impl ContainerEngine for DockerEngine {
    fn backend_name(&self) -> &'static str {
        "docker"
    }

    async fn ping(&self) -> Result<(), EngineError> {
        self.client.ping().await?;
        Ok(())
    }

    async fn runtime_target(&self) -> Result<RuntimeTarget, EngineError> {
        let version = self.client.version().await?;
        let os = version.os.ok_or(EngineError::MissingDaemonPlatform("os"))?;
        let architecture = version
            .arch
            .ok_or(EngineError::MissingDaemonPlatform("architecture"))?;
        Ok(RuntimeTarget::new(os, architecture))
    }

    async fn materialize_image_export(
        &self,
        image: &PinnedImageRef,
        export_path: &str,
        volume_name: &str,
        limits: ArchiveLimits,
    ) -> Result<MaterializedVolume, EngineError> {
        self.ensure_image(&image.to_string()).await?;
        self.materialize_existing_image_export(&image.to_string(), export_path, volume_name, limits)
            .await
    }

    async fn materialize_archive(
        &self,
        carrier_image: &PinnedImageRef,
        archive: &[u8],
        volume_name: &str,
        limits: ArchiveLimits,
    ) -> Result<MaterializedVolume, EngineError> {
        let summary = validate_archive(archive, limits)?;
        self.ensure_image(&carrier_image.to_string()).await?;

        let mut labels = HashMap::new();
        labels.insert("dev.at-run.managed".to_owned(), "true".to_owned());
        labels.insert(
            "dev.at-run.carrier.image".to_owned(),
            carrier_image.to_string(),
        );

        let volume = self
            .client
            .create_volume(VolumeCreateRequest {
                name: Some(volume_name.to_owned()),
                labels: Some(labels),
                ..Default::default()
            })
            .await?;
        let result = self
            .populate_volume_from_archive(
                &carrier_image.to_string(),
                archive.to_vec(),
                &volume.name,
            )
            .await;
        if result.is_err() {
            let _ = self
                .client
                .remove_volume(
                    &volume.name,
                    Some(RemoveVolumeOptionsBuilder::default().force(true).build()),
                )
                .await;
        }
        result?;

        Ok(MaterializedVolume {
            name: volume.name,
            archive: summary,
        })
    }

    async fn create_sandbox(
        &self,
        specification: &SandboxSpec,
    ) -> Result<StoppedSandbox, EngineError> {
        validate_sandbox_spec(specification)?;
        self.ensure_image(&specification.image.to_string()).await?;

        let mounts = specification
            .mounts
            .iter()
            .map(|mount| Mount {
                target: Some(mount.target.clone()),
                source: Some(mount.volume_name.clone()),
                typ: Some(MountType::VOLUME),
                read_only: Some(true),
                ..Default::default()
            })
            .collect();
        let mut tmpfs = HashMap::new();
        tmpfs.insert(
            "/opt/at-run/work".to_owned(),
            format!("rw,nosuid,nodev,size={}", specification.work_bytes),
        );
        let mut labels = HashMap::new();
        labels.insert("dev.at-run.managed".to_owned(), "true".to_owned());
        labels.insert("dev.at-run.role".to_owned(), "sandbox".to_owned());

        let created = self
            .client
            .create_container(
                None,
                ContainerCreateBody {
                    user: Some(specification.user.clone()),
                    attach_stdin: Some(true),
                    attach_stdout: Some(true),
                    attach_stderr: Some(true),
                    tty: Some(false),
                    open_stdin: Some(true),
                    stdin_once: Some(true),
                    env: Some(specification.environment.clone()),
                    cmd: Some(specification.command.clone()),
                    healthcheck: Some(bollard::models::HealthConfig {
                        test: Some(vec!["NONE".to_owned()]),
                        ..Default::default()
                    }),
                    image: Some(specification.image.to_string()),
                    working_dir: Some("/opt/at-run/work".to_owned()),
                    entrypoint: Some(vec![specification.supervisor.clone()]),
                    network_disabled: Some(true),
                    labels: Some(labels),
                    host_config: Some(HostConfig {
                        memory: Some(specification.memory_bytes),
                        memory_swap: Some(specification.memory_bytes),
                        pids_limit: Some(specification.pids),
                        init: Some(false),
                        network_mode: Some("none".to_owned()),
                        auto_remove: Some(false),
                        mounts: Some(mounts),
                        cap_drop: Some(vec!["ALL".to_owned()]),
                        ipc_mode: Some("none".to_owned()),
                        privileged: Some(false),
                        readonly_rootfs: Some(true),
                        security_opt: Some(vec![
                            "no-new-privileges=true".to_owned(),
                            "seccomp=builtin".to_owned(),
                        ]),
                        tmpfs: Some(tmpfs),
                        shm_size: Some(16 * 1024 * 1024),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
            )
            .await?;

        Ok(StoppedSandbox {
            id: created.id,
            warnings: created.warnings,
        })
    }
}

fn validate_sandbox_spec(specification: &SandboxSpec) -> Result<(), EngineError> {
    if specification.user.is_empty()
        || specification.supervisor.is_empty()
        || !specification.supervisor.starts_with('/')
        || specification.command.is_empty()
        || specification.memory_bytes <= 0
        || specification.pids <= 0
        || specification.work_bytes == 0
        || specification
            .environment
            .iter()
            .any(|value| value.contains('\0'))
        || specification.mounts.iter().any(|mount| {
            mount.volume_name.is_empty()
                || !mount.target.starts_with("/opt/at-run/")
                || mount.target.contains("/../")
        })
    {
        return Err(EngineError::InvalidSandboxSpec);
    }
    Ok(())
}

fn validate_export_path(path: &str) -> Result<(), EngineError> {
    let path = Path::new(path);
    if !path.is_absolute()
        || path
            .components()
            .any(|component| matches!(component, std::path::Component::ParentDir))
    {
        return Err(EngineError::InvalidExportPath(
            path.to_string_lossy().into_owned(),
        ));
    }
    Ok(())
}

#[derive(Debug, Error)]
pub enum EngineError {
    #[error("Docker Engine operation failed: {0}")]
    Docker(#[from] bollard::errors::Error),
    #[error(transparent)]
    Archive(#[from] ArchiveValidationError),
    #[error("runtime export path must be an absolute path without parent traversal: {0:?}")]
    InvalidExportPath(String),
    #[error("Docker daemon did not report its {0}")]
    MissingDaemonPlatform(&'static str),
    #[error("invalid sandbox container specification")]
    InvalidSandboxSpec,
}

#[cfg(test)]
mod tests {
    use std::io::Cursor;
    use std::time::{SystemTime, UNIX_EPOCH};

    use bollard::query_parameters::RemoveVolumeOptionsBuilder;

    use super::*;

    #[tokio::test]
    #[ignore = "requires Docker and the local at-run-runtime-fixture:local image"]
    async fn materializes_export_without_starting_selected_image() {
        let engine = DockerEngine::connect_local().unwrap();
        engine.ping().await.unwrap();

        let fixture = engine
            .client
            .inspect_image("at-run-runtime-fixture:local")
            .await
            .unwrap();
        let image: PinnedImageRef = fixture
            .repo_digests
            .unwrap()
            .into_iter()
            .find(|digest| digest.starts_with("at-run-runtime-fixture@sha256:"))
            .unwrap()
            .parse()
            .unwrap();

        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let volume_name = format!("at-run-test-{}-{nonce}", std::process::id());

        let result = engine
            .materialize_image_export(
                &image,
                "/at-run/runtime",
                &volume_name,
                ArchiveLimits::default(),
            )
            .await;

        let verification_container = engine
            .client
            .create_container(
                None,
                ContainerCreateBody {
                    image: Some(image.to_string()),
                    host_config: Some(HostConfig {
                        mounts: Some(vec![Mount {
                            target: Some("/verify".to_owned()),
                            source: Some(volume_name.clone()),
                            typ: Some(MountType::VOLUME),
                            read_only: Some(true),
                            ..Default::default()
                        }]),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        let copied_file = engine
            .archive_from_container(&verification_container.id, "/verify/runtime/runtime.txt")
            .await;
        engine
            .client
            .remove_container(
                &verification_container.id,
                Some(RemoveContainerOptionsBuilder::default().force(true).build()),
            )
            .await
            .unwrap();

        let cleanup = engine
            .client
            .remove_volume(
                &volume_name,
                Some(RemoveVolumeOptionsBuilder::default().force(true).build()),
            )
            .await;

        let materialized = result.unwrap();
        let copied_file = copied_file.unwrap();
        cleanup.unwrap();
        assert_eq!(materialized.name, volume_name);
        assert_eq!(materialized.archive.root, Path::new("runtime"));
        assert!(materialized.archive.entries >= 2);
        assert_eq!(
            validate_archive(&copied_file, ArchiveLimits::default())
                .unwrap()
                .root,
            Path::new("runtime.txt")
        );
    }

    #[tokio::test]
    #[ignore = "requires Docker and the local at-run-runtime-fixture:local image"]
    async fn materializes_pds_archive_without_starting_carrier() {
        let engine = DockerEngine::connect_local().unwrap();
        engine.ping().await.unwrap();
        let fixture = engine
            .client
            .inspect_image("at-run-runtime-fixture:local")
            .await
            .unwrap();
        let image: PinnedImageRef = fixture
            .repo_digests
            .unwrap()
            .into_iter()
            .find(|digest| digest.starts_with("at-run-runtime-fixture@sha256:"))
            .unwrap()
            .parse()
            .unwrap();
        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let volume_name = format!("at-run-code-test-{}-{nonce}", std::process::id());
        let archive = archive_with_file("app/main.py", b"print('hello')");

        let materialized = engine
            .materialize_archive(&image, &archive, &volume_name, ArchiveLimits::default())
            .await
            .unwrap();
        let verification_container = engine
            .client
            .create_container(
                None,
                ContainerCreateBody {
                    image: Some(image.to_string()),
                    host_config: Some(HostConfig {
                        mounts: Some(vec![Mount {
                            target: Some("/verify".to_owned()),
                            source: Some(volume_name.clone()),
                            typ: Some(MountType::VOLUME),
                            read_only: Some(true),
                            ..Default::default()
                        }]),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        let copied = engine
            .archive_from_container(&verification_container.id, "/verify/app/main.py")
            .await
            .unwrap();
        engine
            .client
            .remove_container(
                &verification_container.id,
                Some(RemoveContainerOptionsBuilder::default().force(true).build()),
            )
            .await
            .unwrap();
        engine
            .client
            .remove_volume(
                &volume_name,
                Some(RemoveVolumeOptionsBuilder::default().force(true).build()),
            )
            .await
            .unwrap();

        assert_eq!(materialized.archive.root, Path::new("app"));
        assert_eq!(
            validate_archive(&copied, ArchiveLimits::default())
                .unwrap()
                .root,
            Path::new("main.py")
        );
    }

    #[tokio::test]
    #[ignore = "requires Docker and the local at-run-runtime-fixture:local image"]
    async fn creates_hardened_runner_owned_sandbox_in_stopped_state() {
        let engine = DockerEngine::connect_local().unwrap();
        engine.ping().await.unwrap();
        let fixture = engine
            .client
            .inspect_image("at-run-runtime-fixture:local")
            .await
            .unwrap();
        let image: PinnedImageRef = fixture
            .repo_digests
            .unwrap()
            .into_iter()
            .find(|digest| digest.starts_with("at-run-runtime-fixture@sha256:"))
            .unwrap()
            .parse()
            .unwrap();
        let specification = SandboxSpec {
            image,
            user: "65532:65532".to_owned(),
            supervisor: "/at-run/runtime/never-run".to_owned(),
            command: vec!["/opt/at-run/runtime/runtime/never-run".to_owned()],
            environment: vec!["AT_RUN_TEST=true".to_owned()],
            mounts: vec![],
            memory_bytes: 64 * 1024 * 1024,
            pids: 16,
            work_bytes: 1024 * 1024,
        };

        let sandbox = engine.create_sandbox(&specification).await.unwrap();
        let inspected = engine
            .client
            .inspect_container(&sandbox.id, None)
            .await
            .unwrap();
        engine
            .client
            .remove_container(
                &sandbox.id,
                Some(RemoveContainerOptionsBuilder::default().force(true).build()),
            )
            .await
            .unwrap();

        let state = inspected.state.unwrap();
        let config = inspected.config.unwrap();
        let host = inspected.host_config.unwrap();
        assert_eq!(state.running, Some(false));
        assert_eq!(config.user.as_deref(), Some("65532:65532"));
        assert_eq!(host.readonly_rootfs, Some(true));
        assert_eq!(host.network_mode.as_deref(), Some("none"));
        assert_eq!(host.cap_drop, Some(vec!["ALL".to_owned()]));
        assert_eq!(host.memory, Some(64 * 1024 * 1024));
        let security = host.security_opt.unwrap();
        assert!(
            security
                .iter()
                .any(|option| option.starts_with("no-new-privileges"))
        );
        assert!(security.iter().any(|option| option == "seccomp=builtin"));
    }

    fn archive_with_file(path: &str, contents: &[u8]) -> Vec<u8> {
        let mut bytes = Vec::new();
        {
            let mut builder = tar::Builder::new(&mut bytes);
            let mut header = tar::Header::new_gnu();
            header.set_mode(0o644);
            header.set_size(contents.len() as u64);
            header.set_cksum();
            builder
                .append_data(&mut header, path, Cursor::new(contents))
                .unwrap();
            builder.finish().unwrap();
        }
        bytes
    }
}