oxcache 0.5.0-rc.7

// Copyright (c) 2025-2026 Kirky.X🌠
// SPDX-License-Identifier: MIT
//! Async trait implementations for RedisBackend.

use super::builder::RedisMode;
use super::client::RedisBackend;
use crate::backend::interface::AtomicCacheWriter;
use crate::backend::memory::redis::error;
use crate::backend::{BackendKind, CacheConnector, CacheReader, CacheWriter};
use crate::backend::{BackendScore, Scores};
use crate::core::RedisCommand;
use crate::error::{OxCacheError, OxCacheResult};
use crate::i18n::messages::{
    MSG_DETAIL_REDIS_TTL_EXCEEDS_MAX, MSG_DETAIL_REDIS_TTL_MIN_MILLIS,
    MSG_DETAIL_REDIS_UNEXPECTED_INFO_REPLY, t,
};
use std::time::Duration;

/// Redis 最大 TTL 上界（秒）。Redis SETEX/EXPIRE 仅接受 i32::MAX 秒（~68 年）；
/// 毫秒口径（SET PX/PEXPIRE）沿用同一上界换算，避免隐式放大可写时长。
const REDIS_MAX_TTL_SECS: u64 = i32::MAX as u64;

/// 校验 TTL 值是否在 Redis 可接受范围内，成功返回毫秒值。
///
/// SET PX 自 Redis 2.6.12、PEXPIRE 自 2.6.0 起可用，以毫秒为最小粒度，亚秒 TTL 合法；
/// 仅当整体时长不足 1ms 时才视为无效。pipeline 模块的批量写入复用同一口径。
pub(super) fn validate_redis_ttl(ttl: Duration) -> OxCacheResult<u64> {
    let millis = ttl.as_millis();
    if millis == 0 {
        return Err(OxCacheError::InvalidInput(t(
            MSG_DETAIL_REDIS_TTL_MIN_MILLIS,
            &[],
        )));
    }
    if millis > (REDIS_MAX_TTL_SECS * 1000) as u128 {
        return Err(OxCacheError::InvalidInput(t(
            MSG_DETAIL_REDIS_TTL_EXCEEDS_MAX,
            &[
                ("millis", millis.to_string()),
                ("max", (REDIS_MAX_TTL_SECS * 1000).to_string()),
            ],
        )));
    }
    Ok(millis as u64)
}

/// 将 INFO 应答归一为单节点文本。
///
/// 单机/哨兵连接应答为整段 bulk string；集群连接下 redis-rs 把 INFO 广播到
/// 全部主节点，应答为 节点地址 → 文本 的 map——取地址字典序最小的节点作为
/// 代表样本（确定性选取）。非文本应答返回 None 由调用方显性报错。
fn normalize_info_reply(reply: redis::Value) -> Option<String> {
    fn value_as_text(value: redis::Value) -> Option<String> {
        match value {
            redis::Value::BulkString(bytes) => String::from_utf8(bytes).ok(),
            redis::Value::SimpleString(text) | redis::Value::VerbatimString { text, .. } => {
                Some(text)
            }
            _ => None,
        }
    }

    match reply {
        redis::Value::Map(pairs) => pairs
            .into_iter()
            .filter_map(|(addr, info)| Some((value_as_text(addr)?, value_as_text(info)?)))
            .min_by(|a, b| a.0.cmp(&b.0))
            .map(|(_, info)| info),
        other => value_as_text(other),
    }
}
use crate::security;
use async_trait::async_trait;
use std::collections::HashMap;

