#[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;

impl RedisCache {
    /// Connect to Redis using the provided configuration
    pub async fn connect(config: &CacheConfig) -> CacheResult<Self> {
        #[cfg(feature = "redis_cache")]
        {
            // 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| {
                map_redis_error(e, &format!("Failed to connect to Redis at {}", redis_url))
            })?;

            // Create connection manager
            let connection = ConnectionManager::new(client)
                .await
                .map_err(|e| map_redis_error(e, "Failed to create Redis connection manager"))?;

            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)),
            })
        }

        #[cfg(not(feature = "redis_cache"))]
        {
            Err(CacheError::Connection("Redis support is not enabled. Recompile with the 'redis_cache' 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()
        }
    }

    /// Convert Redis error to CacheError
    fn convert_error(err: RedisError) -> CacheError {
        match err.kind() {
            redis::ErrorKind::IoError => {
                CacheError::Connection(format!("Redis I/O error: {}", err))
            }
            redis::ErrorKind::AuthenticationFailed => {
                CacheError::Connection(format!("Redis authentication failed: {}", err))
            }
            redis::ErrorKind::ResponseError => {
                CacheError::Operation(format!("Redis response error: {}", err))
            }
            redis::ErrorKind::ClientError => {
                CacheError::Operation(format!("Redis client error: {}", err))
            }
            redis::ErrorKind::ExtensionError => {
                CacheError::Operation(format!("Redis extension error: {}", err))
            }
            redis::ErrorKind::TypeError => {
                CacheError::Operation(format!("Redis type error: {}", err))
            }
            redis::ErrorKind::ExecAbortError => {
                CacheError::Operation(format!("Redis exec abort error: {}", err))
            }
            redis::ErrorKind::BusyLoadingError => {
                CacheError::Operation(format!("Redis busy loading: {}", err))
            }
            redis::ErrorKind::InvalidClientConfig => {
                CacheError::Configuration(format!("Redis invalid client config: {}", err))
            }
            redis::ErrorKind::Moved => CacheError::Operation(format!("Redis moved error: {}", err)),
            redis::ErrorKind::Ask => CacheError::Operation(format!("Redis ask error: {}", err)),
            redis::ErrorKind::TryAgain => {
                CacheError::Operation(format!("Redis try again: {}", err))
            }
            redis::ErrorKind::ClusterDown => {
                CacheError::Connection(format!("Redis cluster down: {}", err))
            }
            redis::ErrorKind::CrossSlot => {
                CacheError::Operation(format!("Redis cross slot: {}", err))
            }
            redis::ErrorKind::MasterDown => {
                CacheError::Connection(format!("Redis master down: {}", err))
            }
            redis::ErrorKind::NotBusy => CacheError::Operation(format!("Redis not busy: {}", err)),
            _ => CacheError::Operation(format!("Redis error: {}", err)),
        }
    }
}

impl From<RedisError> for CacheError {
    fn from(err: RedisError) -> Self {
        CacheError::Connection(format!("Redis error: {}", err))
    }
}

/// Map Redis errors to CacheError
#[cfg(feature = "redis_cache")]
fn map_redis_error(err: RedisError, context: &str) -> CacheError {
    match err.kind() {
        ErrorKind::IoError => {
            CacheError::Connection(format!("{}: {} (IO error)", context, err))
        }
        ErrorKind::AuthenticationFailed => {
            CacheError::Connection(format!("{}: {} (Authentication failed)", context, err))
        }
        ErrorKind::ResponseError => {
            CacheError::Operation(format!("{}: {} (Response error)", context, err))
        }
        ErrorKind::ClientError => {
            CacheError::Operation(format!("{}: {} (Client error)", context, err))
        }
        ErrorKind::ExtensionError => {
            CacheError::Operation(format!("{}: {} (Extension error)", context, err))
        }
        ErrorKind::TypeError => {
            CacheError::Operation(format!("{}: {} (Type error)", context, err))
        }
        ErrorKind::ExecAbortError => {
            CacheError::Operation(format!("{}: {} (Exec abort error)", context, err))
        }
        ErrorKind::BusyLoadingError => {
            CacheError::Connection(format!("{}: {} (Busy loading)", context, err))
        }
        ErrorKind::InvalidClientConfig => {
            CacheError::Connection(format!("{}: {} (Invalid client config)", context, err))
        }
        ErrorKind::Moved => CacheError::Operation(format!("Redis moved error: {}", err)),
        ErrorKind::Ask => CacheError::Operation(format!("Redis ask error: {}", err)),
        ErrorKind::TryAgain => {
            CacheError::Operation(format!("{}: {} (Try again)", context, err))
        }
        ErrorKind::ClusterDown => {
            CacheError::Connection(format!("{}: {} (Cluster down)", context, err))
        }
        ErrorKind::CrossSlot => {
            CacheError::Operation(format!("{}: {} (Cross slot)", context, err))
        }
        ErrorKind::MasterDown => {
            CacheError::Connection(format!("{}: {} (Master down)", context, err))
        }
        ErrorKind::NotBusy => CacheError::Operation(format!("Redis not busy: {}", err)),
        _ => CacheError::Operation(format!("{}: {}", context, err)),
    }
}

