//! Redis connection pool and caching layer for HFT operations
//!
//! This module provides high-performance Redis connectivity optimized for
//! sub-millisecond caching operations in high-frequency trading.

use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::time::{Duration, Instant};
use thiserror::Error;
use tokio::sync::RwLock;

// Using redis-rs for Redis connectivity with optimized connection management
use redis::aio::ConnectionManager;
use redis::{AsyncCommands, Client, Pipeline, RedisResult};
use std::collections::VecDeque;
use tokio::sync::Semaphore;

/// Redis-specific errors
#[derive(Debug, Error)]
/// RedisError
///
/// Auto-generated documentation placeholder - enhance with specifics
pub enum RedisError {
    #[error("Connection failed: {0}")]
    // Connection variant
    Connection(#[from] redis::RedisError),
    #[error("Timeout: operation took {actual_ms}ms, max allowed {max_ms}ms")]
    Timeout { actual_ms: u64, max_ms: u64 },
    #[error("Serialization error: {0}")]
    // Serialization variant
    Serialization(String),
    #[error("Pool exhausted: no connections available")]
    // PoolExhausted variant
    PoolExhausted,
    #[error("Configuration error: {0}")]
    // Configuration variant
    Configuration(String),
    #[error("Semaphore acquire error: {0}")]
    // SemaphoreAcquire variant
    SemaphoreAcquire(String),
}

// Implement From<tokio::sync::AcquireError> for RedisError
impl From<tokio::sync::AcquireError> for RedisError {
    fn from(err: tokio::sync::AcquireError) -> Self {
        RedisError::SemaphoreAcquire(err.to_string())
    }
}

/// `Redis` configuration optimized for `HFT` caching
#[derive(Debug, Clone, Deserialize, Serialize)]
/// RedisConfig
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct RedisConfig {
    /// `Redis` connection URL
    pub url: String,
    /// Maximum number of connections in the pool
    pub max_connections: u32,
    /// Minimum number of connections to maintain
    pub min_connections: u32,
    /// Connection timeout in milliseconds
    pub connect_timeout_ms: u64,
    /// Command timeout in microseconds (`HFT` optimized)
    pub command_timeout_micros: u64,
    /// Connection acquire timeout in milliseconds
    pub acquire_timeout_ms: u64,
    /// Maximum connection lifetime in seconds
    pub max_lifetime_seconds: u64,
    /// Idle timeout in seconds
    pub idle_timeout_seconds: u64,
    /// Enable connection prewarming
    pub enable_prewarming: bool,
    /// Enable pipelining for batch operations
    pub enable_pipelining: bool,
    /// Pipeline batch size
    pub pipeline_batch_size: usize,
    /// Default TTL for cached items in seconds
    pub default_ttl_seconds: u64,
    /// Enable compression for large values
    pub enable_compression: bool,
    /// Compression threshold in bytes
    pub compression_threshold_bytes: usize,
}

impl Default for RedisConfig {
    fn default() -> Self {
        Self {
            url: "redis://localhost:6379".to_owned(),
            max_connections: 20,
            min_connections: 5,
            connect_timeout_ms: 100,     // Fast connection establishment
            command_timeout_micros: 500, // <1ms for HFT operations
            acquire_timeout_ms: 50,      // Fast pool acquisition
            max_lifetime_seconds: 3600,  // 1 hour connection lifetime
            idle_timeout_seconds: 300,   // 5 minutes idle timeout
            enable_prewarming: true,
            enable_pipelining: true,
            pipeline_batch_size: 100,
            default_ttl_seconds: 300,  // 5 minutes default TTL
            enable_compression: false, // Disabled for HFT performance
            compression_threshold_bytes: 1024,
        }
    }
}

/// `Redis` connection pool with `HFT` optimizations and proper connection management
pub struct RedisPool {
    connection_manager: ConnectionManager,
    connection_semaphore: Arc<Semaphore>,
    config: RedisConfig,
    metrics: Arc<RwLock<RedisMetrics>>,
    /// Pre-warmed connections for ultra-low latency (currently unused)
    _warm_connections: Arc<RwLock<VecDeque<ConnectionManager>>>,
}

impl std::fmt::Debug for RedisPool {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisPool")
            .field("config", &self.config)
            .field(
                "connection_semaphore_available",
                &self.connection_semaphore.available_permits(),
            )
            .finish_non_exhaustive()
    }
}

