fraiseql-core 2.19.0

//! Tests for the `cache` module.

#![allow(clippy::unwrap_used, clippy::print_stdout, clippy::print_stderr)] // Reason: test code, panics are acceptable

mod config_tests {
    use crate::cache::*;

    #[test]
    fn test_default_config() {
        let config = CacheConfig::default();
        assert!(!config.enabled); // Disabled by default as of rc.12
        assert_eq!(config.max_entries, 10_000);
        assert_eq!(config.ttl_seconds, 86_400);
        assert!(config.cache_list_queries);
    }

    #[test]
    fn test_with_max_entries() {
        let config = CacheConfig::with_max_entries(50_000);
        assert_eq!(config.max_entries, 50_000);
        assert!(!config.enabled); // Disabled by default as of rc.12
        assert_eq!(config.ttl_seconds, 86_400);
    }

    #[test]
    fn test_with_ttl() {
        let config = CacheConfig::with_ttl(3_600);
        assert_eq!(config.ttl_seconds, 3_600);
        assert!(!config.enabled); // Disabled by default as of rc.12
        assert_eq!(config.max_entries, 10_000);
    }

    #[test]
    fn test_enabled() {
        let config = CacheConfig::enabled();
        assert!(config.enabled);
        assert_eq!(config.max_entries, 10_000);
        assert_eq!(config.ttl_seconds, 86_400);
    }

    #[test]
    fn test_disabled() {
        let config = CacheConfig::disabled();
        assert!(!config.enabled);
    }

    #[test]
    fn test_estimated_memory() {
        let config = CacheConfig::with_max_entries(10_000);
        let estimated = config.estimated_memory_bytes();
        // Should be roughly 100 MB (10,000 * 10 KB)
        assert_eq!(estimated, 100_000_000);
    }

    #[test]
    fn test_from_bool_true() {
        let config = CacheConfig::from(true);
        assert!(config.enabled);
        assert_eq!(config.max_entries, 10_000);
        assert_eq!(config.ttl_seconds, 86_400);
    }

    #[test]
    fn test_from_bool_false() {
        let config = CacheConfig::from(false);
        assert!(!config.enabled);
    }

    #[test]
    fn test_serialization() {
        let config = CacheConfig::default();
        let json = serde_json::to_string(&config).unwrap();
        let deserialized: CacheConfig = serde_json::from_str(&json).unwrap();

        assert_eq!(config.enabled, deserialized.enabled);
        assert_eq!(config.max_entries, deserialized.max_entries);
        assert_eq!(config.ttl_seconds, deserialized.ttl_seconds);
    }
}

mod principal_hash_tests {
    use crate::{cache::hash_security_context, security::SecurityContext};

    // ========================================================================
    // Security Context Hash Tests
    // ========================================================================

    #[test]
    fn test_hash_security_context_none_returns_zero() {
        assert_eq!(hash_security_context(None), 0);
    }

    #[test]
    fn test_hash_security_context_same_context_same_hash() {
        let ctx = make_security_context("alice", &["admin"], Some("tenant-1"), &["read:user"]);
        let hash1 = hash_security_context(Some(&ctx));
        let hash2 = hash_security_context(Some(&ctx));
        assert_eq!(hash1, hash2, "Same context must produce same hash");
    }

    #[test]
    fn test_hash_security_context_different_user_different_hash() {
        let alice = make_security_context("alice", &["admin"], Some("tenant-1"), &[]);
        let bob = make_security_context("bob", &["admin"], Some("tenant-1"), &[]);

        assert_ne!(
            hash_security_context(Some(&alice)),
            hash_security_context(Some(&bob)),
            "Different user_id must produce different hash"
        );
    }

    #[test]
    fn test_hash_security_context_different_roles_different_hash() {
        let admin = make_security_context("alice", &["admin"], None, &[]);
        let viewer = make_security_context("alice", &["viewer"], None, &[]);

        assert_ne!(
            hash_security_context(Some(&admin)),
            hash_security_context(Some(&viewer)),
            "Different roles must produce different hash"
        );
    }

    #[test]
    fn test_hash_security_context_role_order_independent() {
        let ctx1 = make_security_context("alice", &["admin", "viewer"], None, &[]);
        let ctx2 = make_security_context("alice", &["viewer", "admin"], None, &[]);

        assert_eq!(
            hash_security_context(Some(&ctx1)),
            hash_security_context(Some(&ctx2)),
            "Role order must not affect hash (sorted internally)"
        );
    }

    #[test]
    fn test_hash_security_context_different_tenant_different_hash() {
        let t1 = make_security_context("alice", &[], Some("tenant-1"), &[]);
        let t2 = make_security_context("alice", &[], Some("tenant-2"), &[]);
        let none = make_security_context("alice", &[], None, &[]);

        assert_ne!(hash_security_context(Some(&t1)), hash_security_context(Some(&t2)),);
        assert_ne!(hash_security_context(Some(&t1)), hash_security_context(Some(&none)),);
    }

    #[test]
    fn test_hash_security_context_different_scopes_different_hash() {
        let read = make_security_context("alice", &[], None, &["read:user"]);
        let write = make_security_context("alice", &[], None, &["write:user"]);
        let both = make_security_context("alice", &[], None, &["read:user", "write:user"]);

        assert_ne!(hash_security_context(Some(&read)), hash_security_context(Some(&write)),);
        assert_ne!(hash_security_context(Some(&read)), hash_security_context(Some(&both)),);
    }

    #[test]
    fn test_hash_security_context_scope_order_independent() {
        let ctx1 = make_security_context("alice", &[], None, &["read:user", "write:post"]);
        let ctx2 = make_security_context("alice", &[], None, &["write:post", "read:user"]);

        assert_eq!(
            hash_security_context(Some(&ctx1)),
            hash_security_context(Some(&ctx2)),
            "Scope order must not affect hash (sorted internally)"
        );
    }

    #[test]
    fn test_hash_security_context_different_attributes_different_hash() {
        let mut ctx1 = make_security_context("alice", &["admin"], None, &[]);
        ctx1.attributes
            .insert("department".to_string(), serde_json::json!("engineering"));

        let mut ctx2 = make_security_context("alice", &["admin"], None, &[]);
        ctx2.attributes.insert("department".to_string(), serde_json::json!("sales"));

        let ctx_no_attrs = make_security_context("alice", &["admin"], None, &[]);

        assert_ne!(
            hash_security_context(Some(&ctx1)),
            hash_security_context(Some(&ctx2)),
            "Different attribute values must produce different hashes"
        );
        assert_ne!(
            hash_security_context(Some(&ctx1)),
            hash_security_context(Some(&ctx_no_attrs)),
            "Attributes vs no attributes must produce different hashes"
        );
    }

    // ========================================================================
    // Helper: SecurityContext builder for tests
    // ========================================================================

    fn make_security_context(
        user_id: &str,
        roles: &[&str],
        tenant_id: Option<&str>,
        scopes: &[&str],
    ) -> SecurityContext {
        use chrono::Utc;
        SecurityContext {
            user_id:          user_id.into(),
            roles:            roles.iter().map(|s| (*s).to_string()).collect(),
            tenant_id:        tenant_id.map(Into::into),
            scopes:           scopes.iter().map(|s| (*s).to_string()).collect(),
            attributes:       std::collections::HashMap::new(),
            request_id:       "test-request".to_string(),
            ip_address:       None,
            authenticated_at: Utc::now(),
            expires_at:       Utc::now() + chrono::Duration::hours(1),
            issuer:           None,
            audience:         None,
            email:            None,
            display_name:     None,
        }
    }
}

mod cascade_metadata_tests {
    use crate::cache::*;

    #[test]
    fn test_build_from_mutations() {
        let mut metadata = CascadeMetadata::new();
        metadata.add_mutation("createUser", "User");
        metadata.add_mutation("updateUser", "User");
        metadata.add_mutation("deleteUser", "User");

        assert_eq!(metadata.count(), 3);
    }

    #[test]
    fn test_map_mutation_to_entity_type() {
        let mut metadata = CascadeMetadata::new();
        metadata.add_mutation("createUser", "User");
        metadata.add_mutation("createPost", "Post");

        assert_eq!(metadata.get_entity_type("createUser"), Some("User"));
        assert_eq!(metadata.get_entity_type("createPost"), Some("Post"));
    }

    #[test]
    fn test_handle_unknown_mutation() {
        let metadata = CascadeMetadata::new();
        assert_eq!(metadata.get_entity_type("unknownMutation"), None);
    }

    #[test]
    fn test_multiple_mutations_same_entity() {
        let mut metadata = CascadeMetadata::new();
        metadata.add_mutation("createUser", "User");
        metadata.add_mutation("updateUser", "User");
        metadata.add_mutation("deleteUser", "User");

        let mutations = metadata.get_mutations_for_entity("User");
        assert_eq!(mutations.len(), 3);
        assert!(mutations.contains(&"createUser".to_string()));
        assert!(mutations.contains(&"updateUser".to_string()));
        assert!(mutations.contains(&"deleteUser".to_string()));
    }

    #[test]
    fn test_contains_mutation() {
        let mut metadata = CascadeMetadata::new();
        metadata.add_mutation("createUser", "User");

        assert!(metadata.contains_mutation("createUser"));
        assert!(!metadata.contains_mutation("unknownMutation"));
    }
}

mod fact_table_version_tests {
    use std::time::Duration;

    use crate::cache::{
        fact_table_version::{CachedVersion, generate_version_key_component},
        *,
    };

    #[test]
    fn test_strategy_default_is_disabled() {
        let strategy = FactTableVersionStrategy::default();
        assert_eq!(strategy, FactTableVersionStrategy::Disabled);
        assert!(!strategy.is_caching_enabled());
    }

    #[test]
    fn test_strategy_time_based() {
        let strategy = FactTableVersionStrategy::time_based(300);
        assert!(strategy.is_caching_enabled());
        assert_eq!(strategy.ttl_seconds(), Some(300));
    }

    #[test]
    fn test_strategy_version_table() {
        let strategy = FactTableVersionStrategy::VersionTable;
        assert!(strategy.is_caching_enabled());
        assert_eq!(strategy.ttl_seconds(), None);
    }

    #[test]
    fn test_strategy_schema_version() {
        let strategy = FactTableVersionStrategy::SchemaVersion;
        assert!(strategy.is_caching_enabled());
        assert_eq!(strategy.ttl_seconds(), None);
    }

    #[test]
    fn test_config_default_strategy() {
        let config = FactTableCacheConfig::default();
        assert_eq!(config.get_strategy("tf_sales"), &FactTableVersionStrategy::Disabled);
    }

    #[test]
    fn test_config_per_table_strategy() {
        let mut config = FactTableCacheConfig::default();
        config.set_strategy("tf_sales", FactTableVersionStrategy::VersionTable);
        config.set_strategy(
            "tf_page_views",
            FactTableVersionStrategy::TimeBased { ttl_seconds: 300 },
        );

        assert_eq!(config.get_strategy("tf_sales"), &FactTableVersionStrategy::VersionTable);
        assert_eq!(
            config.get_strategy("tf_page_views"),
            &FactTableVersionStrategy::TimeBased { ttl_seconds: 300 }
        );
        // Unconfigured table uses default
        assert_eq!(config.get_strategy("tf_other"), &FactTableVersionStrategy::Disabled);
    }