#[async_trait]
impl CacheReader for RedisBackend {
    async fn get(&self, key: &str) -> OxCacheResult<Option<Vec<u8>>> {
        security::validate_redis_key(key)?;
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                redis::cmd(RedisCommand::Get.as_str())
                    .arg(key)
                    .query_async(&mut conn)
                    .await
                    .map_err(error::map_redis_error)
            }
        })
        .await
    }

    async fn exists(&self, key: &str) -> OxCacheResult<bool> {
        security::validate_redis_key(key)?;
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                let n: i64 = redis::cmd(RedisCommand::Exists.as_str())
                    .arg(key)
                    .query_async(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                Ok(n > 0)
            }
        })
        .await
    }

    async fn ttl(&self, key: &str) -> OxCacheResult<Option<Duration>> {
        security::validate_redis_key(key)?;
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                let n: i64 = redis::cmd(RedisCommand::Ttl.as_str())
                    .arg(key)
                    .query_async(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                if n <= 0 {
                    Ok(None)
                } else {
                    Ok(Some(Duration::from_secs(n as u64)))
                }
            }
        })
        .await
    }

    // DBSIZE counts every key in the connected DB index, not only keys
    // written through this instance — see CacheReader::len docs.
    async fn len(&self) -> OxCacheResult<u64> {
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                let len: i64 = redis::cmd(RedisCommand::Dbsize.as_str())
                    .query_async(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                Ok(len as u64)
            }
        })
        .await
    }

    async fn is_empty(&self) -> OxCacheResult<bool> {
        Ok(self.len().await?.eq(&0))
    }

    async fn capacity(&self) -> OxCacheResult<u64> {
        Ok(0)
    }

    /// Collect backend statistics from Redis INFO command.
    ///
    /// Returns a map containing:
    /// - `"memory_info"`: raw output of `INFO memory`
    /// - `"connected_clients"`: current client count (parsed from `INFO clients`)
    /// - `"maxclients"`: maximum client limit (parsed from `INFO clients`)
    ///
    /// 集群连接下 redis-rs 将 INFO 广播到全部主节点并返回节点地址 → 文本
    /// 的 map，此处取地址字典序最小节点作为代表样本（确定性选取，主节点
    /// 角色等价）；单机连接直接使用整段应答。
    async fn stats(&self) -> OxCacheResult<HashMap<String, String>> {
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                let mut stats = HashMap::new();
                stats.insert("type".to_string(), "redis".to_string());

                // INFO memory
                let memory_reply = redis::cmd(RedisCommand::Info.as_str())
                    .arg("memory")
                    .query_async::<redis::Value>(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                let memory_info = normalize_info_reply(memory_reply).ok_or_else(|| {
                    crate::error::OxCacheError::Operation(t(
                        MSG_DETAIL_REDIS_UNEXPECTED_INFO_REPLY,
                        &[("section", "memory".to_string())],
                    ))
                })?;
                stats.insert("memory_info".to_string(), memory_info);

                // INFO clients — parse connected_clients and maxclients
                let clients_reply = redis::cmd(RedisCommand::Info.as_str())
                    .arg("clients")
                    .query_async::<redis::Value>(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                let clients_info = normalize_info_reply(clients_reply).ok_or_else(|| {
                    crate::error::OxCacheError::Operation(t(
                        MSG_DETAIL_REDIS_UNEXPECTED_INFO_REPLY,
                        &[("section", "clients".to_string())],
                    ))
                })?;
                for line in clients_info.lines() {
                    let line = line.trim();
                    if let Some((key, value)) = line.split_once(':') {
                        let key = key.trim();
                        let value = value.trim();
                        if key == "connected_clients" || key == "maxclients" {
                            stats.insert(key.to_string(), value.to_string());
                        }
                    }
                }

                Ok(stats)
            }
        })
        .await
    }

    async fn get_many(&self, keys: &[String]) -> OxCacheResult<Vec<Option<Vec<u8>>>> {
        if keys.is_empty() {
            return Ok(vec![]);
        }
        let keys_slice: Vec<&str> = keys.iter().map(|s| s.as_str()).collect();
        self.get_many_pipeline(&keys_slice).await
    }

    async fn keys(&self, pattern: &str) -> OxCacheResult<Vec<String>> {
        security::validate_scan_pattern(pattern)?;
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                let mut all_keys = Vec::new();
                let mut cursor = 0i64;
                loop {
                    let (new_cursor, batch): (i64, Vec<String>) =
                        redis::cmd(RedisCommand::Scan.as_str())
                            .arg(cursor)
                            .arg("MATCH")
                            .arg(pattern)
                            .arg("COUNT")
                            .arg(100)
                            .query_async(&mut conn)
                            .await
                            .map_err(error::map_redis_error)?;
                    all_keys.extend(batch);
                    cursor = new_cursor;
                    if cursor == 0 {
                        break;
                    }
                }
                Ok(all_keys)
            }
        })
        .await
    }
}

