cachekit/backend/

redis.rs

use std::collections::HashMap;
use std::time::Duration;

use async_trait::async_trait;
use fred::interfaces::LuaInterface;
use fred::prelude::*;
use fred::types::ConnectHandle;

use crate::backend::{Backend, HealthStatus, LockableBackend, TtlInspectable};
use crate::error::{BackendError, BackendErrorKind};

// ── Error mapping ─────────────────────────────────────────────────────────────

/// Sanitize Redis error messages to avoid leaking connection URL passwords.
fn sanitize_redis_message(msg: &str) -> String {
    // Redis URLs may contain passwords: redis://:password@host:6379
    // Strip anything between :// and @ to remove embedded credentials.
    if let Some(proto_end) = msg.find("://") {
        if let Some(at_pos) = msg[proto_end..].find('@') {
            let mut sanitized = String::with_capacity(msg.len());
            sanitized.push_str(&msg[..proto_end + 3]);
            sanitized.push_str("[REDACTED]");
            sanitized.push_str(&msg[proto_end + at_pos..]);
            return sanitized;
        }
    }
    msg.to_string()
}

fn redis_err(e: RedisError) -> BackendError {
    let kind = match e.kind() {
        RedisErrorKind::Auth => BackendErrorKind::Authentication,
        RedisErrorKind::IO => BackendErrorKind::Transient,
        RedisErrorKind::Timeout => BackendErrorKind::Timeout,
        RedisErrorKind::Canceled => BackendErrorKind::Transient,
        _ => BackendErrorKind::Permanent,
    };
    BackendError {
        kind,
        message: sanitize_redis_message(&e.to_string()),
        source: Some(Box::new(e)),
    }
}

// ── RedisBackend ──────────────────────────────────────────────────────────────

/// Redis backend powered by the `fred` client.
///
/// Call [`connect`](RedisBackend::connect) before issuing any cache operations.
pub struct RedisBackend {
    client: RedisClient,
}

impl std::fmt::Debug for RedisBackend {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisBackend").finish_non_exhaustive()
    }
}

impl RedisBackend {
    /// Start building a [`RedisBackend`].
    pub fn builder() -> RedisBackendBuilder {
        RedisBackendBuilder::default()
    }

    /// Connect to Redis and wait until the connection is established.
    ///
    /// This must be called before any cache operations. The returned
    /// `ConnectHandle` drives the connection task — hold onto it for the
    /// lifetime of the backend (or join it on shutdown).
    pub async fn connect(&self) -> Result<ConnectHandle, BackendError> {
        self.client.init().await.map_err(redis_err)
    }
}

// ── Backend impl ──────────────────────────────────────────────────────────────

#[cfg(not(target_arch = "wasm32"))]
#[cfg_attr(not(feature = "unsync"), async_trait)]
#[cfg_attr(feature = "unsync", async_trait(?Send))]
impl Backend for RedisBackend {
    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
        let result: Option<bytes::Bytes> = self.client.get(key).await.map_err(redis_err)?;
        Ok(result.map(|b| b.to_vec()))
    }

    async fn set(
        &self,
        key: &str,
        value: Vec<u8>,
        ttl: Option<Duration>,
    ) -> Result<(), BackendError> {
        let expiration = ttl.map(|d| {
            let secs = i64::try_from(d.as_secs().max(1)).unwrap_or(i64::MAX);
            Expiration::EX(secs)
        });
        self.client
            .set::<(), _, _>(key, value.as_slice(), expiration, None, false)
            .await
            .map_err(redis_err)
    }

    async fn delete(&self, key: &str) -> Result<bool, BackendError> {
        let removed: i64 = self.client.del(key).await.map_err(redis_err)?;
        Ok(removed > 0)
    }

    async fn exists(&self, key: &str) -> Result<bool, BackendError> {
        let count: i64 = self.client.exists(key).await.map_err(redis_err)?;
        Ok(count > 0)
    }

    async fn health(&self) -> Result<HealthStatus, BackendError> {
        let start = std::time::Instant::now();
        let _pong: String = self.client.ping().await.map_err(redis_err)?;
        let latency = start.elapsed();

        let mut details = HashMap::new();
        details.insert("latency_ms".to_string(), latency.as_millis().to_string());
        Ok(HealthStatus {
            is_healthy: true,
            latency_ms: latency.as_secs_f64() * 1000.0,
            backend_type: "redis".to_string(),
            details,
        })
    }

    fn as_lockable(&self) -> Option<&dyn LockableBackend> {
        Some(self)
    }
}