    #[test]
    fn test_config_with_default() {
        let config = FactTableCacheConfig::with_default(FactTableVersionStrategy::SchemaVersion);
        assert_eq!(config.get_strategy("tf_any"), &FactTableVersionStrategy::SchemaVersion);
    }

    #[test]
    fn test_generate_version_key_disabled() {
        let key = generate_version_key_component(
            "tf_sales",
            &FactTableVersionStrategy::Disabled,
            Some(42),
            "1.0.0",
        );
        assert!(key.is_none());
    }

    #[test]
    fn test_generate_version_key_version_table() {
        let key = generate_version_key_component(
            "tf_sales",
            &FactTableVersionStrategy::VersionTable,
            Some(42),
            "1.0.0",
        );
        assert_eq!(key, Some("tv:42".to_string()));

        // No version available - should return None
        let key = generate_version_key_component(
            "tf_sales",
            &FactTableVersionStrategy::VersionTable,
            None,
            "1.0.0",
        );
        assert!(key.is_none());
    }

    #[test]
    fn test_generate_version_key_time_based() {
        let key = generate_version_key_component(
            "tf_sales",
            &FactTableVersionStrategy::TimeBased { ttl_seconds: 300 },
            None,
            "1.0.0",
        );
        assert!(key.is_some());
        assert!(key.unwrap().starts_with("tb:"));
    }

    /// #784: `time_based(0)` is constructible via the public API and serde config;
    /// the version bucket must not divide by it. Zero TTL means "nothing is ever
    /// fresh", which is `Disabled` — the component must be `None`, not a panic.
    #[test]
    fn test_generate_version_key_time_based_zero_ttl_is_disabled() {
        let key = generate_version_key_component(
            "tf_sales",
            &FactTableVersionStrategy::TimeBased { ttl_seconds: 0 },
            None,
            "1.0.0",
        );
        assert!(key.is_none());
    }

    #[test]
    fn test_generate_version_key_schema_version() {
        let key = generate_version_key_component(
            "tf_sales",
            &FactTableVersionStrategy::SchemaVersion,
            None,
            "1.0.0",
        );
        assert_eq!(key, Some("sv:1.0.0".to_string()));
    }

    #[test]
    fn test_version_provider_caching() {
        let provider = FactTableVersionProvider::new(Duration::from_secs(10));

        // Initially no cached version
        assert!(provider.get_cached_version("tf_sales").is_none());

        // Set version
        provider.set_cached_version("tf_sales", 42);
        assert_eq!(provider.get_cached_version("tf_sales"), Some(42));

        // Clear version
        provider.clear_cached_version("tf_sales");
        assert!(provider.get_cached_version("tf_sales").is_none());
    }

    #[test]
    fn test_version_provider_clear_all() {
        let provider = FactTableVersionProvider::new(Duration::from_secs(10));

        provider.set_cached_version("tf_sales", 1);
        provider.set_cached_version("tf_orders", 2);

        provider.clear_all();

        assert!(provider.get_cached_version("tf_sales").is_none());
        assert!(provider.get_cached_version("tf_orders").is_none());
    }

    #[test]
    fn test_cached_version_freshness() {
        let cached = CachedVersion::new(42);

        // Should be fresh immediately
        assert!(cached.is_fresh(Duration::from_secs(1)));

        // Should be fresh for a longer duration
        assert!(cached.is_fresh(Duration::from_mins(1)));
    }

    #[test]
    fn test_strategy_serialization() {
        let strategies = vec![
            FactTableVersionStrategy::Disabled,
            FactTableVersionStrategy::VersionTable,
            FactTableVersionStrategy::TimeBased { ttl_seconds: 300 },
            FactTableVersionStrategy::SchemaVersion,
        ];

        for strategy in strategies {
            let json = serde_json::to_string(&strategy).unwrap();
            let deserialized: FactTableVersionStrategy = serde_json::from_str(&json).unwrap();
            assert_eq!(strategy, deserialized);
        }
    }

    #[test]
    fn test_config_serialization() {
        let mut config =
            FactTableCacheConfig::with_default(FactTableVersionStrategy::SchemaVersion);
        config.set_strategy("tf_sales", FactTableVersionStrategy::VersionTable);
        config.set_strategy("tf_events", FactTableVersionStrategy::TimeBased { ttl_seconds: 60 });

        let json = serde_json::to_string_pretty(&config).unwrap();
        let deserialized: FactTableCacheConfig = serde_json::from_str(&json).unwrap();

        assert_eq!(deserialized.default_strategy, FactTableVersionStrategy::SchemaVersion);
        assert_eq!(deserialized.get_strategy("tf_sales"), &FactTableVersionStrategy::VersionTable);
    }
}

mod entity_key_tests {
    use crate::{cache::*, error::FraiseQLError};

    #[test]
    fn test_create_valid_entity_key() {
        let key = EntityKey::new("User", "550e8400-e29b-41d4-a716-446655440000").unwrap();
        assert_eq!(key.entity_type, "User");
        assert_eq!(key.entity_id, "550e8400-e29b-41d4-a716-446655440000");
    }

    #[test]
    fn test_reject_empty_entity_type() {
        let result = EntityKey::new("", "550e8400-e29b-41d4-a716-446655440000");
        assert!(
            matches!(result, Err(FraiseQLError::Validation { .. })),
            "expected Validation error for empty entity_type, got: {result:?}"
        );
    }

    #[test]
    fn test_reject_empty_entity_id() {
        let result = EntityKey::new("User", "");
        assert!(
            matches!(result, Err(FraiseQLError::Validation { .. })),
            "expected Validation error for empty entity_id, got: {result:?}"
        );
    }

    #[test]
    fn test_serialize_to_cache_key_format() {
        let key = EntityKey::new("User", "550e8400-e29b-41d4-a716-446655440000").unwrap();
        let cache_key = key.to_cache_key();
        assert_eq!(cache_key, "User:550e8400-e29b-41d4-a716-446655440000");
    }

    #[test]
    fn test_deserialize_from_cache_key_format() {
        let cache_key = "User:550e8400-e29b-41d4-a716-446655440000";
        let key = EntityKey::from_cache_key(cache_key).unwrap();
        assert_eq!(key.entity_type, "User");
        assert_eq!(key.entity_id, "550e8400-e29b-41d4-a716-446655440000");
    }

    #[test]
    fn test_reject_colon_in_entity_type() {
        let result = EntityKey::new("User:Admin", "550e8400-e29b-41d4-a716-446655440000");
        assert!(result.is_err(), "colon in entity_type must be rejected");
        let msg = result.unwrap_err().to_string();
        assert!(
            msg.contains("colon") || msg.contains("separator"),
            "error should mention the separator: {msg}"
        );
    }

    #[test]
    fn test_reject_colon_only_in_entity_type() {
        let result = EntityKey::new(":", "some-id");
        assert!(
            matches!(result, Err(FraiseQLError::Validation { .. })),
            "expected Validation error for colon-only entity_type, got: {result:?}"
        );
    }

    #[test]
    fn test_entity_id_may_contain_colon() {
        // Entity IDs can contain colons (e.g. URNs) — only entity_type is restricted.
        let result = EntityKey::new("User", "urn:uuid:550e8400-e29b-41d4-a716-446655440000");
        assert!(result.is_ok(), "colon in entity_id must be accepted");
        // from_cache_key uses splitn(2, ':'), so it should reconstruct correctly.
        let key = result.unwrap();
        let cache_key = key.to_cache_key();
        let parsed = EntityKey::from_cache_key(&cache_key).unwrap();
        assert_eq!(parsed.entity_type, "User");
        assert_eq!(parsed.entity_id, "urn:uuid:550e8400-e29b-41d4-a716-446655440000");
    }

    #[test]
    fn test_hash_consistency_for_hashmap() {
        use std::collections::HashMap;

        let key1 = EntityKey::new("User", "550e8400-e29b-41d4-a716-446655440000").unwrap();
        let key2 = EntityKey::new("User", "550e8400-e29b-41d4-a716-446655440000").unwrap();

        let mut map = HashMap::new();
        map.insert(key1, "value1");

        // Same key should retrieve same value
        assert_eq!(map.get(&key2), Some(&"value1"));

        // Different key should not match
        let key3 = EntityKey::new("User", "550e8400-e29b-41d4-a716-446655440001").unwrap();
        assert_eq!(map.get(&key3), None);
    }
}

mod key_tests {
    use std::collections::{HashMap, HashSet};

    use indexmap::IndexMap;
    use serde_json::{Value as JsonValue, json};

    use crate::{
        backend::{WhereOperator, where_clause::WhereClause},
        cache::{key::verify_deterministic, *},
        schema::{CursorType, QueryDefinition},
    };

    // ========================================================================
    // Security Tests (CRITICAL)
    // ========================================================================

    #[test]
    fn test_different_variables_produce_different_keys() {
        // SECURITY CRITICAL: Different variables MUST produce different keys
        // to prevent User A from seeing User B's cached data
        let query = "query getUser($id: ID!) { user(id: $id) { name email } }";

        let key_alice = generate_cache_key(query, &json!({"id": "alice"}), None, "v1");
        let key_bob = generate_cache_key(query, &json!({"id": "bob"}), None, "v1");

        assert_ne!(
            key_alice, key_bob,
            "SECURITY: Different variables MUST produce different cache keys"
        );
    }

    #[test]
    fn test_different_variable_values_produce_different_keys() {
        let query = "query getUsers($limit: Int!) { users(limit: $limit) { id } }";

        let key_10 = generate_cache_key(query, &json!({"limit": 10}), None, "v1");
        let key_20 = generate_cache_key(query, &json!({"limit": 20}), None, "v1");

        assert_ne!(
            key_10, key_20,
            "SECURITY: Different variable values MUST produce different keys"
        );
    }

    #[test]
    fn test_empty_vs_non_empty_variables() {
        let query = "query { users { id } }";

        let key_empty = generate_cache_key(query, &json!({}), None, "v1");
        let key_with_vars = generate_cache_key(query, &json!({"limit": 10}), None, "v1");

        assert_ne!(
            key_empty, key_with_vars,
            "Empty variables must produce different key than non-empty"
        );
    }

    #[test]
    fn test_variable_order_independence() {
        // Object keys are sorted before hashing, so insertion order should
        // not affect the result. serde_json's default Map is BTreeMap (sorted),
        // but we sort explicitly in hash_json_value to be safe regardless.
        let query = "query($a: Int, $b: Int) { users { id } }";

        let key1 = generate_cache_key(query, &json!({"a": 1, "b": 2}), None, "v1");
        let key2 = generate_cache_key(query, &json!({"a": 1, "b": 2}), None, "v1");

        assert_eq!(key1, key2, "Same variables must produce same key");
    }

    // ========================================================================
    // Determinism Tests
    // ========================================================================

    #[test]
    fn test_cache_key_deterministic() {
        // Same inputs must always produce same output
        let query = "query { users { id } }";
        let vars = json!({"limit": 10});

        let key1 = generate_cache_key(query, &vars, None, "v1");
        let key2 = generate_cache_key(query, &vars, None, "v1");

        assert_eq!(key1, key2, "Cache keys must be deterministic");
    }