impl RedisPool {
    /// Create a new `Redis` connection pool optimized for `HFT`
    pub async fn new(config: RedisConfig) -> Result<Self, RedisError> {
        // Create Redis client with connection options
        let client = Client::open(config.url.as_str()).map_err(RedisError::Connection)?;

        // Create connection manager for proper pooling
        let connection_manager = ConnectionManager::new(client)
            .await
            .map_err(RedisError::Connection)?;

        // Create semaphore for connection limiting
        let connection_semaphore = Arc::new(Semaphore::new(config.max_connections as usize));

        let metrics = Arc::new(RwLock::new(RedisMetrics::new()));
        let _warm_connections = Arc::new(RwLock::new(VecDeque::new()));

        let pool = Self {
            connection_manager,
            connection_semaphore,
            config: config.clone(),
            metrics,
            _warm_connections,
        };

        // Pre-warm connections if enabled (currently disabled)
        // if config.enable_prewarming {
        //     pool.prewarm_connections().await?;
        // }
        // Test connection
        pool.health_check().await?;

        // Ok variant
        Ok(pool)
    }
    /// Get a value from `Redis` with performance monitoring and optimized connection handling
    pub async fn get<T>(&self, key: &str) -> Result<Option<T>, RedisError>
    where
        T: serde::de::DeserializeOwned,
    {
        let start = Instant::now();

        // Acquire connection from pool with timeout
        let _permit = tokio::time::timeout(
            Duration::from_millis(self.config.acquire_timeout_ms),
            self.connection_semaphore.acquire(),
        )
        .await
        .map_err(|_| RedisError::PoolExhausted)??; // RAII: permit auto-returned when dropped

        let mut conn = self.get_connection().await?;

        let result: Result<Option<String>, _> = tokio::time::timeout(
            Duration::from_micros(self.config.command_timeout_micros),
            conn.get(key),
        )
        .await
        .map_err(|_| RedisError::Timeout {
            actual_ms: start.elapsed().as_millis() as u64,
            max_ms: self.config.command_timeout_micros / 1000,
        })?;

        let elapsed = start.elapsed();

        match result {
            Ok(Some(value)) => {
                self.update_metrics("get", elapsed, true, false).await;
                let deserialized: T = serde_json::from_str(&value)
                    .map_err(|e| RedisError::Serialization(e.to_string()))?;
                Ok(Some(deserialized))
            },
            Ok(None) => {
                self.update_metrics("get", elapsed, true, false).await;
                // Ok variant
                Ok(None)
            },
            Err(e) => {
                self.update_metrics("get", elapsed, false, false).await;
                Err(RedisError::Connection(e))
            },
        }
    }

    /// Set a value in `Redis` with TTL and performance monitoring using optimized connection handling
    pub async fn set<T>(
        &self,
        key: &str,
        value: &T,
        ttl: Option<Duration>,
    ) -> Result<(), RedisError>
    where
        T: Serialize,
    {
        let start = Instant::now();

        // Acquire connection from pool with timeout
        let _permit = tokio::time::timeout(
            Duration::from_millis(self.config.acquire_timeout_ms),
            self.connection_semaphore.acquire(),
        )
        .await
        .map_err(|_| RedisError::PoolExhausted)??; // RAII: permit auto-returned when dropped

        let mut conn = self.get_connection().await?;

        let serialized =
            serde_json::to_string(value).map_err(|e| RedisError::Serialization(e.to_string()))?;

        let result = if let Some(ttl) = ttl {
            tokio::time::timeout(
                Duration::from_micros(self.config.command_timeout_micros),
                conn.set_ex::<_, _, ()>(key, serialized, ttl.as_secs()),
            )
            .await
        } else {
            tokio::time::timeout(
                Duration::from_micros(self.config.command_timeout_micros),
                conn.set::<_, _, ()>(key, serialized),
            )
            .await
        };

        let elapsed = start.elapsed();

        match result {
            Ok(Ok(_)) => {
                self.update_metrics("set", elapsed, true, false).await;
                Ok(())
            },
            Ok(Err(e)) => {
                self.update_metrics("set", elapsed, false, false).await;
                Err(RedisError::Connection(e))
            },
            Err(_) => {
                self.update_metrics("set", elapsed, false, false).await;
                Err(RedisError::Timeout {
                    actual_ms: elapsed.as_millis() as u64,
                    max_ms: self.config.command_timeout_micros / 1000,
                })
            },
        }
    }

