reddb_server/storage/cache/

result.rs

//! Query Result Cache
//!
//! Caches query results for expensive cross-modal queries.
//! Supports TTL-based expiration and dependency-based invalidation.
//!
//! # Features
//!
//! - **Result Caching**: Cache expensive query results
//! - **TTL Expiration**: Automatic expiry based on time
//! - **Dependency Tracking**: Invalidate when underlying data changes
//! - **Size Management**: LRU eviction based on memory usage
//! - **Statistics**: Cache hit/miss tracking
//!
//! # Example
//!
//! ```ignore
//! use storage::cache::result::{ResultCache, CacheKey, CachePolicy};
//!
//! let mut cache = ResultCache::new(100_000_000); // 100MB max
//!
//! let key = CacheKey::new("attack_paths")
//!     .param("from", "external")
//!     .param("to", "database");
//!
//! if let Some(result) = cache.get(&key) {
//!     return result;
//! }
//!
//! let result = expensive_query();
//! cache.insert(key, result.clone(), CachePolicy::default()
//!     .ttl(Duration::from_secs(300))
//!     .depends_on(&["hosts", "vulnerabilities"]));
//! ```

use std::collections::{HashMap, HashSet};
use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

// ============================================================================
// Cache Key
// ============================================================================

/// Cache key for query results
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CacheKey {
    /// Query type identifier
    pub query_type: String,
    /// Query parameters (sorted for consistent hashing)
    pub params: Vec<(String, String)>,
    /// Hash for fast comparison
    hash: u64,
}

impl CacheKey {
    /// Create a new cache key
    pub fn new(query_type: impl Into<String>) -> Self {
        let query_type = query_type.into();
        let mut key = Self {
            query_type,
            params: Vec::new(),
            hash: 0,
        };
        key.rehash();
        key
    }

    /// Add a parameter to the key
    pub fn param(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.params.push((name.into(), value.into()));
        self.params.sort_by(|a, b| a.0.cmp(&b.0));
        self.rehash();
        self
    }

    /// Add multiple parameters
    pub fn params(mut self, params: impl IntoIterator<Item = (String, String)>) -> Self {
        self.params.extend(params);
        self.params.sort_by(|a, b| a.0.cmp(&b.0));
        self.rehash();
        self
    }

    fn rehash(&mut self) {
        use std::collections::hash_map::DefaultHasher;
        let mut hasher = DefaultHasher::new();
        self.query_type.hash(&mut hasher);
        for (k, v) in &self.params {
            k.hash(&mut hasher);
            v.hash(&mut hasher);
        }
        self.hash = hasher.finish();
    }
}

impl Hash for CacheKey {
    fn hash<H: Hasher>(&self, state: &mut H) {
        state.write_u64(self.hash);
    }
}

// ============================================================================
// Cache Policy
// ============================================================================

/// Policy for cache entry behavior
#[derive(Debug, Clone)]
pub struct CachePolicy {
    /// Time to live
    pub ttl: Duration,
    /// Tables/collections this result depends on
    pub dependencies: HashSet<String>,
    /// Priority for eviction (higher = keep longer)
    pub priority: u8,
    /// Whether to refresh on access (sliding expiration)
    pub sliding: bool,
}

impl Default for CachePolicy {
    fn default() -> Self {
        Self {
            ttl: Duration::from_secs(300), // 5 minutes
            dependencies: HashSet::new(),
            priority: 50,
            sliding: false,
        }
    }
}

impl CachePolicy {
    /// Set TTL
    pub fn ttl(mut self, ttl: Duration) -> Self {
        self.ttl = ttl;
        self
    }

    /// Add dependencies
    pub fn depends_on(mut self, deps: &[&str]) -> Self {
        for dep in deps {
            self.dependencies.insert(dep.to_string());
        }
        self
    }

    /// Set priority
    pub fn priority(mut self, priority: u8) -> Self {
        self.priority = priority;
        self
    }

    /// Enable sliding expiration
    pub fn sliding(mut self) -> Self {
        self.sliding = true;
        self
    }
}

// ============================================================================
// Cache Entry
// ============================================================================