    #[test]
    fn test_verify_deterministic_helper() {
        assert!(
            verify_deterministic("query { users }", &json!({}), "v1"),
            "Helper should verify determinism"
        );
    }

    // ========================================================================
    // WHERE Clause Tests
    // ========================================================================

    #[test]
    fn test_different_where_clauses_produce_different_keys() {
        let query = "query { users { id } }";

        let where1 = WhereClause::Field {
            path:     vec!["email".to_string()],
            operator: WhereOperator::Eq,
            value:    json!("alice@example.com"),
        };

        let where2 = WhereClause::Field {
            path:     vec!["email".to_string()],
            operator: WhereOperator::Eq,
            value:    json!("bob@example.com"),
        };

        let key1 = generate_cache_key(query, &json!({}), Some(&where1), "v1");
        let key2 = generate_cache_key(query, &json!({}), Some(&where2), "v1");

        assert_ne!(key1, key2, "Different WHERE clauses must produce different keys");
    }

    #[test]
    fn test_different_where_operators_produce_different_keys() {
        let query = "query { users { id } }";

        let where_eq = WhereClause::Field {
            path:     vec!["age".to_string()],
            operator: WhereOperator::Eq,
            value:    json!(30),
        };

        let where_gt = WhereClause::Field {
            path:     vec!["age".to_string()],
            operator: WhereOperator::Gt,
            value:    json!(30),
        };

        let key_eq = generate_cache_key(query, &json!({}), Some(&where_eq), "v1");
        let key_gt = generate_cache_key(query, &json!({}), Some(&where_gt), "v1");

        assert_ne!(key_eq, key_gt, "Different operators must produce different keys");
    }

    #[test]
    fn test_with_and_without_where_clause() {
        let query = "query { users { id } }";

        let where_clause = WhereClause::Field {
            path:     vec!["active".to_string()],
            operator: WhereOperator::Eq,
            value:    json!(true),
        };

        let key_without = generate_cache_key(query, &json!({}), None, "v1");
        let key_with = generate_cache_key(query, &json!({}), Some(&where_clause), "v1");

        assert_ne!(key_without, key_with, "Presence of WHERE clause must change key");
    }

    #[test]
    fn test_complex_where_clause() {
        let query = "query { users { id } }";

        let where_clause = WhereClause::And(vec![
            WhereClause::Field {
                path:     vec!["age".to_string()],
                operator: WhereOperator::Gte,
                value:    json!(18),
            },
            WhereClause::Field {
                path:     vec!["active".to_string()],
                operator: WhereOperator::Eq,
                value:    json!(true),
            },
        ]);

        // Should not panic; produces a valid u64.
        let _key = generate_cache_key(query, &json!({}), Some(&where_clause), "v1");
    }

    // ========================================================================
    // Schema Version Tests
    // ========================================================================

    #[test]
    fn test_different_schema_versions_produce_different_keys() {
        let query = "query { users { id } }";

        let key_v1 = generate_cache_key(query, &json!({}), None, "v1");
        let key_v2 = generate_cache_key(query, &json!({}), None, "v2");

        assert_ne!(key_v1, key_v2, "Different schema versions must produce different keys");
    }

    #[test]
    fn test_schema_version_invalidation() {
        // When schema changes, all cache keys change (automatic invalidation)
        let query = "query { users { id } }";

        let old_schema = "abc123";
        let new_schema = "def456";

        let key_old = generate_cache_key(query, &json!({}), None, old_schema);
        let key_new = generate_cache_key(query, &json!({}), None, new_schema);

        assert_ne!(key_old, key_new, "Schema changes should invalidate cache");
    }

    // ========================================================================
    // Collision Avoidance Test
    // ========================================================================

    #[test]
    fn test_no_collisions_in_sample() {
        // Generate a sample of cache keys from varied inputs and verify
        // that no two distinct inputs produce the same u64.
        let mut keys = HashSet::new();
        let mut count = 0u32;

        let queries = [
            "query { users { id } }",
            "query { posts { id } }",
            "query { users { id name } }",
            "query getUser($id: ID!) { user(id: $id) { name } }",
            "",
        ];
        let variable_sets: &[JsonValue] = &[
            json!({}),
            json!(null),
            json!({"id": 1}),
            json!({"id": 2}),
            json!({"id": "alice"}),
            json!({"limit": 10, "offset": 0}),
            json!({"filter": {"active": true}}),
        ];
        let schema_versions = ["v1", "v2", "abc123"];

        for query in &queries {
            for vars in variable_sets {
                for sv in &schema_versions {
                    let key = generate_cache_key(query, vars, None, sv);
                    keys.insert(key);
                    count += 1;
                }
            }
        }

        assert_eq!(
            keys.len(),
            count as usize,
            "Collision detected among {count} sample cache keys"
        );
    }

    // ========================================================================
    // Extract Views Tests
    // ========================================================================

    #[test]
    fn test_extract_accessed_views_with_sql_source() {
        use crate::schema::AutoParams;

        let query_def = QueryDefinition {
            function: None,

            requires_actor:      Vec::new(),
            returns_count:       false,
            name:                "users".to_string(),
            return_type:         "User".to_string(),
            returns_list:        true,
            nullable:            false,
            arguments:           vec![],
            sql_source:          Some("v_user".to_string()),
            sql_source_kind:     crate::schema::SourceKind::Relation,
            description:         None,
            auto_params:         AutoParams {
                has_where:    true,
                has_order_by: false,
                has_limit:    true,
                has_offset:   false,
            },
            deprecation:         None,
            jsonb_column:        "data".to_string(),
            relay:               false,
            relay_cursor_column: None,
            relay_cursor_type:   CursorType::default(),
            inject_params:       IndexMap::default(),
            read_routing:        crate::backend::types::ReadRouting::default(),
            cache_ttl_seconds:   None,
            additional_views:    vec![],
            requires_role:       None,
            rest_path:           None,
            rest_method:         None,
            rest_stream:         false,
            native_columns:      HashMap::new(),
            pagination_order:    None,
        };

        let views = extract_accessed_views(&query_def);
        assert_eq!(views, vec!["v_user"]);
    }

    #[test]
    fn test_extract_accessed_views_without_sql_source() {
        use crate::schema::AutoParams;

        let query_def = QueryDefinition {
            function: None,

            requires_actor:      Vec::new(),
            returns_count:       false,
            name:                "customQuery".to_string(),
            return_type:         "Custom".to_string(),
            returns_list:        false,
            nullable:            false,
            arguments:           vec![],
            sql_source:          None, // No SQL source (custom resolver)
            sql_source_kind:     crate::schema::SourceKind::Relation,
            description:         None,
            auto_params:         AutoParams {
                has_where:    false,
                has_order_by: false,
                has_limit:    false,
                has_offset:   false,
            },
            deprecation:         None,
            jsonb_column:        "data".to_string(),
            relay:               false,
            relay_cursor_column: None,
            relay_cursor_type:   CursorType::default(),
            inject_params:       IndexMap::default(),
            read_routing:        crate::backend::types::ReadRouting::default(),
            cache_ttl_seconds:   None,
            additional_views:    vec![],
            requires_role:       None,
            rest_path:           None,
            rest_method:         None,
            rest_stream:         false,
            native_columns:      HashMap::new(),
            pagination_order:    None,
        };

        let views = extract_accessed_views(&query_def);
        assert_eq!(views, Vec::<String>::new());
    }

    #[test]
    fn test_extract_accessed_views_with_additional_views() {
        use crate::schema::AutoParams;

        let query_def = QueryDefinition {
            function: None,

            requires_actor:      Vec::new(),
            returns_count:       false,
            name:                "usersWithPosts".to_string(),
            return_type:         "UserWithPosts".to_string(),
            returns_list:        true,
            nullable:            false,
            arguments:           vec![],
            sql_source:          Some("v_user_with_posts".to_string()),
            sql_source_kind:     crate::schema::SourceKind::Relation,
            description:         None,
            auto_params:         AutoParams::default(),
            deprecation:         None,
            jsonb_column:        "data".to_string(),
            relay:               false,
            relay_cursor_column: None,
            relay_cursor_type:   CursorType::default(),
            inject_params:       IndexMap::default(),
            read_routing:        crate::backend::types::ReadRouting::default(),
            cache_ttl_seconds:   None,
            additional_views:    vec!["v_post".to_string(), "v_tag".to_string()],
            requires_role:       None,
            rest_path:           None,
            rest_method:         None,
            rest_stream:         false,
            native_columns:      HashMap::new(),
            pagination_order:    None,
        };

        let views = extract_accessed_views(&query_def);
        assert_eq!(views, vec!["v_user_with_posts", "v_post", "v_tag"]);
    }

    // ========================================================================
    // Edge Case Tests
    // ========================================================================

    #[test]
    fn test_empty_query_string() {
        // Should not panic; produces a valid u64.
        let _key = generate_cache_key("", &json!({}), None, "v1");
    }

    #[test]
    fn test_null_variables() {
        // Should not panic; produces a valid u64.
        let _key = generate_cache_key("query { users }", &json!(null), None, "v1");
    }

    #[test]
    fn test_large_variable_object() {
        let large_vars = json!({
            "filter": {
                "age": 30,
                "active": true,
                "tags": ["rust", "graphql", "database"],
                "metadata": {
                    "created_after": "2024-01-01",
                    "updated_before": "2024-12-31"
                }
            }
        });

        // Should not panic; produces a valid u64.
        let _key = generate_cache_key("query { users }", &large_vars, None, "v1");
    }

    #[test]
    fn test_special_characters_in_query() {
        let query = r#"query { user(email: "test@example.com") { name } }"#;
        // Should not panic; produces a valid u64.
        let _key = generate_cache_key(query, &json!({}), None, "v1");
    }

    // ========================================================================
    // ORDER BY Cache Key Tests
    // ========================================================================

    #[test]
    fn test_view_key_different_order_by_produces_different_keys() {
        use crate::backend::{OrderByClause, OrderDirection};

        let asc = [OrderByClause::new("name".into(), OrderDirection::Asc)];
        let desc = [OrderByClause::new("name".into(), OrderDirection::Desc)];

        let key_asc =
            generate_view_query_key("v_user", None, None, None, Some(&asc), &[], None, "v1");
        let key_desc =
            generate_view_query_key("v_user", None, None, None, Some(&desc), &[], None, "v1");

        assert_ne!(key_asc, key_desc, "Different order directions must produce different keys");
    }

    /// Ruling AH: a guard carries the caller's row predicate over a related row, so two
    /// filters that differ only in their guard — two callers who may read different related
    /// rows — must not share an entry.
    #[test]
    fn test_view_key_differs_by_the_guard_of_a_relation_filter() {
        use crate::db::{WhereClause, WhereOperator};

        let guarded = |tenant: &str| WhereClause::Guarded {
            under: vec!["team".to_string()],
            guard: Box::new(WhereClause::Field {
                path:     vec!["team".to_string(), "tenant_id".to_string()],
                operator: WhereOperator::Eq,
                value:    serde_json::json!(tenant),
            }),
            inner: Box::new(WhereClause::Field {
                path:     vec!["team".to_string(), "name".to_string()],
                operator: WhereOperator::Eq,
                value:    serde_json::json!("blue"),
            }),
        };
        let key = |tenant: &str| {
            generate_view_query_key(
                "v_member",
                Some(&guarded(tenant)),
                None,
                None,
                None,
                &[],
                None,
                "v1",
            )
        };
        assert_ne!(key("A"), key("B"), "callers with different guards must not share an entry");
        assert_eq!(key("A"), key("A"));
    }

