oxcache 0.5.0-rc.9

// Copyright (c) 2025-2026 Kirky.X🌠
// SPDX-License-Identifier: MIT
//! Redis backend core: struct definition and essential methods.
//!
//! Trait implementations are split into dedicated modules:
//! - `async_traits` — CacheReader / CacheWriter / CacheConnector / BackendScore / AtomicCacheWriter
//! - `sync_traits` — SyncCacheReader / SyncCacheWriter / SyncCacheConnector / SyncAtomicCacheWriter
//! - `pipeline` — batch pipeline operations
//! - `lua_executor` — Lua script execution
//! - `namespace` — prefix-scoped key deletion
//! - `builder` — RedisBackendBuilder with extended configuration

use super::builder::{RedisBackendBuilder, RedisMode};
use super::circuit_breaker::CircuitBreaker;
use super::retry::retry_with_backoff;
use super::sentinel::SentinelConnection;
use crate::core::RedisCommand;
use crate::error::{OxCacheError, OxCacheResult};
#[cfg(feature = "metrics")]
use crate::infra::metrics::unified::GLOBAL_UNIFIED_METRICS;
use redis::Client;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

/// Redis async connection abstraction.
///
/// 单机场景走 [`redis::aio::ConnectionManager`]；集群场景走
/// [`redis::cluster_async::ClusterConnection`]（按槽位路由并自动跟随
/// MOVED/ASK 重定向）；哨兵场景由 [`SentinelConnection`] 持有当前 master
/// 连接，可恢复错误后重新发现 master（failover 跟随）。三者都实现
/// `aio::ConnectionLike`，调用方以 `cmd.query_async(&mut conn)` 统一驱动，
/// 无需感知分支。
#[derive(Clone)]
pub(crate) enum RedisConnection {
    Single(redis::aio::ConnectionManager),
    Cluster(redis::cluster_async::ClusterConnection),
    Sentinel(SentinelConnection),
}

impl redis::aio::ConnectionLike for RedisConnection {
    fn req_packed_command<'a>(
        &'a mut self,
        cmd: &'a redis::Cmd,
    ) -> redis::RedisFuture<'a, redis::Value> {
        Box::pin(async move {
            match self {
                RedisConnection::Single(conn) => conn.req_packed_command(cmd).await,
                RedisConnection::Cluster(conn) => conn.req_packed_command(cmd).await,
                RedisConnection::Sentinel(sentinel) => sentinel.exec_cmd(cmd).await,
            }
        })
    }

    fn req_packed_commands<'a>(
        &'a mut self,
        pipeline: &'a redis::Pipeline,
        offset: usize,
        count: usize,
    ) -> redis::RedisFuture<'a, Vec<redis::Value>> {
        Box::pin(async move {
            match self {
                RedisConnection::Single(conn) => {
                    conn.req_packed_commands(pipeline, offset, count).await
                }
                RedisConnection::Cluster(conn) => {
                    conn.req_packed_commands(pipeline, offset, count).await
                }
                RedisConnection::Sentinel(sentinel) => {
                    sentinel.exec_pipeline(pipeline, offset, count).await
                }
            }
        })
    }

    fn get_db(&self) -> i64 {
        match self {
            RedisConnection::Single(conn) => conn.get_db(),
            RedisConnection::Cluster(conn) => conn.get_db(),
            RedisConnection::Sentinel(sentinel) => sentinel.db(),
        }
    }
}

/// Redis cache backend.
///
/// This backend provides a distributed cache using Redis.
/// It supports standalone, sentinel, and cluster modes.
/// 单机用 ConnectionManager；哨兵模式发现 master 且 failover 后自动重新
/// 发现；集群自动切换 ClusterConnection（显式 `mode(Cluster)` 或默认模式
/// 下探测到 `cluster_enabled`）。
#[derive(Clone)]
pub struct RedisBackend {
    client: Arc<Client>,
    mode: RedisMode,
    connection: RedisConnection,
    dangerous_clear_enabled: bool,
    /// Maximum retry attempts for recoverable operations.
    retry_count: u32,
    /// Base delay between retries (exponential backoff).
    retry_delay: Duration,
    /// Circuit breaker for cascading failure protection.
    circuit_breaker: Arc<CircuitBreaker>,
    /// Lua 脚本命令响应超时（EVAL/EVALSHA/SCRIPT LOAD）。
    ///
    /// 历史教训：Lua 路径曾无任何超时——Redis 挂起时调用被无限期拖住，
    /// 且 oxcache 的 connection_timeout 只覆盖建连。`Duration::ZERO` =
    /// 不限时（对齐 limiteron `with_timeout(ZERO)` 惯用法）。
    /// lua feature 关闭时字段随 builder/from_parts 通路存在但无读取方。
    #[cfg_attr(not(feature = "lua"), allow(dead_code))]
    lua_timeout: Duration,
}

