//! An interface to abstract the `redis` commands
//!
//! The folder provides generic functions for providing serialization
//! and deserialization while calling redis.
//! It also includes instruments to provide tracing.

use std::fmt::Debug;

use common_utils::{
    errors::CustomResult,
    ext_traits::{ByteSliceExt, Encode, StringExt},
    fp_utils,
};
use error_stack::{report, ResultExt};
use redis::{
    streams::StreamReadOptions, AsyncCommands, ExistenceCheck, FromRedisValue, SetExpiry,
    SetOptions, ToSingleRedisArg,
};
use tracing::instrument;

use super::types::redis_value_to_option_string;
use crate::{
    constant::redis_rs_commands::{
        REDIS_ARG_COUNT, REDIS_ARG_EX, REDIS_ARG_MATCH, REDIS_ARG_NX, REDIS_ARG_TYPE,
        REDIS_COMMAND_GET, REDIS_COMMAND_HSCAN, REDIS_COMMAND_SCAN, REDIS_COMMAND_SET,
    },
    errors,
    metrics::{track_redis_call, RedisOperation},
    types::{
        DelReply, HsetnxReply, MsetnxReply, RedisEntryId, RedisKey, SaddReply, SetGetReply,
        SetnxReply, StreamEntries, StreamReadResult, StreamTrimConfig,
    },
};

#[cfg(feature = "deja")]
#[derive(serde::Serialize, serde::Deserialize)]
enum DejaRedisValue {
    Null,
    Int(i64),
    BulkString(Vec<u8>),
    Array(Vec<Self>),
    SimpleString(String),
    Okay,
    Map(Vec<(Self, Self)>),
    Attribute {
        data: Box<Self>,
        attributes: Vec<(Self, Self)>,
    },
    Set(Vec<Self>),
    Double(f64),
    Boolean(bool),
    VerbatimString {
        format: String,
        text: String,
    },
    Push {
        kind: String,
        data: Vec<Self>,
    },
    UnsupportedSuccessfulValue {
        variant: String,
    },
}

#[cfg(feature = "deja")]
impl From<redis::Value> for DejaRedisValue {
    fn from(value: redis::Value) -> Self {
        match value {
            redis::Value::Nil => Self::Null,
            redis::Value::Int(value) => Self::Int(value),
            redis::Value::BulkString(value) => Self::BulkString(value),
            redis::Value::Array(values) => {
                Self::Array(values.into_iter().map(Self::from).collect())
            }
            redis::Value::SimpleString(value) => Self::SimpleString(value),
            redis::Value::Okay => Self::Okay,
            redis::Value::Map(values) => Self::Map(
                values
                    .into_iter()
                    .map(|(key, value)| (Self::from(key), Self::from(value)))
                    .collect(),
            ),
            redis::Value::Attribute { data, attributes } => Self::Attribute {
                data: Box::new(Self::from(*data)),
                attributes: attributes
                    .into_iter()
                    .map(|(key, value)| (Self::from(key), Self::from(value)))
                    .collect(),
            },
            redis::Value::Set(values) => Self::Set(values.into_iter().map(Self::from).collect()),
            redis::Value::Double(value) => Self::Double(value),
            redis::Value::Boolean(value) => Self::Boolean(value),
            redis::Value::VerbatimString { format, text } => Self::VerbatimString {
                format: format.to_string(),
                text,
            },
            redis::Value::Push { kind, data } => Self::Push {
                kind: kind.to_string(),
                data: data.into_iter().map(Self::from).collect(),
            },
            redis::Value::BigNumber(_) => Self::UnsupportedSuccessfulValue {
                variant: "BigNumber".to_string(),
            },
            redis::Value::ServerError(error) => Self::UnsupportedSuccessfulValue {
                variant: match error.details() {
                    Some(details) => format!("ServerError({}: {details})", error.code()),
                    None => format!("ServerError({})", error.code()),
                },
            },
            _ => Self::UnsupportedSuccessfulValue {
                variant: "unknown non-exhaustive redis::Value variant".to_string(),
            },
        }
    }
}

#[cfg(feature = "deja")]
impl DejaRedisValue {
    fn into_supported_redis_value(self) -> Result<redis::Value, redis::RedisError> {
        match self {
            Self::Null => Ok(redis::Value::Nil),
            Self::Int(value) => Ok(redis::Value::Int(value)),
            Self::BulkString(value) => Ok(redis::Value::BulkString(value)),
            Self::Array(values) => Ok(redis::Value::Array(
                values
                    .into_iter()
                    .map(Self::into_supported_redis_value)
                    .collect::<Result<Vec<_>, _>>()?,
            )),
            Self::SimpleString(value) => Ok(redis::Value::SimpleString(value)),
            Self::Okay => Ok(redis::Value::Okay),
            Self::Map(values) => Ok(redis::Value::Map(
                values
                    .into_iter()
                    .map(|(key, value)| {
                        Ok((
                            key.into_supported_redis_value()?,
                            value.into_supported_redis_value()?,
                        ))
                    })
                    .collect::<Result<Vec<_>, redis::RedisError>>()?,
            )),
            Self::Attribute { data, attributes } => Ok(redis::Value::Attribute {
                data: Box::new(data.into_supported_redis_value()?),
                attributes: attributes
                    .into_iter()
                    .map(|(key, value)| {
                        Ok((
                            key.into_supported_redis_value()?,
                            value.into_supported_redis_value()?,
                        ))
                    })
                    .collect::<Result<Vec<_>, redis::RedisError>>()?,
            }),
            Self::Set(values) => Ok(redis::Value::Set(
                values
                    .into_iter()
                    .map(Self::into_supported_redis_value)
                    .collect::<Result<Vec<_>, _>>()?,
            )),
            Self::Double(value) => Ok(redis::Value::Double(value)),
            Self::Boolean(value) => Ok(redis::Value::Boolean(value)),
            Self::VerbatimString { format, text } => Ok(redis::Value::VerbatimString {
                format: match format.as_str() {
                    "mkd" => redis::VerbatimFormat::Markdown,
                    "txt" => redis::VerbatimFormat::Text,
                    other => redis::VerbatimFormat::Unknown(other.to_string()),
                },
                text,
            }),
            Self::Push { kind, data } => Ok(redis::Value::Push {
                kind: match kind.as_str() {
                    "disconnection" => redis::PushKind::Disconnection,
                    "invalidate" => redis::PushKind::Invalidate,
                    "message" => redis::PushKind::Message,
                    "pmessage" => redis::PushKind::PMessage,
                    "smessage" => redis::PushKind::SMessage,
                    "unsubscribe" => redis::PushKind::Unsubscribe,
                    "punsubscribe" => redis::PushKind::PUnsubscribe,
                    "sunsubscribe" => redis::PushKind::SUnsubscribe,
                    "subscribe" => redis::PushKind::Subscribe,
                    "psubscribe" => redis::PushKind::PSubscribe,
                    "ssubscribe" => redis::PushKind::SSubscribe,
                    other => redis::PushKind::Other(other.to_string()),
                },
                data: data
                    .into_iter()
                    .map(Self::into_supported_redis_value)
                    .collect::<Result<Vec<_>, _>>()?,
            }),
            Self::UnsupportedSuccessfulValue { variant } => Err((
                redis::ErrorKind::UnexpectedReturnType,
                "unsupported Redis value variant recorded by Deja replay",
                variant,
            )
                .into()),
        }
    }
}

