use crate::cache::{CacheConfig, CacheError, CacheResult, CacheService, CacheStats, CacheType};
use async_trait::async_trait;
#[cfg(feature = "memcached")]
use memcache::Client;
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use uuid::Uuid;

/// Memcached cache implementation
pub struct MemcachedCache {
    /// Memcached client
    #[cfg(feature = "memcached")]
    client: Arc<Mutex<Client>>,
    #[cfg(not(feature = "memcached"))]
    client: Arc<Mutex<()>>,
    /// Default TTL for cache entries
    default_ttl: Duration,
    /// Namespace prefix for cache keys
    namespace: Option<String>,
    /// Cache hit counter
    hits: Arc<AtomicU64>,
    /// Cache miss counter
    misses: Arc<AtomicU64>,
}

impl MemcachedCache {
    /// Connect to Memcached using the provided configuration
    pub async fn connect(config: &CacheConfig) -> CacheResult<Self> {
        #[cfg(feature = "memcached")]
        {
            // Build server list
            let servers = if config.hosts.is_empty() {
                vec!["127.0.0.1".to_string()]
            } else {
                config.hosts.clone()
            };

            // Add port if specified
            let servers_with_port: Vec<String> = servers
                .iter()
                .map(|host| {
                    if let Some(port) = config.port {
                        format!("{}:{}", host, port)
                    } else {
                        // Default memcached port
                        format!("{}:11211", host)
                    }
                })
                .collect();

            // Join server strings for client creation
            let servers_str = servers_with_port.join(",");

            // Create client
            let client = Client::connect(servers_str).map_err(|e| {
                CacheError::Connection(format!("Failed to connect to Memcached: {}", e))
            })?;

            // Check connection with a ping
            client
                .stats()
                .map_err(|e| CacheError::Connection(format!("Failed to ping Memcached: {}", e)))?;

            Ok(Self {
                client: Arc::new(Mutex::new(client)),
                default_ttl: config.get_default_ttl(),
                namespace: config.namespace.clone(),
                hits: Arc::new(AtomicU64::new(0)),
                misses: Arc::new(AtomicU64::new(0)),
            })
        }

        #[cfg(not(feature = "memcached"))]
        {
            Err(CacheError::Connection("Memcached support is not enabled. Recompile with the 'memcached' feature.".to_string()))
        }
    }

    /// Add namespace prefix to key if configured
    fn prefix_key(&self, key: &str) -> String {
        if let Some(ns) = &self.namespace {
            format!("{}:{}", ns, key)
        } else {
            key.to_string()
        }
    }

    /// Strip namespace prefix from key if needed
    fn strip_prefix(&self, key: &str) -> String {
        if let Some(ns) = &self.namespace {
            let prefix = format!("{}:", ns);
            if key.starts_with(&prefix) {
                key[prefix.len()..].to_string()
            } else {
                key.to_string()
            }
        } else {
            key.to_string()
        }
    }
}

#[async_trait]
impl CacheService for MemcachedCache {
    async fn get(&self, key: &str) -> CacheResult<Option<Vec<u8>>> {
        let key = self.prefix_key(key);

        #[cfg(feature = "memcached")]
        {
            let client = self.client.lock().unwrap();
            match client.get::<Vec<u8>>(&key) {
                Ok(value) => {
                    if value.is_some() {
                        self.hits.fetch_add(1, Ordering::Relaxed);
                    } else {
                        self.misses.fetch_add(1, Ordering::Relaxed);
                    }
                    Ok(value)
                }
                Err(e) => {
                    // Handle not found error gracefully
                    if e.to_string().contains("not found") {
                        self.misses.fetch_add(1, Ordering::Relaxed);
                        Ok(None)
                    } else {
                        Err(CacheError::Operation(format!("Memcached get error: {}", e)))
                    }
                }
            }
        }

        #[cfg(not(feature = "memcached"))]
        {
            Err(CacheError::Operation("Memcached support is not enabled. Recompile with the 'memcached' feature.".to_string()))
        }
    }

    async fn set(&self, key: &str, value: &[u8], ttl: Option<Duration>) -> CacheResult<()> {
        let key = self.prefix_key(key);
        let expiry = ttl.unwrap_or(self.default_ttl);
        let expiry_secs = expiry.as_secs() as u32;

        let client = self.client.lock().unwrap();
        client
            .set(&key, value, expiry_secs)
            .map_err(|e| CacheError::Operation(format!("Memcached set error: {}", e)))?;

        Ok(())
    }

    async fn delete(&self, key: &str) -> CacheResult<bool> {
        let key = self.prefix_key(key);

        let client = self.client.lock().unwrap();
        match client.delete(&key) {
            Ok(_) => Ok(true),
            Err(e) => {
                // Handle not found error gracefully
                if e.to_string().contains("not found") {
                    Ok(false)
                } else {
                    Err(CacheError::Operation(format!(
                        "Memcached delete error: {}",
                        e
                    )))
                }
            }
        }
    }