#[async_trait]
impl CacheWriter for RedisBackend {
    async fn set(
        &self,
        key: std::sync::Arc<str>,
        value: std::sync::Arc<Vec<u8>>,
        ttl: Option<Duration>,
    ) -> OxCacheResult<()> {
        let key_ref = key.as_ref();
        security::validate_redis_key(key_ref)?;

        // Validate TTL before entering retry closure
        let ttl_millis = match ttl {
            Some(t) => Some(validate_redis_ttl(t)?),
            None => None,
        };

        self.execute_with_retry(|| {
            let mut conn = self.conn();
            let key = key.clone();
            let value = value.clone();
            async move {
                if let Some(ttl_millis) = ttl_millis {
                    redis::cmd(RedisCommand::Set.as_str())
                        .arg(key.as_ref())
                        .arg(value.as_ref())
                        .arg("PX")
                        .arg(ttl_millis)
                        .query_async::<()>(&mut conn)
                        .await
                        .map_err(error::map_redis_error)?;
                } else {
                    redis::cmd(RedisCommand::Set.as_str())
                        .arg(key.as_ref())
                        .arg(value.as_ref())
                        .query_async::<()>(&mut conn)
                        .await
                        .map_err(error::map_redis_error)?;
                }
                Ok(())
            }
        })
        .await
    }

    async fn delete(&self, key: &str) -> OxCacheResult<()> {
        security::validate_redis_key(key)?;
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                redis::cmd(RedisCommand::Del.as_str())
                    .arg(key)
                    .query_async::<()>(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                Ok(())
            }
        })
        .await
    }

    /// Clear all keys from the Redis database.
    ///
    /// **Safety gate**: requires `dangerous_clear_enabled = true` in the builder.
    /// Uses SCAN + pipeline DEL for batch deletion (not per-key DEL).
    async fn clear(&self) -> OxCacheResult<()> {
        if !self.dangerous_clear_enabled() {
            return Err(OxCacheError::NotSupported(
                "Full-database clear() is disabled by default. \
                 Use RedisBackend::builder().dangerous_clear_enabled(true) to enable, \
                 or use clear_namespace(prefix) for safe prefix-based deletion."
                    .to_string(),
            ));
        }

        security::validate_scan_pattern("*")?;

        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                let mut cursor = 0i64;
                loop {
                    let (new_cursor, keys): (i64, Vec<String>) =
                        redis::cmd(RedisCommand::Scan.as_str())
                            .arg(cursor)
                            .arg("MATCH")
                            .arg("*")
                            .arg("COUNT")
                            .arg(100)
                            .query_async(&mut conn)
                            .await
                            .map_err(error::map_redis_error)?;

                    if !keys.is_empty() {
                        let mut pipe = redis::pipe();
                        for key in &keys {
                            pipe.cmd(RedisCommand::Del.as_str()).arg(key);
                        }
                        pipe.query_async::<()>(&mut conn)
                            .await
                            .map_err(error::map_redis_error)?;
                    }

                    cursor = new_cursor;
                    if cursor == 0 {
                        break;
                    }
                }
                Ok(())
            }
        })
        .await
    }

    async fn expire(&self, key: &str, ttl: Duration) -> OxCacheResult<bool> {
        security::validate_redis_key(key)?;
        let ttl_millis = validate_redis_ttl(ttl)?;
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                let result: i64 = redis::cmd(RedisCommand::PExpire.as_str())
                    .arg(key)
                    .arg(ttl_millis)
                    .query_async(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                Ok(result > 0)
            }
        })
        .await
    }

    async fn set_many(&self, items: &[crate::backend::CacheSetItem]) -> OxCacheResult<()> {
        if items.is_empty() {
            return Ok(());
        }
        for (key, _, _) in items {
            security::validate_redis_key(key)?;
        }

        // Validate TTLs before entering retry closure
        let prepared: OxCacheResult<
            Vec<(std::sync::Arc<str>, std::sync::Arc<Vec<u8>>, Option<u64>)>,
        > = items
            .iter()
            .map(|(key, value, ttl)| {
                let ttl_millis = match ttl {
                    Some(t) => Some(validate_redis_ttl(*t)?),
                    None => None,
                };
                Ok((key.clone(), value.clone(), ttl_millis))
            })
            .collect();
        let prepared = prepared?;

        self.execute_with_retry(|| {
            let mut conn = self.conn();
            let prepared = prepared.clone();
            async move {
                let mut pipe = redis::pipe();
                for (key, value, ttl_millis) in prepared.iter() {
                    if let Some(ms) = ttl_millis {
                        pipe.cmd(RedisCommand::Set.as_str())
                            .arg(key.as_ref())
                            .arg(value.as_ref().as_slice())
                            .arg("PX")
                            .arg(ms);
                    } else {
                        pipe.cmd(RedisCommand::Set.as_str())
                            .arg(key.as_ref())
                            .arg(value.as_ref().as_slice());
                    }
                }
                pipe.query_async::<()>(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                Ok(())
            }
        })
        .await
    }

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

        let keys_slice: Vec<&str> = keys.iter().map(|s| s.as_str()).collect();
        self.delete_many_pipeline(&keys_slice).await
    }
}