    /// Ruling AL: a guard over the related view carries the caller's predicate too — two
    /// callers whose predicates differ must not share an entry.
    #[test]
    fn test_view_key_differs_by_the_view_guard_of_a_relation_filter() {
        use crate::db::{WhereClause, WhereOperator};

        let guarded = |tenant: &str| WhereClause::Guarded {
            under: vec!["team".to_string()],
            guard: Box::new(WhereClause::KeyIn {
                path:       vec!["fk_team".to_string()],
                key_type:   crate::db::types::sql_hints::ScalarFieldType::Integer,
                view:       "v_team".to_string(),
                target_key: vec!["id".to_string()],
                predicate:  Box::new(WhereClause::Field {
                    path:     vec!["tenant_id".to_string()],
                    operator: WhereOperator::Eq,
                    value:    serde_json::json!(tenant),
                }),
            }),
            inner: Box::new(WhereClause::Field {
                path:     vec!["team".to_string(), "name".to_string()],
                operator: WhereOperator::Eq,
                value:    serde_json::json!("blue"),
            }),
        };
        let key = |tenant: &str| {
            generate_view_query_key(
                "v_member",
                Some(&guarded(tenant)),
                None,
                None,
                None,
                &[],
                None,
                "v1",
            )
        };
        assert_ne!(key("A"), key("B"), "callers with different predicates must not share an entry");
        assert_eq!(key("A"), key("A"));
    }

    #[test]
    fn test_view_key_same_order_by_produces_same_key() {
        use crate::backend::{OrderByClause, OrderDirection};

        let clauses = [OrderByClause::new("createdAt".into(), OrderDirection::Desc)];

        let key1 =
            generate_view_query_key("v_user", None, None, None, Some(&clauses), &[], None, "v1");
        let key2 =
            generate_view_query_key("v_user", None, None, None, Some(&clauses), &[], None, "v1");

        assert_eq!(key1, key2, "Same order_by must produce identical keys");
    }

    #[test]
    fn test_view_key_with_and_without_order_by() {
        use crate::backend::{OrderByClause, OrderDirection};

        let clauses = [OrderByClause::new("name".into(), OrderDirection::Asc)];

        let key_with =
            generate_view_query_key("v_user", None, None, None, Some(&clauses), &[], None, "v1");
        let key_without =
            generate_view_query_key("v_user", None, None, None, None, &[], None, "v1");

        assert_ne!(key_with, key_without, "Presence of order_by must change key");
    }

    #[test]
    fn test_view_key_different_fields_produce_different_keys() {
        use crate::backend::{OrderByClause, OrderDirection};

        let by_name = [OrderByClause::new("name".into(), OrderDirection::Asc)];
        let by_date = [OrderByClause::new("createdAt".into(), OrderDirection::Asc)];

        let key_name =
            generate_view_query_key("v_user", None, None, None, Some(&by_name), &[], None, "v1");
        let key_date =
            generate_view_query_key("v_user", None, None, None, Some(&by_date), &[], None, "v1");

        assert_ne!(key_name, key_date, "Different order_by fields must produce different keys");
    }

    #[test]
    fn test_projection_key_includes_order_by() {
        use crate::backend::{OrderByClause, OrderDirection};

        let clauses = [OrderByClause::new("name".into(), OrderDirection::Asc)];

        let ordered = crate::backend::ProjectionRequest {
            order_by: Some(&clauses),
            ..crate::backend::ProjectionRequest::new("v_user")
        };
        let key_with = generate_projection_query_key(&ordered, &[], None, "v1");
        let key_without = generate_projection_query_key(
            &crate::backend::ProjectionRequest::new("v_user"),
            &[],
            None,
            "v1",
        );

        assert_ne!(key_with, key_without, "Projection key must include order_by");
    }

    /// #1314: a page carrying its match count is keyed apart from the same page without
    /// one, so a cached plain page is never served to a counting read.
    #[test]
    fn a_counting_read_is_keyed_apart_from_a_plain_one() {
        let plain = crate::backend::ProjectionRequest::new("v_doc");
        let counting = crate::backend::ProjectionRequest {
            matched_up_to: Some(10),
            ..crate::backend::ProjectionRequest::new("v_doc")
        };
        let key = |request| generate_projection_query_key(request, &[], None, "v1");
        assert_ne!(key(&plain), key(&counting));
    }

    /// #1512: the request locale is a section of its own. Identical session variables (a
    /// path that did not carry the locale as a setting) still key two locales apart, and
    /// "no locale" is a third key.
    #[test]
    fn the_request_locale_is_its_own_key_section() {
        let request = crate::backend::ProjectionRequest::new("v_product");
        let fr = generate_projection_query_key(&request, &[], Some("fr"), "v1");
        let de = generate_projection_query_key(&request, &[], Some("de-DE"), "v1");
        let none = generate_projection_query_key(&request, &[], None, "v1");
        assert_ne!(fr, de);
        assert_ne!(fr, none);
        let view = |locale| {
            generate_view_query_key("v_product", None, None, None, None, &[], locale, "v1")
        };
        assert_ne!(view(Some("fr")), view(Some("de-DE")));
        assert_ne!(view(Some("fr")), view(None));
    }
}

mod cascade_invalidator_tests {
    use crate::cache::*;

    #[test]
    fn test_add_single_dependency() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();

        assert!(invalidator.get_direct_dependencies("v_user").contains("v_raw_user"));
        assert!(invalidator.get_direct_dependents("v_raw_user").contains("v_user"));
    }

    #[test]
    fn test_self_dependency_fails() {
        let mut invalidator = CascadeInvalidator::new();
        let result = invalidator.add_dependency("v_user", "v_user");
        assert!(
            matches!(result, Err(crate::error::FraiseQLError::Validation { .. })),
            "expected Validation error for self-dependency, got: {result:?}"
        );
    }

    #[test]
    fn test_cascade_invalidate_single_level() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();

        let invalidated = invalidator.cascade_invalidate("v_raw_user").unwrap();
        assert_eq!(invalidated.len(), 2);
        assert!(invalidated.contains("v_raw_user"));
        assert!(invalidated.contains("v_user"));
    }

    #[test]
    fn test_cascade_invalidate_multiple_levels() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();
        invalidator.add_dependency("v_dashboard", "v_analytics").unwrap();

        let invalidated = invalidator.cascade_invalidate("v_raw_user").unwrap();
        assert_eq!(invalidated.len(), 4);
        assert!(invalidated.contains("v_raw_user"));
        assert!(invalidated.contains("v_user"));
        assert!(invalidated.contains("v_analytics"));
        assert!(invalidated.contains("v_dashboard"));
    }

    #[test]
    fn test_cascade_invalidate_branching() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_post", "v_raw_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();
        invalidator.add_dependency("v_dashboard", "v_post").unwrap();

        let invalidated = invalidator.cascade_invalidate("v_raw_user").unwrap();
        assert_eq!(invalidated.len(), 5);
        assert!(invalidated.contains("v_raw_user"));
        assert!(invalidated.contains("v_user"));
        assert!(invalidated.contains("v_post"));
        assert!(invalidated.contains("v_analytics"));
        assert!(invalidated.contains("v_dashboard"));
    }

    #[test]
    fn test_get_direct_dependents() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_post", "v_raw_user").unwrap();

        let dependents = invalidator.get_direct_dependents("v_raw_user");
        assert_eq!(dependents.len(), 2);
        assert!(dependents.contains("v_user"));
        assert!(dependents.contains("v_post"));
    }

    #[test]
    fn test_get_direct_dependencies() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_post").unwrap();

        let deps = invalidator.get_direct_dependencies("v_analytics");
        assert_eq!(deps.len(), 2);
        assert!(deps.contains("v_user"));
        assert!(deps.contains("v_post"));
    }

    #[test]
    fn test_get_transitive_dependents() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();
        invalidator.add_dependency("v_dashboard", "v_analytics").unwrap();

        let transitive = invalidator.get_transitive_dependents("v_raw_user");
        assert_eq!(transitive.len(), 4);
        assert!(transitive.contains("v_raw_user"));
        assert!(transitive.contains("v_user"));
        assert!(transitive.contains("v_analytics"));
        assert!(transitive.contains("v_dashboard"));
    }

    #[test]
    fn test_has_dependency_path() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();

        assert!(invalidator.has_dependency_path("v_analytics", "v_raw_user"));
        assert!(invalidator.has_dependency_path("v_analytics", "v_user"));
        assert!(invalidator.has_dependency_path("v_user", "v_raw_user"));
        assert!(!invalidator.has_dependency_path("v_raw_user", "v_analytics"));
        assert!(!invalidator.has_dependency_path("v_raw_user", "v_user"));
    }

    #[test]
    fn test_stats_tracking() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();

        invalidator.cascade_invalidate("v_raw_user").unwrap();
        invalidator.cascade_invalidate("v_user").unwrap();

        let stats = invalidator.stats();
        assert_eq!(stats.total_cascades, 2);
        assert_eq!(stats.total_invalidated, 5); // 3 (raw_user + user + analytics) + 2 (user + analytics)
        assert_eq!(stats.max_affected, 3);
    }

    #[test]
    fn test_clear() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        assert_eq!(invalidator.view_count(), 2);

        invalidator.clear();
        assert_eq!(invalidator.view_count(), 0);
        assert_eq!(invalidator.dependency_count(), 0);
    }

    #[test]
    fn test_view_and_dependency_count() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_post", "v_raw_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();

        assert_eq!(invalidator.view_count(), 4);
        assert_eq!(invalidator.dependency_count(), 3);
    }

    #[test]
    fn test_diamond_dependency() {
        let mut invalidator = CascadeInvalidator::new();
        // Diamond: raw → [user, post] → analytics
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_post", "v_raw_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_post").unwrap();

        let invalidated = invalidator.cascade_invalidate("v_raw_user").unwrap();
        // raw_user, user, post, analytics (4 total)
        assert_eq!(invalidated.len(), 4);
        assert!(invalidated.contains("v_raw_user"));
        assert!(invalidated.contains("v_user"));
        assert!(invalidated.contains("v_post"));
        assert!(invalidated.contains("v_analytics"));
    }

    #[test]
    fn test_multiple_independent_chains() {
        let mut invalidator = CascadeInvalidator::new();
        // Chain 1: raw1 → user1 → analytics1
        invalidator.add_dependency("v_user_1", "v_raw_1").unwrap();
        invalidator.add_dependency("v_analytics_1", "v_user_1").unwrap();
        // Chain 2: raw2 → user2 → analytics2
        invalidator.add_dependency("v_user_2", "v_raw_2").unwrap();
        invalidator.add_dependency("v_analytics_2", "v_user_2").unwrap();

        let invalidated = invalidator.cascade_invalidate("v_raw_1").unwrap();
        assert_eq!(invalidated.len(), 3); // Only chain 1
        assert!(!invalidated.contains("v_raw_2"));
        assert!(!invalidated.contains("v_user_2"));
    }

    #[test]
    fn test_cycle_detection_via_has_dependency_path() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();
        invalidator.add_dependency("v_analytics", "v_user").unwrap();

        // Verify no forward path from leaf to root
        assert!(!invalidator.has_dependency_path("v_raw_user", "v_analytics"));
    }

    #[test]
    fn a_cycle_is_accepted_and_invalidates_each_member_once() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("B", "A").unwrap(); // B depends on A
        invalidator.add_dependency("C", "B").unwrap(); // C depends on B
        // A depends on C closes A→C→B→A: views that read each other's rows.
        invalidator.add_dependency("A", "C").unwrap();

        for start in ["A", "B", "C"] {
            let invalidated = invalidator.cascade_invalidate(start).unwrap();
            assert_eq!(invalidated.len(), 3, "{start}: {invalidated:?}");
        }
    }

    #[test]
    fn a_self_dependency_is_rejected() {
        let mut invalidator = CascadeInvalidator::new();
        let result = invalidator.add_dependency("A", "A");
        assert!(
            matches!(result, Err(crate::error::FraiseQLError::Validation { .. })),
            "{result:?}"
        );
    }

    #[test]
    fn test_serialization() {
        let mut invalidator = CascadeInvalidator::new();
        invalidator.add_dependency("v_user", "v_raw_user").unwrap();

        let json = serde_json::to_string(&invalidator).expect("serialize should work");
        let restored: CascadeInvalidator =
            serde_json::from_str(&json).expect("deserialize should work");

        assert_eq!(
            restored.get_direct_dependents("v_raw_user"),
            invalidator.get_direct_dependents("v_raw_user")
        );
    }

    #[test]
    fn cascade_invalidator_is_send_sync() {
        // Invariant: CascadeInvalidator is read-only (graph data) after construction.
        // Stats use an internal Mutex for thread-safe writes.
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<CascadeInvalidator>();
    }

    #[test]
    fn cascade_invalidate_takes_shared_ref() {
        let mut inv = CascadeInvalidator::new();
        inv.add_dependency("v_b", "v_a").unwrap();
        // Should work with &inv (not &mut inv) — cascade_invalidate takes &self.
        let result = inv.cascade_invalidate("v_a").unwrap();
        assert!(result.contains("v_b"));
    }
}

