#[cfg(feature = "redis_cache")]
use crate::cache::{CacheConfig, CacheError, CacheResult, CacheService, CacheStats, CacheType};
#[cfg(feature = "redis_cache")]
use async_trait::async_trait;
#[cfg(feature = "redis_cache")]
use redis::{
    aio::ConnectionManager, AsyncCommands, Client, ErrorKind, FromRedisValue, RedisError,
    RedisResult, Script, ToRedisArgs, Value as RedisValue,
};
#[cfg(feature = "redis_cache")]
use std::collections::HashMap;
#[cfg(feature = "redis_cache")]
use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(feature = "redis_cache")]
use std::sync::Arc;
#[cfg(feature = "redis_cache")]
use std::time::Duration;
#[cfg(feature = "redis_cache")]
use uuid::Uuid;

/// Redis cache implementation
#[cfg(feature = "redis_cache")]
pub struct RedisCache {
    /// Redis connection manager
    connection: Arc<ConnectionManager>,
    /// 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>,
}

#[cfg(not(feature = "redis_cache"))]
pub struct RedisCache;

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

        // Add port if specified
        let server_with_port = if let Some(port) = config.port {
            format!("{}:{}", server, port)
        } else {
            // Default Redis port
            format!("{}:6379", server)
        };

        // Build Redis URL
        let mut redis_url = format!("redis://{}", server_with_port);

        // Add credentials if provided
        if let (Some(username), Some(password)) = (&config.username, &config.password) {
            redis_url = format!("redis://{}:{}@{}", username, password, server_with_port);
        } else if let Some(password) = &config.password {
            redis_url = format!("redis://:{}@{}", password, server_with_port);
        }

        // Create client
        let client = Client::open(redis_url.clone())
            .map_err(|e| Self::convert_error(e))?;

        // Create connection manager
        let connection = ConnectionManager::new(client)
            .await
            .map_err(|e| Self::convert_error(e))?;

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

    /// 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()
        }
    }

    /// Convert Redis errors to CacheError
    fn convert_error(err: RedisError) -> CacheError {
        match err.kind() {
            ErrorKind::IoError => {
                CacheError::Connection(format!("Redis I/O error: {}", err))
            }
            ErrorKind::AuthenticationFailed => {
                CacheError::Connection(format!("Redis authentication failed: {}", err))
            }
            ErrorKind::BusyLoadingError => {
                CacheError::Connection(format!("Redis busy loading: {}", err))
            }
            ErrorKind::InvalidClientConfig => {
                CacheError::Configuration(format!("Redis invalid client config: {}", err))
            }
            ErrorKind::ClusterDown => {
                CacheError::Connection(format!("Redis cluster down: {}", err))
            }
            ErrorKind::MasterDown => {
                CacheError::Connection(format!("Redis master down: {}", err))
            }
            _ => CacheError::Operation(format!("Redis error: {}", err)),
        }
    }
}

