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

use bollard::{
    Docker,
    container::{
        Config, CreateContainerOptions, ListContainersOptions, RemoveContainerOptions,
        RestartContainerOptions, StartContainerOptions,
    },
    secret::HostConfig,
};
use photon_etcd_cluster::{ClusterNode, ClusterNodeBuilder, MetricsCollector, NodeMetadata};
use once_cell::sync::Lazy;
use serde_json::json;
use tracing::info;

static DOCKER: Lazy<Docker> = Lazy::new(|| Docker::connect_with_local_defaults().unwrap());

pub struct Etcd {
    container_id: String,
    pub endpoint: String,
    port: u16,
}

impl Etcd {
    pub async fn restart(&mut self) {
        info!("Restarting etcd container {}", self.container_id);

        // Stop the container
        DOCKER
            .restart_container(&self.container_id, Some(RestartContainerOptions { t: 0 }))
            .await
            .unwrap();

        // Wait for container to be ready
        wait_for_port(self.port).await;

        info!("Etcd restarted at {}", self.endpoint);
    }
}

pub async fn start_etcd() -> Etcd {
    let exp_port = 2377;
    let image = "docker.io/bitnami/etcd:3.5.21";
    let container_name = "etcd-test-container";

    let filters = HashMap::from([("name", vec![container_name])]);
    let existing = DOCKER
        .list_containers(Some(ListContainersOptions {
            all: true,
            filters,
            ..Default::default()
        }))
        .await
        .unwrap();

    if !existing.is_empty() {
        info!("Cleaning up existing container");
        DOCKER
            .remove_container(
                container_name,
                Some(RemoveContainerOptions {
                    force: true,
                    v: true,
                    ..Default::default()
                }),
            )
            .await
            .unwrap();
    }

    // Create container configuration
    let config = Config {
        image: Some(image.to_string()),
        env: Some(vec![
            "ALLOW_NONE_AUTHENTICATION=yes".to_string(),
            "ETCDCTL_API=3".to_string(),
            "ETCD_DATA_DIR=/bitnami/etcd/data".to_string(),
            format!("ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:{}", exp_port),
            format!("ETCD_ADVERTISE_CLIENT_URLS=http://0.0.0.0:{}", exp_port),
        ]),
        exposed_ports: Some({
            let mut ports = HashMap::new();
            ports.insert(format!("{}/tcp", exp_port), HashMap::new());
            ports
        }),
        host_config: Some(HostConfig {
            port_bindings: Some({
                let mut bindings = HashMap::new();
                bindings.insert(
                    format!("{}/tcp", exp_port),
                    Some(vec![bollard::models::PortBinding {
                        host_ip: None,
                        host_port: Some(exp_port.to_string()),
                    }]),
                );
                bindings
            }),
            ..Default::default()
        }),
        ..Default::default()
    };

    // Create and start container
    let container = DOCKER
        .create_container(
            Some(CreateContainerOptions::<&str> {
                name: container_name,
                platform: None,
            }),
            config,
        )
        .await
        .unwrap();
    DOCKER
        .start_container(&container.id, None::<StartContainerOptions<String>>)
        .await
        .unwrap();

    wait_for_port(exp_port).await;

    let endpoint = format!("http://127.0.0.1:{}", exp_port);

    Etcd {
        container_id: container.id,
        endpoint,
        port: exp_port,
    }
}

async fn wait_for_port(port: u16) {
    use std::time::{Duration, Instant};
    use tokio::net::TcpStream;

    let timeout = Duration::from_secs(10);
    let start = Instant::now();

    loop {
        match TcpStream::connect(("127.0.0.1", port)).await {
            Ok(_) => break,
            Err(_) if start.elapsed() < timeout => {
                tokio::time::sleep(Duration::from_millis(50)).await;
            }
            Err(e) => panic!("Port {port} not ready: {e}"),
        }
    }
}

pub fn produce_cluster_node(endpoints: Vec<String>, id: &str) -> ClusterNode {
    ClusterNode::new(
        endpoints,
        id.to_string(),
        "127.0.0.1".parse().unwrap(),
        "default".into(),
        Some(3),
    )
}

/// A test metrics collector that returns configurable static metrics.
pub struct TestMetricsCollector {
    metrics: Arc<std::sync::RwLock<NodeMetadata>>,
}

impl TestMetricsCollector {
    pub fn new(initial_metrics: NodeMetadata) -> Self {
        Self {
            metrics: Arc::new(std::sync::RwLock::new(initial_metrics)),
        }
    }

    /// Get a clone of the internal Arc for sharing with tests.
    pub fn metrics_handle(&self) -> Arc<std::sync::RwLock<NodeMetadata>> {
        Arc::clone(&self.metrics)
    }
}

impl MetricsCollector for TestMetricsCollector {
    fn collect(&self) -> NodeMetadata {
        self.metrics.read().unwrap().clone()
    }
}

/// Creates a cluster node with custom metrics collector.
pub fn produce_cluster_node_with_metrics(
    endpoints: Vec<String>,
    id: &str,
    collector: impl MetricsCollector + 'static,
    metrics_update_interval: u64,
) -> ClusterNode {
    ClusterNodeBuilder::new(
        endpoints,
        id.to_string(),
        "127.0.0.1".parse().unwrap(),
        "default".into(),
    )
    .ttl(3)
    .metrics_collector(collector)
    .metrics_update_interval(metrics_update_interval)
    .build()
}

/// Creates test metrics with typical system values.
pub fn create_test_metrics(cpu: f64, memory: f64, queue_depth: u32) -> NodeMetadata {
    json!({
        "cpu_usage_percent": cpu,
        "memory_usage_percent": memory,
        "queue_depth": queue_depth,
        "custom_field": "test_value"
    })
}