// ── TtlInspectable impl ───────────────────────────────────────────────────────

#[cfg(not(target_arch = "wasm32"))]
#[cfg_attr(not(feature = "unsync"), async_trait)]
#[cfg_attr(feature = "unsync", async_trait(?Send))]
impl TtlInspectable for RedisBackend {
    async fn ttl(&self, key: &str) -> Result<Option<Duration>, BackendError> {
        // Redis TTL return values:
        //   -2 → key does not exist
        //   -1 → key exists but has no TTL
        //   N  → remaining seconds
        let secs: i64 = self.client.ttl(key).await.map_err(redis_err)?;
        match secs {
            ..0 => Ok(None),
            n => Ok(Some(Duration::from_secs(n.unsigned_abs()))),
        }
    }
}

// ── LockableBackend impl ──────────────────────────────────────────────────────

/// Callers pass the bare cache key; the backend derives the lock key on the
/// wire. `<key>:lock` matches cachekit-py's Redis lock namespace so py and rs
/// workloads contend on the same lock.
fn lock_key(key: &str) -> String {
    format!("{key}:lock")
}

/// Compare-and-delete: only the holder of the lock_id token may release.
/// Atomic via Lua — a GET-then-DEL from the client would race with expiry
/// and delete a lock some other client has since acquired.
const RELEASE_LOCK_SCRIPT: &str = r#"if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) else return 0 end"#;

#[cfg(not(target_arch = "wasm32"))]
#[cfg_attr(not(feature = "unsync"), async_trait)]
#[cfg_attr(feature = "unsync", async_trait(?Send))]
impl LockableBackend for RedisBackend {
    async fn acquire_lock(
        &self,
        key: &str,
        timeout_ms: u64,
    ) -> Result<Option<String>, BackendError> {
        let lock_id = uuid::Uuid::new_v4().to_string();
        // PX must be >= 1 (Redis rejects 0); clamp like Backend::set does for EX.
        let px = i64::try_from(timeout_ms.max(1)).unwrap_or(i64::MAX);

        // SET NX PX: returns OK when the lock was free, nil when contested.
        let result: Option<String> = self
            .client
            .set(
                lock_key(key),
                lock_id.as_str(),
                Some(Expiration::PX(px)),
                Some(SetOptions::NX),
                false,
            )
            .await
            .map_err(redis_err)?;

        Ok(result.map(|_| lock_id))
    }

    async fn release_lock(&self, key: &str, lock_id: &str) -> Result<bool, BackendError> {
        let deleted: i64 = self
            .client
            .eval(RELEASE_LOCK_SCRIPT, vec![lock_key(key)], vec![lock_id])
            .await
            .map_err(redis_err)?;
        Ok(deleted == 1)
    }
}

// ── Builder ───────────────────────────────────────────────────────────────────

/// Per-command deadline. fred's default is 0 (no timeout), so a stalled or
/// black-holed server would hold every op until TCP gives up and the circuit
/// breaker could never open. 5 s matches cachekit-py and cachekit-ts.
pub(crate) const COMMAND_TIMEOUT: Duration = Duration::from_secs(5);