/// Cached query result
struct CacheEntry {
    /// Serialized result data
    data: Vec<u8>,
    /// Estimated size in bytes
    size: usize,
    /// When this entry was created
    created_at: Instant,
    /// Last access time
    last_accessed: Instant,
    /// Access count
    access_count: AtomicU64,
    /// Cache policy
    policy: CachePolicy,
}

impl CacheEntry {
    fn new(data: Vec<u8>, policy: CachePolicy) -> Self {
        let size = data.len();
        let now = Instant::now();
        Self {
            data,
            size,
            created_at: now,
            last_accessed: now,
            access_count: AtomicU64::new(0),
            policy,
        }
    }

    fn is_expired(&self) -> bool {
        let elapsed = if self.policy.sliding {
            self.last_accessed.elapsed()
        } else {
            self.created_at.elapsed()
        };
        elapsed > self.policy.ttl
    }

    fn touch(&mut self) {
        self.access_count.fetch_add(1, Ordering::Relaxed);
        self.last_accessed = Instant::now();
    }

    /// Calculate eviction score (lower = more likely to evict)
    fn eviction_score(&self) -> u64 {
        let frequency = self.access_count.load(Ordering::Relaxed);
        let recency = self.last_accessed.elapsed().as_secs();
        let priority = self.policy.priority as u64;

        // Score: frequency * priority / recency
        // Higher = keep longer
        frequency
            .saturating_mul(priority)
            .checked_div(recency)
            .unwrap_or_else(|| frequency.saturating_mul(priority).saturating_mul(1000))
    }
}

// ============================================================================
// Cache Statistics
// ============================================================================

/// Cache statistics
#[derive(Debug, Clone, Default)]
pub struct ResultCacheStats {
    /// Cache hits
    pub hits: u64,
    /// Cache misses
    pub misses: u64,
    /// Entries evicted
    pub evictions: u64,
    /// Current entry count
    pub entry_count: usize,
    /// Current memory usage in bytes
    pub memory_bytes: usize,
    /// Maximum memory limit
    pub max_memory_bytes: usize,
    /// Entries expired
    pub expirations: u64,
    /// Invalidations by dependency
    pub invalidations: u64,
}

impl ResultCacheStats {
    /// Calculate hit rate
    pub fn hit_rate(&self) -> f64 {
        let total = self.hits + self.misses;
        if total == 0 {
            0.0
        } else {
            self.hits as f64 / total as f64
        }
    }

    /// Calculate memory utilization
    pub fn memory_utilization(&self) -> f64 {
        if self.max_memory_bytes == 0 {
            0.0
        } else {
            self.memory_bytes as f64 / self.max_memory_bytes as f64
        }
    }
}

// ============================================================================
// Result Cache
// ============================================================================

/// LRU cache for query results with memory management
pub struct ResultCache {
    /// Cached entries
    entries: HashMap<CacheKey, CacheEntry>,
    /// Dependency index: table -> keys depending on it
    dependency_index: HashMap<String, HashSet<CacheKey>>,
    /// Maximum memory in bytes
    max_memory: usize,
    /// Current memory usage
    current_memory: usize,
    /// Statistics
    stats: ResultCacheStats,
}

impl ResultCache {
    /// Create a new result cache with max memory limit
    pub fn new(max_memory_bytes: usize) -> Self {
        Self {
            entries: HashMap::new(),
            dependency_index: HashMap::new(),
            max_memory: max_memory_bytes,
            current_memory: 0,
            stats: ResultCacheStats {
                max_memory_bytes,
                ..Default::default()
            },
        }
    }

    /// Get a cached result
    pub fn get(&mut self, key: &CacheKey) -> Option<Vec<u8>> {
        // Remove expired entry if present
        if let Some(entry) = self.entries.get(key) {
            if entry.is_expired() {
                self.remove(key);
                self.stats.expirations += 1;
                self.stats.misses += 1;
                return None;
            }
        }

        if let Some(entry) = self.entries.get_mut(key) {
            entry.touch();
            self.stats.hits += 1;
            Some(entry.data.clone())
        } else {
            self.stats.misses += 1;
            None
        }
    }

