use std::collections::HashMap;
use std::sync::Arc;
use anyhow::Result;
use bollard::container::{
    Config, CreateContainerOptions, LogsOptions, RemoveContainerOptions, StartContainerOptions,
    StopContainerOptions, StatsOptions, Stats,
};
use bollard::image::CreateImageOptions;
use bollard::models::{HostConfig, PortBinding};
use futures::StreamExt;
use tracing::info;

use crate::docker::DockerManager;
use crate::server::ServerSpec;

/// Container name prefix for all managed game servers.
const CONTAINER_PREFIX: &str = "gp_";

pub fn container_name(server_uuid: &str) -> String {
    format!("{}{}", CONTAINER_PREFIX, server_uuid)
}

impl DockerManager {
    /// Pull a Docker image if not already present.
    pub async fn pull_image(&self, image: &str) -> Result<()> {
        info!(image = %image, "Pulling Docker image");

        let options = CreateImageOptions {
            from_image: image,
            ..Default::default()
        };

        let mut stream = self.client().create_image(Some(options), None, None);
        while let Some(result) = stream.next().await {
            match result {
                Ok(info) => {
                    if let Some(status) = &info.status {
                        tracing::debug!(status = %status, "Image pull progress");
                    }
                }
                Err(e) => return Err(e.into()),
            }
        }

        info!(image = %image, "Image pulled successfully");
        Ok(())
    }

    /// Create and configure a container for a game server.
    pub async fn create_container(&self, spec: &ServerSpec) -> Result<String> {
        let name = container_name(&spec.uuid);

        // Build port bindings
        let mut port_bindings: HashMap<String, Option<Vec<PortBinding>>> = HashMap::new();
        for port_map in &spec.ports {
            let container_port = format!("{}/{}", port_map.container_port, port_map.protocol);
            port_bindings.insert(
                container_port,
                Some(vec![PortBinding {
                    host_ip: Some("0.0.0.0".to_string()),
                    host_port: Some(port_map.host_port.to_string()),
                }]),
            );
        }

        // Build exposed ports
        let mut exposed_ports: HashMap<String, HashMap<(), ()>> = HashMap::new();
        for port_map in &spec.ports {
            let container_port = format!("{}/{}", port_map.container_port, port_map.protocol);
            exposed_ports.insert(container_port, HashMap::new());
        }

        // Convert env map to Docker format
        let env: Vec<String> = spec
            .environment
            .iter()
            .map(|(k, v)| format!("{}={}", k, v))
            .collect();

        let host_config = HostConfig {
            memory: Some(spec.memory_limit),
            memory_swap: Some(spec.memory_limit), // no swap
            nano_cpus: Some((spec.cpu_limit as i64) * 10_000_000), // cpu_limit=100 means 1 core
            port_bindings: Some(port_bindings),
            network_mode: Some(self.network_name().to_string()),
            binds: Some(vec![format!(
                "{}:/data",
                spec.data_path.display()
            )]),
            ..Default::default()
        };

        let config = Config {
            image: Some(spec.docker_image.clone()),
            hostname: Some(spec.uuid.clone()),
            env: Some(env),
            exposed_ports: Some(exposed_ports),
            host_config: Some(host_config),
            working_dir: Some("/data".to_string()),
            cmd: if spec.startup_command.is_empty() {
                None
            } else {
                Some(
                    spec.startup_command
                        .split_whitespace()
                        .map(String::from)
                        .collect(),
                )
            },
            tty: Some(true),
            attach_stdin: Some(true),
            attach_stdout: Some(true),
            attach_stderr: Some(true),
            open_stdin: Some(true),
            ..Default::default()
        };

        let options = CreateContainerOptions { name: name.as_str(), platform: None };
        let response = self.client().create_container(Some(options), config).await?;

        info!(container_id = %response.id, uuid = %spec.uuid, "Container created");
        Ok(response.id)
    }

    /// Start a container.
    pub async fn start_container(&self, server_uuid: &str) -> Result<()> {
        let name = container_name(server_uuid);
        self.client()
            .start_container(&name, None::<StartContainerOptions<String>>)
            .await?;
        info!(uuid = %server_uuid, "Container started");
        Ok(())
    }

    /// Stop a container gracefully.
    pub async fn stop_container(&self, server_uuid: &str, timeout_secs: i64) -> Result<()> {
        let name = container_name(server_uuid);
        self.client()
            .stop_container(
                &name,
                Some(StopContainerOptions {
                    t: timeout_secs,
                }),
            )
            .await?;
        info!(uuid = %server_uuid, "Container stopped");
        Ok(())
    }

    /// Kill a container immediately.
    pub async fn kill_container(&self, server_uuid: &str) -> Result<()> {
        let name = container_name(server_uuid);
        self.client()
            .kill_container::<String>(&name, None)
            .await?;
        info!(uuid = %server_uuid, "Container killed");
        Ok(())
    }

    /// Remove a container and its volumes.
    pub async fn remove_container(&self, server_uuid: &str) -> Result<()> {
        let name = container_name(server_uuid);
        self.client()
            .remove_container(
                &name,
                Some(RemoveContainerOptions {
                    force: true,
                    v: true,
                    ..Default::default()
                }),
            )
            .await?;
        info!(uuid = %server_uuid, "Container removed");
        Ok(())
    }

    /// Get container stats (CPU, memory, network).
    pub async fn container_stats(
        &self,
        server_uuid: &str,
    ) -> Result<Stats> {
        let name = container_name(server_uuid);
        let mut stream = self.client().stats(
            &name,
            Some(StatsOptions {
                stream: false,
                one_shot: true,
                ..Default::default()
            }),
        );

        match stream.next().await {
            Some(Ok(stats)) => Ok(stats),
            Some(Err(e)) => Err(e.into()),
            None => Err(anyhow::anyhow!("No stats returned")),
        }
    }

    /// Check if a container exists and return its state.
    pub async fn container_state(
        &self,
        server_uuid: &str,
    ) -> Result<Option<String>> {
        let name = container_name(server_uuid);
        match self.client().inspect_container(&name, None).await {
            Ok(info) => {
                let state = info
                    .state
                    .and_then(|s| s.status)
                    .map(|s| format!("{:?}", s));
                Ok(state)
            }
            Err(bollard::errors::Error::DockerResponseServerError {
                status_code: 404, ..
            }) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    /// Stream container logs (stdout + stderr). Returns an owned stream.
    pub fn stream_logs(
        self: &Arc<Self>,
        server_uuid: &str,
    ) -> impl futures::Stream<Item = Result<String, bollard::errors::Error>> + Send + 'static {
        let name = container_name(server_uuid);
        let options = LogsOptions::<String> {
            follow: true,
            stdout: true,
            stderr: true,
            tail: "100".to_string(),
            ..Default::default()
        };

        let client = self.client().clone();
        client.logs(&name, Some(options)).map(|result| {
            result.map(|output| output.to_string())
        })
    }

    /// Send a command to a container via exec (attach to stdin).
    pub async fn send_command(&self, server_uuid: &str, command: &str) -> Result<()> {
        let name = container_name(server_uuid);

        let exec = self
            .client()
            .create_exec(
                &name,
                bollard::exec::CreateExecOptions {
                    cmd: Some(vec!["sh", "-c", &format!("echo '{}' > /proc/1/fd/0", command)]),
                    attach_stdout: Some(true),
                    attach_stderr: Some(true),
                    ..Default::default()
                },
            )
            .await?;

        self.client()
            .start_exec(&exec.id, None::<bollard::exec::StartExecOptions>)
            .await?;

        Ok(())
    }
}