#![cfg(feature = "cache-redis")]

use rs_zero::{
    cache::{CacheKey, CacheStore},
    cache_redis::{
        RedisBreakerConfig, RedisCacheConfig, RedisCacheError, RedisCacheStore,
        RedisDegradationEvent, RedisDegradedAction, RedisOperation, RedisUnavailablePolicy,
    },
    resil::BreakerState,
};
use std::time::Duration;

#[test]
fn default_policy_keeps_existing_fail_closed_behavior() {
    let config = RedisCacheConfig::default();
    let error = RedisCacheError::Connection("redis down".to_string());

    assert_eq!(
        config
            .unavailable_policy
            .action_for(RedisOperation::Get, &error),
        RedisDegradedAction::ReturnError
    );
    assert_eq!(
        config
            .unavailable_policy
            .action_for(RedisOperation::Set, &error),
        RedisDegradedAction::ReturnError
    );
    assert_eq!(
        config
            .unavailable_policy
            .action_for(RedisOperation::Delete, &error),
        RedisDegradedAction::ReturnError
    );
}

#[test]
fn fail_open_policy_can_return_miss_and_skip_writes() {
    let policy = RedisUnavailablePolicy::fail_open_for_cache();
    let error = RedisCacheError::Timeout("GET".to_string());

    assert_eq!(
        policy.action_for(RedisOperation::Get, &error),
        RedisDegradedAction::ReturnMiss
    );
    assert_eq!(
        policy.action_for(RedisOperation::Set, &error),
        RedisDegradedAction::SkipWrite
    );
    assert_eq!(
        policy.action_for(RedisOperation::Delete, &error),
        RedisDegradedAction::SkipWrite
    );
}

#[test]
fn redis_breaker_defaults_to_disabled_for_compatibility() {
    let config = RedisCacheConfig::default();

    assert!(!config.breaker.enabled);
}

#[tokio::test]
async fn redis_breaker_fail_closed_rejects_after_connection_failure() {
    let store = RedisCacheStore::new(RedisCacheConfig {
        url: "redis://127.0.0.1:1".to_string(),
        connect_timeout: Duration::from_millis(50),
        command_timeout: Duration::from_millis(50),
        breaker: RedisBreakerConfig::fast_failure(1, Duration::from_secs(60)),
        ..RedisCacheConfig::default()
    })
    .expect("store");
    let key = CacheKey::new(store.namespace(), ["breaker", "closed"]);

    let first = store
        .get_raw(&key)
        .await
        .expect_err("first call should fail closed");
    assert!(first.to_string().contains("redis"));
    assert_eq!(
        store.breaker_snapshot().await.expect("snapshot").state,
        BreakerState::Open
    );

    let second = store
        .get_raw(&key)
        .await
        .expect_err("open breaker should reject");
    assert!(second.to_string().contains("circuit breaker"));
}

#[tokio::test]
async fn redis_breaker_fail_open_degrades_rejections_without_touching_caller() {
    let store = RedisCacheStore::new(RedisCacheConfig {
        url: "redis://127.0.0.1:1".to_string(),
        connect_timeout: Duration::from_millis(50),
        command_timeout: Duration::from_millis(50),
        unavailable_policy: RedisUnavailablePolicy::fail_open_for_cache(),
        breaker: RedisBreakerConfig::fast_failure(1, Duration::from_secs(60)),
        ..RedisCacheConfig::default()
    })
    .expect("store");
    let key = CacheKey::new(store.namespace(), ["breaker", "open"]);

    assert!(store.get_raw(&key).await.expect("degraded miss").is_none());
    assert_eq!(
        store.breaker_snapshot().await.expect("snapshot").state,
        BreakerState::Open
    );
    assert!(store.get_raw(&key).await.expect("rejected miss").is_none());
    store
        .set_raw(&key, b"skipped".to_vec(), Some(Duration::from_secs(1)))
        .await
        .expect("rejected set is skipped");
    store
        .delete(&key)
        .await
        .expect("rejected delete is skipped");
}

#[test]
fn degradation_event_labels_do_not_include_keys_or_urls() {
    let event = RedisDegradationEvent::new(RedisOperation::Get, RedisDegradedAction::ReturnMiss);

    assert_eq!(event.operation.as_str(), "get");
    assert_eq!(event.action.as_str(), "return_miss");
    assert!(!format!("{event:?}").contains("redis://"));
    assert!(!format!("{event:?}").contains("users:42"));
}

#[cfg(feature = "observability")]
#[test]
fn degradation_metrics_are_exported_without_sensitive_values() {
    let registry = rs_zero::observability::MetricsRegistry::new();
    registry.record_redis_degradation(rs_zero::observability::RedisDegradationLabels::new(
        RedisOperation::Get.as_str(),
        RedisDegradedAction::ReturnMiss.as_str(),
        "primary",
    ));

    let text = registry.render_prometheus();

    assert!(text.contains("rs_zero_redis_degradations_total"));
    assert!(text.contains("operation=\"get\",action=\"return_miss\",shard=\"primary\""));
    assert!(!text.contains("redis://"));
    assert!(!text.contains("users:42"));
    assert!(!text.contains("password"));
}