    /// Check if a key exists (without touching)
    pub fn contains(&self, key: &CacheKey) -> bool {
        if let Some(entry) = self.entries.get(key) {
            !entry.is_expired()
        } else {
            false
        }
    }

    /// Insert a result into the cache
    pub fn insert(&mut self, key: CacheKey, data: Vec<u8>, policy: CachePolicy) {
        let entry_size = data.len() + std::mem::size_of::<CacheEntry>();

        // Remove existing entry if present
        if self.entries.contains_key(&key) {
            self.remove(&key);
        }

        // Evict until we have space
        while self.current_memory + entry_size > self.max_memory && !self.entries.is_empty() {
            self.evict_one();
        }

        // Index dependencies
        for dep in &policy.dependencies {
            self.dependency_index
                .entry(dep.clone())
                .or_default()
                .insert(key.clone());
        }

        let entry = CacheEntry::new(data, policy);
        self.current_memory += entry.size;
        self.entries.insert(key, entry);
        self.stats.entry_count = self.entries.len();
        self.stats.memory_bytes = self.current_memory;
    }

    /// Remove an entry
    pub fn remove(&mut self, key: &CacheKey) -> Option<Vec<u8>> {
        if let Some(entry) = self.entries.remove(key) {
            self.current_memory = self.current_memory.saturating_sub(entry.size);

            // Remove from dependency index
            for dep in &entry.policy.dependencies {
                if let Some(keys) = self.dependency_index.get_mut(dep) {
                    keys.remove(key);
                }
            }

            self.stats.entry_count = self.entries.len();
            self.stats.memory_bytes = self.current_memory;
            Some(entry.data)
        } else {
            None
        }
    }

    /// Invalidate all entries depending on a table/collection
    pub fn invalidate_by_dependency(&mut self, dependency: &str) {
        if let Some(keys) = self.dependency_index.remove(dependency) {
            for key in keys {
                if self.entries.remove(&key).is_some() {
                    self.stats.invalidations += 1;
                }
            }
            self.stats.entry_count = self.entries.len();
            // Recalculate memory
            self.current_memory = self.entries.values().map(|e| e.size).sum();
            self.stats.memory_bytes = self.current_memory;
        }
    }

    /// Invalidate entries matching a predicate
    pub fn invalidate_where<F>(&mut self, predicate: F)
    where
        F: Fn(&CacheKey) -> bool,
    {
        let keys_to_remove: Vec<CacheKey> = self
            .entries
            .keys()
            .filter(|k| predicate(k))
            .cloned()
            .collect();

        for key in keys_to_remove {
            self.remove(&key);
            self.stats.invalidations += 1;
        }
    }

    /// Prune all expired entries
    pub fn prune_expired(&mut self) {
        let expired: Vec<CacheKey> = self
            .entries
            .iter()
            .filter(|(_, v)| v.is_expired())
            .map(|(k, _)| k.clone())
            .collect();

        for key in expired {
            self.remove(&key);
            self.stats.expirations += 1;
        }
    }

    /// Clear all entries
    pub fn clear(&mut self) {
        self.entries.clear();
        self.dependency_index.clear();
        self.current_memory = 0;
        self.stats.entry_count = 0;
        self.stats.memory_bytes = 0;
    }

    /// Get cache statistics
    pub fn stats(&self) -> &ResultCacheStats {
        &self.stats
    }

    /// Evict one entry (lowest eviction score)
    fn evict_one(&mut self) {
        let victim = self
            .entries
            .iter()
            .min_by_key(|(_, v)| v.eviction_score())
            .map(|(k, _)| k.clone());

        if let Some(key) = victim {
            self.remove(&key);
            self.stats.evictions += 1;
        }
    }
}

// ============================================================================
// Materialized View Cache
// ============================================================================

/// Definition of a materialized view
#[derive(Debug, Clone)]
pub struct MaterializedViewDef {
    /// View name
    pub name: String,
    /// Query that populates the view
    pub query: String,
    /// Tables this view depends on
    pub dependencies: Vec<String>,
    /// Refresh policy
    pub refresh: RefreshPolicy,
    /// `WITH RETENTION <duration>` clause on CREATE MATERIALIZED VIEW
    /// (issue #584 slice 12). Persisted on the view definition; the
    /// physical sweep against view-backing rows activates once the
    /// slice-9 row-storage follow-up lands.
    pub retention_duration_ms: Option<u64>,
}

