use std::{
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use anyhow::{Result, anyhow, bail};
use etcd_client::{Client, ConnectOptions};
use log::{debug, info, warn};
use tokio::{
    sync::{Mutex, RwLock},
    task::JoinHandle,
    time::interval,
};

/// EtcdClient that supports automatic member discovery and failover
#[derive(Clone)]
pub struct EtcdClient {
    inner: Arc<EtcdClientInner>,
}

struct EtcdClientInner {
    /// Current active etcd client
    client: RwLock<Option<Client>>,

    /// Current list of etcd endpoints
    endpoints: RwLock<Vec<String>>,

    /// Connection options for etcd client
    connect_options: Option<ConnectOptions>,

    /// Member discovery background task handle
    member_discovery_handle: Mutex<Option<JoinHandle<()>>>,

    /// Member refresh interval in seconds
    member_refresh_interval: Duration,

    /// Whether the client is running
    running: AtomicBool,
}

impl EtcdClient {
    /// Create a new etcd client with initial endpoints
    fn new(initial_endpoints: Vec<String>, connect_options: Option<ConnectOptions>) -> Self {
        Self {
            inner: Arc::new(EtcdClientInner {
                client: RwLock::new(None),
                endpoints: RwLock::new(initial_endpoints.clone()),
                connect_options,
                member_discovery_handle: Mutex::new(None),
                member_refresh_interval: Duration::from_secs(30),
                running: AtomicBool::new(false),
            }),
        }
    }

    pub fn new_tls(
        initial_endpoints: &[String],
        sec_config: &tici_common::security::SecurityConfig,
    ) -> Result<Self> {
        if sec_config.is_valid() {
            let http_pd_addrs: Vec<String> = initial_endpoints
                .iter()
                .map(|addr| {
                    if addr.starts_with("https://") {
                        addr.clone()
                    } else {
                        format!("https://{}", addr)
                    }
                })
                .collect();
            let ca = std::fs::read_to_string(sec_config.ca_path.as_str())?;
            let cert = std::fs::read_to_string(sec_config.cert_path.as_str())?;
            let key = std::fs::read_to_string(sec_config.key_path.as_str())?;

            info!(
                "new_tls: config={:?}, len={}/{}/{}",
                sec_config,
                ca.len(),
                cert.len(),
                key.len()
            );

            let tls = etcd_client::TlsOptions::new()
                .ca_certificate(etcd_client::Certificate::from_pem(ca))
                .identity(etcd_client::Identity::from_pem(cert, key));

            Ok(EtcdClient::new(
                http_pd_addrs,
                Some(etcd_client::ConnectOptions::new().with_tls(tls)),
            ))
        } else {
            let http_pd_addrs: Vec<String> = initial_endpoints
                .iter()
                .map(|addr| {
                    if addr.starts_with("http://") {
                        addr.clone()
                    } else {
                        format!("http://{}", addr)
                    }
                })
                .collect();
            Ok(EtcdClient::new(http_pd_addrs, None))
        }
    }

    pub fn new_for_test(initial_endpoints: Vec<String>) -> Self {
        EtcdClient::new(initial_endpoints, None)
    }

    /// Start the etcd client with automatic member discovery and health
    /// checking
    pub async fn start(&self) -> Result<()> {
        if self.inner.running.load(Ordering::Acquire) {
            return Ok(());
        }

        info!("Starting etcd client");

        // Try to establish initial connection
        self.reconnect().await?;

        self.inner.running.store(true, Ordering::Release);

        // Start background tasks
        self.start_member_discovery().await;

        info!("etcd client started successfully");
        Ok(())
    }

    /// Stop the etcd client and cleanup background tasks
    pub async fn stop(&self) -> Result<()> {
        if !self.inner.running.load(Ordering::Acquire) {
            return Ok(());
        }

        info!("Stopping etcd client");
        self.inner.running.store(false, Ordering::Release);

        // Stop background tasks
        if let Some(handle) = self.inner.member_discovery_handle.lock().await.take() {
            handle.abort();
        }

        info!("etcd client stopped");
        Ok(())
    }

    /// Get a client instance, automatically reconnecting if necessary
    pub async fn get_client(&self) -> Result<Client> {
        // Try to get existing client first
        if let Some(mut client) = self.inner.client.read().await.as_ref().map(|c| c.clone()) {
            // Quick health check - try a simple operation
            if self.is_client_healthy(&mut client).await {
                return Ok(client);
            }
        }

        // Client is not available or unhealthy, try to reconnect
        info!("etcd client unavailable or unhealthy, attempting reconnect");
        self.reconnect().await?;

        // Return the new client
        self.inner
            .client
            .read()
            .await
            .as_ref()
            .map(|c| c.clone())
            .ok_or_else(|| anyhow!("Failed to establish etcd connection after reconnect"))
    }

    /// Get current endpoints
    pub async fn get_endpoints(&self) -> Vec<String> {
        self.inner.endpoints.read().await.clone()
    }

    /// Attempt to reconnect to etcd cluster
    async fn reconnect(&self) -> Result<()> {
        let endpoints = self.get_endpoints().await;
        match self.try_connect(endpoints.clone()).await {
            Ok(client) => {
                *self.inner.client.write().await = Some(client);
                Ok(())
            }
            Err(e) => {
                bail!(
                    "Failed to connect to etcd using endpoints {:?}: {}",
                    endpoints,
                    e
                );
            }
        }
    }

    /// Try to connect to etcd with given endpoints
    async fn try_connect(&self, endpoints: Vec<String>) -> Result<Client> {
        let connect_fut = Client::connect(endpoints.clone(), self.inner.connect_options.clone());

        // Add timeout for connection attempt
        match tokio::time::timeout(Duration::from_secs(10), connect_fut).await {
            Ok(Ok(client)) => Ok(client),
            Ok(Err(e)) => Err(anyhow!("etcd connection failed: {}", e)),
            Err(_) => Err(anyhow!("etcd connection timed out after 10 seconds")),
        }
    }

    /// Check if the current client is healthy
    async fn is_client_healthy(&self, client: &mut Client) -> bool {
        tokio::time::timeout(Duration::from_secs(3), client.status())
            .await
            .is_ok_and(|res| res.is_ok())
    }

    /// Start background member discovery task
    async fn start_member_discovery(&self) {
        let inner = self.inner.clone();
        let handle = tokio::spawn(async move {
            let mut interval = interval(inner.member_refresh_interval);

            while inner.running.load(Ordering::Acquire) {
                interval.tick().await;

                if let Err(e) = inner.discover_cluster_members().await {
                    warn!("Failed to discover cluster members: {}", e);
                }
            }
        });

        *self.inner.member_discovery_handle.lock().await = Some(handle);
    }
}

impl EtcdClientInner {
    /// Discover etcd cluster members and update endpoint list
    async fn discover_cluster_members(&self) -> Result<()> {
        debug!("Discovering etcd cluster members");

        let mut client = match self.client.read().await.as_ref() {
            Some(client) => client.clone(),
            None => {
                return Ok(());
            }
        };

        // Get member list from etcd
        match tokio::time::timeout(Duration::from_secs(5), client.member_list()).await {
            Ok(Ok(response)) => {
                let mut new_endpoints = Vec::new();

                for member in response.members() {
                    if !member.client_urls().is_empty() {
                        // Use the first client URL for each member
                        if let Some(url) = member.client_urls().first() {
                            new_endpoints.push(url.clone());
                        }
                    }
                }

                if !new_endpoints.is_empty() {
                    let old_endpoints = self.endpoints.read().await.clone();

                    // Update endpoints if they have changed
                    if new_endpoints != old_endpoints {
                        info!("Discovered new etcd members: {:?}", new_endpoints);
                        *self.endpoints.write().await = new_endpoints.clone();
                    }
                } else {
                    warn!("No healthy etcd members discovered");
                }
            }
            Ok(Err(e)) => {
                warn!("Failed to get etcd member list: {}", e);
            }
            Err(_) => {
                warn!("etcd member list request timed out");
            }
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use tokio::time::sleep;

    use super::*;
    use crate::test_utils::setup_etcd;

    #[tokio::test]
    async fn test_client_basic_operations() {
        let endpoint = match setup_etcd().await {
            Ok(endpoint) => endpoint,
            Err(e) => {
                eprintln!("Failed to set up etcd, err={}", e);
                return;
            }
        };

        let etcd_client = EtcdClient::new_for_test(vec![endpoint]);

        // Start the client
        assert!(etcd_client.start().await.is_ok());

        // Test getting a client
        let client = etcd_client.get_client().await;
        assert!(client.is_ok());

        // Test basic operations
        let mut client = client.unwrap();
        let put_result = client.put("test_key", "test_value", None).await;
        assert!(put_result.is_ok());

        let get_result = client.get("test_key", None).await;
        assert!(get_result.is_ok());

        // Cleanup
        let _ = client.delete("test_key", None).await;

        // Stop the client
        assert!(etcd_client.stop().await.is_ok());
    }

    #[tokio::test]
    async fn test_endpoint_discovery() {
        let endpoint = match setup_etcd().await {
            Ok(endpoint) => endpoint,
            Err(e) => {
                eprintln!("Failed to set up etcd, err={}", e);
                return;
            }
        };

        let etcd_client = EtcdClient::new_for_test(vec![endpoint.clone()]);

        assert!(etcd_client.start().await.is_ok());

        // Wait for member discovery to run
        sleep(Duration::from_millis(100)).await;

        let endpoints = etcd_client.get_endpoints().await;
        assert!(!endpoints.is_empty());

        assert!(etcd_client.stop().await.is_ok());
    }
}