fraiseql_core/cache/

response_cache.rs

//! Executor-level response cache.
//!
//! Caches the final projected GraphQL response value (after RBAC filtering,
//! projection, and envelope wrapping) to skip all redundant work on cache
//! hits for the same user + query combination.
//!
//! This is a **second cache tier** above the adapter-level row cache:
//! - Row cache: raw JSONB rows, shared across projection shapes
//! - Response cache: final `serde_json::Value`, keyed per (query + security context)
//!
//! ## Performance characteristics
//!
//! Backed by `moka::sync::Cache` (W-TinyLFU, lock-free reads). Invalidation
//! uses a `DashMap` reverse index (view name → set of keys), enabling O(k)
//! eviction without scanning the full cache.
//!
//! ## Security
//!
//! The response cache key includes a hash of the `SecurityContext` fields
//! that affect response content (`user_id`, roles, `tenant_id`, scopes,
//! attributes). Different RBAC scopes produce different cache entries.

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

use dashmap::{DashMap, DashSet};
use fraiseql_db::ViewName;
use moka::sync::Cache as MokaCache;
use serde_json::Value;

use crate::{error::Result, security::SecurityContext};

/// Configuration for the response cache.
#[derive(Debug, Clone, Copy)]
pub struct ResponseCacheConfig {
    /// Enable the response cache.
    pub enabled: bool,

    /// Maximum number of cached responses.
    pub max_entries: usize,

    /// TTL in seconds (0 = no time-based expiry, live until invalidated).
    pub ttl_seconds: u64,
}

impl Default for ResponseCacheConfig {
    fn default() -> Self {
        Self {
            enabled:     false, // opt-in
            max_entries: 10_000,
            ttl_seconds: 300,
        }
    }
}

/// Per-entry value stored in moka alongside the response value.
///
/// Contains the accessed views so the eviction listener can clean up
/// the reverse index when an entry is evicted by TTL or LFU policy.
struct ResponseEntry {
    /// The projected GraphQL response value.
    response: Arc<Value>,

    /// Views accessed by this query (for invalidation).
    ///
    /// Stored as a boxed slice of [`ViewName`] (each backed by `Arc<str>`)
    /// so cloning a name into the view index is a cheap atomic ref-count
    /// bump rather than a fresh heap allocation.
    accessed_views: Box<[ViewName]>,

    /// Process-unique identity of this entry instance.
    ///
    /// See [`crate::cache::result`]'s module docs: the reverse index registers
    /// `(key, epoch)` so the eviction listener deregisters the displaced entry
    /// and never the live replacement under the same key (#740).
    epoch: u64,
}

/// A reverse-index registration: the `(query_hash, security_hash)` cache key
/// plus the epoch of the entry instance that registered it.
type IndexRef = ((u64, u64), u64);

/// Executor-level cache for projected GraphQL responses.
///
/// Stores the final serialized response keyed by `(query_hash, security_hash)`.
/// On hit, the entire projection + RBAC + serialization pipeline is skipped.
///
/// # Thread Safety
///
/// `moka::sync::Cache` is `Send + Sync` with lock-free reads. The view reverse
/// index uses `DashMap` (fine-grained shard locking). There is no global mutex
/// on the read path.
pub struct ResponseCache {
    store: MokaCache<(u64, u64), Arc<ResponseEntry>>,

    /// Reverse index: view name → set of `((query_hash, sec_hash), epoch)`.
    ///
    /// Maintained in `put()` and pruned by the moka eviction listener.
    /// Enables O(k) invalidation without scanning the full cache.
    /// Keyed by [`ViewName`] so the entry shares its `Arc<str>` allocation
    /// with the names stored on each cache entry.
    view_index: Arc<DashMap<ViewName, DashSet<IndexRef>>>,

    enabled:    bool,
    hits:       AtomicU64,
    misses:     AtomicU64,
    next_epoch: AtomicU64,

    /// Bumped by every invalidation, so a read that straddled one can tell (#1079).
    ///
    /// Same defect and same fence as `QueryResultCache`: registering the key before
    /// `store.insert` makes it *visible* to a concurrent `invalidate_views`, but does not
    /// stop the insert that follows. Whole GraphQL responses are stranded here, not just
    /// row sets, so fixing only the row cache would leave the user-visible half live.
    invalidation_generation: AtomicU64,
}