/// How to refresh a materialized view
#[derive(Debug, Clone)]
pub enum RefreshPolicy {
    /// Refresh on demand only
    Manual,
    /// Refresh when dependencies change
    OnChange,
    /// Refresh periodically
    Periodic(Duration),
    /// Refresh after N invalidations
    AfterWrites(usize),
}

/// Materialized view cache entry
struct MaterializedView {
    /// The cached result
    data: Vec<u8>,
    /// View definition
    def: MaterializedViewDef,
    /// When last refreshed (logical clock — `Instant::now()` in
    /// production, fed by callers for fake-clock unit tests).
    last_refresh: Instant,
    /// Wall-clock timestamp of the last successful or attempted
    /// refresh — surfaced via `red.materialized_views.last_refresh_at`.
    last_refresh_at_ms: u64,
    /// Duration of the last refresh attempt (success or failure).
    last_refresh_duration_ms: u64,
    /// Last refresh error, cleared on next successful refresh.
    last_error: Option<String>,
    /// Row count of the most recent successful refresh.
    current_row_count: u64,
    /// Scheduled refresh cadence in milliseconds (mirrors
    /// `RefreshPolicy::Periodic`). `None` means refresh-on-demand.
    refresh_every_ms: Option<u64>,
    /// `WITH RETENTION <duration>` from CREATE MATERIALIZED VIEW
    /// (issue #584 slice 12).
    view_retention_ms: Option<u64>,
    /// Write count since last refresh
    writes_since_refresh: usize,
    /// Whether the view is stale
    stale: bool,
}

/// Public snapshot of a materialized view's runtime state. Returned
/// by `MaterializedViewCache::metadata` for the `red.materialized_views`
/// virtual table.
#[derive(Debug, Clone)]
pub struct MaterializedViewMetadata {
    pub name: String,
    pub query_text: String,
    pub refresh_every_ms: Option<u64>,
    pub last_refresh_at_ms: u64,
    pub last_refresh_duration_ms: u64,
    pub last_error: Option<String>,
    pub current_row_count: u64,
    /// `WITH RETENTION <duration>` clause from CREATE MATERIALIZED VIEW
    /// (issue #584 slice 12).
    pub retention_duration_ms: Option<u64>,
}

/// Cache for materialized views
pub struct MaterializedViewCache {
    /// Views by name
    views: HashMap<String, MaterializedView>,
    /// Dependency index: table -> view names
    dependency_index: HashMap<String, HashSet<String>>,
}

impl MaterializedViewCache {
    /// Create a new materialized view cache
    pub fn new() -> Self {
        Self {
            views: HashMap::new(),
            dependency_index: HashMap::new(),
        }
    }

    /// Register a view definition
    pub fn register(&mut self, def: MaterializedViewDef) {
        // Index dependencies
        for dep in &def.dependencies {
            self.dependency_index
                .entry(dep.clone())
                .or_default()
                .insert(def.name.clone());
        }

        let refresh_every_ms = match &def.refresh {
            RefreshPolicy::Periodic(d) => Some(d.as_millis() as u64),
            _ => None,
        };
        let view_retention_ms = def.retention_duration_ms;

        let view = MaterializedView {
            data: Vec::new(),
            def,
            last_refresh: Instant::now(),
            last_refresh_at_ms: 0,
            last_refresh_duration_ms: 0,
            last_error: None,
            current_row_count: 0,
            refresh_every_ms,
            view_retention_ms,
            writes_since_refresh: 0,
            stale: true,
        };

        self.views.insert(view.def.name.clone(), view);
    }