mod result_tests {
    use std::sync::Arc;

    use serde_json::json;

    use crate::{cache::*, db::types::JsonbValue};

    // Helper to create test result
    fn test_result() -> Vec<JsonbValue> {
        vec![JsonbValue::new(json!({"id": 1, "name": "test"}))]
    }

    // ========================================================================
    // Cache Hit/Miss Tests
    // ========================================================================

    #[test]
    fn test_cache_miss() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        let result = cache.get(999_u64).unwrap();
        assert!(result.is_none(), "Should be cache miss");

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.misses, 1);
        assert_eq!(metrics.hits, 0);
    }

    #[test]
    fn test_cache_put_and_get() {
        let cache = QueryResultCache::new(CacheConfig::enabled());
        let result = test_result();

        // Put
        cache.put(1_u64, result, vec!["v_user".to_string()], None, None, None).unwrap();

        // Get
        let cached = cache.get(1_u64).unwrap();
        assert!(cached.is_some(), "Should be cache hit");
        assert_eq!(cached.unwrap().len(), 1);

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.hits, 1);
        assert_eq!(metrics.misses, 0);
        assert_eq!(metrics.total_cached, 1);
    }

    #[test]
    fn test_cache_hit_updates_hit_count() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        // First hit
        cache.get(1_u64).unwrap();
        // Second hit
        cache.get(1_u64).unwrap();

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.hits, 2);
    }

    // ========================================================================
    // TTL Expiry Tests
    // ========================================================================

    #[test]
    fn test_ttl_expiry() {
        let config = CacheConfig {
            ttl_seconds: 1,
            enabled: true,
            ..Default::default()
        };

        let cache = QueryResultCache::new(config);

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        // Wait for expiry
        std::thread::sleep(std::time::Duration::from_secs(2));
        cache.run_pending_tasks();

        // Should be expired
        let result = cache.get(1_u64).unwrap();
        assert!(result.is_none(), "Entry should be expired");

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.misses, 1); // Expired counts as miss
    }

    #[test]
    fn test_per_entry_ttl_override_expires_early() {
        // Global config has 1-hour TTL but entry overrides to 1 second
        let config = CacheConfig {
            ttl_seconds: 3600,
            enabled: true,
            ..Default::default()
        };
        let cache = QueryResultCache::new(config);

        cache
            .put(
                1_u64,
                test_result(),
                vec!["v_ref".to_string()],
                Some(1), // 1-second per-entry override
                None,
                None,
            )
            .unwrap();

        std::thread::sleep(std::time::Duration::from_secs(2));
        cache.run_pending_tasks();

        let result = cache.get(1_u64).unwrap();
        assert!(result.is_none(), "Entry with per-entry TTL=1s should have expired");
    }

    #[test]
    fn test_per_entry_ttl_zero_cached_indefinitely() {
        // TTL=0 = no time-based expiry; entry lives until mutation invalidation.
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(1_u64, test_result(), vec!["v_live".to_string()], Some(0), None, None)
            .unwrap();

        let result = cache.get(1_u64).unwrap();
        assert!(result.is_some(), "Entry with TTL=0 should be cached indefinitely");
    }

    #[test]
    fn test_ttl_not_expired() {
        let config = CacheConfig {
            ttl_seconds: 3600, // 1 hour TTL
            enabled: true,
            ..Default::default()
        };

        let cache = QueryResultCache::new(config);

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        // Should still be valid
        let result = cache.get(1_u64).unwrap();
        assert!(result.is_some(), "Entry should not be expired");
    }

    // ========================================================================
    // Eviction Tests (capacity-based)
    // ========================================================================

    #[test]
    fn test_capacity_eviction() {
        let config = CacheConfig {
            max_entries: 2,
            enabled: true,
            ..Default::default()
        };

        let cache = QueryResultCache::new(config);

        // Add 3 entries (max is 2); moka will evict one
        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();
        cache
            .put(2_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();
        cache
            .put(3_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        // Run pending tasks to flush evictions
        cache.run_pending_tasks();

        let metrics = cache.metrics().unwrap();
        assert!(metrics.size <= 2, "Cache size should not exceed max capacity");
    }

    // ========================================================================
    // Cache Disabled Tests
    // ========================================================================

    #[test]
    fn test_cache_disabled() {
        let config = CacheConfig::disabled();
        let cache = QueryResultCache::new(config);

        // Put should be no-op
        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        // Get should return None
        assert!(cache.get(1_u64).unwrap().is_none(), "Cache disabled should always miss");

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.total_cached, 0);
    }

    // ========================================================================
    // Invalidation Tests
    // ========================================================================

    #[test]
    fn test_invalidate_single_view() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();
        cache
            .put(2_u64, test_result(), vec!["v_post".to_string()], None, None, None)
            .unwrap();

        // Invalidate v_user
        cache.invalidate_views(&[ViewName::from("v_user")]).unwrap();

        // v_user entry gone, v_post remains
        assert!(cache.get(1_u64).unwrap().is_none());
        assert!(cache.get(2_u64).unwrap().is_some());
    }

    #[test]
    fn test_invalidate_multiple_views() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();
        cache
            .put(2_u64, test_result(), vec!["v_post".to_string()], None, None, None)
            .unwrap();
        cache
            .put(3_u64, test_result(), vec!["v_product".to_string()], None, None, None)
            .unwrap();

        // Invalidate v_user and v_post
        cache
            .invalidate_views(&[ViewName::from("v_user"), ViewName::from("v_post")])
            .unwrap();

        assert!(cache.get(1_u64).unwrap().is_none());
        assert!(cache.get(2_u64).unwrap().is_none());
        assert!(cache.get(3_u64).unwrap().is_some());
    }

    #[test]
    fn test_invalidate_entry_with_multiple_views() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        // Entry accesses both v_user and v_post
        cache
            .put(
                1_u64,
                test_result(),
                vec!["v_user".to_string(), "v_post".to_string()],
                None,
                None,
                None,
            )
            .unwrap();

        // Invalidating either view should remove the entry
        cache.invalidate_views(&[ViewName::from("v_user")]).unwrap();

        assert!(cache.get(1_u64).unwrap().is_none());
    }

    #[test]
    fn test_invalidate_nonexistent_view() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        // Invalidate view that doesn't exist
        cache.invalidate_views(&[ViewName::from("v_nonexistent")]).unwrap();

        // Entry should remain
        assert!(cache.get(1_u64).unwrap().is_some());
    }

    // ========================================================================
    // Clear Tests
    // ========================================================================

    #[test]
    fn test_clear() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();
        cache
            .put(2_u64, test_result(), vec!["v_post".to_string()], None, None, None)
            .unwrap();

        cache.clear().unwrap();

        // Run pending tasks to flush moka's eviction pipeline
        cache.run_pending_tasks();

        assert!(cache.get(1_u64).unwrap().is_none());
        assert!(cache.get(2_u64).unwrap().is_none());

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.size, 0);
    }

    // ========================================================================
    // Metrics Tests
    // ========================================================================

    #[test]
    fn test_metrics_tracking() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        // Miss
        cache.get(999_u64).unwrap();

        // Put
        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        // Hit
        cache.get(1_u64).unwrap();

        // moka::sync::Cache entry_count() is eventually consistent — flush pending
        // write operations before asserting on size.
        cache.run_pending_tasks();

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.hits, 1);
        assert_eq!(metrics.misses, 1);
        assert_eq!(metrics.size, 1);
        assert_eq!(metrics.total_cached, 1);
    }

    #[test]
    fn test_metrics_hit_rate() {
        let metrics = CacheMetrics {
            hits:          80,
            misses:        20,
            total_cached:  100,
            invalidations: 5,
            size:          95,
            memory_bytes:  1_000_000,
        };

        assert!((metrics.hit_rate() - 0.8).abs() < f64::EPSILON);
        assert!(metrics.is_healthy());
    }

    #[test]
    fn test_metrics_hit_rate_zero_requests() {
        let metrics = CacheMetrics {
            hits:          0,
            misses:        0,
            total_cached:  0,
            invalidations: 0,
            size:          0,
            memory_bytes:  0,
        };

        assert!((metrics.hit_rate() - 0.0).abs() < f64::EPSILON);
        assert!(!metrics.is_healthy());
    }

    #[test]
    fn test_metrics_is_healthy() {
        let good = CacheMetrics {
            hits:          70,
            misses:        30,
            total_cached:  100,
            invalidations: 5,
            size:          95,
            memory_bytes:  1_000_000,
        };
        assert!(good.is_healthy()); // 70% > 60%

        let bad = CacheMetrics {
            hits:          50,
            misses:        50,
            total_cached:  100,
            invalidations: 5,
            size:          95,
            memory_bytes:  1_000_000,
        };
        assert!(!bad.is_healthy()); // 50% < 60%
    }

    // ========================================================================
    // Entity-Aware Invalidation Tests
    // ========================================================================

    fn entity_result(id: &str) -> Vec<JsonbValue> {
        vec![JsonbValue::new(
            serde_json::json!({"id": id, "name": "test"}),
        )]
    }

    #[test]
    fn test_invalidate_by_entity_only_removes_matching_entries() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        // Cache User A and User B as separate entries
        cache
            .put(
                1_u64,
                entity_result("uuid-a"),
                vec!["v_user".to_string()],
                None,
                Some("User"),
                None,
            )
            .unwrap();
        cache
            .put(
                2_u64,
                entity_result("uuid-b"),
                vec!["v_user".to_string()],
                None,
                Some("User"),
                None,
            )
            .unwrap();

        // Invalidate User A — User B must remain
        let evicted = cache.invalidate_by_entity("User", "uuid-a").unwrap();
        assert_eq!(evicted, 1);
        assert!(cache.get(1_u64).unwrap().is_none(), "User A should be evicted");
        assert!(cache.get(2_u64).unwrap().is_some(), "User B should remain");
    }

    #[test]
    fn test_invalidate_by_entity_removes_list_containing_entity() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        // Cache a single-entity entry (entity_ref uses first row's id)
        cache
            .put(
                1_u64,
                entity_result("uuid-a"),
                vec!["v_user".to_string()],
                None,
                Some("User"),
                None,
            )
            .unwrap();

        // Invalidate by User A
        let evicted = cache.invalidate_by_entity("User", "uuid-a").unwrap();
        assert_eq!(evicted, 1);
        assert!(cache.get(1_u64).unwrap().is_none(), "Entry for A should be evicted");
    }

    #[test]
    fn test_invalidate_by_entity_leaves_unrelated_types() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        // Cache a User entry and a Post entry
        cache
            .put(
                1_u64,
                entity_result("uuid-user"),
                vec!["v_user".to_string()],
                None,
                Some("User"),
                None,
            )
            .unwrap();
        cache
            .put(
                2_u64,
                entity_result("uuid-post"),
                vec!["v_post".to_string()],
                None,
                Some("Post"),
                None,
            )
            .unwrap();

        // Invalidate the User — Post entry must remain untouched
        let evicted = cache.invalidate_by_entity("User", "uuid-user").unwrap();
        assert_eq!(evicted, 1);
        assert!(cache.get(1_u64).unwrap().is_none(), "User entry should be evicted");
        assert!(cache.get(2_u64).unwrap().is_some(), "Post entry should remain");
    }

    #[test]
    fn test_put_builds_entity_id_index() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(
                1_u64,
                entity_result("uuid-1"),
                vec!["v_user".to_string()],
                None,
                Some("User"),
                None,
            )
            .unwrap();

        // Invalidating by uuid-1 should evict the entry
        let evicted = cache.invalidate_by_entity("User", "uuid-1").unwrap();
        assert_eq!(evicted, 1);
        assert!(cache.get(1_u64).unwrap().is_none());
    }

    #[test]
    fn test_put_without_entity_type_not_indexed() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(
                1_u64,
                entity_result("uuid-1"),
                vec!["v_user".to_string()],
                None,
                None, // no entity type
                None,
            )
            .unwrap();

        // invalidate_by_entity should not match (no index was built)
        let evicted = cache.invalidate_by_entity("User", "uuid-1").unwrap();
        assert_eq!(evicted, 0);
        assert!(cache.get(1_u64).unwrap().is_some(), "Non-indexed entry should remain");
    }

    // ========================================================================
    // Multi-entity indexing + view-sweep invalidation tests
    // ========================================================================

    fn list_result(ids: &[&str]) -> Vec<JsonbValue> {
        ids.iter()
            .map(|id| JsonbValue::new(serde_json::json!({"id": id, "name": "test"})))
            .collect()
    }

    #[test]
    fn test_put_indexes_all_entities_in_list() {
        let cache = QueryResultCache::new(CacheConfig::enabled());
        let rows = list_result(&["uuid-A", "uuid-B", "uuid-C"]);
        cache
            .put(0xABC, rows, vec!["v_user".to_string()], None, Some("User"), None)
            .unwrap();

        let evicted_a = cache.invalidate_by_entity("User", "uuid-A").unwrap();
        assert_eq!(evicted_a, 1, "uuid-A must be indexed and evictable");

        // Re-insert to test uuid-C
        let rows2 = list_result(&["uuid-A", "uuid-B", "uuid-C"]);
        cache
            .put(0xDEF, rows2, vec!["v_user".to_string()], None, Some("User"), None)
            .unwrap();
        let evicted_c = cache.invalidate_by_entity("User", "uuid-C").unwrap();
        assert_eq!(evicted_c, 1, "uuid-C at position 2 must also be indexed");
    }

    #[test]
    fn test_update_evicts_list_query_via_non_first_entity() {
        let cache = QueryResultCache::new(CacheConfig::enabled());
        let rows = list_result(&["uuid-A", "uuid-B"]);
        cache
            .put(0x111, rows, vec!["v_user".to_string()], None, Some("User"), None)
            .unwrap();

        // uuid-B is at position 1 — must still be evicted
        let evicted = cache.invalidate_by_entity("User", "uuid-B").unwrap();
        assert_eq!(evicted, 1);
        assert!(cache.get(0x111).unwrap().is_none(), "list entry containing uuid-B must be gone");
    }

    /// A view sweep evicts every entry for the view — including the one-row
    /// results the deleted `invalidate_list_queries` used to spare (#742).
    ///
    /// Row count never told us whether an entry was affected by a write: a
    /// one-row result can be a filtered list about to gain a second row, and a
    /// zero-row result is exactly what a CREATE makes stale. The sparing was an
    /// optimisation whose safety condition was never checked.
    #[test]
    fn test_invalidate_views_evicts_regardless_of_row_count() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        let empty: Vec<JsonbValue> = vec![];
        cache
            .put(0x001, empty, vec!["v_user".to_string()], None, Some("User"), None)
            .unwrap();
        let single = vec![JsonbValue::new(serde_json::json!({"id": "uuid-X"}))];
        cache
            .put(0x002, single, vec!["v_user".to_string()], None, Some("User"), None)
            .unwrap();
        let list = list_result(&["uuid-A", "uuid-B"]);
        cache
            .put(0x003, list, vec!["v_user".to_string()], None, Some("User"), None)
            .unwrap();

        cache.invalidate_views(&[ViewName::from("v_user")]).unwrap();
        for key in [0x001_u64, 0x002, 0x003] {
            assert!(cache.get(key).unwrap().is_none(), "entry {key:#x} must be evicted");
        }
    }

    #[test]
    fn test_invalidate_by_entity_short_circuits_on_empty_index() {
        let cache = QueryResultCache::new(CacheConfig::enabled());
        // Nothing cached — must return 0 without panicking
        let count = cache.invalidate_by_entity("User", "uuid-X").unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn test_eviction_listener_cleans_all_entity_refs() {
        let cache = QueryResultCache::new(CacheConfig::enabled());
        let rows = list_result(&["uuid-A", "uuid-B"]);
        cache
            .put(0x001, rows, vec!["v_user".to_string()], None, Some("User"), None)
            .unwrap();

        // Evict it by one of its rows
        assert_eq!(cache.invalidate_by_entity("User", "uuid-A").unwrap(), 1);
        // Flush moka's async eviction pipeline
        cache.run_pending_tasks();

        // After eviction the entity_index must be cleaned up (no dangling refs)
        let count_a = cache.invalidate_by_entity("User", "uuid-A").unwrap();
        let count_b = cache.invalidate_by_entity("User", "uuid-B").unwrap();
        assert_eq!(count_a, 0, "entity_index must be clean after eviction");
        assert_eq!(count_b, 0, "entity_index must be clean after eviction");
    }

    // ========================================================================
    // Thread Safety Tests
    // ========================================================================

    #[test]
    fn test_concurrent_access() {
        use std::{sync::Arc, thread};

        let cache = Arc::new(QueryResultCache::new(CacheConfig::enabled()));

        // Spawn multiple threads accessing cache
        let handles: Vec<_> = (0_u64..10)
            .map(|key| {
                let cache_clone = cache.clone();
                thread::spawn(move || {
                    cache_clone
                        .put(key, test_result(), vec!["v_user".to_string()], None, None, None)
                        .unwrap();
                    cache_clone.get(key).unwrap();
                })
            })
            .collect();

        for handle in handles {
            handle.join().unwrap();
        }

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.total_cached, 10);
        assert_eq!(metrics.hits, 10);
    }

    // ========================================================================
    // Sentinel tests — boundary guards for mutation testing
    // ========================================================================

    /// Sentinel: `cache_list_queries = false` must skip results with >1 row.
    ///
    /// Kills the `> → >=` mutation at the list-query guard: `result.len() > 1`.
    #[test]
    fn test_cache_list_queries_false_skips_multi_row() {
        let config = CacheConfig {
            enabled: true,
            cache_list_queries: false,
            ..CacheConfig::default()
        };
        let cache = QueryResultCache::new(config);

        // Two-row result: must be skipped (killed by > → >= mutant)
        let two_rows = vec![
            JsonbValue::new(json!({"id": 1})),
            JsonbValue::new(json!({"id": 2})),
        ];
        cache
            .put(1_u64, two_rows, vec!["v_user".to_string()], None, None, None)
            .unwrap();
        assert!(
            cache.get(1_u64).unwrap().is_none(),
            "multi-row result must not be cached when cache_list_queries=false"
        );
    }

    /// Sentinel: `cache_list_queries = false` must still store single-row results.
    ///
    /// Complements the above: the single-row path must remain unaffected.
    #[test]
    fn test_cache_list_queries_false_allows_single_row() {
        let config = CacheConfig {
            enabled: true,
            cache_list_queries: false,
            ..CacheConfig::default()
        };
        let cache = QueryResultCache::new(config);

        // One-row result: must be stored
        let one_row = vec![JsonbValue::new(json!({"id": 1}))];
        cache.put(1_u64, one_row, vec!["v_user".to_string()], None, None, None).unwrap();
        assert!(
            cache.get(1_u64).unwrap().is_some(),
            "single-row result must be cached even when cache_list_queries=false"
        );
    }

    /// Sentinel: entries exceeding `max_entry_bytes` must be silently skipped.
    ///
    /// Kills mutations on the `estimated > max_entry` guard.
    #[test]
    fn test_max_entry_bytes_skips_oversized_entry() {
        let config = CacheConfig {
            enabled: true,
            max_entry_bytes: Some(10), // 10 bytes — smaller than any JSON row
            ..CacheConfig::default()
        };
        let cache = QueryResultCache::new(config);

        // A typical row serialises to far more than 10 bytes
        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();
        assert!(cache.get(1_u64).unwrap().is_none(), "oversized entry must be silently skipped");
    }

    /// Sentinel: entries within `max_entry_bytes` must be stored normally.
    ///
    /// Complements the above to pin both sides of the size boundary.
    #[test]
    fn test_max_entry_bytes_allows_small_entry() {
        let config = CacheConfig {
            enabled: true,
            max_entry_bytes: Some(100_000), // 100 KB — plenty for a test row
            ..CacheConfig::default()
        };
        let cache = QueryResultCache::new(config);

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();
        assert!(
            cache.get(1_u64).unwrap().is_some(),
            "small entry must be cached when within max_entry_bytes"
        );
    }

    /// Sentinel: `put()` must skip new entries when `max_total_bytes` budget is exhausted.
    ///
    /// Kills mutations on the `current >= max_total` guard.
    #[test]
    fn test_max_total_bytes_skips_when_budget_exhausted() {
        let config = CacheConfig {
            enabled: true,
            max_total_bytes: Some(0), // 0 bytes — always exhausted
            ..CacheConfig::default()
        };
        let cache = QueryResultCache::new(config);

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();
        assert!(
            cache.get(1_u64).unwrap().is_none(),
            "entry must be skipped when max_total_bytes budget is already exhausted"
        );
    }

    // ========================================================================
    // Cross-key invalidation Tests (replaces cross-shard tests)
    // ========================================================================

    /// `invalidate_views` clears matching entries regardless of cache key.
    #[test]
    fn test_cross_key_view_invalidation() {
        let config = CacheConfig {
            max_entries: 10_000,
            enabled: true,
            ..CacheConfig::default()
        };
        let cache = QueryResultCache::new(config);

        // Insert many entries
        for i in 0_u64..200 {
            let view = if i % 2 == 0 { "v_user" } else { "v_post" };
            cache.put(i, test_result(), vec![view.to_string()], None, None, None).unwrap();
        }

        // Invalidate v_user
        cache.invalidate_views(&[ViewName::from("v_user")]).unwrap();

        // All v_user entries gone, all v_post entries remain
        for i in 0_u64..200 {
            if i % 2 == 0 {
                assert!(cache.get(i).unwrap().is_none(), "v_user entry should be invalidated");
            } else {
                assert!(cache.get(i).unwrap().is_some(), "v_post entry should remain");
            }
        }
    }

    /// Cross-key entity invalidation works across all cache keys.
    #[test]
    fn test_cross_key_entity_invalidation() {
        let config = CacheConfig {
            max_entries: 10_000,
            enabled: true,
            ..CacheConfig::default()
        };
        let cache = QueryResultCache::new(config);

        // Insert entries for the same entity across different cache keys
        for i in 0_u64..50 {
            cache
                .put(
                    i,
                    entity_result("uuid-target"),
                    vec!["v_user".to_string()],
                    None,
                    Some("User"),
                    None,
                )
                .unwrap();
        }

        // Also insert an unrelated entry
        cache
            .put(
                999_u64,
                entity_result("uuid-other"),
                vec!["v_user".to_string()],
                None,
                Some("User"),
                None,
            )
            .unwrap();

        let evicted = cache.invalidate_by_entity("User", "uuid-target").unwrap();
        assert_eq!(evicted, 50);
        assert!(cache.get(999_u64).unwrap().is_some(), "unrelated entity should remain");
    }

    /// Clear works for all entries.
    #[test]
    fn test_clear_all() {
        let config = CacheConfig {
            max_entries: 10_000,
            enabled: true,
            ..CacheConfig::default()
        };
        let cache = QueryResultCache::new(config);

        for i in 0_u64..200 {
            cache
                .put(i, test_result(), vec!["v_user".to_string()], None, None, None)
                .unwrap();
        }

        cache.clear().unwrap();
        cache.run_pending_tasks();

        let metrics = cache.metrics().unwrap();
        assert_eq!(metrics.size, 0);

        for i in 0_u64..200 {
            assert!(cache.get(i).unwrap().is_none());
        }
    }

    /// `memory_bytes` is tracked and reported via `metrics()`.
    #[test]
    fn test_memory_bytes_tracked() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(1_u64, test_result(), vec!["v".to_string()], None, None, None)
            .unwrap();
        cache
            .put(2_u64, test_result(), vec!["v".to_string()], None, None, None)
            .unwrap();

        let before = cache.metrics().unwrap().memory_bytes;
        assert!(before > 0, "memory_bytes should be tracked");
    }

    /// `memory_bytes` decreases after invalidation (synchronously via clear).
    #[test]
    fn test_memory_bytes_decreases_on_clear() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(1_u64, test_result(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        let before = cache.metrics().unwrap().memory_bytes;
        assert!(before > 0);

        cache.clear().unwrap();

        let after = cache.metrics().unwrap().memory_bytes;
        assert_eq!(after, 0, "memory_bytes should be zero after clear()");
    }

    // ========================================================================
    // Concurrency regression test (#185)
    // ========================================================================

    /// Concurrency guard + manual scaling benchmark for #185 (LRU+Mutex serialized
    /// all hot-key reads through one shard's mutex; moka reads are lock-free).
    ///
    /// **The hard assertion is correctness under concurrent load**, not wall-clock:
    /// every concurrent reader must see the entry and its exact value, and the cache
    /// must survive the hammering uncorrupted. That is a robust, deterministic guard
    /// against a re-serialization regression that also corrupts (deadlock, lost
    /// entry, torn value).
    ///
    /// The **scaling ratio is reported, not asserted (#600)**. A wall-clock ratio of
    /// `multi / single` is not a reliable serialization signal: a moka `get` is a
    /// few nanoseconds, so on low-core CI runners *and* on hybrid P/E-core desktops
    /// (asymmetric cores + E-core scheduling) thread-scheduling overhead dominates
    /// the actual work — lock-free reads produced ~29× ratios on an idle i7-13700K,
    /// nearly indistinguishable from serialization. The numbers are printed for a
    /// human to interpret when investigating scaling; they are not a pass/fail.
    #[test]
    #[ignore = "manual scaling benchmark — asserts concurrency correctness, reports throughput"]
    // Reason: the throughput figures are a human-read benchmark report; f64 precision
    // loss on the small op counts is irrelevant and never asserted on.
    #[allow(clippy::cast_precision_loss)]
    fn test_concurrent_reads_do_not_serialize() {
        const ITERS: usize = 10_000;

        // `JsonbValue` is not `PartialEq`; compare the inner JSON values.
        fn same(a: &[JsonbValue], b: &[JsonbValue]) -> bool {
            a.len() == b.len() && a.iter().zip(b).all(|(x, y)| x.as_value() == y.as_value())
        }

        // Match threads to available parallelism instead of a fixed 40 to avoid
        // oversubscription noise; clamp so the benchmark stays bounded on big boxes.
        let cores = std::thread::available_parallelism().map_or(1, |n| n.get());
        let threads = cores.clamp(2, 32);

        let config = CacheConfig::enabled();
        let cache = Arc::new(QueryResultCache::new(config));
        let key = 42_u64;
        let expected = test_result();
        cache
            .put(key, expected.clone(), vec!["v_user".to_string()], None, None, None)
            .unwrap();

        // Single-threaded baseline.
        let start = std::time::Instant::now();
        for _ in 0..ITERS {
            assert!(cache.get(key).unwrap().is_some(), "baseline read must hit");
        }
        let single_elapsed = start.elapsed();

        // Concurrent readers: the HARD guard. Each reader asserts every read hits and
        // returns the exact stored value — a serialized-and-corrupting regression, a
        // lost entry, or a torn read fails here deterministically, on any hardware.
        let start = std::time::Instant::now();
        let handles: Vec<_> = (0..threads)
            .map(|_| {
                let c = Arc::clone(&cache);
                let want = expected.clone();
                std::thread::spawn(move || {
                    for _ in 0..ITERS {
                        let got = c.get(key).unwrap().expect("concurrent read must hit");
                        assert!(same(&got, &want), "concurrent read returned an unexpected value");
                    }
                })
            })
            .collect();
        for h in handles {
            h.join().expect("reader thread must not panic (deadlock/corruption)");
        }
        let multi_elapsed = start.elapsed();

        // The cache is intact after concurrent hammering.
        assert!(
            same(&cache.get(key).unwrap().expect("entry survives concurrency"), &expected),
            "entry value survives concurrency"
        );

        // Reported, NOT asserted (#600): aggregate throughput and the raw ratio, for
        // manual interpretation of lock-free scaling.
        let single_tput = ITERS as f64 / single_elapsed.as_secs_f64();
        let multi_tput = (threads * ITERS) as f64 / multi_elapsed.as_secs_f64();
        eprintln!(
            "concurrent-read scaling ({threads} threads, {ITERS} iters/thread): \
             single {single_elapsed:?} ({single_tput:.0} ops/s), \
             multi {multi_elapsed:?} ({multi_tput:.0} ops/s), \
             aggregate speedup {:.2}×, wall ratio {:.1}×",
            multi_tput / single_tput,
            multi_elapsed.as_secs_f64() / single_elapsed.as_secs_f64(),
        );
    }
}