/// Lua 脚本命令默认响应超时（对齐 limiteron `EVAL_TIMEOUT` 口径）。
pub(crate) const DEFAULT_LUA_TIMEOUT: Duration = Duration::from_secs(5);

impl RedisBackend {
    /// Construct a `RedisBackend` from its constituent parts.
    ///
    /// Called by `RedisBackendBuilder::build()`; all fields are builder-fed,
    /// a params struct would just re-pack them one-to-one.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn from_parts(
        client: Arc<Client>,
        mode: RedisMode,
        connection: RedisConnection,
        dangerous_clear_enabled: bool,
        retry_count: u32,
        retry_delay: Duration,
        circuit_breaker: Arc<CircuitBreaker>,
        lua_timeout: Duration,
    ) -> Self {
        Self {
            client,
            mode,
            connection,
            dangerous_clear_enabled,
            retry_count,
            retry_delay,
            circuit_breaker,
            lua_timeout,
        }
    }

    /// Whether dangerous full-database `clear()` is enabled.
    pub(crate) fn dangerous_clear_enabled(&self) -> bool {
        self.dangerous_clear_enabled
    }

    /// Create a new Redis backend with connection string.
    pub async fn new(connection_string: &str) -> OxCacheResult<Self> {
        Self::builder()
            .connection_string(connection_string)
            .build()
            .await
    }

    /// Create a new Redis backend with connection pool.
    ///
    /// # Parameters
    ///
    /// - `connection_string`: Redis connection URL.
    /// - `_pool_size`: Currently unused. The underlying `redis` crate’s
    ///   `ConnectionManager` manages its own connection pool internally.
    ///   This parameter is retained for API compatibility and will be
    ///   wired through when a custom pool backend is introduced.
    pub async fn with_pool(connection_string: &str, _pool_size: usize) -> OxCacheResult<Self> {
        Self::builder()
            .connection_string(connection_string)
            .build()
            .await
    }

    /// Create a new Redis backend builder.
    pub fn builder() -> RedisBackendBuilder {
        RedisBackendBuilder::default()
    }

    /// Redact sensitive information from connection string for logging.
    ///
    /// # Example
    /// ```
    /// // pragma: allowlist secret
    /// use oxcache::backend::memory::RedisBackend;
    /// let conn_str = "redis://:secret_password@localhost:6379/0";
    /// let redacted = RedisBackend::redact_connection_string(conn_str);
    /// assert!(!redacted.contains("secret_password"));
    /// ```
    pub fn redact_connection_string(conn_str: &str) -> String {
        if let Some(start) = conn_str.find("://") {
            let protocol = &conn_str[..start + 3];
            let rest = &conn_str[start + 3..];

            // Check for userinfo (username[:password]@host). Use the last '@':
            // passwords may legally contain '@' (and '/'); splitting at the
            // first '@' would leave password fragments in the clear. A path
            // '@' without userinfo is not a valid Redis URL; over-redacting
            // it errs toward hiding rather than leaking.
            if let Some(at_pos) = rest.rfind('@') {
                // Found userinfo section - redact it
                return format!("{}[REDACTED]@{}", protocol, &rest[at_pos + 1..]);
            }
        }
        conn_str.to_string()
    }

    /// Get the Redis mode.
    pub fn mode(&self) -> RedisMode {
        self.mode
    }

    /// Get the Redis client.
    pub fn client(&self) -> &Client {
        &self.client
    }

    /// Get a cloned connection handle.
    ///
    /// ConnectionManager / ClusterConnection 内部均为 Arc 共享，clone 廉价。
    pub(crate) fn conn(&self) -> RedisConnection {
        self.connection.clone()
    }

    /// Lua 脚本命令响应超时。
    #[cfg_attr(not(feature = "lua"), allow(dead_code))]
    pub(crate) fn lua_timeout(&self) -> Duration {
        self.lua_timeout
    }

    /// 熔断器预检：Open 时返回 `Degraded`，Closed 放行。
    fn check_circuit(breaker: &CircuitBreaker) -> OxCacheResult<()> {
        if breaker.is_open() {
            return Err(OxCacheError::Degraded(
                "Redis circuit breaker is open".to_string(),
            ));
        }
        Ok(())
    }

    /// 执行结果记账：成功重置、失败计数，跳闸瞬间记录降级 metrics。
    fn record_outcome<T>(breaker: &CircuitBreaker, result: &OxCacheResult<T>) {
        match result {
            Ok(_) => breaker.record_success(),
            Err(_) => {
                if breaker.record_failure() {
                    // Circuit breaker just transitioned to Open
                    #[cfg(feature = "metrics")]
                    GLOBAL_UNIFIED_METRICS.record_l2_degraded();
                }
            }
        }
    }

    /// 单次执行 + 熔断器记账（**无重试**）。
    ///
    /// 供 Lua 脚本路径使用：脚本可有副作用，失败重试存在双应用风险，
    /// 故只取「Open 拒绝 + 成功/失败记账」，与 `execute_with_retry` 的
    /// 重试语义解耦。
    #[cfg_attr(not(feature = "lua"), allow(dead_code))]
    pub(crate) async fn execute_once_with_breaker<F, Fut, T>(
        &self,
        operation: F,
    ) -> OxCacheResult<T>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = OxCacheResult<T>>,
    {
        Self::check_circuit(&self.circuit_breaker)?;
        let result = operation().await;
        Self::record_outcome(&self.circuit_breaker, &result);
        result
    }

    /// Execute an operation with circuit breaker protection and retry.
    ///
    /// 熔断器 Open 时立即返回 `Degraded`；否则经 `retry_with_backoff` 执行并记录成功/失败。
    pub(crate) async fn execute_with_retry<F, Fut, T>(&self, operation: F) -> OxCacheResult<T>
    where
        F: Fn() -> Fut + Send + Sync,
        Fut: Future<Output = OxCacheResult<T>> + Send,
    {
        Self::check_circuit(&self.circuit_breaker)?;

        let result = retry_with_backoff(operation, self.retry_count, self.retry_delay).await;

        Self::record_outcome(&self.circuit_breaker, &result);

        result
    }

    /// Ping the Redis server.
    pub async fn ping(&self) -> OxCacheResult<String> {
        let mut conn = self.conn();
        let result: String = redis::cmd(RedisCommand::Ping.as_str())
            .query_async(&mut conn)
            .await
            .map_err(super::error::map_redis_error)?;
        Ok(result)
    }
}

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

    #[test]
    fn test_redact_connection_string_password_with_slash() {
        // 密码含 / 时不得绕过脱敏
        let redacted = RedisBackend::redact_connection_string("redis://user:p@ss/w0rd@host:6379"); /* pragma: allowlist secret */
        assert!(!redacted.contains("p@ss/w0rd"));
        assert!(!redacted.contains("w0rd"));
        assert!(redacted.contains("[REDACTED]"));
    }

    #[test]
    fn test_redact_connection_string_no_userinfo_unchanged() {
        let redacted = RedisBackend::redact_connection_string("redis://localhost:6379/0");
        assert_eq!(redacted, "redis://localhost:6379/0");
    }

    /// check_circuit：Closed 放行、Open 拒绝（Degraded）
    #[test]
    fn test_check_circuit_rejects_when_open() {
        let breaker = CircuitBreaker::new(1, Duration::from_secs(60));
        assert!(RedisBackend::check_circuit(&breaker).is_ok());

        breaker.record_failure(); // threshold=1，一次失败即跳闸
        assert!(breaker.is_open());
        let err = RedisBackend::check_circuit(&breaker).unwrap_err();
        assert!(matches!(err, OxCacheError::Degraded(_)));
    }

    /// record_outcome：成功/失败记账并按阈值联动跳闸
    #[test]
    fn test_record_outcome_counts_toward_trip() {
        let breaker = CircuitBreaker::new(2, Duration::from_secs(60));

        RedisBackend::record_outcome::<u8>(&breaker, &Ok(1));
        assert!(!breaker.is_open());

        RedisBackend::record_outcome::<u8>(
            &breaker,
            &Err(OxCacheError::Connection("x".to_string())),
        );
        assert!(!breaker.is_open());

        RedisBackend::record_outcome::<u8>(
            &breaker,
            &Err(OxCacheError::Connection("y".to_string())),
        );
        assert!(breaker.is_open());
    }
}