    /// Atomically claim the set of views due for periodic refresh at
    /// the given logical instant. Returned views have their
    /// `last_refresh` pre-advanced to `now` so a concurrent caller on
    /// the same tick sees them as not-due (the contract requested by
    /// the "exactly once per tick, no duplicates" acceptance criterion
    /// in issue #583 slice 10).
    pub fn claim_due_at(&mut self, now: Instant) -> Vec<String> {
        let mut due = Vec::new();
        for view in self.views.values_mut() {
            if let RefreshPolicy::Periodic(interval) = &view.def.refresh {
                let elapsed = now.saturating_duration_since(view.last_refresh);
                if elapsed >= *interval {
                    due.push(view.def.name.clone());
                    view.last_refresh = now;
                }
            }
        }
        due
    }

    /// Record a successful refresh — updates cached data, row count,
    /// duration, wall-clock timestamp, and clears any prior error.
    pub fn record_refresh_success(
        &mut self,
        name: &str,
        data: Vec<u8>,
        row_count: u64,
        duration_ms: u64,
        at_unix_ms: u64,
    ) {
        if let Some(view) = self.views.get_mut(name) {
            view.data = data;
            view.last_refresh = Instant::now();
            view.last_refresh_at_ms = at_unix_ms;
            view.last_refresh_duration_ms = duration_ms;
            view.last_error = None;
            view.current_row_count = row_count;
            view.writes_since_refresh = 0;
            view.stale = false;
        }
    }

    /// Record a refresh failure — captures the error string and
    /// duration but leaves `data` / `current_row_count` intact so
    /// the prior content remains readable (per acceptance criterion).
    pub fn record_refresh_failure(
        &mut self,
        name: &str,
        error: String,
        duration_ms: u64,
        at_unix_ms: u64,
    ) {
        if let Some(view) = self.views.get_mut(name) {
            view.last_refresh = Instant::now();
            view.last_refresh_at_ms = at_unix_ms;
            view.last_refresh_duration_ms = duration_ms;
            view.last_error = Some(error);
        }
    }

    /// Per-view runtime state snapshot — feeds `red.materialized_views`.
    pub fn metadata(&self) -> Vec<MaterializedViewMetadata> {
        self.views
            .values()
            .map(|v| MaterializedViewMetadata {
                name: v.def.name.clone(),
                query_text: v.def.query.clone(),
                refresh_every_ms: v.refresh_every_ms,
                last_refresh_at_ms: v.last_refresh_at_ms,
                last_refresh_duration_ms: v.last_refresh_duration_ms,
                last_error: v.last_error.clone(),
                current_row_count: v.current_row_count,
                retention_duration_ms: v.view_retention_ms,
            })
            .collect()
    }

    /// Get view data (if not stale)
    pub fn get(&self, name: &str) -> Option<&[u8]> {
        self.views
            .get(name)
            .filter(|v| !v.stale && !v.data.is_empty())
            .map(|v| v.data.as_slice())
    }

    /// Check if view needs refresh
    pub fn needs_refresh(&self, name: &str) -> bool {
        self.views.get(name).map(|v| v.stale).unwrap_or(false)
    }

    /// Refresh a view with new data
    pub fn refresh(&mut self, name: &str, data: Vec<u8>) {
        if let Some(view) = self.views.get_mut(name) {
            view.data = data;
            view.last_refresh = Instant::now();
            view.writes_since_refresh = 0;
            view.stale = false;
        }
    }

    /// Mark views depending on a table as stale
    pub fn mark_stale(&mut self, table: &str) {
        if let Some(view_names) = self.dependency_index.get(table) {
            for name in view_names.clone() {
                if let Some(view) = self.views.get_mut(&name) {
                    view.writes_since_refresh += 1;

                    match &view.def.refresh {
                        RefreshPolicy::OnChange => {
                            view.stale = true;
                        }
                        RefreshPolicy::AfterWrites(threshold)
                            if view.writes_since_refresh >= *threshold =>
                        {
                            view.stale = true;
                        }
                        _ => {}
                    }
                }
            }
        }
    }

    /// Get views needing periodic refresh
    pub fn due_for_refresh(&self) -> Vec<String> {
        self.views
            .values()
            .filter(|v| {
                if let RefreshPolicy::Periodic(interval) = &v.def.refresh {
                    v.last_refresh.elapsed() >= *interval
                } else {
                    false
                }
            })
            .map(|v| v.def.name.clone())
            .collect()
    }