/// Eviction-listener lifecycle: an entry re-cached under the same key must stay
/// reachable from every invalidation index (#740).
///
/// `put_arc` registers the key in `view_index` / `entity_index` *before*
/// `store.insert`, so moka fires the eviction listener for the entry that was
/// just displaced. If that listener cleans the indexes by key alone it deletes
/// the registration the *live* entry depends on, and no mutation can ever evict
/// it again — permanently, for views annotated `cache_ttl_seconds = 0`.
mod eviction_lifecycle_tests {
    use serde_json::json;

    use crate::{cache::*, db::types::JsonbValue};

    fn one_row() -> Vec<JsonbValue> {
        vec![JsonbValue::new(json!({"id": "u1", "name": "Alice"}))]
    }

    /// Two concurrent misses for the same hot query both `put`. The second
    /// insert displaces the first; the displaced entry must not take the live
    /// entry's index registrations with it.
    #[test]
    fn recached_entry_is_still_reachable_by_view_invalidation() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(42, one_row(), vec!["v_user".to_string()], Some(0), Some("User"), None)
            .unwrap();
        cache
            .put(42, one_row(), vec!["v_user".to_string()], Some(0), Some("User"), None)
            .unwrap();
        cache.run_pending_tasks();

        cache.invalidate_views(&[ViewName::from("v_user")]).unwrap();
        cache.run_pending_tasks();
        assert!(
            cache.get(42).unwrap().is_none(),
            "#740: the live entry must actually be gone after invalidate_views"
        );
    }

    #[test]
    fn recached_entry_is_still_reachable_by_entity_invalidation() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(43, one_row(), vec!["v_user".to_string()], Some(0), Some("User"), None)
            .unwrap();
        cache
            .put(43, one_row(), vec!["v_user".to_string()], Some(0), Some("User"), None)
            .unwrap();
        cache.run_pending_tasks();

        assert_eq!(
            cache.invalidate_by_entity("User", "u1").unwrap(),
            1,
            "#740: a re-cached entry must still be reachable through entity_index"
        );
        cache.run_pending_tasks();
        assert!(
            cache.get(43).unwrap().is_none(),
            "#740: the live entry must actually be gone after invalidate_by_entity"
        );
    }

    /// A genuinely evicted entry must *not* leave its registration behind, or
    /// `entity_index` grows without bound on a long-running TTL-based cache.
    #[test]
    fn explicitly_evicted_entry_releases_its_index_registration() {
        let cache = QueryResultCache::new(CacheConfig::enabled());

        cache
            .put(44, one_row(), vec!["v_user".to_string()], Some(0), Some("User"), None)
            .unwrap();
        cache.run_pending_tasks();
        assert_eq!(cache.invalidate_by_entity("User", "u1").unwrap(), 1);
        cache.run_pending_tasks();

        assert_eq!(
            cache.invalidate_by_entity("User", "u1").unwrap(),
            0,
            "#740: an evicted entry must not linger in entity_index"
        );
    }
}

