hydracache 0.40.0

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use hydracache_core::{CacheKeyBuilder, CacheOptions, TagSet};
use tokio::sync::oneshot;

use crate::tests::common::{user, LoaderError, User};
use crate::HydraCache;

#[tokio::test]
async fn typed_cache_puts_gets_and_namespaces_keys() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");
    let admins = cache.typed::<User>("admins");

    users.put("1", user(1), CacheOptions::new()).await.unwrap();
    admins.put("1", user(2), CacheOptions::new()).await.unwrap();

    assert_eq!(users.namespace(), "users");
    assert_eq!(users.key("1"), "users:1");
    assert_eq!(users.get("1").await.unwrap(), Some(user(1)));
    assert_eq!(admins.get("1").await.unwrap(), Some(user(2)));
    assert!(cache.contains_key("users:1").await);
    assert!(cache.contains_key("admins:1").await);
}

#[tokio::test]
async fn typed_cache_remove_contains_ttl_and_flush_delegate_to_shared_cache() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");
    let admins = cache.typed::<User>("admins");

    users
        .put(
            "1",
            user(1),
            CacheOptions::new().ttl(Duration::from_millis(20)),
        )
        .await
        .unwrap();
    admins.put("1", user(2), CacheOptions::new()).await.unwrap();

    assert!(users.contains_key("1").await);
    tokio::time::sleep(Duration::from_millis(40)).await;
    assert!(!users.contains_key("1").await);
    assert!(admins.remove("1").await.unwrap());
    assert!(!admins.invalidate_key("1").await.unwrap());

    users.put("2", user(2), CacheOptions::new()).await.unwrap();
    admins.put("2", user(3), CacheOptions::new()).await.unwrap();
    users.flush().await.unwrap();

    assert_eq!(users.get("2").await.unwrap(), None);
    assert_eq!(admins.get("2").await.unwrap(), None);
}

#[tokio::test]
async fn typed_cache_loader_helpers_and_errors_work() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");

    let infallible = users
        .get_or_insert_with("1", CacheOptions::new(), || async { user(1) })
        .await
        .unwrap();
    let fallible = users
        .try_get_or_insert_with("2", CacheOptions::new(), || async {
            Ok::<_, LoaderError>(user(2))
        })
        .await
        .unwrap();
    let error = users
        .try_get_or_insert_with("error", CacheOptions::new(), || async {
            Err::<User, _>(LoaderError)
        })
        .await;

    assert_eq!(infallible, user(1));
    assert_eq!(fallible, user(2));
    assert!(error.is_err());
    assert!(!users.contains_key("error").await);
}

#[tokio::test]
async fn typed_cache_diagnostics_delegate_to_shared_cache() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");

    users
        .get_or_insert_with("1", CacheOptions::new(), || async { user(1) })
        .await
        .unwrap();
    users
        .get_or_insert_with("1", CacheOptions::new(), || async { user(2) })
        .await
        .unwrap();

    let diagnostics = users.diagnostics().await;
    assert_eq!(diagnostics.stats.loads, 1);
    assert_eq!(diagnostics.stats.hits, 1);
    assert_eq!(diagnostics.total_requests(), 2);
    assert!(!diagnostics.is_empty());
}

#[tokio::test]
async fn typed_cache_single_flight_uses_namespaced_key() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");
    let calls = Arc::new(AtomicUsize::new(0));
    let mut tasks = Vec::new();

    for _ in 0..8 {
        let users = users.clone();
        let calls = calls.clone();
        tasks.push(tokio::spawn(async move {
            users
                .get_or_load("1", CacheOptions::new(), move || {
                    let calls = calls.clone();
                    async move {
                        calls.fetch_add(1, Ordering::SeqCst);
                        tokio::time::sleep(Duration::from_millis(10)).await;
                        Ok::<_, LoaderError>(user(1))
                    }
                })
                .await
                .unwrap()
        }));
    }

    for task in tasks {
        assert_eq!(task.await.unwrap(), user(1));
    }

    assert_eq!(calls.load(Ordering::SeqCst), 1);
    assert_eq!(users.stats().single_flight_joins, 7);
}

#[tokio::test]
async fn typed_cache_key_builder_and_tag_set_integrate_with_runtime() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");
    let physical_key = users.key_from(CacheKeyBuilder::new().tenant(7).entity("user", 42));
    let tags = TagSet::new().tag("users").tenant(7).entity("user", 42);

    cache
        .put(&physical_key, user(42), CacheOptions::new().tag_set(tags))
        .await
        .unwrap();

    assert_eq!(physical_key, "users:tenant:7:user:42");
    assert_eq!(cache.invalidate_tag("user:42").await.unwrap(), 1);
    let cached: Option<User> = cache.get(&physical_key).await.unwrap();
    assert_eq!(cached, None);
}

#[tokio::test]
async fn typed_cache_escapes_builder_segments_and_handles_empty_builder() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");

    let escaped = users.key_from(
        CacheKeyBuilder::new()
            .tenant("tenant:7")
            .entity("user:type", "42%beta"),
    );

    assert_eq!(users.key_from(CacheKeyBuilder::new()), "users");
    assert_eq!(escaped, "users:tenant:tenant%3A7:user%3Atype:42%25beta");
}

#[tokio::test]
async fn typed_cache_invalidation_during_load_discards_stale_store() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");
    let (started_tx, started_rx) = oneshot::channel();
    let (release_tx, release_rx) = oneshot::channel();
    let load_users = users.clone();

    let task = tokio::spawn(async move {
        load_users
            .get_or_load("1", CacheOptions::new().tag("users"), move || async move {
                started_tx.send(()).unwrap();
                release_rx.await.unwrap();
                Ok::<_, LoaderError>(user(1))
            })
            .await
            .unwrap()
    });

    started_rx.await.unwrap();
    assert_eq!(users.invalidate_tag("users").await.unwrap(), 0);
    release_tx.send(()).unwrap();

    assert_eq!(task.await.unwrap(), user(1));
    assert_eq!(users.get("1").await.unwrap(), None);
    assert_eq!(users.stats().stale_load_discards, 1);
}

#[tokio::test]
async fn typed_cache_post_invalidation_caller_starts_fresh_load() {
    let cache = HydraCache::local().build();
    let users = cache.typed::<User>("users");
    let calls = Arc::new(AtomicUsize::new(0));
    let (started_tx, started_rx) = oneshot::channel();
    let (release_tx, release_rx) = oneshot::channel();
    let stale_users = users.clone();
    let stale_calls = calls.clone();

    let stale_task = tokio::spawn(async move {
        stale_users
            .get_or_load("1", CacheOptions::new().tag("users"), move || async move {
                stale_calls.fetch_add(1, Ordering::SeqCst);
                started_tx.send(()).unwrap();
                release_rx.await.unwrap();
                Ok::<_, LoaderError>(user(1))
            })
            .await
            .unwrap()
    });

    started_rx.await.unwrap();
    assert_eq!(users.invalidate_tag("users").await.unwrap(), 0);

    let fresh_calls = calls.clone();
    let fresh = users
        .get_or_load("1", CacheOptions::new().tag("users"), move || async move {
            fresh_calls.fetch_add(1, Ordering::SeqCst);
            Ok::<_, LoaderError>(user(2))
        })
        .await
        .unwrap();

    release_tx.send(()).unwrap();

    assert_eq!(fresh, user(2));
    assert_eq!(stale_task.await.unwrap(), user(1));
    assert_eq!(calls.load(Ordering::SeqCst), 2);
    assert_eq!(users.get("1").await.unwrap(), Some(user(2)));
}