/// fred client options for [`RedisBackendBuilder::build`].
///
/// The unresponsive-connection check is enabled only with a reconnect policy:
/// fred closes a connection it deems unresponsive, and without a policy it is
/// never re-established, so one stall past the threshold would leave the
/// client failing permanently.
pub(crate) fn fred_options(reconnects: bool) -> (PerformanceConfig, ConnectionConfig) {
    let perf = PerformanceConfig {
        default_command_timeout: COMMAND_TIMEOUT,
        ..Default::default()
    };
    let mut connection = ConnectionConfig::default();
    if reconnects {
        connection.unresponsive.max_timeout = Some(COMMAND_TIMEOUT);
    }
    (perf, connection)
}

/// Builder for [`RedisBackend`].
#[derive(Default)]
#[must_use]
pub struct RedisBackendBuilder {
    url: Option<String>,
    reconnect: Option<ReconnectPolicy>,
}

impl RedisBackendBuilder {
    /// Set the Redis connection URL (required).
    ///
    /// Accepts standard Redis URL formats, e.g. `redis://localhost:6379` or
    /// `redis://:password@host:6379/0`.
    pub fn url(mut self, url: impl Into<String>) -> Self {
        self.url = Some(url.into());
        self
    }

    /// Automatically re-establish dropped connections with exponential
    /// backoff (100 ms → 30 s, retrying indefinitely).
    ///
    /// Initial connections still fail fast — fred's `fail_fast` default stays
    /// on, so the policy only governs reconnecting a connection that was
    /// previously established. Used by the reliability-first intent presets
    /// (`CacheKit::production`, `CacheKit::secure`).
    pub(crate) fn auto_reconnect(mut self) -> Self {
        self.reconnect = Some(ReconnectPolicy::new_exponential(0, 100, 30_000, 2));
        self
    }

    /// Consume the builder and construct a [`RedisBackend`].
    ///
    /// # Errors
    ///
    /// Returns an error if `url` was not provided or is not a valid Redis URL.
    /// The connection itself is lazy — call [`RedisBackend::connect`] to establish it.
    pub fn build(self) -> Result<RedisBackend, crate::error::CachekitError> {
        use crate::error::CachekitError;

        let url = self
            .url
            .filter(|u| !u.is_empty())
            .ok_or_else(|| CachekitError::Config("url is required".to_string()))?;

        let config = RedisConfig::from_url(&url).map_err(|e| {
            CachekitError::Config(format!(
                "invalid Redis URL: {}",
                sanitize_redis_message(&e.to_string())
            ))
        })?;

        let (perf, connection) = fred_options(self.reconnect.is_some());
        let client = RedisClient::new(config, Some(perf), Some(connection), self.reconnect);
        Ok(RedisBackend { client })
    }
}

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

    /// Compile-time proof that RedisBackend implements LockableBackend.
    fn _assert_lockable(_b: &dyn LockableBackend) {}

    #[test]
    fn redis_is_lockable() {
        fn _check(backend: &RedisBackend) {
            _assert_lockable(backend);
        }
    }

    #[test]
    fn lock_key_derives_py_compatible_namespace() {
        // Bare-key contract: callers never build ":lock" themselves; the
        // backend appends it, byte-identical to cachekit-py's Redis lock.
        assert_eq!(
            lock_key("ns:app:func:m.f:args:abc:v1"),
            "ns:app:func:m.f:args:abc:v1:lock"
        );
    }

    #[test]
    fn command_timeout_is_five_seconds_on_every_build() {
        assert_eq!(COMMAND_TIMEOUT, Duration::from_secs(5));
        for reconnects in [false, true] {
            let (perf, _) = fred_options(reconnects);
            assert_eq!(perf.default_command_timeout, COMMAND_TIMEOUT);
        }
    }

    #[test]
    fn unresponsive_recycling_only_with_reconnect_policy() {
        let (_, connection) = fred_options(false);
        assert_eq!(connection.unresponsive.max_timeout, None);
        let (_, connection) = fred_options(true);
        assert_eq!(connection.unresponsive.max_timeout, Some(COMMAND_TIMEOUT));
    }
}