impl ResponseCache {
    /// Create a new response cache from configuration.
    #[must_use]
    pub fn new(config: ResponseCacheConfig) -> Self {
        let view_index: Arc<DashMap<ViewName, DashSet<IndexRef>>> = Arc::new(DashMap::new());
        let vi = Arc::clone(&view_index);

        let mut builder = MokaCache::builder()
            .max_capacity(config.max_entries as u64)
            .eviction_listener(move |key: Arc<(u64, u64)>, value: Arc<ResponseEntry>, _cause| {
                // Deregister this entry INSTANCE, not the key: a re-cached
                // response is live under the same key while moka reports the
                // displaced one, and pruning by key alone would detach it from
                // every invalidation path (#740). `_cause` is deliberately
                // unused — see `crate::cache::result`'s module docs.
                let reference: IndexRef = (*key, value.epoch);
                for view in &value.accessed_views {
                    if let Some(keys) = vi.get(view) {
                        keys.remove(&reference);
                    }
                }
            });

        if config.ttl_seconds > 0 {
            builder = builder.time_to_live(Duration::from_secs(config.ttl_seconds));
        }

        let store = builder.build();

        Self {
            store,
            view_index,
            enabled: config.enabled,
            hits: AtomicU64::new(0),
            misses: AtomicU64::new(0),
            next_epoch: AtomicU64::new(0),
            invalidation_generation: AtomicU64::new(0),
        }
    }

    /// Whether the response cache is enabled.
    #[must_use]
    pub const fn is_enabled(&self) -> bool {
        self.enabled
    }

    /// Look up a cached response.
    ///
    /// # Errors
    ///
    /// This method is infallible with the moka backend and always returns `Ok`.
    pub fn get(&self, query_key: u64, security_hash: u64) -> Result<Option<Arc<Value>>> {
        if !self.enabled {
            return Ok(None);
        }

        let key = (query_key, security_hash);
        if let Some(entry) = self.store.get(&key) {
            self.hits.fetch_add(1, Ordering::Relaxed);
            Ok(Some(Arc::clone(&entry.response)))
        } else {
            self.misses.fetch_add(1, Ordering::Relaxed);
            Ok(None)
        }
    }

    /// Snapshot of the invalidation generation, for fencing a cache write (#1079).
    ///
    /// Take this before the work whose result is being cached, and pass it to
    /// [`put`](Self::put).
    #[must_use]
    pub fn invalidation_generation(&self) -> u64 {
        self.invalidation_generation.load(Ordering::Acquire)
    }

    /// Store a response in the cache.
    ///
    /// `fence` is a snapshot from [`invalidation_generation`](Self::invalidation_generation)
    /// taken before the work that produced `response`. If an invalidation landed since,
    /// the entry is discarded rather than stored — the caller keeps its response, it is
    /// simply not cached. `None` stores unconditionally.
    ///
    /// # Errors
    ///
    /// This method is infallible with the moka backend and always returns `Ok`.
    pub fn put(
        &self,
        query_key: u64,
        security_hash: u64,
        response: Arc<Value>,
        accessed_views: Vec<String>,
        fence: Option<u64>,
    ) -> Result<()> {
        if !self.enabled {
            return Ok(());
        }

        // Cheap pre-check; the authoritative one is after registration (#1079).
        if fence.is_some_and(|snapshot| snapshot != self.invalidation_generation()) {
            return Ok(());
        }

        let key = (query_key, security_hash);

        // Promote owned `String` view names into `ViewName(Arc<str>)` exactly
        // once; the same Arc is then shared by `view_index` and the cached
        // entry.
        let accessed_views: Box<[ViewName]> =
            accessed_views.into_iter().map(ViewName::from).collect();

        // Update view → key reverse index before inserting into the store,
        // so invalidate_views() called concurrently won't miss the key.
        let epoch = self.next_epoch.fetch_add(1, Ordering::Relaxed);
        let reference: IndexRef = (key, epoch);
        for view in &accessed_views {
            self.view_index.entry(view.clone()).or_default().insert(reference);
        }

        // The fence (#1079): authoritative check, after registration and before insert.
        // Deregistration is symmetric with the eviction listener's, so no index row
        // survives pointing at a key that was never stored.
        if fence.is_some_and(|snapshot| snapshot != self.invalidation_generation()) {
            for view in &accessed_views {
                if let Some(keys) = self.view_index.get(view) {
                    keys.remove(&reference);
                }
            }
            return Ok(());
        }

        let entry = Arc::new(ResponseEntry {
            response,
            accessed_views,
            epoch,
        });

        self.store.insert(key, entry);
        Ok(())
    }

