//! Shared data types — backend-neutral.
//!
//! All struct/enum *definitions* live here. Backend-specific trait
//! implementations (`FromRedisValue`, `FromRedis`, `ToRedisArgs`, `Deref`,
//! `From<RedisValue> for …`) live in each backend's `types.rs` instead.
//!
//! The one cfg-gated field that cannot be split is `RedisValue::inner`,
//! because Rust does not allow splitting a struct definition across files.

use common_utils::errors::CustomResult;

use crate::errors;

// ─── RedisValue — wrapper whose inner type depends on the active backend ─────

#[derive(Clone, Debug)]
pub struct RedisValue {
    #[cfg(feature = "redis-rs")]
    pub(crate) inner: redis::Value,
    #[cfg(feature = "fred")]
    pub(crate) inner: fred::types::RedisValue,
}

// Method impls are in module/redis_rs/types.rs and module/fred/types.rs.

// ─── Shared configuration types ─────────────────────────────────────────────

#[derive(Debug, serde::Deserialize, Clone)]
#[serde(default)]
pub struct RedisSettings {
    pub host: String,
    pub port: u16,
    pub cluster_enabled: bool,
    pub cluster_urls: Vec<String>,
    pub use_legacy_version: bool,
    /// Number of reconnection attempts before giving up (default: 5).
    pub pool_size: usize,
    /// Maximum number of connection retry attempts (default: 5).
    pub reconnect_max_attempts: u32,
    /// Initial delay in milliseconds between reconnection attempts (default: 5).
    pub reconnect_delay: u32,
    /// TTL in seconds
    pub default_ttl: u32,
    /// TTL for hash-tables in seconds
    pub default_hash_ttl: u32,
    pub stream_read_count: u64,
    pub auto_pipeline: bool,
    pub disable_auto_backpressure: bool,
    /// Maximum number of in-flight commands before backpressure is applied.
    /// Set to 0 to disable.
    pub max_in_flight_commands: usize,
    /// Command timeout in seconds.
    pub default_command_timeout: u64,
    pub max_feed_count: u64,
    pub unresponsive_timeout: u64,
    pub unresponsive_check_interval: u64,
    /// Capacity of the broadcast channel used for pub/sub message distribution.
    pub broadcast_channel_capacity: usize,
    /// Maximum duration (in seconds) that Redis can be unreachable before the server shuts down.
    pub max_failure_threshold_seconds: u32,
}

impl RedisSettings {
    /// Validates the Redis configuration provided.
    pub fn validate(&self) -> CustomResult<(), errors::RedisError> {
        use common_utils::{ext_traits::ConfigExt, fp_utils::when};

        when(self.host.is_default_or_empty(), || {
            Err(errors::RedisError::InvalidConfiguration(
                "Redis `host` must be specified".into(),
            ))
        })?;

        when(self.cluster_enabled && self.cluster_urls.is_empty(), || {
            Err(errors::RedisError::InvalidConfiguration(
                "Redis `cluster_urls` must be specified if `cluster_enabled` is `true`".into(),
            ))
        })?;

        when(
            self.default_command_timeout < self.unresponsive_timeout,
            || {
                Err(errors::RedisError::InvalidConfiguration(
                    "Unresponsive timeout cannot be greater than the command timeout".into(),
                ))
            },
        )?;

        when(
            self.unresponsive_check_interval > self.max_failure_threshold_seconds.into(),
            || {
                Err(errors::RedisError::InvalidConfiguration(
                    "Unresponsive check interval cannot be greater than the max failure threshold"
                        .into(),
                ))
            },
        )?;

        Ok(())
    }
}

impl Default for RedisSettings {
    fn default() -> Self {
        Self {
            host: "127.0.0.1".to_string(),
            port: 6379,
            cluster_enabled: false,
            cluster_urls: vec![],
            use_legacy_version: false,
            pool_size: 5,
            reconnect_max_attempts: 5,
            reconnect_delay: 5,
            default_ttl: 300,
            stream_read_count: 1,
            default_hash_ttl: 900,
            auto_pipeline: true,
            disable_auto_backpressure: false,
            max_in_flight_commands: 5000,
            default_command_timeout: 30,
            max_feed_count: 200,
            unresponsive_timeout: 10,
            unresponsive_check_interval: 2,
            broadcast_channel_capacity: 32,
            max_failure_threshold_seconds: 5,
        }
    }
}

// ─── RedisEntryId ────────────────────────────────────────────────────────────

#[derive(Debug)]
pub enum RedisEntryId {
    UserSpecifiedID {
        milliseconds: String,
        sequence_number: String,
    },
    AutoGeneratedID,
    AfterLastID,
    /// Applicable only with consumer groups
    UndeliveredEntryID,
}