    /// Delete a key from `Redis` with optimized connection handling
    pub async fn delete(&self, key: &str) -> Result<bool, RedisError> {
        let start = Instant::now();

        // Acquire connection from pool with timeout
        let _permit = tokio::time::timeout(
            Duration::from_millis(self.config.acquire_timeout_ms),
            self.connection_semaphore.acquire(),
        )
        .await
        .map_err(|_| RedisError::PoolExhausted)??; // RAII: permit auto-returned when dropped

        let mut conn = self.get_connection().await?;

        let result: Result<i32, _> = tokio::time::timeout(
            Duration::from_micros(self.config.command_timeout_micros),
            conn.del(key),
        )
        .await
        .map_err(|_| RedisError::Timeout {
            actual_ms: start.elapsed().as_millis() as u64,
            max_ms: self.config.command_timeout_micros / 1000,
        })?;

        let elapsed = start.elapsed();

        match result {
            Ok(deleted_count) => {
                self.update_metrics("del", elapsed, true, false).await;
                // Ok variant
                Ok(deleted_count > 0)
            },
            Err(e) => {
                self.update_metrics("del", elapsed, false, false).await;
                Err(RedisError::Connection(e))
            },
        }
    }

    /// Check if a key exists in `Redis` with optimized connection handling
    pub async fn exists(&self, key: &str) -> Result<bool, RedisError> {
        let start = Instant::now();

        // Acquire connection from pool with timeout
        let _permit = tokio::time::timeout(
            Duration::from_millis(self.config.acquire_timeout_ms),
            self.connection_semaphore.acquire(),
        )
        .await
        .map_err(|_| RedisError::PoolExhausted)??; // RAII: permit auto-returned when dropped

        let mut conn = self.get_connection().await?;

        let result: Result<bool, _> = tokio::time::timeout(
            Duration::from_micros(self.config.command_timeout_micros),
            conn.exists(key),
        )
        .await
        .map_err(|_| RedisError::Timeout {
            actual_ms: start.elapsed().as_millis() as u64,
            max_ms: self.config.command_timeout_micros / 1000,
        })?;

        let elapsed = start.elapsed();

        match result {
            Ok(exists) => {
                self.update_metrics("exists", elapsed, true, false).await;
                // Ok variant
                Ok(exists)
            },
            Err(e) => {
                self.update_metrics("exists", elapsed, false, false).await;
                Err(RedisError::Connection(e))
            },
        }
    }

    /// Execute multiple operations in a pipeline for better performance with optimized connection handling
    pub async fn pipeline_execute<F, R>(&self, operations: F) -> Result<R, RedisError>
    where
        F: FnOnce(&mut Pipeline) -> R,
    {
        let start = Instant::now();

        // Acquire connection from pool with timeout
        let _permit = tokio::time::timeout(
            Duration::from_millis(self.config.acquire_timeout_ms),
            self.connection_semaphore.acquire(),
        )
        .await
        .map_err(|_| RedisError::PoolExhausted)??; // RAII: permit auto-returned when dropped

        let mut conn = self.get_connection().await?;

        let mut pipe = redis::pipe();
        let result_data = operations(&mut pipe);

        let result = tokio::time::timeout(
            Duration::from_micros(self.config.command_timeout_micros * 10), // More time for pipelines
            pipe.query_async::<()>(&mut conn),
        )
        .await;

        let elapsed = start.elapsed();

        match result {
            Ok(Ok(_)) => {
                self.update_metrics("pipeline", elapsed, true, true).await;
                // Ok variant
                Ok(result_data)
            },
            Ok(Err(e)) => {
                self.update_metrics("pipeline", elapsed, false, true).await;
                Err(RedisError::Connection(e))
            },
            Err(_) => {
                self.update_metrics("pipeline", elapsed, false, true).await;
                Err(RedisError::Timeout {
                    actual_ms: elapsed.as_millis() as u64,
                    max_ms: (self.config.command_timeout_micros * 10) / 1000,
                })
            },
        }
    }