    /// Remove a view
    pub fn remove(&mut self, name: &str) {
        if let Some(view) = self.views.remove(name) {
            for dep in &view.def.dependencies {
                if let Some(names) = self.dependency_index.get_mut(dep) {
                    names.remove(name);
                }
            }
        }
    }

    /// List all view names
    pub fn list(&self) -> Vec<&str> {
        self.views.keys().map(|s| s.as_str()).collect()
    }
}

impl Default for MaterializedViewCache {
    fn default() -> Self {
        Self::new()
    }
}

// ============================================================================
// Tests
// ============================================================================

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

    fn mk_periodic_view(name: &str, ms: u64) -> MaterializedViewDef {
        MaterializedViewDef {
            name: name.into(),
            query: "<test>".into(),
            dependencies: vec![],
            refresh: RefreshPolicy::Periodic(Duration::from_millis(ms)),
            retention_duration_ms: None,
        }
    }

    /// Issue #583 slice 10 — `claim_due_at` returns due views and
    /// pre-advances their `last_refresh`, so a second tick at the
    /// same instant returns an empty set.
    #[test]
    fn test_materialized_view_claim_due_exactly_once_per_tick() {
        let mut cache = MaterializedViewCache::new();
        cache.register(mk_periodic_view("v1", 100));

        let t0 = Instant::now();
        // No time has passed — not due yet.
        assert!(cache.claim_due_at(t0).is_empty());

        // Advance past the interval.
        let t1 = t0 + Duration::from_millis(150);
        let due = cache.claim_due_at(t1);
        assert_eq!(due, vec!["v1".to_string()]);

        // Same tick → already claimed, must not duplicate.
        assert!(cache.claim_due_at(t1).is_empty());

        // Advance another full interval → due again exactly once.
        let t2 = t1 + Duration::from_millis(150);
        assert_eq!(cache.claim_due_at(t2), vec!["v1".to_string()]);
    }

    /// `Manual` views never appear in `claim_due_at`'s output.
    #[test]
    fn test_materialized_view_claim_due_skips_manual_views() {
        let mut cache = MaterializedViewCache::new();
        cache.register(MaterializedViewDef {
            name: "m".into(),
            query: "<test>".into(),
            dependencies: vec![],
            refresh: RefreshPolicy::Manual,
            retention_duration_ms: None,
        });
        let t = Instant::now() + Duration::from_secs(60);
        assert!(cache.claim_due_at(t).is_empty());
    }

    /// `record_refresh_failure` keeps prior data + row count intact
    /// and stores the error string; the next success clears it.
    #[test]
    fn test_materialized_view_failure_preserves_prior_content() {
        let mut cache = MaterializedViewCache::new();
        cache.register(mk_periodic_view("v", 100));

        cache.record_refresh_success("v", b"first-payload".to_vec(), 42, 7, 1_000);
        {
            let md = cache.metadata();
            let entry = md.iter().find(|m| m.name == "v").unwrap();
            assert_eq!(entry.current_row_count, 42);
            assert!(entry.last_error.is_none());
        }

        cache.record_refresh_failure("v", "boom".into(), 3, 2_000);
        {
            let md = cache.metadata();
            let entry = md.iter().find(|m| m.name == "v").unwrap();
            // Prior content intact.
            assert_eq!(entry.current_row_count, 42);
            assert_eq!(entry.last_error.as_deref(), Some("boom"));
            assert_eq!(entry.last_refresh_at_ms, 2_000);
        }

        cache.record_refresh_success("v", b"second".to_vec(), 7, 4, 3_000);
        {
            let md = cache.metadata();
            let entry = md.iter().find(|m| m.name == "v").unwrap();
            assert_eq!(entry.current_row_count, 7);
            assert!(entry.last_error.is_none());
        }
    }

    /// DROP cleanup: `remove` drops the view from both indices and
    /// future `claim_due_at` ticks see nothing — proves no leaked
    /// scheduled state remains after the view is dropped.
    #[test]
    fn test_materialized_view_drop_cleans_scheduled_work() {
        let mut cache = MaterializedViewCache::new();
        cache.register(mk_periodic_view("v", 50));
        cache.remove("v");
        let t = Instant::now() + Duration::from_secs(10);
        assert!(cache.claim_due_at(t).is_empty());
        assert!(cache.metadata().is_empty());
    }

    /// `metadata()` reflects the seven columns surfaced by
    /// `red.materialized_views`.
    #[test]
    fn test_materialized_view_metadata_exposes_seven_fields() {
        let mut cache = MaterializedViewCache::new();
        cache.register(mk_periodic_view("v", 500));
        let md = cache.metadata();
        assert_eq!(md.len(), 1);
        let m = &md[0];
        assert_eq!(m.name, "v");
        assert_eq!(m.refresh_every_ms, Some(500));
        assert_eq!(m.last_refresh_at_ms, 0);
        assert_eq!(m.last_refresh_duration_ms, 0);
        assert!(m.last_error.is_none());
        assert_eq!(m.current_row_count, 0);
    }

    #[test]
    fn test_cache_key_hashing() {
        let key1 = CacheKey::new("attack_paths")
            .param("from", "host1")
            .param("to", "host2");

        let key2 = CacheKey::new("attack_paths")
            .param("to", "host2")
            .param("from", "host1"); // Different order

        assert_eq!(key1, key2);
        assert_eq!(key1.hash, key2.hash);
    }

    #[test]
    fn test_result_cache_basic() {
        let mut cache = ResultCache::new(1024 * 1024); // 1MB

        let key = CacheKey::new("test_query").param("id", "123");
        let data = vec![1, 2, 3, 4, 5];

        cache.insert(key.clone(), data.clone(), CachePolicy::default());

        let result = cache.get(&key);
        assert_eq!(result, Some(data));
        assert_eq!(cache.stats().hits, 1);
    }

    #[test]
    fn test_cache_expiration() {
        let mut cache = ResultCache::new(1024 * 1024);

        let key = CacheKey::new("test");
        let data = vec![1, 2, 3];

        // Very short TTL
        cache.insert(
            key.clone(),
            data,
            CachePolicy::default().ttl(Duration::from_millis(1)),
        );

        // Wait for expiration
        std::thread::sleep(Duration::from_millis(10));

        assert!(cache.get(&key).is_none());
        assert_eq!(cache.stats().expirations, 1);
    }

    #[test]
    fn test_dependency_invalidation() {
        let mut cache = ResultCache::new(1024 * 1024);

        let key = CacheKey::new("host_query");
        cache.insert(
            key.clone(),
            vec![1, 2, 3],
            CachePolicy::default().depends_on(&["hosts"]),
        );

        assert!(cache.contains(&key));

        // Invalidate hosts table
        cache.invalidate_by_dependency("hosts");

        assert!(!cache.contains(&key));
        assert_eq!(cache.stats().invalidations, 1);
    }

    #[test]
    fn test_memory_eviction() {
        let mut cache = ResultCache::new(100); // Very small

        // Insert enough to trigger eviction
        for i in 0..10 {
            let key = CacheKey::new("query").param("i", i.to_string());
            cache.insert(key, vec![0u8; 20], CachePolicy::default());
        }

        // Should have evicted some entries
        assert!(cache.stats().evictions > 0);
        assert!(cache.stats().memory_bytes <= 100);
    }

    #[test]
    fn test_materialized_view() {
        let mut cache = MaterializedViewCache::new();

        cache.register(MaterializedViewDef {
            name: "active_hosts".to_string(),
            query: "SELECT * FROM hosts WHERE status = 'active'".to_string(),
            dependencies: vec!["hosts".to_string()],
            refresh: RefreshPolicy::OnChange,
            retention_duration_ms: None,
        });

        // Initially stale
        assert!(cache.needs_refresh("active_hosts"));

        // Refresh it
        cache.refresh("active_hosts", vec![1, 2, 3]);
        assert!(!cache.needs_refresh("active_hosts"));
        assert_eq!(cache.get("active_hosts"), Some(&[1, 2, 3][..]));

        // Mark stale due to dependency
        cache.mark_stale("hosts");
        assert!(cache.needs_refresh("active_hosts"));
    }
}