#[cfg(not(feature = "redis_cache"))]
impl RedisCache {
    /// Connect to Redis using the provided configuration
    pub async fn connect(_config: &crate::cache::CacheConfig) -> crate::cache::CacheResult<Self> {
        Err(crate::cache::CacheError::Connection(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }
}

#[cfg(feature = "redis_cache")]
#[async_trait]
impl CacheService for RedisCache {
    async fn get(&self, key: &str) -> CacheResult<Option<Vec<u8>>> {
        let prefixed_key = self.prefix_key(key);
        let mut conn = self.connection.clone();

        // Execute Redis GET command
        match conn.get::<_, Option<Vec<u8>>>(&prefixed_key).await {
            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 nil responses gracefully
                if e.kind() == ErrorKind::TypeError || e.to_string().contains("nil") {
                    self.misses.fetch_add(1, Ordering::Relaxed);
                    Ok(None)
                } else {
                    Err(Self::convert_error(e))
                }
            }
        }
    }

    async fn set(&self, key: &str, value: &[u8], ttl: Option<Duration>) -> CacheResult<()> {
        let prefixed_key = self.prefix_key(key);
        let mut conn = self.connection.clone();
        let expiry = ttl.unwrap_or(self.default_ttl);

        // For normal TTLs, use second precision
        conn.set_ex(&prefixed_key, value, expiry.as_secs() as usize)
            .await
            .map_err(Self::convert_error)?;

        Ok(())
    }

    async fn delete(&self, key: &str) -> CacheResult<bool> {
        let key = self.prefix_key(key);
        let mut conn = self.connection.clone();
        let result: RedisResult<i64> = conn.del(&key).await;

        match result {
            Ok(count) => Ok(count > 0),
            Err(e) => Err(Self::convert_error(e)),
        }
    }

    async fn exists(&self, key: &str) -> CacheResult<bool> {
        let key = self.prefix_key(key);
        let mut conn = self.connection.clone();
        let result: RedisResult<i64> = conn.exists(&key).await;

        match result {
            Ok(count) => Ok(count > 0),
            Err(e) => Err(Self::convert_error(e)),
        }
    }

    async fn set_nx(
        &self,
        key: &str,
        value: &[u8],
        ttl: Option<Duration>,
    ) -> CacheResult<bool> {
        let key = self.prefix_key(key);
        let mut conn = self.connection.clone();
        let result = conn.set_nx(&key, value).await.map_err(Self::convert_error)?;
        
        // Set TTL if needed (must be done as a separate command)
        if result && ttl.is_some() {
            let expire_result = conn
                .expire(&key, ttl.unwrap().as_secs() as usize)
                .await
                .map_err(Self::convert_error)?;
            if !expire_result {
                // Key disappeared between set_nx and expire, consider this a failure
                return Ok(false);
            }
        }
        
        Ok(result)
    }

    async fn get_set(&self, key: &str, value: &[u8]) -> CacheResult<Option<Vec<u8>>> {
        let key = self.prefix_key(key);
        let mut conn = self.connection.clone();
        let result: RedisResult<Option<Vec<u8>>> = conn.get_set(&key, value).await;

        match result {
            Ok(old_value) => Ok(old_value),
            Err(e) => {
                // Handle nil responses gracefully
                if e.kind() == ErrorKind::TypeError || e.to_string().contains("nil") {
                    Ok(None)
                } else {
                    Err(Self::convert_error(e))
                }
            }
        }
    }

    async fn increment(&self, key: &str, delta: i64) -> CacheResult<i64> {
        let key = self.prefix_key(key);
        let mut conn = self.connection.clone();
        let result: RedisResult<i64> = if delta >= 0 {
            conn.incr(&key, delta).await
        } else {
            conn.decr(&key, -delta).await
        };

        match result {
            Ok(value) => Ok(value),
            Err(e) => Err(Self::convert_error(e)),
        }
    }

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

        let mut conn = self.connection.clone();
        let mut pipe = redis::pipe();
        
        // Add all SET commands to the pipeline
        for (key, value) in items {
            let prefixed_key = self.prefix_key(key);
            if let Some(ttl) = ttl {
                pipe.set_ex(&prefixed_key, value.as_slice(), ttl.as_secs() as usize);
            } else {
                pipe.set(&prefixed_key, value.as_slice());
            }
        }
        
        pipe.query_async(&mut conn).await.map_err(Self::convert_error)?;
        Ok(())
    }

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

        // Prefix all keys
        let prefixed_keys: Vec<String> = keys.iter().map(|k| self.prefix_key(k)).collect();
        
        let mut conn = self.connection.clone();
        let values: Vec<Option<Vec<u8>>> = conn
            .get(prefixed_keys.as_slice())
            .await
            .map_err(Self::convert_error)?;
        
        // Combine keys and values into result map
        let mut result = HashMap::new();
        for (i, value) in values.into_iter().enumerate() {
            if let Some(data) = value {
                let original_key = if i < keys.len() { &keys[i] } else { continue };
                result.insert(original_key.clone(), data);
            }
        }
        