#[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(map_redis_error(
                        e,
                        &format!("Failed to get key {}", prefixed_key),
                    ))
                }
            }
        }
    }

    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 short-lived TTLs (under a second), use millisecond precision
        if expiry.as_secs() == 0 && expiry.subsec_millis() > 0 {
            let mut cmd = redis::cmd("SET");
            cmd.arg(&prefixed_key)
                .arg(value)
                .arg("PX")
                .arg(expiry.as_millis() as u64);

            cmd.query_async(&mut conn)
                .await
                .map_err(|e| {
                    map_redis_error(e, &format!("Failed to set key {} with PX", prefixed_key))
                })?;
        } else {
            // For normal TTLs, use second precision
            conn.set_ex(prefixed_key, value, expiry.as_secs() as usize)
                .await
                .map_err(|e| {
                    map_redis_error(e, &format!("Failed to set key {} with EX", key))
                })?;
        }

        Ok(())
    }

    async fn delete(&self, key: &str) -> CacheResult<bool> {
        let key = self.prefix_key(key);
        let result: RedisResult<i64> = self.connection.clone().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 result: RedisResult<i64> = self.connection.clone().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 expiry = ttl.unwrap_or(self.default_ttl);

        let result: RedisResult<bool> = if expiry.as_secs() > 0 {
            let mut cmd = redis::cmd("SET");
            cmd.arg(&key)
                .arg(value)
                .arg("NX")
                .arg("EX")
                .arg(expiry.as_secs());
            let response: RedisValue = cmd.query_async(&mut self.connection.clone()).await?;

            match response {
                RedisValue::Nil => Ok(false),
                RedisValue::Status(status) if status == "OK" => Ok(true),
                _ => Ok(false),
            }
        } else {
            self.connection.clone().set_nx(&key, value).await
        };

        result.map_err(Self::convert_error)
    }

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

        match result {
            Ok(value) => Ok(value),
            Err(e) => {
                // Handle nil response as None
                if e.kind() == redis::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 result: RedisResult<i64> = if delta >= 0 {
            self.connection.clone().incr(&key, delta).await
        } else {
            self.connection.clone().decr(&key, -delta).await
        };

        result.map_err(Self::convert_error)
    }

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

        let mut pipe = redis::pipe();
        let expiry = ttl.unwrap_or(self.default_ttl);

        for (key, value) in items {
            let key = self.prefix_key(key);
            if expiry.as_secs() > 0 {
                pipe.set_ex(&key, value, expiry.as_secs() as usize);
            } else {
                pipe.set(&key, value);
            }
        }

        let result: RedisResult<()> = pipe.query_async(&mut self.connection.clone()).await;
        result.map_err(Self::convert_error)
    }

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

        let prefixed_keys: Vec<String> = keys.iter().map(|k| self.prefix_key(k)).collect();

        let result: RedisResult<Vec<Option<Vec<u8>>>> = self
            .connection
            .clone()
            .get(prefixed_keys.as_slice())
            .await;

        match result {
            Ok(values) => {
                let mut map = 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 };
                        map.insert(original_key.clone(), data);
                    }
                }
                Ok(map)
            }
            Err(e) => Err(Self::convert_error(e)),
        }
    }

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

        let prefixed_keys: Vec<String> = keys.iter().map(|k| self.prefix_key(k)).collect();

        let result: RedisResult<i64> = self
            .connection
            .clone()
            .del(prefixed_keys.as_slice())
            .await;

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

    async fn clear(&self, namespace: Option<&str>) -> CacheResult<()> {
        // Use namespace from parameter, or object's namespace if not specified
        let pattern = if let Some(ns) = namespace {
            format!("{}:*", ns)
        } else if let Some(ns) = &self.namespace {
            format!("{}:*", ns)
        } else {
            "*".to_string()
        };

        // WARNING: This is a potentially expensive operation in production
        // as it uses the KEYS command
        let keys: RedisResult<Vec<String>> = redis::cmd("KEYS")
            .arg(&pattern)
            .query_async(&mut self.connection.clone())
            .await;

        match keys {
            Ok(keys) => {
                if !keys.is_empty() {
                    let result: RedisResult<i64> =
                        self.connection.clone().del(keys.as_slice()).await;
                    result.map(|_| ()).map_err(Self::convert_error)
                } else {
                    Ok(())
                }
            }
            Err(e) => Err(Self::convert_error(e)),
        }
    }

    async fn lock(&self, key: &str, ttl: Duration) -> CacheResult<Option<String>> {
        // Use Redis to implement distributed locking (based on the Redlock algorithm)
        let lock_key = self.prefix_key(&format!("lock:{}", key));
        let token = uuid::Uuid::new_v4().to_string();

        // Try to acquire the lock with NX
        let result: RedisResult<bool> = self
            .connection
            .clone()
            .set_ex(&lock_key, token.as_bytes(), ttl.as_secs() as usize)
            .await;

        match result {
            Ok(true) => Ok(Some(token)),
            Ok(false) => Ok(None),
            Err(e) => Err(Self::convert_error(e)),
        }
    }

    async fn unlock(&self, key: &str, lock_token: &str) -> CacheResult<bool> {
        // Use Lua script to ensure atomic release - only delete if the token matches
        let lock_key = self.prefix_key(&format!("lock:{}", key));

        let script = r#"
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        "#;

        let result: RedisResult<i64> = redis::Script::new(script)
            .key(&lock_key)
            .arg(lock_token.as_bytes())
            .invoke_async(&mut self.connection.clone())
            .await;

        match result {
            Ok(1) => Ok(true),
            Ok(_) => Ok(false),
            Err(e) => Err(Self::convert_error(e)),
        }
    }

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

    async fn ping(&self) -> CacheResult<()> {
        let result: RedisResult<String> = redis::cmd("PING")
            .query_async(&mut self.connection.clone())
            .await;

        match result {
            Ok(response) if response == "PONG" => Ok(()),
            Ok(_) => Err(CacheError::Connection("Invalid PING response".to_string())),
            Err(e) => Err(Self::convert_error(e)),
        }
    }

    async fn close(&self) -> CacheResult<()> {
        // ConnectionManager handles connection cleanup automatically
        Ok(())
    }

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

    async fn stats(&self) -> CacheResult<CacheStats> {
        let info_cmd: RedisResult<String> = redis::cmd("INFO")
            .query_async(&mut self.connection.clone())
            .await;

        match info_cmd {
            Ok(info) => {
                let mut stats = CacheStats {
                    item_count: None,
                    memory_used_bytes: None,
                    hits: None,
                    misses: None,
                    sets: None,
                    deletes: None,
                    additional_metrics: HashMap::new(),
                    ..Default::default()
                };

                // Parse relevant lines from INFO command output
                for line in info.lines() {
                    // Skip comments and empty lines
                    if line.starts_with('#') || line.trim().is_empty() {
                        continue;
                    }

                    if let Some((key, value)) = line.split_once(':') {
                        match key {
                            "keyspace_hits" => {
                                if let Ok(hits) = value.parse::<u64>() {
                                    stats.hits = Some(hits);
                                }
                            }
                            "keyspace_misses" => {
                                if let Ok(misses) = value.parse::<u64>() {
                                    stats.misses = Some(misses);
                                }
                            }
                            "used_memory" => {
                                if let Ok(memory) = value.parse::<u64>() {
                                    stats.memory_used_bytes = Some(memory);
                                }
                            }
                            "db0" => {
                                // Parse db0 string which looks like: "keys=123,expires=12,avg_ttl=3600"
                                if let Some(keys_part) = value.split(',').next() {
                                    if let Some(keys_str) = keys_part.strip_prefix("keys=") {
                                        if let Ok(keys) = keys_str.parse::<u64>() {
                                            stats.item_count = Some(keys);
                                        }
                                    }
                                }
                            }
                            // Add other interesting metrics to additional_metrics
                            "connected_clients"
                            | "total_connections_received"
                            | "expired_keys"
                            | "evicted_keys"
                            | "uptime_in_seconds" => {
                                stats
                                    .additional_metrics
                                    .insert(key.to_string(), value.to_string());
                            }
                            _ => {}
                        }
                    }
                }

                Ok(stats)
            }
            Err(e) => Err(Self::convert_error(e)),
        }
    }

    async fn flush(&self) -> CacheResult<()> {
        let pattern = if let Some(ns) = &self.namespace {
            format!("{}:*", ns)
        } else {
            "*".to_string()
        };

        // WARNING: This is a potentially expensive operation in production
        // as it uses the KEYS command
        let keys: RedisResult<Vec<String>> = redis::cmd("KEYS")
            .arg(&pattern)
            .query_async(&mut self.connection.clone())
            .await;

        match keys {
            Ok(keys) => {
                if !keys.is_empty() {
                    let result: RedisResult<i64> =
                        self.connection.clone().del(keys.as_slice()).await;
                    result.map(|_| ()).map_err(Self::convert_error)
                } else {
                    Ok(())
                }
            }
            Err(e) => Err(Self::convert_error(e)),
        }
    }
}

#[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()))
    }
}

#[cfg(not(feature = "redis_cache"))]
impl RedisCache {
    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()))
    }
}