    /// Invalidate all entries that access any of the given views.
    ///
    /// Uses the O(k) reverse index — no full-cache scan.
    ///
    /// # Errors
    ///
    /// This method is infallible with the moka backend and always returns `Ok`.
    pub fn invalidate_views(&self, views: &[ViewName]) -> Result<u64> {
        // Bump before collecting keys (#1079) — the window may be too wide, never too narrow.
        self.invalidation_generation.fetch_add(1, Ordering::Release);

        // Dedup by cache key: an entry mid-replacement is registered under two
        // epochs, and one reading several of `views` is registered once per view.
        let mut to_remove: std::collections::HashSet<(u64, u64)> = std::collections::HashSet::new();

        for view in views {
            // ViewName: Borrow<str> — look up by &str without allocating a
            // fresh ViewName.
            if let Some(keys) = self.view_index.get(view.as_str()) {
                for reference in keys.iter() {
                    to_remove.insert(reference.0);
                }
            }
        }

        #[allow(clippy::cast_possible_truncation)]
        // Reason: entry count never exceeds u64
        let total = to_remove.len() as u64;
        for key in to_remove {
            self.store.invalidate(&key);
            // The eviction listener handles index cleanup.
        }

        Ok(total)
    }

    /// Get cache hit/miss counts.
    #[must_use]
    pub fn metrics(&self) -> (u64, u64) {
        (self.hits.load(Ordering::Relaxed), self.misses.load(Ordering::Relaxed))
    }

    /// Flush pending background tasks in the moka store.
    ///
    /// Used in tests to synchronise async eviction/invalidation before assertions.
    #[cfg(test)]
    pub(crate) fn run_pending_tasks(&self) {
        self.store.run_pending_tasks();
    }
}

/// Hash the security context fields that affect response content.
///
/// Fields hashed: `user_id`, roles (sorted), `tenant_id`, scopes (sorted),
/// `attributes` (sorted keys + JSON-serialized values).
///
/// Fields NOT hashed: `request_id`, `ip_address`, `authenticated_at`, `expires_at`,
/// `issuer`, `audience` — these don't affect which data the user can see.
///
/// `attributes` IS hashed because custom RLS policies can key on arbitrary
/// attributes (e.g., "department", "region") to produce different query results
/// for users who otherwise share the same `user_id`/roles/`tenant_id`/scopes.
///
/// Returns `0` when no security context is present (all users share one entry).
#[must_use]
pub fn hash_security_context(ctx: Option<&SecurityContext>) -> u64 {
    use std::hash::{Hash, Hasher};

    let Some(ctx) = ctx else {
        return 0;
    };

    let mut hasher = ahash::AHasher::default();
    ctx.user_id.hash(&mut hasher);

    // Sort roles for determinism (JWT may present them in any order)
    let mut sorted_roles = ctx.roles.clone();
    sorted_roles.sort();
    for role in &sorted_roles {
        role.hash(&mut hasher);
    }

    ctx.tenant_id.hash(&mut hasher);

    let mut sorted_scopes = ctx.scopes.clone();
    sorted_scopes.sort();
    for scope in &sorted_scopes {
        scope.hash(&mut hasher);
    }

    // Hash attributes (custom RLS policies can key on these)
    if !ctx.attributes.is_empty() {
        let mut attr_keys: Vec<&String> = ctx.attributes.keys().collect();
        attr_keys.sort();
        for key in attr_keys {
            key.hash(&mut hasher);
            // Use JSON serialization for deterministic Value hashing
            serde_json::to_string(&ctx.attributes[key])
                .unwrap_or_default()
                .hash(&mut hasher);
        }
    }

    hasher.finish()
}