use alloy::{
    hex,
    primitives::{Address, Bytes},
};
use newton_prover_core::{common::chain::get_block_duration_ms, newton_prover_task_manager::NewtonMessage};
use redis::{aio::ConnectionManager, AsyncCommands, Client};
use serde::{Deserialize, Serialize};
use std::time::{SystemTime, UNIX_EPOCH};
use thiserror::Error;
use tracing::{info, warn};

#[derive(Debug, Error)]
pub enum RedisCacheError {
    #[error("Redis connection error: {0}")]
    ConnectionError(String),
    #[error("Redis operation error: {0}")]
    OperationError(String),
    #[error("Serialization error: {0}")]
    SerializationError(String),
    #[error("Deserialization error: {0}")]
    DeserializationError(String),
    #[error("Cache miss")]
    CacheMiss,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct CachedPolicyTaskData {
    data: NewtonMessage::PolicyTaskData,
    expires_at: u128, // Unix timestamp in milliseconds
}

/// Redis cache for policy task data
pub struct PolicyTaskDataCache {
    client: Client,
    connection_manager: Option<ConnectionManager>,
}

impl std::fmt::Debug for PolicyTaskDataCache {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PolicyTaskDataCache")
            .field("client", &self.client)
            .field("connection_manager", &self.connection_manager.is_some())
            .finish()
    }
}

impl PolicyTaskDataCache {
    /// Create a new Redis cache instance
    pub fn new(redis_url: &str) -> Result<Self, RedisCacheError> {
        info!(
            "Creating Redis client with URL: {}",
            redis_url.split('@').next_back().unwrap_or(redis_url)
        );

        let client = Client::open(redis_url).map_err(|e| {
            warn!("Failed to parse Redis URL: {}", e);
            RedisCacheError::ConnectionError(format!("Failed to parse Redis URL: {}", e))
        })?;

        let connection_info = client.get_connection_info();
        info!(
            "Connecting to Redis at {} (db: {})",
            connection_info.addr, connection_info.redis.db
        );

        Ok(Self {
            client,
            connection_manager: None,
        })
    }

    /// Initialize the connection manager
    pub async fn initialize(&mut self) -> Result<(), RedisCacheError> {
        info!("Initializing policy task data cache");

        info!("Attempting to connect to Redis (this may take up to 30 seconds for TLS connections)...");
        let connection_result =
            tokio::time::timeout(std::time::Duration::from_secs(30), self.client.get_connection_manager()).await;

        let connection_manager = match connection_result {
            Ok(Ok(manager)) => {
                info!("Successfully established Redis connection");
                manager
            }
            Ok(Err(e)) => {
                warn!(error = %e, error_debug = ?e, "Failed to connect to Redis server");
                return Err(RedisCacheError::ConnectionError(format!(
                    "Failed to connect to Redis: {}. If using TLS (rediss://), ensure the redis crate has TLS features enabled and certificates are valid.",
                    e
                )));
            }
            Err(_) => {
                warn!("Redis connection timed out after 30 seconds - check network connectivity and TLS configuration");
                return Err(RedisCacheError::ConnectionError(
                    "Connection timeout after 30 seconds. Check: 1) Redis server is accessible, 2) TLS configuration is correct for rediss:// URLs, 3) Network/firewall allows connection".to_string()
                ));
            }
        };

        self.connection_manager = Some(connection_manager);
        info!("Policy task data cache initialized successfully");
        Ok(())
    }

    /// Generate cache key for policy data address
    fn cache_key(
        &self,
        policy_client_address: Address,
        policy_data_address: Address,
        chain_id: u64,
        wasm_args: Option<Bytes>,
    ) -> String {
        format!(
            "policy_task_data:{}:{}:{}:{}",
            chain_id,
            policy_client_address,
            policy_data_address,
            wasm_args.map(hex::encode).unwrap_or_default()
        )
    }

    /// Convert block-based expiration to system time
    ///
    /// # Arguments
    /// * `expire_after_blocks` - Number of blocks after which data expires
    /// * `chain_id` - Chain ID to determine block time
    ///
    /// # Returns
    /// Unix timestamp when the data expires
    fn blocks_to_expiration_time(&self, expire_after_blocks: u32, chain_id: u64) -> u128 {
        let block_time_ms = get_block_duration_ms(chain_id).unwrap_or(12000);
        let expiration_ms = expire_after_blocks as u128 * block_time_ms as u128;

        SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() + expiration_ms
    }

