crank/crates/crank-runtime/tests/unit/cache.rs

use std::time::Duration;

use crank_core::{
    CacheBackend, CacheScope, CacheStoreError, CachedHeader, CachedResponse,
    CoordinationStateReservation, CoordinationStateStore, CoordinationStateValue,
    RateLimitBucketState, RateLimitStateStore, ReplayGuardStatus, ReplayGuardStore,
    ResponseCacheStore,
};
use crank_runtime::{
    InMemoryCoordinationStateStore, InMemoryRateLimitStateStore, InMemoryReplayGuardStore,
    InMemoryResponseCacheStore, RuntimeCacheConfig, RuntimeCacheConfigError,
    RuntimeCacheStoreInitError, RuntimeCacheStores,
};
use serde_json::json;

#[test]
fn defaults_to_in_memory_cache_without_url() {
    let config = RuntimeCacheConfig::default();

    assert_eq!(config.backend, CacheBackend::Memory);
    assert_eq!(config.url, None);
}

#[test]
fn accepts_validated_valkey_values() {
    let config = RuntimeCacheConfig::try_new(
        CacheBackend::Valkey,
        Some("redis://cache:6379/0".to_owned()),
    )
    .unwrap();

    assert_eq!(config.backend, CacheBackend::Valkey);
    assert_eq!(config.url.as_deref(), Some("redis://cache:6379/0"));
}

#[test]
fn rejects_external_backend_without_url() {
    let error = RuntimeCacheConfig::try_new(CacheBackend::Redis, None).unwrap_err();

    assert!(matches!(
        error,
        RuntimeCacheConfigError::MissingUrl {
            backend: CacheBackend::Redis
        }
    ));
}

#[test]
fn rejects_url_for_memory_backend() {
    let error =
        RuntimeCacheConfig::try_new(CacheBackend::Memory, Some("redis://cache:6379".to_owned()))
            .unwrap_err();
    assert!(matches!(error, RuntimeCacheConfigError::UnexpectedUrl));
}

#[tokio::test]
async fn in_memory_response_cache_roundtrips_values() {
    let store = InMemoryResponseCacheStore::default();
    let value = CachedResponse {
        status: 200,
        headers: vec![CachedHeader {
            name: "content-type".to_owned(),
            value: "application/json".to_owned(),
        }],
        body: br#"{"ok":true}"#.to_vec(),
        data: br#"{"ok":true}"#.to_vec(),
    };

    store
        .put("response:crm:list", value.clone(), Duration::from_secs(60))
        .await
        .unwrap();

    assert_eq!(
        store.get("response:crm:list").await.unwrap(),
        Some(value.clone())
    );

    store.delete("response:crm:list").await.unwrap();

    assert_eq!(store.get("response:crm:list").await.unwrap(), None);
}

#[tokio::test]
async fn in_memory_rate_limit_store_roundtrips_bucket_state() {
    let store = InMemoryRateLimitStateStore::default();
    let bucket = RateLimitBucketState {
        tokens_micros: 1_500_000,
        last_refill_unix_ms: 1_735_689_000_000,
    };

    store
        .put_bucket("workspace:alpha", bucket, Duration::from_secs(30))
        .await
        .unwrap();

    assert_eq!(
        store.get_bucket("workspace:alpha").await.unwrap(),
        Some(bucket)
    );

    store.delete_bucket("workspace:alpha").await.unwrap();

    assert_eq!(store.get_bucket("workspace:alpha").await.unwrap(), None);
}

#[tokio::test]
async fn in_memory_replay_guard_marks_key_only_once_until_cleared() {
    let store = InMemoryReplayGuardStore::default();

    assert_eq!(
        store
            .mark_seen("token:nonce:1", Duration::from_secs(10))
            .await
            .unwrap(),
        ReplayGuardStatus::Fresh
    );
    assert_eq!(
        store
            .mark_seen("token:nonce:1", Duration::from_secs(10))
            .await
            .unwrap(),
        ReplayGuardStatus::AlreadySeen
    );

    store.clear("token:nonce:1").await.unwrap();

    assert_eq!(
        store
            .mark_seen("token:nonce:1", Duration::from_secs(10))
            .await
            .unwrap(),
        ReplayGuardStatus::Fresh
    );
}