#[async_trait]
impl CacheConnector for RedisBackend {
    async fn health_check(&self) -> OxCacheResult<()> {
        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                redis::cmd(RedisCommand::Ping.as_str())
                    .query_async::<String>(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                Ok(())
            }
        })
        .await
    }

    async fn shutdown(&self) {
        // Redis connection is managed by ConnectionManager; no explicit close needed.
    }

    fn backend_kind(&self) -> BackendKind {
        if self.mode() == RedisMode::ValkeyStandalone {
            BackendKind::Valkey
        } else {
            BackendKind::Redis
        }
    }

    #[cfg(feature = "lua")]
    fn as_lua_executor(&self) -> Option<&dyn crate::backend::interface::LuaExecutor> {
        Some(self)
    }

    fn as_atomic_writer(&self) -> Option<&dyn AtomicCacheWriter> {
        Some(self)
    }
}

impl BackendScore for RedisBackend {
    fn score(&self) -> u8 {
        Scores::REDIS
    }

    fn is_persistent(&self) -> bool {
        true
    }

    fn backend_name(&self) -> &'static str {
        "redis"
    }
}

// ============================================================================
// AtomicCacheWriter Implementation
// ============================================================================

#[async_trait]
impl AtomicCacheWriter for RedisBackend {
    async fn incr(&self, key: &str, delta: i64, ttl: Option<Duration>) -> OxCacheResult<i64> {
        security::validate_redis_key(key)?;

        // Validate TTL before entering retry closure
        let ttl_millis = match ttl {
            Some(t) => Some(validate_redis_ttl(t)?),
            None => None,
        };

        self.execute_with_retry(|| {
            let mut conn = self.conn();
            async move {
                if let Some(ttl_millis) = ttl_millis {
                    // Atomic INCR + PEXPIRE via Lua script to avoid non-atomic two-step
                    let script = "local r = redis.call('INCRBY', KEYS[1], ARGV[1]); \
                         redis.call('PEXPIRE', KEYS[1], ARGV[2]); \
                         return r";
                    let result: i64 = redis::cmd(RedisCommand::Eval.as_str())
                        .arg(script)
                        .arg(1)
                        .arg(key)
                        .arg(delta)
                        .arg(ttl_millis)
                        .query_async(&mut conn)
                        .await
                        .map_err(error::map_redis_error)?;
                    Ok(result)
                } else if delta == 1 {
                    let result: i64 = redis::cmd(RedisCommand::Incr.as_str())
                        .arg(key)
                        .query_async(&mut conn)
                        .await
                        .map_err(error::map_redis_error)?;
                    Ok(result)
                } else {
                    let result: i64 = redis::cmd(RedisCommand::IncrBy.as_str())
                        .arg(key)
                        .arg(delta)
                        .query_async(&mut conn)
                        .await
                        .map_err(error::map_redis_error)?;
                    Ok(result)
                }
            }
        })
        .await
    }