    /// Store policy task data in cache
    pub async fn set(
        &self,
        policy_client_address: Address,
        policy_data_address: Address,
        wasm_args: Option<Bytes>,
        data: NewtonMessage::PolicyTaskData,
        expire_after_blocks: u32,
        chain_id: u64,
    ) -> Result<(), RedisCacheError> {
        let connection_manager = self
            .connection_manager
            .as_ref()
            .ok_or_else(|| RedisCacheError::ConnectionError("Cache not initialized".to_string()))?;

        let expires_at = self.blocks_to_expiration_time(expire_after_blocks, chain_id);
        let cached_data = CachedPolicyTaskData { data, expires_at };

        let serialized =
            serde_json::to_string(&cached_data).map_err(|e| RedisCacheError::SerializationError(e.to_string()))?;

        let key = self.cache_key(policy_client_address, policy_data_address, chain_id, wasm_args);
        let current_time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis();

        // Subtract a buffer to account for task processing time (network latency, operator evaluation, aggregation, etc.)
        // This ensures cached data remains valid when actually used during task processing
        const PROCESSING_TIME_BUFFER_MS: u128 = 10_000; // 10 seconds buffer
        let ttl_milliseconds = expires_at
            .saturating_sub(current_time)
            .saturating_sub(PROCESSING_TIME_BUFFER_MS);

        // Redis PSETEX requires TTL > 0. Skip caching if data is already expired or about to expire
        if ttl_milliseconds == 0 {
            warn!(
                "Skipping cache set for {} - data already expired or expires immediately",
                policy_data_address
            );
            return Ok(());
        }

        // Ensure TTL fits in u64 (Redis limitation) and is at least 1ms
        let ttl_ms = ttl_milliseconds.min(u64::MAX as u128) as u64;
        if ttl_ms == 0 {
            warn!(
                "Skipping cache set for {} - TTL too small or overflow",
                policy_data_address
            );
            return Ok(());
        }

        let mut conn = connection_manager.clone();
        conn.pset_ex::<_, _, ()>(&key, &serialized, ttl_ms)
            .await
            .map_err(|e| RedisCacheError::OperationError(e.to_string()))?;

        info!(
            "Cached policy task data for address {} with TTL {} milliseconds ({} milliseconds buffer applied for processing time)",
            policy_data_address, ttl_ms, PROCESSING_TIME_BUFFER_MS
        );

        Ok(())
    }

    /// Get policy task data from cache
    pub async fn get(
        &self,
        policy_client_address: Address,
        policy_data_address: Address,
        chain_id: u64,
        wasm_args: Option<Bytes>,
    ) -> Result<NewtonMessage::PolicyTaskData, RedisCacheError> {
        let connection_manager = self
            .connection_manager
            .as_ref()
            .ok_or_else(|| RedisCacheError::ConnectionError("Cache not initialized".to_string()))?;

        let key = self.cache_key(policy_client_address, policy_data_address, chain_id, wasm_args.clone());
        let mut conn = connection_manager.clone();

        let result: Option<String> = conn
            .get(&key)
            .await
            .map_err(|e| RedisCacheError::OperationError(e.to_string()))?;

        match result {
            Some(serialized) => {
                let cached_data: CachedPolicyTaskData = serde_json::from_str(&serialized)
                    .map_err(|e| RedisCacheError::DeserializationError(e.to_string()))?;

                // Check if data has expired
                let current_time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis();

                if current_time >= cached_data.expires_at {
                    info!(
                        "Cached data for address {} has expired, removing from cache",
                        policy_data_address
                    );
                    self.delete(policy_client_address, policy_data_address, chain_id, wasm_args)
                        .await?;
                    return Err(RedisCacheError::CacheMiss);
                }

                info!("Cache hit for policy data address {}", policy_data_address);
                Ok(cached_data.data)
            }
            None => {
                info!("Cache miss for policy data address {}", policy_data_address);
                Err(RedisCacheError::CacheMiss)
            }
        }
    }

    /// Delete policy task data from cache
    pub async fn delete(
        &self,
        policy_client_address: Address,
        policy_data_address: Address,
        chain_id: u64,
        wasm_args: Option<Bytes>,
    ) -> Result<(), RedisCacheError> {
        let connection_manager = self
            .connection_manager
            .as_ref()
            .ok_or_else(|| RedisCacheError::ConnectionError("Cache not initialized".to_string()))?;

        let key = self.cache_key(policy_client_address, policy_data_address, chain_id, wasm_args);
        let mut conn = connection_manager.clone();

        conn.del::<_, ()>(&key)
            .await
            .map_err(|e| RedisCacheError::OperationError(e.to_string()))?;

        info!("Deleted cached data for policy data address {}", policy_data_address);
        Ok(())
    }

    /// Check if cache is available
    pub async fn is_available(&self) -> bool {
        self.connection_manager.is_some() && {
            if let Some(connection_manager) = &self.connection_manager {
                let mut conn = connection_manager.clone();
                // Use a simple GET command to test connectivity
                conn.get::<&str, Option<String>>("__test__").await.is_ok()
            } else {
                false
            }
        }
    }
}