impl super::RedisConnectionWithContext {
    /// Prefix `key` with the tenant key prefix of the underlying pool.
    pub fn add_prefix(&self, key: &str) -> String {
        self.redis_conn.add_prefix(key)
    }

    // ─── Key Commands ────────────────────────────────────────────────────────

    #[cfg_attr(
        feature = "deja",
        deja::redis(
            operation = "set_key",
            codec = deja::codec::ResultCodec::<(), errors::RedisError>,
            args = {
                serde_json::json!({
                    "key": key.as_str(),
                    "command": "SET",
                    "has_ttl": true,
                })
            },
        )
    )]
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_key<V>(&self, key: &RedisKey, value: V) -> CustomResult<(), errors::RedisError>
    where
        V: redis::ToRedisArgs + Debug + Send + Sync + ToSingleRedisArg,
    {
        let mut conn = self.redis_conn.pool.clone();
        let options = SetOptions::default()
            .with_expiration(SetExpiry::EX(u64::from(self.redis_conn.config.default_ttl)));
        let _: Option<String> = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetKey,
            conn.set_options(key.tenant_aware_key(&self.redis_conn), value, options),
        )
        .await
        .change_context(errors::RedisError::SetFailed)?;
        Ok(())
    }

    #[cfg_attr(
        feature = "deja",
        deja::redis(
            operation = "set_key_without_modifying_ttl",
            codec = deja::codec::ResultCodec::<(), errors::RedisError>,
            args = {
                serde_json::json!({
                    "key": key.as_str(),
                    "command": "SET_KEEP_TTL",
                })
            },
        )
    )]
    pub async fn set_key_without_modifying_ttl<V>(
        &self,
        key: &RedisKey,
        value: V,
    ) -> CustomResult<(), errors::RedisError>
    where
        V: redis::ToRedisArgs + Debug + Send + Sync + ToSingleRedisArg,
    {
        let mut conn = self.redis_conn.pool.clone();
        let options = SetOptions::default().with_expiration(SetExpiry::KEEPTTL);
        let _: Option<String> = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetKeyWithoutModifyingTtl,
            conn.set_options(key.tenant_aware_key(&self.redis_conn), value, options),
        )
        .await
        .change_context(errors::RedisError::SetFailed)?;
        Ok(())
    }

    pub async fn set_multiple_keys_if_not_exist<K, V>(
        &self,
        key_value_pairs: &[(K, V)],
    ) -> CustomResult<MsetnxReply, errors::RedisError>
    where
        K: redis::ToRedisArgs + Debug + Send + Sync,
        V: redis::ToRedisArgs + Debug + Send + Sync,
    {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetMultipleKeysIfNotExist,
            conn.mset_nx(key_value_pairs),
        )
        .await
        .change_context(errors::RedisError::SetFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn serialize_and_set_key_if_not_exist<V>(
        &self,
        key: &RedisKey,
        value: V,
        ttl: Option<i64>,
    ) -> CustomResult<SetnxReply, errors::RedisError>
    where
        V: serde::Serialize + Debug,
    {
        let serialized = value
            .encode_to_vec()
            .change_context(errors::RedisError::JsonSerializationFailed)?;
        self.set_key_if_not_exists_with_expiry(key, serialized.as_slice(), ttl)
            .await
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn serialize_and_set_key<V>(
        &self,
        key: &RedisKey,
        value: V,
    ) -> CustomResult<(), errors::RedisError>
    where
        V: serde::Serialize + Debug,
    {
        let serialized = value
            .encode_to_vec()
            .change_context(errors::RedisError::JsonSerializationFailed)?;

        self.set_key(key, serialized.as_slice()).await
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn serialize_and_set_key_without_modifying_ttl<V>(
        &self,
        key: &RedisKey,
        value: V,
    ) -> CustomResult<(), errors::RedisError>
    where
        V: serde::Serialize + Debug,
    {
        let serialized = value
            .encode_to_vec()
            .change_context(errors::RedisError::JsonSerializationFailed)?;

        self.set_key_without_modifying_ttl(key, serialized.as_slice())
            .await
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn serialize_and_set_key_with_expiry<V>(
        &self,
        key: &RedisKey,
        value: V,
        seconds: i64,
    ) -> CustomResult<(), errors::RedisError>
    where
        V: serde::Serialize + Debug,
    {
        let serialized = value
            .encode_to_vec()
            .change_context(errors::RedisError::JsonSerializationFailed)?;

        #[cfg(feature = "deja")]
        {
            self.set_key_with_expiry(key, serialized.as_slice(), seconds)
                .await
        }

        #[cfg(not(feature = "deja"))]
        {
            let mut conn = self.redis_conn.pool.clone();
            let options = SetOptions::default().with_expiration(SetExpiry::EX(
                u64::try_from(seconds).change_context(errors::RedisError::SetExFailed)?,
            ));
            let _: Option<String> = track_redis_call(
                self.request_id.as_deref(),
                self.redis_conn.event_emitter.as_ref(),
                RedisOperation::SerializeAndSetKeyWithExpiry,
                conn.set_options(
                    key.tenant_aware_key(&self.redis_conn),
                    serialized.as_slice(),
                    options,
                ),
            )
            .await
            .change_context(errors::RedisError::SetExFailed)?;
            Ok(())
        }
    }

    #[cfg(feature = "deja")]
    #[instrument(level = "DEBUG", skip(self))]
    #[deja::redis(
        operation = "get_key",
        codec = deja::codec::ResultCodec::<DejaRedisValue, errors::RedisError>,
        state_read = key.tenant_aware_key(&self.redis_conn),
        args = {
            serde_json::json!({
                "key": key.as_str(),
                "command": "GET",
            })
        },
    )]
    async fn get_key_raw(
        &self,
        key: &RedisKey,
    ) -> CustomResult<DejaRedisValue, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        match track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::GetKey,
            conn.get::<_, redis::Value>(key.tenant_aware_key(&self.redis_conn)),
        )
        .await
        .change_context(errors::RedisError::GetFailed)
        {
            Ok(value) => Ok(DejaRedisValue::from(value)),
            Err(_err) => {
                #[cfg(not(feature = "multitenancy_fallback"))]
                {
                    Err(_err)
                }

                #[cfg(feature = "multitenancy_fallback")]
                {
                    track_redis_call(
                        self.request_id.as_deref(),
                        self.redis_conn.event_emitter.as_ref(),
                        RedisOperation::GetKey,
                        conn.get::<_, redis::Value>(key.tenant_unaware_key(&self.redis_conn)),
                    )
                    .await
                    .map(DejaRedisValue::from)
                    .change_context(errors::RedisError::GetFailed)
                }
            }
        }
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_key<V>(&self, key: &RedisKey) -> CustomResult<V, errors::RedisError>
    where
        V: FromRedisValue + Send + 'static,
    {
        #[cfg(feature = "deja")]
        {
            let raw_value = self
                .get_key_raw(key)
                .await?
                .into_supported_redis_value()
                .map_err(|err| report!(err).change_context(errors::RedisError::GetFailed))?;
            return V::from_redis_value(raw_value)
                .map_err(|err| report!(err).change_context(errors::RedisError::GetFailed));
        }

        #[cfg(not(feature = "deja"))]
        {
            let mut conn = self.redis_conn.pool.clone();
            match track_redis_call(
                self.request_id.as_deref(),
                self.redis_conn.event_emitter.as_ref(),
                RedisOperation::GetKey,
                conn.get::<_, V>(key.tenant_aware_key(&self.redis_conn)),
            )
            .await
            .change_context(errors::RedisError::GetFailed)
            {
                Ok(v) => Ok(v),
                Err(_err) => {
                    #[cfg(not(feature = "multitenancy_fallback"))]
                    {
                        Err(_err)
                    }

                    #[cfg(feature = "multitenancy_fallback")]
                    {
                        track_redis_call(
                            self.request_id.as_deref(),
                            self.redis_conn.event_emitter.as_ref(),
                            RedisOperation::GetKey,
                            conn.get::<_, V>(key.tenant_unaware_key(&self.redis_conn)),
                        )
                        .await
                        .change_context(errors::RedisError::GetFailed)
                    }
                }
            }
        }
    }

    #[instrument(level = "DEBUG", skip(self))]
    async fn get_multiple_keys_with_mget<V>(
        &self,
        keys: &[RedisKey],
    ) -> CustomResult<Vec<Option<V>>, errors::RedisError>
    where
        V: FromRedisValue + Send + 'static,
    {
        if keys.is_empty() {
            return Ok(Vec::new());
        }

        let tenant_aware_keys: Vec<String> = keys
            .iter()
            .map(|key| key.tenant_aware_key(&self.redis_conn))
            .collect();
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::GetMultipleKeys,
            conn.mget(&tenant_aware_keys),
        )
        .await
        .change_context(errors::RedisError::GetFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    async fn get_multiple_keys_with_parallel_get<V>(
        &self,
        keys: &[RedisKey],
    ) -> CustomResult<Vec<Option<V>>, errors::RedisError>
    where
        V: FromRedisValue + Send + 'static,
    {
        if keys.is_empty() {
            return Ok(Vec::new());
        }
        let tenant_aware_keys: Vec<String> = keys
            .iter()
            .map(|key| key.tenant_aware_key(&self.redis_conn))
            .collect();

        let futures = tenant_aware_keys.iter().map(|redis_key| {
            let mut conn = self.redis_conn.pool.clone();
            let key = redis_key.clone();
            async move {
                track_redis_call(
                    self.request_id.as_deref(),
                    self.redis_conn.event_emitter.as_ref(),
                    RedisOperation::GetKey,
                    conn.get::<_, Option<V>>(&key),
                )
                .await
            }
        });

        let results = futures::future::try_join_all(futures)
            .await
            .change_context(errors::RedisError::GetFailed)
            .attach_printable("Failed to get keys in cluster mode")?;

        Ok(results)
    }

    /// Helper method to encapsulate the logic for choosing between cluster and non-cluster modes
    #[instrument(level = "DEBUG", skip(self))]
    async fn get_keys_by_mode<V>(
        &self,
        keys: &[RedisKey],
    ) -> CustomResult<Vec<Option<V>>, errors::RedisError>
    where
        V: FromRedisValue + Send + 'static,
    {
        if self.redis_conn.config.cluster_enabled {
            // Use individual GET commands for cluster mode to avoid CROSSSLOT errors
            self.get_multiple_keys_with_parallel_get(keys).await
        } else {
            // Use MGET for non-cluster mode for better performance
            self.get_multiple_keys_with_mget(keys).await
        }
    }

    #[cfg_attr(
        feature = "deja",
        deja::redis(
            operation = "get_multiple_keys",
            read_set = keys.iter().map(|key| key.tenant_aware_key(&self.redis_conn)).collect::<Vec<_>>(),
            args = {
                serde_json::json!({
                    "key_count": keys.len(),
                    "keys": keys.iter().map(|key| key.as_str()).collect::<Vec<_>>(),
                    "command": "MGET",
                })
            },
            result = {
                (
                    match &__deja_result {
                        Ok(_) => serde_json::json!({"ok": true}),
                        Err(e) => serde_json::json!({"ok": false, "error": format!("{:?}", e)}),
                    },
                    __deja_result.is_err(),
                )
            },
        )
    )]
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_multiple_keys<V>(
        &self,
        keys: &[RedisKey],
    ) -> CustomResult<Vec<Option<V>>, errors::RedisError>
    where
        V: FromRedisValue + Send + 'static,
    {
        if keys.is_empty() {
            return Ok(Vec::new());
        }

        match self.get_keys_by_mode(keys).await {
            Ok(values) => Ok(values),
            Err(_err) => {
                #[cfg(not(feature = "multitenancy_fallback"))]
                {
                    Err(_err)
                }

                #[cfg(feature = "multitenancy_fallback")]
                {
                    let tenant_unaware_keys: Vec<RedisKey> = keys
                        .iter()
                        .map(|key| key.tenant_unaware_key(&self.redis_conn).into())
                        .collect();

                    self.get_keys_by_mode(&tenant_unaware_keys).await
                }
            }
        }
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn exists<V>(&self, key: &RedisKey) -> CustomResult<bool, errors::RedisError>
    where
        V: Send + 'static,
    {
        let mut conn = self.redis_conn.pool.clone();
        match track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::Exists,
            conn.exists::<_, bool>(key.tenant_aware_key(&self.redis_conn)),
        )
        .await
        .change_context(errors::RedisError::GetFailed)
        {
            Ok(v) => Ok(v),
            Err(_err) => {
                #[cfg(not(feature = "multitenancy_fallback"))]
                {
                    Err(_err)
                }

                #[cfg(feature = "multitenancy_fallback")]
                {
                    track_redis_call(
                        self.request_id.as_deref(),
                        self.redis_conn.event_emitter.as_ref(),
                        RedisOperation::Exists,
                        conn.exists::<_, bool>(key.tenant_unaware_key(&self.redis_conn)),
                    )
                    .await
                    .change_context(errors::RedisError::GetFailed)
                }
            }
        }
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_and_deserialize_key<T>(
        &self,
        key: &RedisKey,
        type_name: &'static str,
    ) -> CustomResult<T, errors::RedisError>
    where
        T: serde::de::DeserializeOwned,
    {
        let value_bytes = self.get_key::<Vec<u8>>(key).await?;
        fp_utils::when(value_bytes.is_empty(), || Err(errors::RedisError::NotFound))?;
        value_bytes
            .parse_struct(type_name)
            .change_context(errors::RedisError::JsonDeserializationFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_and_deserialize_multiple_keys<T>(
        &self,
        keys: &[RedisKey],
        type_name: &'static str,
    ) -> CustomResult<Vec<Option<T>>, errors::RedisError>
    where
        T: serde::de::DeserializeOwned,
    {
        let value_bytes_vec = self.get_multiple_keys::<Vec<u8>>(keys).await?;

        let mut results = Vec::with_capacity(value_bytes_vec.len());
        for value_bytes_opt in value_bytes_vec {
            match value_bytes_opt {
                Some(value_bytes) => {
                    if value_bytes.is_empty() {
                        results.push(None);
                    } else {
                        let parsed = value_bytes
                            .parse_struct(type_name)
                            .change_context(errors::RedisError::JsonDeserializationFailed)?;
                        results.push(Some(parsed));
                    }
                }
                None => results.push(None),
            }
        }

        Ok(results)
    }

    #[cfg_attr(
        feature = "deja",
        deja::redis(
            operation = "delete_key",
            codec = deja::codec::ResultCodec::<DelReply, errors::RedisError>,
            args = {
                serde_json::json!({
                    "key": key.as_str(),
                    "command": "DEL",
                })
            },
        )
    )]
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn delete_key(&self, key: &RedisKey) -> CustomResult<DelReply, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        // Redis DEL returns the number of keys that were deleted.
        // 0 means the key did not exist (not an error).
        let deleted_count: usize = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::DeleteKey,
            conn.del(key.tenant_aware_key(&self.redis_conn)),
        )
        .await
        .change_context(errors::RedisError::DeleteFailed)?;

        let reply = if deleted_count > 0 {
            DelReply::KeyDeleted
        } else {
            // Key was not found in tenant-aware namespace.
            // With multitenancy_fallback, try the tenant-unaware namespace.
            // This mirrors the old behavior where a failed DEL in tenant-aware
            // namespace would fall back to tenant-unaware.
            #[cfg(not(feature = "multitenancy_fallback"))]
            {
                DelReply::KeyNotDeleted
            }

            #[cfg(feature = "multitenancy_fallback")]
            {
                let fallback_count: usize = track_redis_call(
                    self.request_id.as_deref(),
                    self.redis_conn.event_emitter.as_ref(),
                    RedisOperation::DeleteKey,
                    conn.del(key.tenant_unaware_key(&self.redis_conn)),
                )
                .await
                .change_context(errors::RedisError::DeleteFailed)?;
                if fallback_count > 0 {
                    DelReply::KeyDeleted
                } else {
                    DelReply::KeyNotDeleted
                }
            }
        };

        Ok(reply)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn delete_multiple_keys(
        &self,
        keys: &[RedisKey],
    ) -> CustomResult<Vec<DelReply>, errors::RedisError> {
        let futures = keys.iter().map(|key| self.delete_key(key));

        let del_result = futures::future::try_join_all(futures)
            .await
            .change_context(errors::RedisError::DeleteFailed)?;

        Ok(del_result)
    }

    #[cfg_attr(
        feature = "deja",
        deja::redis(
            operation = "set_key_with_expiry",
            codec = deja::codec::ResultCodec::<(), errors::RedisError>,
            args = {
                serde_json::json!({
                    "key": key.as_str(),
                    "command": "SETEX",
                    "ttl_seconds": seconds,
                })
            },
        )
    )]
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_key_with_expiry<V>(
        &self,
        key: &RedisKey,
        value: V,
        seconds: i64,
    ) -> CustomResult<(), errors::RedisError>
    where
        V: redis::ToRedisArgs + Debug + Send + Sync + ToSingleRedisArg,
    {
        let mut conn = self.redis_conn.pool.clone();
        let options = SetOptions::default().with_expiration(SetExpiry::EX(
            u64::try_from(seconds).change_context(errors::RedisError::SetExFailed)?,
        ));
        let _: Option<String> = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetKeyWithExpiry,
            conn.set_options(key.tenant_aware_key(&self.redis_conn), value, options),
        )
        .await
        .change_context(errors::RedisError::SetExFailed)?;
        Ok(())
    }

    #[cfg_attr(
        feature = "deja",
        deja::redis(
            operation = "set_key_if_not_exists_with_expiry",
            codec = deja::codec::ResultCodec::<SetnxReply, errors::RedisError>,
            state_write = key.tenant_aware_key(&self.redis_conn),
            args = {
                serde_json::json!({
                    "key": key.as_str(),
                    "command": "SETNX",
                    "ttl_seconds": seconds,
                })
            },
        )
    )]
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_key_if_not_exists_with_expiry<V>(
        &self,
        key: &RedisKey,
        value: V,
        seconds: Option<i64>,
    ) -> CustomResult<SetnxReply, errors::RedisError>
    where
        V: redis::ToRedisArgs + Debug + Send + Sync + ToSingleRedisArg,
    {
        let ttl = seconds.unwrap_or(self.redis_conn.config.default_ttl.into());
        let mut conn = self.redis_conn.pool.clone();
        let options = SetOptions::default()
            .conditional_set(ExistenceCheck::NX)
            .with_expiration(SetExpiry::EX(
                u64::try_from(ttl).change_context(errors::RedisError::SetFailed)?,
            ));
        let result: Option<String> = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetKeyIfNotExistsWithExpiry,
            conn.set_options(key.tenant_aware_key(&self.redis_conn), value, options),
        )
        .await
        .change_context(errors::RedisError::SetFailed)?;

        // SET NX returns OK on success, nil if key already exists
        let reply = if result.is_some() {
            SetnxReply::KeySet
        } else {
            SetnxReply::KeyNotSet
        };
        Ok(reply)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_expiry(
        &self,
        key: &RedisKey,
        seconds: i64,
    ) -> CustomResult<(), errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetExpiry,
            conn.expire::<_, ()>(key.tenant_aware_key(&self.redis_conn), seconds),
        )
        .await
        .change_context(errors::RedisError::SetExpiryFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_expire_at(
        &self,
        key: &RedisKey,
        timestamp: i64,
    ) -> CustomResult<(), errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetExpireAt,
            conn.expire_at::<_, ()>(key.tenant_aware_key(&self.redis_conn), timestamp),
        )
        .await
        .change_context(errors::RedisError::SetExpiryFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_ttl(&self, key: &RedisKey) -> CustomResult<i64, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::GetTtl,
            conn.ttl::<_, i64>(key.tenant_aware_key(&self.redis_conn)),
        )
        .await
        .change_context(errors::RedisError::GetFailed)
    }

    // ─── Hash Commands ───────────────────────────────────────────────────────

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_hash_fields<F, V>(
        &self,
        key: &RedisKey,
        field_value_pairs: Vec<(F, V)>,
        ttl: Option<i64>,
    ) -> CustomResult<(), errors::RedisError>
    where
        F: redis::ToRedisArgs + Debug + Send + Sync,
        V: redis::ToRedisArgs + Debug + Send + Sync,
    {
        let mut conn = self.redis_conn.pool.clone();
        let _: () = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetHashFields,
            conn.hset_multiple(key.tenant_aware_key(&self.redis_conn), &field_value_pairs),
        )
        .await
        .change_context(errors::RedisError::SetHashFailed)?;

        // setting expiry for the key — reuse the same connection
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetHashFields,
            conn.expire::<_, ()>(
                key.tenant_aware_key(&self.redis_conn),
                ttl.unwrap_or(self.redis_conn.config.default_hash_ttl.into()),
            ),
        )
        .await
        .change_context(errors::RedisError::SetExpiryFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_hash_field_if_not_exist<V>(
        &self,
        key: &RedisKey,
        field: &str,
        value: V,
        ttl: Option<u32>,
    ) -> CustomResult<HsetnxReply, errors::RedisError>
    where
        V: redis::ToRedisArgs + ToSingleRedisArg + Debug + Send + Sync,
    {
        let mut conn = self.redis_conn.pool.clone();
        let result: HsetnxReply = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetHashFieldIfNotExist,
            conn.hset_nx::<_, _, _, HsetnxReply>(
                key.tenant_aware_key(&self.redis_conn),
                field,
                value,
            ),
        )
        .await
        .change_context(errors::RedisError::SetHashFieldFailed)?;

        // Only set expiry if the field was actually set
        if matches!(result, HsetnxReply::KeySet) {
            self.set_expiry(
                key,
                ttl.unwrap_or(self.redis_conn.config.default_hash_ttl)
                    .into(),
            )
            .await?;
        }

        Ok(result)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn serialize_and_set_hash_field_if_not_exist<V>(
        &self,
        key: &RedisKey,
        field: &str,
        value: V,
        ttl: Option<u32>,
    ) -> CustomResult<HsetnxReply, errors::RedisError>
    where
        V: serde::Serialize + Debug,
    {
        let serialized = value
            .encode_to_vec()
            .change_context(errors::RedisError::JsonSerializationFailed)?;

        self.set_hash_field_if_not_exist(key, field, serialized.as_slice(), ttl)
            .await
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn serialize_and_set_multiple_hash_field_if_not_exist<V>(
        &self,
        kv: &[(&RedisKey, V)],
        field: &str,
        ttl: Option<u32>,
    ) -> CustomResult<Vec<HsetnxReply>, errors::RedisError>
    where
        V: serde::Serialize + Debug,
    {
        let mut hsetnx: Vec<HsetnxReply> = Vec::with_capacity(kv.len());
        for (key, val) in kv {
            hsetnx.push(
                self.serialize_and_set_hash_field_if_not_exist(key, field, val, ttl)
                    .await?,
            );
        }
        Ok(hsetnx)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn increment_fields_in_hash<T>(
        &self,
        key: &RedisKey,
        fields_to_increment: &[(T, i64)],
    ) -> CustomResult<Vec<usize>, errors::RedisError>
    where
        T: Debug + ToString,
    {
        let mut conn = self.redis_conn.pool.clone();
        let mut values_after_increment = Vec::with_capacity(fields_to_increment.len());
        for (field, increment) in fields_to_increment {
            values_after_increment.push(
                track_redis_call(
                    self.request_id.as_deref(),
                    self.redis_conn.event_emitter.as_ref(),
                    RedisOperation::IncrementFieldsInHash,
                    conn.hincr::<_, _, _, usize>(
                        key.tenant_aware_key(&self.redis_conn),
                        field.to_string(),
                        *increment,
                    ),
                )
                .await
                .change_context(errors::RedisError::IncrementHashFieldFailed)?,
            )
        }

        Ok(values_after_increment)
    }

    /// Manually builds the HSCAN command because redis-rs's `hscan` / `hscan_match`
    /// helpers don't expose the `COUNT` argument. We need `MATCH` + `COUNT` together
    /// to bound per-iteration server work on large hashes.
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn hscan(
        &self,
        key: &RedisKey,
        pattern: &str,
        count: Option<u32>,
    ) -> CustomResult<Vec<String>, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();

        let mut results: Vec<String> = Vec::new();
        let mut cursor: u64 = 0;
        let mut iterations: u32 = 0;

        loop {
            // Guard against a corrupted cursor that never returns to 0
            iterations += 1;
            if iterations > crate::constant::redis_rs_commands::MAX_SCAN_ITERATIONS {
                tracing::warn!(key = ?key, iterations, "HSCAN exceeded max iterations — returning partial results");
                break;
            }

            // Build HSCAN command: HSCAN key cursor MATCH pattern [COUNT count]
            let mut command = redis::cmd(REDIS_COMMAND_HSCAN);
            command
                .arg(key.tenant_aware_key(&self.redis_conn))
                .arg(cursor)
                .arg(REDIS_ARG_MATCH)
                .arg(pattern);

            if let Some(count_value) = count {
                command.arg(REDIS_ARG_COUNT).arg(count_value);
            }

            // HSCAN returns: [cursor, [field1, value1, field2, value2, ...]]
            let reply: (u64, Vec<redis::Value>) = track_redis_call(
                self.request_id.as_deref(),
                self.redis_conn.event_emitter.as_ref(),
                RedisOperation::Hscan,
                command.query_async(&mut conn),
            )
            .await
            .change_context(errors::RedisError::GetHashFieldFailed)?;

            cursor = reply.0;

            // Extract values (odd indices) from field/value pairs
            let pairs = reply.1;
            for (index, value) in pairs.into_iter().enumerate() {
                if index % 2 == 1 {
                    if let Some(s) = redis_value_to_option_string(&value) {
                        results.push(s);
                    }
                }
            }

            // Cursor 0 means iteration is complete
            if cursor == 0 {
                break;
            }
        }

        Ok(results)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn scan(
        &self,
        pattern: &RedisKey,
        count: Option<u32>,
        scan_type: Option<crate::types::RedisScanType>,
    ) -> CustomResult<Vec<String>, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();

        let mut results: Vec<String> = Vec::new();
        let mut cursor: u64 = 0;
        let mut iterations: u32 = 0;

        loop {
            // Guard against a corrupted cursor that never returns to 0
            iterations += 1;
            if iterations > crate::constant::redis_rs_commands::MAX_SCAN_ITERATIONS {
                tracing::warn!(pattern = ?pattern, iterations, "SCAN exceeded max iterations — returning partial results");
                break;
            }

            let mut command = redis::cmd(REDIS_COMMAND_SCAN);
            command
                .arg(cursor)
                .arg(REDIS_ARG_MATCH)
                .arg(pattern.tenant_aware_key(&self.redis_conn));

            if let Some(count_value) = count {
                command.arg(REDIS_ARG_COUNT).arg(count_value);
            }

            if let Some(scan_type_value) = scan_type {
                command.arg(REDIS_ARG_TYPE).arg(scan_type_value.as_ref());
            }

            let reply: (u64, Vec<String>) = track_redis_call(
                self.request_id.as_deref(),
                self.redis_conn.event_emitter.as_ref(),
                RedisOperation::Scan,
                command.query_async(&mut conn),
            )
            .await
            .change_context(errors::RedisError::GetFailed)?;

            cursor = reply.0;
            results.extend(reply.1);

            if cursor == 0 {
                break;
            }
        }

        Ok(results)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn hscan_and_deserialize<T>(
        &self,
        key: &RedisKey,
        pattern: &str,
        count: Option<u32>,
    ) -> CustomResult<Vec<T>, errors::RedisError>
    where
        T: serde::de::DeserializeOwned,
    {
        let redis_results = self.hscan(key, pattern, count).await?;
        Ok(redis_results
            .iter()
            .filter_map(|redis_value| {
                let r: T = redis_value.parse_struct(std::any::type_name::<T>()).ok()?;
                Some(r)
            })
            .collect())
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_hash_field<V>(
        &self,
        key: &RedisKey,
        field: &str,
    ) -> CustomResult<V, errors::RedisError>
    where
        V: FromRedisValue + Send + 'static,
    {
        let mut conn = self.redis_conn.pool.clone();
        match track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::GetHashField,
            conn.hget::<_, _, V>(key.tenant_aware_key(&self.redis_conn), field),
        )
        .await
        .change_context(errors::RedisError::GetHashFieldFailed)
        {
            Ok(v) => Ok(v),
            Err(_err) => {
                #[cfg(feature = "multitenancy_fallback")]
                {
                    track_redis_call(
                        self.request_id.as_deref(),
                        self.redis_conn.event_emitter.as_ref(),
                        RedisOperation::GetHashField,
                        conn.hget::<_, _, V>(key.tenant_unaware_key(&self.redis_conn), field),
                    )
                    .await
                    .change_context(errors::RedisError::GetHashFieldFailed)
                }

                #[cfg(not(feature = "multitenancy_fallback"))]
                {
                    Err(_err)
                }
            }
        }
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_hash_fields<V>(&self, key: &RedisKey) -> CustomResult<V, errors::RedisError>
    where
        V: FromRedisValue + Send + 'static,
    {
        let mut conn = self.redis_conn.pool.clone();
        match track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::GetHashFields,
            conn.hgetall::<_, V>(key.tenant_aware_key(&self.redis_conn)),
        )
        .await
        .change_context(errors::RedisError::GetHashFieldFailed)
        {
            Ok(v) => Ok(v),
            Err(_err) => {
                #[cfg(feature = "multitenancy_fallback")]
                {
                    track_redis_call(
                        self.request_id.as_deref(),
                        self.redis_conn.event_emitter.as_ref(),
                        RedisOperation::GetHashFields,
                        conn.hgetall::<_, V>(key.tenant_unaware_key(&self.redis_conn)),
                    )
                    .await
                    .change_context(errors::RedisError::GetHashFieldFailed)
                }

                #[cfg(not(feature = "multitenancy_fallback"))]
                {
                    Err(_err)
                }
            }
        }
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_hash_field_and_deserialize<V>(
        &self,
        key: &RedisKey,
        field: &str,
        type_name: &'static str,
    ) -> CustomResult<V, errors::RedisError>
    where
        V: serde::de::DeserializeOwned,
    {
        let value_bytes = self.get_hash_field::<Vec<u8>>(key, field).await?;

        if value_bytes.is_empty() {
            return Err(errors::RedisError::NotFound.into());
        }

        value_bytes
            .parse_struct(type_name)
            .change_context(errors::RedisError::JsonDeserializationFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn delete_hash_fields<F>(
        &self,
        key: &RedisKey,
        fields: F,
    ) -> CustomResult<usize, errors::RedisError>
    where
        F: redis::ToRedisArgs + Debug + Send + Sync,
    {
        let mut conn = self.redis_conn.pool.clone();
        // HDEL returns the number of fields removed (missing fields are ignored, not an error).
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::DeleteHashFields,
            conn.hdel(key.tenant_aware_key(&self.redis_conn), fields),
        )
        .await
        .change_context(errors::RedisError::DeleteHashFieldFailed)
    }

    // ─── Set Commands ────────────────────────────────────────────────────────

    #[cfg_attr(
        feature = "deja",
        deja::redis(
            replay = Substitute,
            operation = "sadd",
            codec = deja::codec::ResultCodec::<SaddReply, errors::RedisError>,
            state_write = key.tenant_aware_key(&self.redis_conn),
            args = {
                serde_json::json!({
                    "key": key.as_str(),
                    "command": "SADD",
                })
            },
        )
    )]
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn sadd<V>(
        &self,
        key: &RedisKey,
        members: V,
    ) -> CustomResult<SaddReply, errors::RedisError>
    where
        V: redis::ToRedisArgs + Debug + Send + Sync,
    {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::Sadd,
            conn.sadd::<_, _, SaddReply>(key.tenant_aware_key(&self.redis_conn), members),
        )
        .await
        .change_context(errors::RedisError::SetAddMembersFailed)
    }

    // ─── Stream Commands ─────────────────────────────────────────────────────

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn stream_append_entry<F, V>(
        &self,
        stream: &RedisKey,
        entry_id: &RedisEntryId,
        fields: Vec<(F, V)>,
    ) -> CustomResult<(), errors::RedisError>
    where
        F: Into<String> + Debug + Send + Sync,
        V: Into<String> + Debug + Send + Sync,
    {
        let pairs: Vec<(String, String)> = fields
            .into_iter()
            .map(|(f, v)| (f.into(), v.into()))
            .collect();

        let mut conn = self.redis_conn.pool.clone();
        let _: Option<String> = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::StreamAppendEntry,
            conn.xadd_map(
                stream.tenant_aware_key(&self.redis_conn),
                entry_id.to_stream_id(),
                &pairs,
            ),
        )
        .await
        .change_context(errors::RedisError::StreamAppendFailed)?;
        Ok(())
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn stream_delete_entries(
        &self,
        stream: &RedisKey,
        ids: Vec<String>,
    ) -> CustomResult<usize, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::StreamDeleteEntries,
            conn.xdel(stream.tenant_aware_key(&self.redis_conn), &ids),
        )
        .await
        .change_context(errors::RedisError::StreamDeleteFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn stream_trim_entries(
        &self,
        stream: &RedisKey,
        config: StreamTrimConfig,
    ) -> CustomResult<usize, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();

        let options = config
            .to_trim_options()
            .change_context(errors::RedisError::StreamTrimFailed)?;

        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::StreamTrimEntries,
            conn.xtrim_options(stream.tenant_aware_key(&self.redis_conn), &options),
        )
        .await
        .change_context(errors::RedisError::StreamTrimFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn stream_acknowledge_entries(
        &self,
        stream: &RedisKey,
        group: &str,
        ids: Vec<String>,
    ) -> CustomResult<usize, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::StreamAcknowledgeEntries,
            conn.xack(stream.tenant_aware_key(&self.redis_conn), group, &ids),
        )
        .await
        .change_context(errors::RedisError::StreamAcknowledgeFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn stream_get_length(
        &self,
        stream: &RedisKey,
    ) -> CustomResult<usize, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::StreamGetLength,
            conn.xlen(stream.tenant_aware_key(&self.redis_conn)),
        )
        .await
        .change_context(errors::RedisError::GetLengthFailed)
    }

    /// Read entries from one or more streams using XREAD.
    /// Returns a [`StreamReadResult`] (stream key → list of `(entry_id, fields)`),
    /// so callers never need to parse backend-specific reply types.
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn stream_read_entries(
        &self,
        streams: &[RedisKey],
        ids: Vec<String>,
        read_count: Option<u64>,
    ) -> CustomResult<StreamReadResult, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        let stream_keys: Vec<String> = streams
            .iter()
            .map(|stream| stream.tenant_aware_key(&self.redis_conn))
            .collect();

        let count = read_count.unwrap_or(self.redis_conn.config.default_stream_read_count);

        let options = StreamReadOptions::default()
            .count(usize::try_from(count).change_context(errors::RedisError::StreamReadFailed)?);

        let reply: redis::streams::StreamReadReply = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::StreamReadEntries,
            conn.xread_options(&stream_keys, &ids, &options),
        )
        .await
        .map_err(|err| {
            let kind = err.kind();
            match kind {
                redis::ErrorKind::UnexpectedReturnType | redis::ErrorKind::Parse => {
                    report!(err).change_context(errors::RedisError::StreamEmptyOrNotAvailable)
                }
                _ => report!(err).change_context(errors::RedisError::StreamReadFailed),
            }
        })?;

        Ok(reply
            .keys
            .into_iter()
            .map(|stream_response| {
                let stream_entries: StreamEntries = stream_response
                    .ids
                    .into_iter()
                    .map(|entry_response| {
                        let fields_by_redis_value: std::collections::HashMap<
                            String,
                            crate::RedisValue,
                        > = entry_response
                            .map
                            .into_iter()
                            .map(|(field_name, raw_redis_value)| {
                                (field_name, crate::RedisValue::new(raw_redis_value))
                            })
                            .collect();
                        (entry_response.id, fields_by_redis_value)
                    })
                    .collect();
                (stream_response.key, stream_entries)
            })
            .collect())
    }

    /// Read stream entries and return them grouped by stream key with optional field values.
    /// Read entries from streams with options (XREAD / XREADGROUP)
    /// Returns a [`StreamReadResult`] (stream key → list of `(entry_id, fields)`),
    /// so callers never need to parse backend-specific reply types.
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn stream_read_with_options(
        &self,
        streams: &[RedisKey],
        ids: Vec<String>,
        count: Option<u64>,
        block: Option<u64>,
        group: Option<(&str, &str)>,
    ) -> CustomResult<StreamReadResult, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        let stream_keys: Vec<String> = streams
            .iter()
            .map(|stream| stream.tenant_aware_key(&self.redis_conn))
            .collect();

        let mut options = StreamReadOptions::default();

        if let Some(count_val) = count {
            options = options.count(
                usize::try_from(count_val).change_context(errors::RedisError::StreamReadFailed)?,
            );
        }
        if let Some(block_ms) = block {
            options = options.block(
                usize::try_from(block_ms).change_context(errors::RedisError::StreamReadFailed)?,
            );
        }
        if let Some((group_name, consumer_name)) = group {
            options = options.group(group_name, consumer_name);
        }

        let reply: redis::streams::StreamReadReply = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::StreamReadWithOptions,
            conn.xread_options(&stream_keys, &ids, &options),
        )
        .await
        .map_err(|err| {
            let kind = err.kind();
            match kind {
                redis::ErrorKind::UnexpectedReturnType | redis::ErrorKind::Parse => {
                    report!(err).change_context(errors::RedisError::StreamEmptyOrNotAvailable)
                }
                _ => report!(err).change_context(errors::RedisError::StreamReadFailed),
            }
        })?;

        Ok(reply
            .keys
            .into_iter()
            .map(|stream_response| {
                let stream_entries: StreamEntries = stream_response
                    .ids
                    .into_iter()
                    .map(|entry_response| {
                        let fields_by_redis_value: std::collections::HashMap<
                            String,
                            crate::RedisValue,
                        > = entry_response
                            .map
                            .into_iter()
                            .map(|(field_name, raw_redis_value)| {
                                (field_name, crate::RedisValue::new(raw_redis_value))
                            })
                            .collect();
                        (entry_response.id, fields_by_redis_value)
                    })
                    .collect();
                (stream_response.key, stream_entries)
            })
            .collect())
    }

    // ─── List Commands ───────────────────────────────────────────────────────

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn append_elements_to_list<V>(
        &self,
        key: &RedisKey,
        elements: V,
    ) -> CustomResult<(), errors::RedisError>
    where
        V: redis::ToRedisArgs + Debug + Send + Sync,
    {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::AppendElementsToList,
            conn.rpush::<_, _, ()>(key.tenant_aware_key(&self.redis_conn), elements),
        )
        .await
        .change_context(errors::RedisError::AppendElementsToListFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_list_elements(
        &self,
        key: &RedisKey,
        start: i64,
        stop: i64,
    ) -> CustomResult<Vec<String>, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::GetListElements,
            conn.lrange::<_, Vec<String>>(
                key.tenant_aware_key(&self.redis_conn),
                isize::try_from(start).change_context(errors::RedisError::GetListElementsFailed)?,
                isize::try_from(stop).change_context(errors::RedisError::GetListElementsFailed)?,
            ),
        )
        .await
        .change_context(errors::RedisError::GetListElementsFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn get_list_length(&self, key: &RedisKey) -> CustomResult<usize, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::GetListLength,
            conn.llen::<_, usize>(key.tenant_aware_key(&self.redis_conn)),
        )
        .await
        .change_context(errors::RedisError::GetListLengthFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn lpop_list_elements(
        &self,
        key: &RedisKey,
        count: Option<usize>,
    ) -> CustomResult<Vec<String>, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        let non_zero_count = count.and_then(std::num::NonZeroUsize::new);
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::LpopListElements,
            conn.lpop::<_, Vec<String>>(key.tenant_aware_key(&self.redis_conn), non_zero_count),
        )
        .await
        .change_context(errors::RedisError::PopListElementsFailed)
    }

    //                                              Consumer Group API

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn consumer_group_create(
        &self,
        stream: &RedisKey,
        group: &str,
        id: &RedisEntryId,
    ) -> CustomResult<(), errors::RedisError> {
        if matches!(
            id,
            RedisEntryId::AutoGeneratedID | RedisEntryId::UndeliveredEntryID
        ) {
            Err(errors::RedisError::InvalidRedisEntryId)?;
        }

        let mut conn = self.redis_conn.pool.clone();
        let _: () = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::ConsumerGroupCreate,
            conn.xgroup_create_mkstream(
                stream.tenant_aware_key(&self.redis_conn),
                group,
                id.to_stream_id(),
            ),
        )
        .await
        .change_context(errors::RedisError::ConsumerGroupCreateFailed)?;
        Ok(())
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn consumer_group_destroy(
        &self,
        stream: &RedisKey,
        group: &str,
    ) -> CustomResult<crate::types::ConsumerGroupDestroyReply, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        let reply: crate::types::ConsumerGroupDestroyReply = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::ConsumerGroupDestroy,
            conn.xgroup_destroy(stream.tenant_aware_key(&self.redis_conn), group),
        )
        .await
        .change_context(errors::RedisError::ConsumerGroupDestroyFailed)?;
        Ok(reply)
    }

    // the number of pending messages that the consumer had before it was deleted
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn consumer_group_delete_consumer(
        &self,
        stream: &RedisKey,
        group: &str,
        consumer: &str,
    ) -> CustomResult<usize, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::ConsumerGroupDeleteConsumer,
            conn.xgroup_delconsumer(stream.tenant_aware_key(&self.redis_conn), group, consumer),
        )
        .await
        .change_context(errors::RedisError::ConsumerGroupRemoveConsumerFailed)
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn consumer_group_set_last_id(
        &self,
        stream: &RedisKey,
        group: &str,
        id: &RedisEntryId,
    ) -> CustomResult<String, errors::RedisError> {
        let mut conn = self.redis_conn.pool.clone();
        let _: () = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::ConsumerGroupSetLastId,
            conn.xgroup_setid(
                stream.tenant_aware_key(&self.redis_conn),
                group,
                id.to_stream_id(),
            ),
        )
        .await
        .change_context(errors::RedisError::ConsumerGroupSetIdFailed)?;
        Ok(id.to_stream_id().to_string())
    }

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn consumer_group_set_message_owner<R>(
        &self,
        stream: &RedisKey,
        group: &str,
        consumer: &str,
        min_idle_time: u64,
        ids: Vec<String>,
    ) -> CustomResult<R, errors::RedisError>
    where
        R: FromRedisValue + Send + 'static,
    {
        let mut conn = self.redis_conn.pool.clone();
        track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::ConsumerGroupSetMessageOwner,
            conn.xclaim(
                stream.tenant_aware_key(&self.redis_conn),
                group,
                consumer,
                min_idle_time,
                &ids,
            ),
        )
        .await
        .change_context(errors::RedisError::ConsumerGroupClaimFailed)
    }

    // ─── Lua Scripting ───────────────────────────────────────────────────────

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn evaluate_redis_script<V, T>(
        &self,
        lua_script: &'static str,
        key: Vec<String>,
        values: V,
    ) -> CustomResult<T, errors::RedisError>
    where
        V: redis::ToRedisArgs + Debug + Send + Sync,
        T: serde::de::DeserializeOwned + FromRedisValue,
    {
        let mut conn = self.redis_conn.pool.clone();
        let script = redis::Script::new(lua_script);
        let mut invocation = script.prepare_invoke();

        for script_key in &key {
            invocation.key(script_key);
        }
        invocation.arg(values);

        let val: T = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::EvaluateRedisScript,
            invocation.invoke_async(&mut conn),
        )
        .await
        .change_context(errors::RedisError::ScriptExecutionFailed)?;
        Ok(val)
    }

    // ─── Transactions ────────────────────────────────────────────────────────

    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_multiple_keys_if_not_exists_and_get_values<V>(
        &self,
        keys: &[(RedisKey, V)],
        ttl: Option<i64>,
    ) -> CustomResult<Vec<SetGetReply<V>>, errors::RedisError>
    where
        V: redis::ToRedisArgs
            + Debug
            + FromRedisValue
            + ToOwned<Owned = V>
            + Send
            + Sync
            + serde::de::DeserializeOwned,
    {
        let futures = keys.iter().map(|(key, value)| {
            self.set_key_if_not_exists_and_get_value(key, (*value).to_owned(), ttl)
        });

        let del_result = futures::future::try_join_all(futures)
            .await
            .change_context(errors::RedisError::SetFailed)?;

        Ok(del_result)
    }

    /// Sets a value in Redis if not already present, and returns the value (either existing or newly set).
    /// This operation is atomic using Redis transactions (MULTI/EXEC pipeline).
    #[instrument(level = "DEBUG", skip(self))]
    pub async fn set_key_if_not_exists_and_get_value<V>(
        &self,
        key: &RedisKey,
        value: V,
        ttl: Option<i64>,
    ) -> CustomResult<SetGetReply<V>, errors::RedisError>
    where
        V: redis::ToRedisArgs + Debug + FromRedisValue + Send + Sync + serde::de::DeserializeOwned,
    {
        let redis_key = key.tenant_aware_key(&self.redis_conn);
        let ttl_seconds = ttl.unwrap_or(self.redis_conn.config.default_ttl.into());

        let mut conn = self.redis_conn.pool.clone();

        // Build an atomic pipeline (MULTI/EXEC)
        let mut pipe = redis::pipe();
        pipe.atomic();

        // SET key value EX ttl NX
        pipe.cmd(REDIS_COMMAND_SET)
            .arg(&redis_key)
            .arg(&value)
            .arg(REDIS_ARG_EX)
            .arg(ttl_seconds)
            .arg(REDIS_ARG_NX);

        pipe.cmd(REDIS_COMMAND_GET).arg(&redis_key);

        // Execute the transaction.
        // Redis MULTI/EXEC guarantees results are returned in the same order
        // as the commands were queued: [SET result, GET result].
        let results: Vec<redis::Value> = track_redis_call(
            self.request_id.as_deref(),
            self.redis_conn.event_emitter.as_ref(),
            RedisOperation::SetKeyIfNotExistsAndGetValue,
            pipe.query_async(&mut conn),
        )
        .await
        .change_context(errors::RedisError::SetFailed)
        .attach_printable("Failed to execute the redis transaction")?;

        let msg = "Got unexpected number of results from transaction";
        if results.len() < 2 {
            return Err(report!(errors::RedisError::SetFailed).attach_printable(msg));
        }

        let set_result = results
            .first()
            .cloned()
            .ok_or_else(|| report!(errors::RedisError::SetFailed).attach_printable(msg))?;
        let get_result = results
            .get(1)
            .cloned()
            .ok_or_else(|| report!(errors::RedisError::SetFailed).attach_printable(msg))?;

        // Parse the GET result to get the actual value
        let actual_value: V = FromRedisValue::from_redis_value(get_result)
            .change_context(errors::RedisError::SetFailed)
            .attach_printable("Failed to convert from redis value")?;

        // Check if SET NX succeeded or failed using the existing SetnxReply type
        let setnx_reply = SetnxReply::from_redis_value(set_result)
            .change_context(errors::RedisError::SetFailed)
            .attach_printable("Unexpected result from SET NX operation")?;

        Ok(match setnx_reply {
            SetnxReply::KeySet => SetGetReply::ValueSet(actual_value),
            SetnxReply::KeyNotSet => SetGetReply::ValueExists(actual_value),
        })
    }
}