    /// Set a value with default TTL
    pub async fn set_with_default_ttl<T>(&self, key: &str, value: &T) -> Result<(), RedisError>
    where
        T: Serialize,
    {
        self.set(
            key,
            value,
            Some(Duration::from_secs(self.config.default_ttl_seconds)),
        )
        .await
    }

    /// Health check for `Redis` connection using connection manager
    pub async fn health_check(&self) -> Result<(), RedisError> {
        let start = Instant::now();
        let mut conn = self.connection_manager.clone();

        let result: RedisResult<String> = tokio::time::timeout(
            Duration::from_millis(1000), // 1 second health check timeout
            redis::cmd("PING").query_async(&mut conn),
        )
        .await
        .map_err(|_| RedisError::Timeout {
            actual_ms: start.elapsed().as_millis() as u64,
            max_ms: 1000,
        })?;

        match result {
            Ok(_) => Ok(()),
            Err(e) => Err(RedisError::Connection(e)),
        }
    }

    /// Get current performance metrics
    pub async fn get_metrics(&self) -> Result<RedisMetrics, RedisError> {
        Ok(self.metrics.read().await.clone())
    }

    /// Get an optimized connection, preferring pre-warmed connections for `HFT` performance
    async fn get_connection(&self) -> Result<ConnectionManager, RedisError> {
        // For now, just return a clone of the connection manager
        // Pre-warmed connections could be added later if needed
        Ok(self.connection_manager.clone())
    }

    /// Pre-warm connections for `HFT` performance (currently disabled)
    async fn _prewarm_connections(&self) -> Result<(), RedisError> {
        // Currently disabled since we're using ConnectionManager directly
        // This could be re-enabled if needed for performance optimization
        Ok(())
    }

    /// Batch operations for improved `HFT` performance
    pub async fn batch_get<T>(&self, keys: &[&str]) -> Result<Vec<Option<T>>, RedisError>
    where
        T: serde::de::DeserializeOwned,
    {
        if keys.is_empty() {
            return Ok(Vec::new());
        }

        let start = Instant::now();
        let mut conn = self.get_connection().await?;

        let result: Result<Vec<Option<String>>, _> = tokio::time::timeout(
            Duration::from_micros(self.config.command_timeout_micros * keys.len() as u64),
            conn.get(keys),
        )
        .await
        .map_err(|_| RedisError::Timeout {
            actual_ms: start.elapsed().as_millis() as u64,
            max_ms: self.config.command_timeout_micros * keys.len() as u64 / 1000,
        })?;

        let elapsed = start.elapsed();

        match result {
            Ok(values) => {
                self.update_metrics("batch_get", elapsed, true, true).await;
                let mut deserialized = Vec::with_capacity(values.len());
                for value in values {
                    if let Some(val) = value {
                        let deser: T = serde_json::from_str(&val)
                            .map_err(|e| RedisError::Serialization(e.to_string()))?;
                        deserialized.push(Some(deser));
                    } else {
                        deserialized.push(None);
                    }
                }
                // Ok variant
                Ok(deserialized)
            },
            Err(e) => {
                self.update_metrics("batch_get", elapsed, false, true).await;
                Err(RedisError::Connection(e))
            },
        }
    }