impl RedisEntryId {
    /// Convert to the string representation used by Redis stream commands
    pub fn to_stream_id(&self) -> String {
        match self {
            Self::UserSpecifiedID {
                milliseconds,
                sequence_number,
            } => format!("{milliseconds}-{sequence_number}"),
            Self::AutoGeneratedID => "*".to_string(),
            Self::AfterLastID => "$".to_string(),
            Self::UndeliveredEntryID => ">".to_string(),
        }
    }
}

// Trait impls live in module/redis_rs/types.rs and module/fred/types.rs.

// ─── Reply type enums ────────────────────────────────────────────────────────

#[derive(Debug, Eq, PartialEq)]
#[cfg_attr(feature = "deja", derive(serde::Serialize, serde::Deserialize))]
pub enum SetnxReply {
    KeySet,
    KeyNotSet, // Existing key
}

#[derive(Debug, Eq, PartialEq)]
#[cfg_attr(feature = "deja", derive(serde::Serialize, serde::Deserialize))]
pub enum HsetnxReply {
    KeySet,
    KeyNotSet, // Existing key
}

#[derive(Debug, Eq, PartialEq)]
#[cfg_attr(feature = "deja", derive(serde::Serialize, serde::Deserialize))]
pub enum MsetnxReply {
    KeysSet,
    KeysNotSet, // At least one existing key
}

#[derive(Debug, Eq, PartialEq)]
#[cfg_attr(feature = "deja", derive(serde::Serialize, serde::Deserialize))]
pub enum DelReply {
    KeyDeleted,
    KeyNotDeleted, // Key not found
}

impl DelReply {
    pub fn is_key_deleted(&self) -> bool {
        matches!(self, Self::KeyDeleted)
    }

    pub fn is_key_not_deleted(&self) -> bool {
        matches!(self, Self::KeyNotDeleted)
    }
}

/// Reply from SADD command
#[derive(Debug, Eq, PartialEq)]
#[cfg_attr(feature = "deja", derive(serde::Serialize, serde::Deserialize))]
pub enum SaddReply {
    /// Returned when atleast 1 value was inserted to Set
    /// i64 value represent the total number of values that were inserted.
    KeySet(i64),
    /// Returned when no value was inserted to the Set
    KeyNotSet,
}

/// Reply from `XGROUP DESTROY`.
#[derive(Debug, Eq, PartialEq)]
pub enum ConsumerGroupDestroyReply {
    Destroyed,
    NotFound,
}

/// Redis key types used with the `TYPE` filter in `SCAN`.
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum RedisScanType {
    String,
    List,
    Set,
    Zset,
    Hash,
    Stream,
}

impl AsRef<str> for RedisScanType {
    fn as_ref(&self) -> &str {
        match self {
            Self::String => "STRING",
            Self::List => "LIST",
            Self::Set => "SET",
            Self::Zset => "ZSET",
            Self::Hash => "HASH",
            Self::Stream => "STREAM",
        }
    }
}

#[derive(Debug)]
pub enum SetGetReply<T> {
    ValueSet(T),    // Value was set and this is the value that was set
    ValueExists(T), // Value already existed and this is the existing value
}

impl<T> SetGetReply<T> {
    pub fn get_value(&self) -> &T {
        match self {
            Self::ValueSet(value) => value,
            Self::ValueExists(value) => value,
        }
    }
}

// ─── Stream types ────────────────────────────────────────────────────────────

/// Entries within a single stream, as `(entry_id, fields)`.
pub type StreamEntries = Vec<(String, std::collections::HashMap<String, RedisValue>)>;

/// Grouped result of a stream read: stream key → list of `(entry_id, fields)`.
pub type StreamReadResult = std::collections::HashMap<String, StreamEntries>;

#[derive(Debug, Clone)]
pub enum StreamCapKind {
    MinID,
    MaxLen,
}

#[derive(Debug, Clone)]
pub enum StreamCapTrim {
    Exact,
    AlmostExact,
}

// Trait impls for StreamCapKind/StreamCapTrim live in module/fred/types.rs.

// ─── Stream trim types ────────────────────────────────────────────────────────

/// Configuration for trimming a Redis stream via `XTRIM`.
///
/// Encapsulates the kind (MaxLen or MinID), precision (Exact or AlmostExact),
/// and the numeric or ID threshold.
///
/// # Examples
///
/// ```
/// use redis_interface::types::{StreamCapKind, StreamCapTrim, StreamTrimConfig};
///
/// let config = StreamTrimConfig::new(StreamCapKind::MaxLen, StreamCapTrim::AlmostExact, "1000");
/// ```
#[derive(Debug, Clone)]
pub struct StreamTrimConfig {
    /// What to compare against — entry count or minimum ID.
    pub kind: StreamCapKind,
    /// How precisely to match the threshold.
    pub trim: StreamCapTrim,
    /// The threshold value: a numeric string for `MaxLen`, a stream ID for `MinID`.
    pub threshold: String,
}

impl StreamTrimConfig {
    /// Create a new trim configuration.
    pub fn new(kind: StreamCapKind, trim: StreamCapTrim, threshold: impl Into<String>) -> Self {
        Self {
            kind,
            trim,
            threshold: threshold.into(),
        }
    }