// ── invalidation_fence_tests ────────────────────────────────────────────────

/// #1079: a cache write whose read straddled an invalidation must be discarded.
///
/// A read is `get` → miss → **await the database** → `put`. An invalidation landing
/// inside that await evicts nothing — the key is not registered yet — and the `put` then
/// stores rows fetched *before* the mutation committed. The client that just wrote sees
/// its own write vanish on the next read.
///
/// #740's "register the key before inserting" note makes the key *visible* to a concurrent
/// `invalidate_views`, but it does not fence the insert: an invalidation landing between
/// registration and `store.insert` collects a key that is not in the store yet — a no-op —
/// and the insert lands stale anyway. So the fence has to cover the insert itself.
///
/// These tests drive the interleaving **as a sequence, not a sleep**: the invalidation is
/// performed between the snapshot and the put, exactly where the race puts it. A
/// probabilistic version would be a flake in CI and a false green on a fast runner.
mod invalidation_fence_tests {
    use fraiseql_db::ViewName;

    use crate::cache::{CacheConfig, QueryResultCache};

    fn row_cache() -> QueryResultCache {
        QueryResultCache::new(CacheConfig::enabled())
    }

    fn rows() -> Vec<crate::backend::types::JsonbValue> {
        vec![crate::backend::types::JsonbValue::new(
            serde_json::json!({"id": "u-1", "n": "pre"}),
        )]
    }