#[tokio::test]
async fn in_memory_coordination_store_scopes_keys() {
    let store = InMemoryCoordinationStateStore::default();
    let response_value = CoordinationStateValue {
        payload: json!({ "cursor": "abc" }),
    };
    let session_value = CoordinationStateValue {
        payload: json!({ "cursor": "xyz" }),
    };

    store
        .put_value(
            CacheScope::Response,
            "job-1",
            response_value.clone(),
            Duration::from_secs(20),
        )
        .await
        .unwrap();
    store
        .put_value(
            CacheScope::Coordination,
            "job-1",
            session_value.clone(),
            Duration::from_secs(20),
        )
        .await
        .unwrap();

    assert_eq!(
        store
            .get_value(CacheScope::Response, "job-1")
            .await
            .unwrap(),
        Some(response_value)
    );
    assert_eq!(
        store
            .get_value(CacheScope::Coordination, "job-1")
            .await
            .unwrap(),
        Some(session_value)
    );
}

#[tokio::test]
async fn in_memory_coordination_store_atomically_takes_and_reserves_values() {
    let store = InMemoryCoordinationStateStore::default();
    let value = CoordinationStateValue {
        payload: json!({ "state": "pending" }),
    };
    store
        .put_value(
            CacheScope::Coordination,
            "atomic-job",
            value.clone(),
            Duration::from_secs(20),
        )
        .await
        .unwrap();

    let (first, second) = tokio::join!(
        store.take_value(CacheScope::Coordination, "atomic-job"),
        store.take_value(CacheScope::Coordination, "atomic-job")
    );
    assert_eq!(
        usize::from(first.unwrap().is_some()) + usize::from(second.unwrap().is_some()),
        1
    );

    assert_eq!(
        store
            .reserve_value(
                CacheScope::Coordination,
                "reservation",
                value.clone(),
                Duration::from_secs(20),
            )
            .await
            .unwrap(),
        CoordinationStateReservation::Reserved
    );
    assert_eq!(
        store
            .reserve_value(
                CacheScope::Coordination,
                "reservation",
                CoordinationStateValue {
                    payload: json!({ "state": "other" }),
                },
                Duration::from_secs(20),
            )
            .await
            .unwrap(),
        CoordinationStateReservation::Existing(value.clone())
    );
    let completed = CoordinationStateValue {
        payload: json!({ "state": "completed" }),
    };
    assert!(
        store
            .compare_and_set_value(
                CacheScope::Coordination,
                "reservation",
                &value,
                completed.clone(),
                Duration::from_secs(20),
            )
            .await
            .unwrap()
    );
    assert_eq!(
        store
            .get_value(CacheScope::Coordination, "reservation")
            .await
            .unwrap(),
        Some(completed)
    );
}

#[tokio::test]
async fn in_memory_stores_reject_empty_keys() {
    let response_store = InMemoryResponseCacheStore::default();
    let replay_store = InMemoryReplayGuardStore::default();

    let error = response_store.get("").await.unwrap_err();
    assert!(matches!(error, CacheStoreError::InvalidKey { .. }));

    let error = replay_store
        .mark_seen("", Duration::from_secs(1))
        .await
        .unwrap_err();
    assert!(matches!(error, CacheStoreError::InvalidKey { .. }));
}

#[tokio::test]
async fn runtime_cache_stores_default_to_memory_backend() {
    let stores = RuntimeCacheStores::from_config(&RuntimeCacheConfig::default())
        .await
        .unwrap();

    assert_eq!(stores.backend, CacheBackend::Memory);

    stores
        .response
        .put(
            "response:health",
            CachedResponse {
                status: 200,
                headers: vec![],
                body: b"ok".to_vec(),
                data: br#"null"#.to_vec(),
            },
            Duration::from_secs(5),
        )
        .await
        .unwrap();

    assert!(
        stores
            .response
            .get("response:health")
            .await
            .unwrap()
            .is_some()
    );
}

#[test]
fn redis_cache_store_roundtrips_serialized_values() {
    let response = CachedResponse {
        status: 202,
        headers: vec![CachedHeader {
            name: "x-cache".to_owned(),
            value: "hit".to_owned(),
        }],
        body: br#"{"queued":true}"#.to_vec(),
        data: br#"null"#.to_vec(),
    };

    let encoded = serde_json::to_vec(&response).unwrap();
    let decoded: CachedResponse = serde_json::from_slice(&encoded).unwrap();

    assert_eq!(decoded, response);
}

#[test]
fn runtime_cache_stores_report_missing_external_url() {
    let future = RuntimeCacheStores::from_config(&RuntimeCacheConfig {
        backend: CacheBackend::Valkey,
        url: None,
    });

    let runtime = tokio::runtime::Runtime::new().unwrap();
    let error = match runtime.block_on(future) {
        Ok(_) => panic!("expected missing external cache url error"),
        Err(error) => error,
    };

    assert_eq!(
        error,
        RuntimeCacheStoreInitError::MissingUrl {
            backend: CacheBackend::Valkey
        }
    );
}