    async fn exists(&self, key: &str) -> CacheResult<bool> {
        // Memcached doesn't have a direct exists operation, so we use get
        let result = self.get(key).await?;
        Ok(result.is_some())
    }

    async fn set_nx(&self, key: &str, value: &[u8], ttl: Option<Duration>) -> CacheResult<bool> {
        let key = self.prefix_key(key);
        let expiry = ttl.unwrap_or(self.default_ttl);
        let expiry_secs = expiry.as_secs() as u32;

        let client = self.client.lock().unwrap();
        match client.add(&key, value, expiry_secs) {
            Ok(_) => Ok(true),
            Err(e) => {
                // Handle already exists error gracefully
                if e.to_string().contains("already exists") || e.to_string().contains("not stored")
                {
                    Ok(false)
                } else {
                    Err(CacheError::Operation(format!("Memcached add error: {}", e)))
                }
            }
        }
    }

    async fn get_set(&self, key: &str, value: &[u8]) -> CacheResult<Option<Vec<u8>>> {
        let key = self.prefix_key(key);

        // Memcached doesn't have a direct get-and-set operation
        // So we need to get first, then set
        let client = self.client.lock().unwrap();

        // Get the current value
        let current_value = match client.get::<Vec<u8>>(&key) {
            Ok(value) => {
                if value.is_some() {
                    self.hits.fetch_add(1, Ordering::Relaxed);
                } else {
                    self.misses.fetch_add(1, Ordering::Relaxed);
                }
                value
            }
            Err(e) => {
                // Handle not found error gracefully
                if e.to_string().contains("not found") {
                    self.misses.fetch_add(1, Ordering::Relaxed);
                    None
                } else {
                    return Err(CacheError::Operation(format!("Memcached get error: {}", e)));
                }
            }
        };

        // Set the new value
        let expiry = self.default_ttl.as_secs() as u32;
        client
            .set(&key, value, expiry)
            .map_err(|e| CacheError::Operation(format!("Memcached set error: {}", e)))?;

        Ok(current_value)
    }

    async fn increment(&self, key: &str, delta: i64) -> CacheResult<i64> {
        let key = self.prefix_key(key);

        let client = self.client.lock().unwrap();

        // Memcached doesn't handle negative increments directly
        let result = if delta >= 0 {
            client.increment(&key, delta as u64)
        } else {
            // For decrement, we need to use the absolute value
            client.decrement(&key, delta.unsigned_abs())
        };

        match result {
            Ok(value) => Ok(value as i64),
            Err(e) => {
                // Handle not found error - initialize with delta if positive, or 0 if negative
                if e.to_string().contains("not found") {
                    let init_value = if delta >= 0 { delta } else { 0 };
                    let expiry = self.default_ttl.as_secs() as u32;
                    client
                        .set(&key, init_value.to_string().as_bytes(), expiry)
                        .map_err(|e| {
                            CacheError::Operation(format!("Memcached set error: {}", e))
                        })?;
                    Ok(init_value)
                } else {
                    Err(CacheError::Operation(format!(
                        "Memcached increment error: {}",
                        e
                    )))
                }
            }
        }
    }

    async fn set_many(
        &self,
        items: &HashMap<String, Vec<u8>>,
        ttl: Option<Duration>,
    ) -> CacheResult<()> {
        if items.is_empty() {
            return Ok(());
        }

        let expiry = ttl.unwrap_or(self.default_ttl);
        let expiry_secs = expiry.as_secs() as u32;

        let client = self.client.lock().unwrap();

        // Memcached client doesn't have a direct multi-set, so we set each individually
        for (key, value) in items {
            let key = self.prefix_key(key);
            client
                .set(&key, value.as_slice(), expiry_secs)
                .map_err(|e| CacheError::Operation(format!("Memcached set error: {}", e)))?;
        }

        Ok(())
    }

    async fn get_many(&self, keys: &[String]) -> CacheResult<HashMap<String, Vec<u8>>> {
        if keys.is_empty() {
            return Ok(HashMap::new());
        }

        let mut result_map = HashMap::new();
        
        // Client doesn't support atomic multi-get, so we need to do it one by one
        let client = self.client.lock().unwrap();
        
        for key in keys {
            let prefixed_key = self.prefix_key(key);
            
            // The memcached client returns Result<Option<Vec<u8>>> from get
            match client.get(&prefixed_key) {
                Ok(Some(value)) => {
                    // We got a value, unwrap the Option
                    result_map.insert(self.strip_prefix(key), value);
                    self.hits.fetch_add(1, Ordering::Relaxed);
                }
                Ok(None) => {
                    // No value found, don't add to result map
                    self.misses.fetch_add(1, Ordering::Relaxed);
                }
                Err(e) => {
                    if e.to_string().contains("not found") {
                        // Key not found is not an error
                        self.misses.fetch_add(1, Ordering::Relaxed);
                    } else {
                        return Err(CacheError::Operation(format!("Memcached get error: {}", e)));
                    }
                }
            }
        }
        
        Ok(result_map)
    }