    /// Update internal metrics
    async fn update_metrics(
        &self,
        operation: &str,
        duration: Duration,
        success: bool,
        _is_pipeline: bool,
    ) {
        let mut metrics = self.metrics.write().await;

        match operation {
            "get" => {
                metrics.total_gets += 1;
                if success {
                    metrics.successful_gets += 1;
                } else {
                    metrics.failed_gets += 1;
                }
            },
            "set" => {
                metrics.total_sets += 1;
                if success {
                    metrics.successful_sets += 1;
                } else {
                    metrics.failed_sets += 1;
                }
            },
            "del" => {
                metrics.total_deletes += 1;
                if success {
                    metrics.successful_deletes += 1;
                } else {
                    metrics.failed_deletes += 1;
                }
            },
            "exists" => {
                metrics.total_exists += 1;
                if success {
                    metrics.successful_exists += 1;
                } else {
                    metrics.failed_exists += 1;
                }
            },
            "pipeline" => {
                metrics.total_pipelines += 1;
                if success {
                    metrics.successful_pipelines += 1;
                } else {
                    metrics.failed_pipelines += 1;
                }
            },
            _ => {},
        }

        metrics.total_operations += 1;
        metrics.total_duration_micros += duration.as_micros() as u64;

        if success {
            metrics.successful_operations += 1;
        } else {
            metrics.failed_operations += 1;
        }

        // Track latency distribution
        if duration.as_micros() < 500 {
            metrics.sub_500_micros += 1;
        } else if duration.as_micros() < 1000 {
            metrics.sub_1ms += 1;
        } else {
            metrics.over_1ms += 1;
        }
    }
}

/// `Redis` performance metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
/// RedisMetrics
///
/// Auto-generated documentation placeholder - enhance with specifics
pub struct RedisMetrics {
    /// Total Operations
    pub total_operations: u64,
    /// Successful Operations
    pub successful_operations: u64,
    /// Failed Operations
    pub failed_operations: u64,
    /// Total `Duration` Micros
    pub total_duration_micros: u64,
    /// Total Gets
    pub total_gets: u64,
    /// Successful Gets
    pub successful_gets: u64,
    /// Failed Gets
    pub failed_gets: u64,
    /// Total Sets
    pub total_sets: u64,
    /// Successful Sets
    pub successful_sets: u64,
    /// Failed Sets
    pub failed_sets: u64,
    /// Total Deletes
    pub total_deletes: u64,
    /// Successful Deletes
    pub successful_deletes: u64,
    /// Failed Deletes
    pub failed_deletes: u64,
    /// Total Exists
    pub total_exists: u64,
    /// Successful Exists
    pub successful_exists: u64,
    /// Failed Exists
    pub failed_exists: u64,
    /// Total Pipelines
    pub total_pipelines: u64,
    /// Successful Pipelines
    pub successful_pipelines: u64,
    /// Failed Pipelines
    pub failed_pipelines: u64,
    /// Sub 500 Micros
    pub sub_500_micros: u64,
    /// Sub 1Ms
    pub sub_1ms: u64,
    /// Over 1Ms
    pub over_1ms: u64,
}

impl RedisMetrics {
    const fn new() -> Self {
        Self {
            total_operations: 0,
            successful_operations: 0,
            failed_operations: 0,
            total_duration_micros: 0,
            total_gets: 0,
            successful_gets: 0,
            failed_gets: 0,
            total_sets: 0,
            successful_sets: 0,
            failed_sets: 0,
            total_deletes: 0,
            successful_deletes: 0,
            failed_deletes: 0,
            total_exists: 0,
            successful_exists: 0,
            failed_exists: 0,
            total_pipelines: 0,
            successful_pipelines: 0,
            failed_pipelines: 0,
            sub_500_micros: 0,
            sub_1ms: 0,
            over_1ms: 0,
        }
    }

    /// Calculate average operation latency in microseconds
    pub fn average_latency_micros(&self) -> f64 {
        if self.total_operations == 0 {
            0.0
        } else {
            self.total_duration_micros as f64 / self.total_operations as f64
        }
    }

    /// Calculate success rate as percentage
    pub fn success_rate(&self) -> f64 {
        if self.total_operations == 0 {
            0.0
        } else {
            (self.successful_operations as f64 / self.total_operations as f64) * 100.0
        }
    }

    /// Calculate percentage of operations under 1ms
    pub fn sub_1ms_percentage(&self) -> f64 {
        if self.total_operations == 0 {
            0.0
        } else {
            ((self.sub_500_micros + self.sub_1ms) as f64 / self.total_operations as f64) * 100.0
        }
    }

    /// Calculate cache hit rate (gets that succeed)
    pub fn cache_hit_rate(&self) -> f64 {
        if self.total_gets == 0 {
            0.0
        } else {
            (self.successful_gets as f64 / self.total_gets as f64) * 100.0
        }
    }
}
