use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;

use anyhow::Result;
use bollard::Docker;
use bollard::network::CreateNetworkOptions;
use tokio::io::{AsyncWrite, AsyncWriteExt};
use tokio::sync::{Mutex, RwLock};
use tokio::task::JoinHandle;
use tracing::info;

use crate::config::DaemonConfig;

type AttachedInput = Pin<Box<dyn AsyncWrite + Send>>;

pub(crate) struct CommandStreamHandle {
    /// Docker id of the container this stdin stream was opened against. A
    /// container that gets recreated (or restarted) keeps the same name but
    /// gets a new id, and writes to the old hijacked socket are silently
    /// swallowed — so the id is what makes a cached stream reusable.
    container_id: String,
    input: Mutex<AttachedInput>,
    drain_task: JoinHandle<()>,
}

impl CommandStreamHandle {
    pub(crate) fn new(
        container_id: String,
        input: AttachedInput,
        drain_task: JoinHandle<()>,
    ) -> Self {
        Self {
            container_id,
            input: Mutex::new(input),
            drain_task,
        }
    }

    pub(crate) fn container_id(&self) -> &str {
        &self.container_id
    }

    pub(crate) async fn write_all(&self, bytes: &[u8]) -> Result<()> {
        let mut input = self.input.lock().await;
        input.write_all(bytes).await?;
        input.flush().await?;
        Ok(())
    }

    pub(crate) fn abort(&self) {
        self.drain_task.abort();
    }
}

/// Manages the Docker client and network setup.
#[derive(Clone)]
pub struct DockerManager {
    client: Docker,
    network_name: String,
    data_root: PathBuf,
    host_data_root: PathBuf,
    command_streams: Arc<RwLock<HashMap<String, Arc<CommandStreamHandle>>>>,
}

impl DockerManager {
    pub async fn new(config: &DaemonConfig) -> Result<Self> {
        let client = Docker::connect_with_socket(
            &config.docker.socket,
            120, // timeout
            bollard::API_DEFAULT_VERSION,
        )?;

        // Verify connection
        let version = client.version().await?;
        info!(
            docker_version = version.version.as_deref().unwrap_or("unknown"),
            "Connected to Docker"
        );

        let data_root = config.data_path.clone();
        let host_data_root = config.host_data_path();
        if data_root != host_data_root {
            info!(
                data_root = %data_root.display(),
                host_data_root = %host_data_root.display(),
                "Server data directories are bind-mounted from a different host path",
            );
        }

        let manager = Self {
            client,
            network_name: config.docker.network.clone(),
            data_root,
            host_data_root,
            command_streams: Arc::new(RwLock::new(HashMap::new())),
        };

        manager.ensure_network(&config.docker.network_subnet).await?;

        Ok(manager)
    }

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

    pub fn network_name(&self) -> &str {
        &self.network_name
    }

    /// Translate a daemon-local server data directory into the path the Docker
    /// host must bind-mount. These differ when the daemon runs in a container.
    pub fn host_bind_source(&self, data_path: &Path) -> PathBuf {
        if self.data_root == self.host_data_root {
            return data_path.to_path_buf();
        }

        match data_path.strip_prefix(&self.data_root) {
            Ok(relative) => self.host_data_root.join(relative),
            Err(_) => data_path.to_path_buf(),
        }
    }

    /// Inverse of [`Self::host_bind_source`]: turn a bind-mount source reported
    /// by Docker back into a path the daemon can read and write itself.
    pub fn daemon_data_path(&self, host_path: &Path) -> PathBuf {
        if self.data_root == self.host_data_root {
            return host_path.to_path_buf();
        }

        match host_path.strip_prefix(&self.host_data_root) {
            Ok(relative) => self.data_root.join(relative),
            Err(_) => host_path.to_path_buf(),
        }
    }

    pub(crate) fn command_streams(&self) -> &Arc<RwLock<HashMap<String, Arc<CommandStreamHandle>>>> {
        &self.command_streams
    }

    async fn ensure_network(&self, subnet: &str) -> Result<()> {
        let networks = self.client.list_networks::<String>(None).await?;
        let exists = networks
            .iter()
            .any(|n| n.name.as_deref() == Some(&self.network_name));

        if !exists {
            info!(network = %self.network_name, "Creating Docker network");
            let ipam_config = bollard::models::IpamConfig {
                subnet: Some(subnet.to_string()),
                ..Default::default()
            };
            let ipam = bollard::models::Ipam {
                config: Some(vec![ipam_config]),
                ..Default::default()
            };
            self.client
                .create_network(CreateNetworkOptions {
                    name: self.network_name.clone(),
                    driver: "bridge".to_string(),
                    ipam,
                    ..Default::default()
                })
                .await?;
            info!(network = %self.network_name, "Docker network created");
        }

        Ok(())
    }
}