    /// The defect, in one sequence.
    #[test]
    fn a_read_that_straddles_an_invalidation_does_not_cache() {
        let cache = row_cache();

        // The read snapshots, then awaits the database…
        let fence = cache.invalidation_generation();

        // …and while it is awaiting, a mutation commits and invalidates.
        cache.invalidate_views(&[ViewName::from("v_user")]).unwrap();

        // …then the read returns and stores its pre-mutation rows.
        cache
            .put(1_u64, rows(), vec!["v_user".to_string()], None, None, Some(fence))
            .unwrap();

        assert!(
            cache.get(1_u64).unwrap().is_none(),
            "a read whose round trip straddled an invalidation must not populate the cache — \
             storing it re-caches pre-mutation rows and the client's own write disappears (#1079)"
        );
    }

    /// The counterweight: with nothing invalidating, the write must still land.
    ///
    /// Without this, a fence that refused *every* write would satisfy the test above and
    /// silently disable the cache.
    #[test]
    fn an_uncontended_read_still_caches() {
        let cache = row_cache();
        let fence = cache.invalidation_generation();

        cache
            .put(1_u64, rows(), vec!["v_user".to_string()], None, None, Some(fence))
            .unwrap();

        assert!(
            cache.get(1_u64).unwrap().is_some(),
            "an uncontended read must still be cached — the fence must not cost hit rate \
             when nothing raced it"
        );
    }

    /// A refused write must leave no index rows behind, or the next invalidation walks
    /// references to a key that was never stored.
    #[test]
    fn a_fenced_out_write_deregisters_itself() {
        let cache = row_cache();
        let fence = cache.invalidation_generation();
        cache.invalidate_by_entity("User", "u-1").unwrap();
        cache
            .put(1_u64, rows(), vec!["v_user".to_string()], None, Some("User"), Some(fence))
            .unwrap();
        cache.run_pending_tasks();

        // If the refused write left its registration behind, this invalidation would
        // report evicting an entry that does not exist.
        let evicted = cache.invalidate_by_entity("User", "u-1").unwrap();
        assert_eq!(
            evicted, 0,
            "a fenced-out write must deregister itself; a stale index row makes the next \
             invalidation count a key that was never stored"
        );
    }

    /// The straddle, generalised: snapshot, run `during`, then put a read of `views` holding
    /// `entity_type`. Returns whether the fill was stored.
    fn straddled_fill_is_stored(
        views: &[&str],
        entity_type: Option<&str>,
        during: impl FnOnce(&QueryResultCache),
    ) -> bool {
        let cache = row_cache();
        let fence = cache.invalidation_generation();
        during(&cache);
        let views = views.iter().map(|v| (*v).to_string()).collect();
        cache.put(1_u64, rows(), views, None, entity_type, Some(fence)).unwrap();
        cache.get(1_u64).unwrap().is_some()
    }

    /// #1584: the fence is per view. A write to a view this read cannot see is no reason
    /// to refuse its fill: under steady writes some invalidation is always in flight, and a
    /// global fence then caches nothing at all.
    #[test]
    fn a_read_straddling_an_unrelated_invalidation_still_caches() {
        assert!(
            straddled_fill_is_stored(&["v_user"], None, |c| {
                c.invalidate_views(&[ViewName::from("v_post")]).unwrap();
            }),
            "a write to v_post must not refuse the fill of a read of v_user (#1584)"
        );
    }

    /// The read is fenced over every view it is registered under, not only its source: a
    /// write to a secondary view (#761) evicts the entry, so it must refuse the fill too.
    #[test]
    fn a_read_straddling_an_invalidation_of_a_secondary_view_does_not_cache() {
        assert!(
            straddled_fill_is_stored(&["v_user", "v_post"], None, |_| {}),
            "precondition: the two-view fill is stored when nothing races it"
        );
        assert!(
            !straddled_fill_is_stored(&["v_user", "v_post"], None, |c| {
                c.invalidate_views(&[ViewName::from("v_post")]).unwrap();
            }),
            "a write to a secondary view of the read must refuse its fill"
        );
    }

    /// An entity eviction reaches only the fills holding its row: it evicts the entries
    /// whose entity references hold that `(type, id)`, and no other (#1605).
    #[test]
    fn an_entity_eviction_fences_only_fills_holding_its_row() {
        assert!(
            !straddled_fill_is_stored(&["v_user"], Some("User"), |c| {
                c.invalidate_by_entity("User", "u-1").unwrap();
            }),
            "a read holding User u-1 must not cache across an eviction of u-1"
        );
        assert!(
            straddled_fill_is_stored(&["v_user"], Some("User"), |c| {
                c.invalidate_by_entity("User", "u-2").unwrap();
            }),
            "an eviction of User u-2 must not refuse the fill of a read holding only u-1 \
             (#1605): stored first, it would have been left in place"
        );
        assert!(
            straddled_fill_is_stored(&["v_user"], Some("User"), |c| {
                c.invalidate_by_entity("Post", "p-1").unwrap();
            }),
            "an eviction of a Post must not refuse the fill of a read holding Users (#1584)"
        );
    }

    /// A list holding many rows is refused by an eviction of any one of them, and by none of
    /// another row.
    #[test]
    fn an_entity_eviction_fences_a_list_holding_its_row_wherever_it_sits() {
        let list = || {
            (1..=50)
                .map(|n| {
                    crate::backend::types::JsonbValue::new(
                        serde_json::json!({"id": format!("u-{n}")}),
                    )
                })
                .collect::<Vec<_>>()
        };
        for (evicted, stored) in [
            ("u-1", false),
            ("u-37", false),
            ("u-50", false),
            ("u-51", true),
        ] {
            let cache = row_cache();
            let fence = cache.invalidation_generation();
            cache.invalidate_by_entity("User", evicted).unwrap();
            cache
                .put(1_u64, list(), vec!["v_user".to_string()], None, Some("User"), Some(fence))
                .unwrap();
            assert_eq!(cache.get(1_u64).unwrap().is_some(), stored, "eviction of {evicted}");
        }
    }

    /// `clear` reaches every fill.
    #[test]
    fn a_read_straddling_a_clear_does_not_cache() {
        assert!(
            !straddled_fill_is_stored(&["v_user"], Some("User"), |c| c.clear().unwrap()),
            "a read in flight across a clear must not repopulate it"
        );
    }

    /// A fill whose read straddled an invalidation reaching it is never visible, not even
    /// for the instant between its insert and the authoritative check: a concurrent reader
    /// there would be served the pre-write rows. The pre-check refuses it before the store.
    #[test]
    fn a_straddled_fill_is_never_visible_in_the_store() {
        let cache = row_cache();
        let fence = cache.invalidation_generation();
        cache.invalidate_views(&[ViewName::from("v_post")]).unwrap();
        cache
            .put_arc_with(
                1_u64,
                std::sync::Arc::new(rows()),
                vec!["v_user".to_string(), "v_post".to_string()],
                None,
                None,
                Some(fence),
                || {},
                || {
                    assert!(
                        cache.get(1_u64).unwrap().is_none(),
                        "a reader concurrent with the put must not be served a fill whose read \
                         straddled a write to one of its views"
                    );
                },
            )
            .unwrap();
    }

    /// As above, for an eviction of a row the fill holds (#1605): the pre-check reads the
    /// rows' references, so the straddled fill never reaches the store.
    #[test]
    fn a_fill_straddling_an_eviction_of_a_row_it_holds_is_never_visible() {
        let cache = row_cache();
        let fence = cache.invalidation_generation();
        cache.invalidate_by_entity("User", "u-1").unwrap();
        cache
            .put_arc_with(
                1_u64,
                std::sync::Arc::new(rows()),
                vec!["v_user".to_string()],
                None,
                Some("User"),
                Some(fence),
                || {},
                || {
                    assert!(
                        cache.get(1_u64).unwrap().is_none(),
                        "a reader concurrent with the put must not be served a fill whose read \
                         straddled an eviction of a row it holds"
                    );
                },
            )
            .unwrap();
    }

    /// An invalidation that lands inside `put` itself, after the pre-check and before the
    /// entry is in the store, collects a key that is not stored yet: it evicts nothing. The
    /// fill must notice and withdraw, or the pre-write rows stay cached (at TTL 0, forever).
    /// Each way an invalidation reaches a fill is raced there: its source view, a secondary
    /// view, a row it holds, a clear.
    #[test]
    fn an_invalidation_landing_inside_the_put_does_not_leave_the_fill_cached() {
        let racers: [(&str, fn(&QueryResultCache)); 4] = [
            ("its source view", |c| {
                c.invalidate_views(&[ViewName::from("v_user")]).unwrap();
            }),
            ("a secondary view", |c| {
                c.invalidate_views(&[ViewName::from("v_post")]).unwrap();
            }),
            ("a row it holds", |c| {
                c.invalidate_by_entity("User", "u-1").unwrap();
            }),
            ("a clear", |c| c.clear().unwrap()),
        ];
        for (what, racer) in racers {
            let cache = row_cache();
            let fence = cache.invalidation_generation();
            cache
                .put_arc_with(
                    1_u64,
                    std::sync::Arc::new(rows()),
                    vec!["v_user".to_string(), "v_post".to_string()],
                    None,
                    Some("User"),
                    Some(fence),
                    || racer(&cache),
                    || {},
                )
                .unwrap();
            assert!(
                cache.get(1_u64).unwrap().is_none(),
                "an invalidation of {what} landing between the pre-check and the insert must \
                 not leave the pre-write rows cached"
            );
        }
    }
}