    async fn delete_many(&self, keys: &[String]) -> CacheResult<u64> {
        if keys.is_empty() {
            return Ok(0);
        }

        let client = self.client.lock().unwrap();
        let mut count = 0;

        // Memcached client doesn't have a direct multi-delete, so we delete each individually
        for key in keys {
            let key = self.prefix_key(key);
            match client.delete(&key) {
                Ok(_) => count += 1,
                Err(e) => {
                    // Ignore "not found" errors
                    if !e.to_string().contains("not found") {
                        return Err(CacheError::Operation(format!(
                            "Memcached delete error: {}",
                            e
                        )));
                    }
                }
            }
        }

        Ok(count)
    }

    async fn clear(&self, namespace: Option<&str>) -> CacheResult<()> {
        // Memcached doesn't support pattern-based deletion
        // If we're clearing a specific namespace and it matches our current namespace,
        // or if we're clearing all (no namespace specified) and we don't have a namespace set,
        // then we flush the entire cache

        if (namespace.is_some() && self.namespace.as_deref() == namespace)
            || (namespace.is_none() && self.namespace.is_none())
        {
            let client = self.client.lock().unwrap();
            client
                .flush()
                .map_err(|e| CacheError::Operation(format!("Memcached flush error: {}", e)))?;
            return Ok(());
        }

        // If we have a namespace mismatch or we're trying to clear a specific namespace
        // within our namespace, we can't do it efficiently with Memcached
        // We would need to maintain a list of keys for each namespace, which is beyond
        // the scope of this implementation

        Err(CacheError::Operation(
            "Clearing specific namespace not directly supported by Memcached implementation"
                .to_string(),
        ))
    }

    async fn lock(&self, key: &str, ttl: Duration) -> CacheResult<Option<String>> {
        let lock_key = self.prefix_key(&format!("lock:{}", key));
        let token = Uuid::new_v4().to_string();

        // Try to add the lock (add will fail if the key already exists)
        if self.set_nx(&lock_key, token.as_bytes(), Some(ttl)).await? {
            Ok(Some(token))
        } else {
            Ok(None)
        }
    }

    async fn unlock(&self, key: &str, lock_token: &str) -> CacheResult<bool> {
        let lock_key = self.prefix_key(&format!("lock:{}", key));

        // Check if lock exists and token matches
        let current_token = match self.get(&lock_key).await? {
            Some(data) => String::from_utf8_lossy(&data).to_string(),
            None => return Ok(false), // Lock doesn't exist
        };

        if current_token == lock_token {
            // Remove the lock if token matches
            self.delete(&lock_key).await
        } else {
            // Token doesn't match
            Ok(false)
        }
    }

    fn get_cache_type(&self) -> CacheType {
        CacheType::Memcached
    }

    async fn ping(&self) -> CacheResult<()> {
        let client = self.client.lock().unwrap();
        client
            .stats()
            .map_err(|e| CacheError::Connection(format!("Failed to ping Memcached: {}", e)))?;

        Ok(())
    }

    async fn close(&self) -> CacheResult<()> {
        // No explicit close needed for memcache client
        Ok(())
    }

    fn get_default_ttl(&self) -> Duration {
        self.default_ttl
    }

    async fn stats(&self) -> CacheResult<CacheStats> {
        let client = self.client.lock().unwrap();

        match client.stats() {
            Ok(stats_map) => {
                let mut stats = CacheStats {
                    item_count: None,
                    memory_used_bytes: None,
                    hits: Some(self.hits.load(Ordering::Relaxed)),
                    misses: Some(self.misses.load(Ordering::Relaxed)),
                    sets: None,
                    deletes: None,
                    additional_metrics: HashMap::new(),
                    ..Default::default()
                };

                // Process stats from all servers
                for (server, server_stats) in stats_map {
                    // Extract common metrics
                    for (key, value) in server_stats {
                        match key.as_str() {
                            "curr_items" => {
                                if let Ok(count) = value.parse::<u64>() {
                                    stats.item_count = Some(count);
                                }
                            }
                            "bytes" => {
                                if let Ok(bytes) = value.parse::<u64>() {
                                    stats.memory_used_bytes = Some(bytes);
                                }
                            }
                            "get_hits" | "get_misses" | "evictions" | "cmd_get" | "cmd_set"
                            | "bytes_read" | "bytes_written" => {
                                stats
                                    .additional_metrics
                                    .insert(format!("{}.{}", server, key), value);
                            }
                            _ => {}
                        }
                    }
                }

                Ok(stats)
            }
            Err(e) => Err(CacheError::Operation(format!(
                "Failed to get Memcached stats: {}",
                e
            ))),
        }
    }

    async fn flush(&self) -> CacheResult<()> {
        let client = self.client.lock().map_err(|e| CacheError::Internal(format!("Memcached client lock poisoned: {}", e)))?;
        client.flush().map_err(|e| CacheError::Operation(format!("Memcached flush error: {}", e)))?;
        Ok(())
    }
}