    async fn compare_and_swap(
        &self,
        key: &str,
        expected: Option<&[u8]>,
        new: Vec<u8>,
        ttl: Option<Duration>,
    ) -> OxCacheResult<bool> {
        security::validate_redis_key(key)?;

        // Static Lua scripts — avoid per-call heap allocation
        static NO_EXPECT_WITH_TTL: &str = "if redis.call('EXISTS', KEYS[1]) == 0 then \
             redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2]) \
             return 1 else return 0 end";
        static NO_EXPECT_WITHOUT_TTL: &str = "if redis.call('EXISTS', KEYS[1]) == 0 then \
             redis.call('SET', KEYS[1], ARGV[1]) \
             return 1 else return 0 end";
        static WITH_EXPECT_WITH_TTL: &str = "if redis.call('GET', KEYS[1]) == ARGV[1] then \
             redis.call('SET', KEYS[1], ARGV[2], 'PX', ARGV[3]) \
             return 1 else return 0 end";
        static WITH_EXPECT_WITHOUT_TTL: &str = "if redis.call('GET', KEYS[1]) == ARGV[1] then \
             redis.call('SET', KEYS[1], ARGV[2]) \
             return 1 else return 0 end";

        let lua_script: &str = match expected {
            None => {
                if ttl.is_some() {
                    NO_EXPECT_WITH_TTL
                } else {
                    NO_EXPECT_WITHOUT_TTL
                }
            }
            Some(_) => {
                if ttl.is_some() {
                    WITH_EXPECT_WITH_TTL
                } else {
                    WITH_EXPECT_WITHOUT_TTL
                }
            }
        };

        // Validate TTL before entering retry closure
        let ttl_millis = match ttl {
            Some(t) => Some(validate_redis_ttl(t)?),
            None => None,
        };

        self.execute_with_retry(|| {
            let mut conn = self.conn();
            let new = new.clone();
            async move {
                let mut cmd = redis::cmd(RedisCommand::Eval.as_str());
                cmd.arg(lua_script).arg(1).arg(key);
                match expected {
                    None => {
                        cmd.arg(new.as_slice());
                        if let Some(millis) = ttl_millis {
                            cmd.arg(millis);
                        }
                    }
                    Some(exp_bytes) => {
                        cmd.arg(exp_bytes);
                        cmd.arg(new.as_slice());
                        if let Some(millis) = ttl_millis {
                            cmd.arg(millis);
                        }
                    }
                }
                let result: i64 = cmd
                    .query_async(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                Ok(result == 1)
            }
        })
        .await
    }

    async fn set_if_absent(
        &self,
        key: &str,
        value: Vec<u8>,
        ttl: Option<Duration>,
    ) -> OxCacheResult<bool> {
        security::validate_redis_key(key)?;

        // Validate TTL before entering retry closure
        let ttl_millis = match ttl {
            Some(t) => Some(validate_redis_ttl(t)?),
            None => None,
        };

        self.execute_with_retry(|| {
            let mut conn = self.conn();
            let value = value.clone();
            async move {
                let mut cmd = redis::cmd(RedisCommand::Set.as_str());
                cmd.arg(key).arg(value.as_slice()).arg("NX");
                if let Some(millis) = ttl_millis {
                    cmd.arg("PX").arg(millis);
                }
                let result: Option<redis::Value> = cmd
                    .query_async(&mut conn)
                    .await
                    .map_err(error::map_redis_error)?;
                Ok(result.is_some())
            }
        })
        .await
    }
}