    /// Convert to the `redis` crate's `StreamTrimOptions` (redis-rs backend).
    ///
    /// Returns an error if `kind` is `MaxLen` and the threshold
    /// cannot be parsed as a `usize`.
    #[cfg(feature = "redis-rs")]
    pub fn to_trim_options(
        self,
    ) -> Result<redis::streams::StreamTrimOptions, StreamTrimThresholdError> {
        let trim_mode = match self.trim {
            StreamCapTrim::Exact => redis::streams::StreamTrimmingMode::Exact,
            StreamCapTrim::AlmostExact => redis::streams::StreamTrimmingMode::Approx,
        };

        match self.kind {
            StreamCapKind::MaxLen => {
                let max_len: usize = self
                    .threshold
                    .parse()
                    .map_err(|_| StreamTrimThresholdError(self.threshold.clone()))?;
                Ok(redis::streams::StreamTrimOptions::maxlen(
                    trim_mode, max_len,
                ))
            }
            StreamCapKind::MinID => Ok(redis::streams::StreamTrimOptions::minid(
                trim_mode,
                self.threshold,
            )),
        }
    }
}

/// Error returned when a `StreamTrimConfig` threshold cannot be parsed.
#[derive(Debug)]
pub struct StreamTrimThresholdError(String);

impl std::fmt::Display for StreamTrimThresholdError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(formatter, "Invalid stream trim threshold: {}", self.0)
    }
}

impl std::error::Error for StreamTrimThresholdError {}

// ─── RedisKey ────────────────────────────────────────────────────────────────

#[derive(Debug, Clone)]
pub struct RedisKey(String);

impl RedisKey {
    #[cfg(feature = "deja")]
    pub fn as_str(&self) -> &str {
        self.0.as_str()
    }

    pub fn tenant_aware_key(&self, pool: &crate::RedisConnectionPool) -> String {
        pool.add_prefix(&self.0)
    }

    pub fn tenant_unaware_key(&self, _pool: &crate::RedisConnectionPool) -> String {
        self.0.clone()
    }
}

impl<T: AsRef<str>> From<T> for RedisKey {
    fn from(value: T) -> Self {
        let value = value.as_ref();
        Self(value.to_string())
    }
}

// ─── Tests (backend-neutral only) ────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_entry_id_user_specified() {
        let id = RedisEntryId::UserSpecifiedID {
            milliseconds: "1234567890".to_string(),
            sequence_number: "0".to_string(),
        };
        assert_eq!(id.to_stream_id(), "1234567890-0");
    }

    #[test]
    fn test_entry_id_auto_generated() {
        assert_eq!(RedisEntryId::AutoGeneratedID.to_stream_id(), "*");
    }

    #[test]
    fn test_entry_id_after_last() {
        assert_eq!(RedisEntryId::AfterLastID.to_stream_id(), "$");
    }

    #[test]
    fn test_entry_id_undelivered() {
        assert_eq!(RedisEntryId::UndeliveredEntryID.to_stream_id(), ">");
    }

    #[test]
    fn test_redis_settings_validate_valid_defaults() {
        let settings = RedisSettings::default();
        assert!(settings.validate().is_ok());
    }

    #[test]
    fn test_redis_settings_validate_empty_host() {
        let settings = RedisSettings {
            host: String::new(),
            ..RedisSettings::default()
        };
        assert!(settings.validate().is_err());
    }

    #[test]
    fn test_redis_settings_validate_cluster_without_urls() {
        let settings = RedisSettings {
            cluster_enabled: true,
            cluster_urls: vec![],
            ..RedisSettings::default()
        };
        assert!(settings.validate().is_err());
    }

    #[test]
    fn test_redis_settings_validate_unresponsive_timeout_exceeds_command_timeout() {
        let settings = RedisSettings {
            unresponsive_timeout: 60,
            default_command_timeout: 30,
            ..RedisSettings::default()
        };
        assert!(settings.validate().is_err());
    }

    #[test]
    fn test_redis_settings_default_values() {
        let settings = RedisSettings::default();
        assert_eq!(settings.host, "127.0.0.1");
        assert_eq!(settings.port, 6379);
        assert!(!settings.cluster_enabled);
        assert!(settings.cluster_urls.is_empty());
        assert!(!settings.use_legacy_version);
        assert_eq!(settings.reconnect_max_attempts, 5);
        assert_eq!(settings.reconnect_delay, 5);
        assert_eq!(settings.default_ttl, 300);
        assert_eq!(settings.default_hash_ttl, 900);
        assert_eq!(settings.broadcast_channel_capacity, 32);
        assert_eq!(settings.max_failure_threshold_seconds, 5);
    }

    #[test]
    fn test_del_reply_is_key_deleted() {
        assert!(DelReply::KeyDeleted.is_key_deleted());
        assert!(!DelReply::KeyNotDeleted.is_key_deleted());
    }
}