        Ok(result)
    }

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

        // Prefix all keys
        let prefixed_keys: Vec<String> = keys.iter().map(|k| self.prefix_key(k)).collect();
        
        let mut conn = self.connection.clone();
        let count: i64 = conn
            .del(prefixed_keys.as_slice())
            .await
            .map_err(Self::convert_error)?;
        
        Ok(count as u64)
    }

    async fn clear(&self, namespace: Option<&str>) -> CacheResult<()> {
        let mut conn = self.connection.clone();
        
        // If namespace is not specified, use the current namespace
        let target_namespace = namespace.or_else(|| self.namespace.as_deref());
        
        // Build the pattern for keys to remove
        let pattern = if let Some(ns) = target_namespace {
            format!("{}:*", ns)
        } else {
            "*".to_string()
        };
        
        // Get all keys matching the pattern
        let keys: Vec<String> = conn
            .keys(&pattern)
            .await
            .map_err(Self::convert_error)?;
        
        // If there are keys, delete them
        if !keys.is_empty() {
            conn.del(keys.as_slice())
                .await
                .map_err(Self::convert_error)?;
        }
        
        Ok(())
    }

    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();
        let mut conn = self.connection.clone();
        
        // Try to acquire lock with NX (only if not exists)
        let acquired: bool = conn
            .set_nx(&lock_key, token.as_bytes())
            .await
            .map_err(Self::convert_error)?;
        
        if acquired {
            // Set TTL
            conn.expire(&lock_key, ttl.as_secs() as usize)
                .await
                .map_err(Self::convert_error)?;
            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));
        let mut conn = self.connection.clone();
        
        // Use Lua script for atomic check-and-delete operation
        let script = r#"
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
        "#;
        
        let result: i64 = Script::new(script)
            .key(&lock_key)
            .arg(lock_token.as_bytes())
            .invoke_async(&mut conn)
            .await
            .map_err(Self::convert_error)?;
        
        Ok(result > 0)
    }

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

    async fn ping(&self) -> CacheResult<()> {
        let mut conn = self.connection.clone();
        let result: String = conn.ping().await.map_err(Self::convert_error)?;
        
        if result == "PONG" {
            Ok(())
        } else {
            Err(CacheError::Connection(format!("Unexpected ping response: {}", result)))
        }
    }

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

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

    async fn stats(&self) -> CacheResult<CacheStats> {
        let mut conn = self.connection.clone();
        let info: String = conn.info().await.map_err(Self::convert_error)?;
        
        // Parse Redis INFO response
        let mut stats = CacheStats {
            hits: Some(self.hits.load(Ordering::Relaxed)),
            misses: Some(self.misses.load(Ordering::Relaxed)),
            ..Default::default()
        };
        
        // Extract key stats from INFO output
        for line in info.lines() {
            let line = line.trim();
            if line.starts_with('#') || line.is_empty() {
                continue;
            }
            
            if let Some(pos) = line.find(':') {
                let key = &line[0..pos];
                let value = &line[pos + 1..];
                
                match key {
                    "used_memory" => {
                        if let Ok(val) = value.parse::<u64>() {
                            stats.memory_used_bytes = Some(val);
                        }
                    }
                    "db0" => {
                        // Format: db0:keys=1,expires=0,avg_ttl=0
                        if let Some(pos) = value.find("keys=") {
                            if let Some(end) = value[pos + 5..].find(',') {
                                if let Ok(val) = value[pos + 5..pos + 5 + end].parse::<u64>() {
                                    stats.item_count = Some(val);
                                }
                            }
                        }
                    }
                    _ => {
                        // Add other metrics to additional_metrics
                        stats.additional_metrics.insert(key.to_string(), value.to_string());
                    }
                }
            }
        }
        
        Ok(stats)
    }

    async fn flush(&self) -> CacheResult<()> {
        let mut conn = self.connection.clone();
        conn.flushdb(true).await.map_err(Self::convert_error)
    }
}

#[cfg(not(feature = "redis_cache"))]
#[async_trait::async_trait]
impl crate::cache::CacheService for RedisCache {
    async fn get(&self, _key: &str) -> crate::cache::CacheResult<Option<Vec<u8>>> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn set(
        &self,
        _key: &str,
        _value: &[u8],
        _ttl: Option<std::time::Duration>,
    ) -> crate::cache::CacheResult<()> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn delete(&self, _key: &str) -> crate::cache::CacheResult<bool> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn exists(&self, _key: &str) -> crate::cache::CacheResult<bool> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn set_nx(
        &self,
        _key: &str,
        _value: &[u8],
        _ttl: Option<std::time::Duration>,
    ) -> crate::cache::CacheResult<bool> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn get_set(&self, _key: &str, _value: &[u8]) -> crate::cache::CacheResult<Option<Vec<u8>>> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn increment(&self, _key: &str, _delta: i64) -> crate::cache::CacheResult<i64> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn set_many(
        &self,
        _items: &std::collections::HashMap<String, Vec<u8>>,
        _ttl: Option<std::time::Duration>,
    ) -> crate::cache::CacheResult<()> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn get_many(
        &self,
        _keys: &[String],
    ) -> crate::cache::CacheResult<std::collections::HashMap<String, Vec<u8>>> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn delete_many(&self, _keys: &[String]) -> crate::cache::CacheResult<u64> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn clear(&self, _namespace: Option<&str>) -> crate::cache::CacheResult<()> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn lock(
        &self,
        _key: &str,
        _ttl: std::time::Duration,
    ) -> crate::cache::CacheResult<Option<String>> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn unlock(&self, _key: &str, _lock_token: &str) -> crate::cache::CacheResult<bool> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    fn get_cache_type(&self) -> crate::cache::CacheType {
        crate::cache::CacheType::Redis
    }

    async fn ping(&self) -> crate::cache::CacheResult<()> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn close(&self) -> crate::cache::CacheResult<()> {
        Ok(())
    }

    fn get_default_ttl(&self) -> std::time::Duration {
        std::time::Duration::from_secs(300)
    }

    async fn stats(&self) -> crate::cache::CacheResult<crate::cache::CacheStats> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }

    async fn flush(&self) -> crate::cache::CacheResult<()> {
        Err(crate::cache::CacheError::Operation(
            "Redis support is not enabled. Recompile with the 'redis_cache' feature.".to_string(),
        ))
    }
}
