sochdb-storage 0.4.7

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//! Application-Level Page Cache with Clock-Pro (Recommendation 8)
//!
//! ## Problem
//!
//! SQLite maintains a sophisticated page cache with:
//! - LRU-K eviction policy
//! - Dirty page tracking
//! - Read-ahead for sequential scans
//!
//! SochDB currently relies on OS page cache (mmap) which:
//! - Has no application-level control
//! - Cannot prioritize hot pages
//! - Has 4KB granularity (vs optimal 64KB for SSTable blocks)
//!
//! ## Solution
//!
//! Clock-Pro algorithm (adaptive replacement cache):
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────┐
//! │                    Clock-Pro Cache                       │
//! ├─────────────────────────────────────────────────────────┤
//! │  Hot Ring (frequently accessed)                         │
//! │    ↻ Clock hand for hot pages                           │
//! ├─────────────────────────────────────────────────────────┤
//! │  Cold Ring (recently accessed once)                      │
//! │    ↻ Clock hand for cold pages                          │
//! ├─────────────────────────────────────────────────────────┤
//! │  Test Ring (ghost entries for adaptive sizing)          │
//! │    Tracks recently evicted to detect reuse              │
//! └─────────────────────────────────────────────────────────┘
//! ```
//!
//! ## Performance Analysis
//!
//! Clock-Pro complexity:
//! - Insert: O(1)
//! - Lookup: O(1)
//! - Eviction: O(1) amortized
//!
//! Hit rate improvement over LRU:
//! - LRU: 90% hit rate (typical)
//! - Clock-Pro: 95-99% hit rate (adaptive)
//!
//! I/O reduction:
//! ```text
//! With 1GB cache, 100GB dataset, 1% access skew:
//! LRU: 10% miss × 100M accesses = 10M I/O ops
//! Clock-Pro: 2% miss × 100M accesses = 2M I/O ops (5x reduction)
//! ```

use std::hash::Hash;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;

use dashmap::DashMap;
use parking_lot::RwLock;

/// Default page size (64KB - optimized for SSTable blocks)
pub const DEFAULT_PAGE_SIZE: usize = 64 * 1024;

/// Default cache capacity in pages
pub const DEFAULT_CACHE_PAGES: usize = 16_384; // 1GB cache

/// Minimum hot ring size (fraction of total)
pub const MIN_HOT_RATIO: f64 = 0.05;

/// Maximum hot ring size (fraction of total)
pub const MAX_HOT_RATIO: f64 = 0.95;

// =============================================================================
// Page Identifier
// =============================================================================

/// Unique identifier for a page
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct PageId {
    /// File identifier (could be table_id, sstable_id, etc.)
    pub file_id: u64,
    /// Page number within file
    pub page_no: u64,
}

impl PageId {
    pub fn new(file_id: u64, page_no: u64) -> Self {
        Self { file_id, page_no }
    }
}

// =============================================================================
// Page Entry
// =============================================================================

/// State of a page in the cache
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PageState {
    /// In hot ring (frequently accessed)
    Hot,
    /// In cold ring (recently accessed once)
    Cold,
    /// In test ring (ghost entry, metadata only)
    Test,
}

/// A cached page
pub struct CachedPage {
    /// Page identifier
    #[allow(dead_code)]
    id: PageId,
    /// Page data
    data: Vec<u8>,
    /// Reference bit (for clock algorithm)
    referenced: AtomicBool,
    /// Current state
    state: RwLock<PageState>,
    /// Dirty flag
    dirty: AtomicBool,
    /// Access count (for statistics)
    access_count: AtomicU64,
    /// Pin count (prevents eviction while > 0)
    pin_count: AtomicUsize,
}

impl CachedPage {
    pub fn new(id: PageId, data: Vec<u8>) -> Self {
        Self {
            id,
            data,
            referenced: AtomicBool::new(true),
            state: RwLock::new(PageState::Cold),
            dirty: AtomicBool::new(false),
            access_count: AtomicU64::new(1),
            pin_count: AtomicUsize::new(0),
        }
    }

    /// Get page data
    pub fn data(&self) -> &[u8] {
        &self.data
    }

    /// Get mutable page data
    pub fn data_mut(&mut self) -> &mut [u8] {
        self.dirty.store(true, Ordering::Release);
        &mut self.data
    }

    /// Check if dirty
    pub fn is_dirty(&self) -> bool {
        self.dirty.load(Ordering::Acquire)
    }

    /// Mark as clean
    pub fn mark_clean(&self) {
        self.dirty.store(false, Ordering::Release);
    }

    /// Pin the page (prevent eviction)
    pub fn pin(&self) {
        self.pin_count.fetch_add(1, Ordering::AcqRel);
    }

    /// Unpin the page
    pub fn unpin(&self) {
        self.pin_count.fetch_sub(1, Ordering::AcqRel);
    }

    /// Check if pinned
    pub fn is_pinned(&self) -> bool {
        self.pin_count.load(Ordering::Acquire) > 0
    }

    /// Touch (mark as referenced)
    pub fn touch(&self) {
        self.referenced.store(true, Ordering::Release);
        self.access_count.fetch_add(1, Ordering::Relaxed);
    }

    /// Get state
    pub fn state(&self) -> PageState {
        *self.state.read()
    }

    /// Set state
    pub fn set_state(&self, state: PageState) {
        *self.state.write() = state;
    }
}

// =============================================================================
// Clock Ring
// =============================================================================

/// A clock ring for FIFO-like eviction with reference bits
struct ClockRing<K: Clone + Eq + Hash> {
    /// Entries in ring order
    entries: Vec<K>,
    /// Current hand position
    hand: AtomicUsize,
    /// Maximum size
    capacity: usize,
}

impl<K: Clone + Eq + Hash> ClockRing<K> {
    fn new(capacity: usize) -> Self {
        Self {
            entries: Vec::with_capacity(capacity),
            hand: AtomicUsize::new(0),
            capacity,
        }
    }

    /// Insert entry at current position
    fn insert(&mut self, key: K) -> Option<K> {
        if self.entries.len() < self.capacity {
            self.entries.push(key);
            return None;
        }

        // Ring is full, replace at hand position
        let pos = self.hand.load(Ordering::Relaxed) % self.entries.len();
        let evicted = std::mem::replace(&mut self.entries[pos], key);
        self.advance_hand();
        Some(evicted)
    }

    /// Remove entry
    fn remove(&mut self, key: &K) -> bool {
        if let Some(pos) = self.entries.iter().position(|k| k == key) {
            self.entries.remove(pos);
            // Adjust hand if needed
            let hand = self.hand.load(Ordering::Relaxed);
            if pos < hand && hand > 0 {
                self.hand.fetch_sub(1, Ordering::Relaxed);
            }
            return true;
        }
        false
    }

    /// Advance clock hand
    fn advance_hand(&self) {
        if !self.entries.is_empty() {
            self.hand.fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Get entry at hand
    fn current(&self) -> Option<K> {
        if self.entries.is_empty() {
            return None;
        }
        let pos = self.hand.load(Ordering::Relaxed) % self.entries.len();
        Some(self.entries[pos].clone())
    }

    /// Length
    fn len(&self) -> usize {
        self.entries.len()
    }

    /// Check if empty
    #[allow(dead_code)]
    fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// Resize capacity
    fn set_capacity(&mut self, new_capacity: usize) {
        self.capacity = new_capacity;
    }
}

// =============================================================================
// Clock-Pro Page Cache
// =============================================================================

/// Clock-Pro adaptive page cache
///
/// ## Algorithm Overview
///
/// 1. **Cold pages**: Recently accessed once. These may be promoted
///    to hot if accessed again before eviction (reuse distance < test size).
///
/// 2. **Hot pages**: Frequently accessed. Protected from eviction
///    until demoted to cold.
///
/// 3. **Test pages**: Ghost entries tracking recently evicted cold pages.
///    If a test page is accessed, it indicates we should increase hot space.
///
/// ## Adaptive Sizing
///
/// The hot/cold ratio adjusts based on access patterns:
/// - Access to test page → increase hot space (working set growing)
/// - Cold page eviction without test hit → decrease hot space
pub struct ClockProCache {
    /// All cached pages
    pages: DashMap<PageId, Arc<CachedPage>>,
    /// Hot ring (frequently accessed)
    hot_ring: RwLock<ClockRing<PageId>>,
    /// Cold ring (recently accessed once)
    cold_ring: RwLock<ClockRing<PageId>>,
    /// Test ring (ghost entries)
    test_ring: RwLock<ClockRing<PageId>>,
    /// Total capacity in pages
    capacity: usize,
    /// Current hot size target
    hot_target: AtomicUsize,
    /// Statistics
    stats: CacheStats,
    /// Page size
    page_size: usize,
}

/// Cache statistics
#[derive(Debug, Default)]
pub struct CacheStats {
    /// Total hits
    pub hits: AtomicU64,
    /// Total misses
    pub misses: AtomicU64,
    /// Hot hits
    pub hot_hits: AtomicU64,
    /// Cold hits (promoted to hot)
    pub cold_hits: AtomicU64,
    /// Test hits (adaptive)
    pub test_hits: AtomicU64,
    /// Evictions
    pub evictions: AtomicU64,
    /// Dirty evictions (requiring flush)
    pub dirty_evictions: AtomicU64,
}

impl CacheStats {
    pub fn hit_rate(&self) -> f64 {
        let hits = self.hits.load(Ordering::Relaxed);
        let misses = self.misses.load(Ordering::Relaxed);
        let total = hits + misses;
        if total == 0 {
            return 0.0;
        }
        hits as f64 / total as f64
    }

    pub fn reset(&self) {
        self.hits.store(0, Ordering::Relaxed);
        self.misses.store(0, Ordering::Relaxed);
        self.hot_hits.store(0, Ordering::Relaxed);
        self.cold_hits.store(0, Ordering::Relaxed);
        self.test_hits.store(0, Ordering::Relaxed);
        self.evictions.store(0, Ordering::Relaxed);
        self.dirty_evictions.store(0, Ordering::Relaxed);
    }
}

impl ClockProCache {
    /// Create new cache with default settings
    pub fn new(capacity: usize) -> Self {
        Self::with_page_size(capacity, DEFAULT_PAGE_SIZE)
    }

    /// Create with specific page size
    pub fn with_page_size(capacity: usize, page_size: usize) -> Self {
        let initial_hot = capacity / 2;
        let cold_capacity = capacity - initial_hot;

        Self {
            pages: DashMap::new(),
            hot_ring: RwLock::new(ClockRing::new(initial_hot)),
            cold_ring: RwLock::new(ClockRing::new(cold_capacity)),
            test_ring: RwLock::new(ClockRing::new(capacity)), // Test can grow to full size
            capacity,
            hot_target: AtomicUsize::new(initial_hot),
            stats: CacheStats::default(),
            page_size,
        }
    }

    /// Get a page from cache
    pub fn get(&self, page_id: &PageId) -> Option<Arc<CachedPage>> {
        if let Some(page) = self.pages.get(page_id) {
            let page = page.clone();
            page.touch();

            match page.state() {
                PageState::Hot => {
                    self.stats.hot_hits.fetch_add(1, Ordering::Relaxed);
                }
                PageState::Cold => {
                    // Promote to hot
                    self.promote_to_hot(page_id);
                    self.stats.cold_hits.fetch_add(1, Ordering::Relaxed);
                }
                PageState::Test => {
                    // Test hit - need to fetch from storage and increase hot
                    self.stats.test_hits.fetch_add(1, Ordering::Relaxed);
                    self.adapt_increase_hot();
                    return None; // Caller must fetch from storage
                }
            }

            self.stats.hits.fetch_add(1, Ordering::Relaxed);
            return Some(page);
        }

        // Check test ring
        if self.is_in_test(page_id) {
            self.stats.test_hits.fetch_add(1, Ordering::Relaxed);
            self.adapt_increase_hot();
        }

        self.stats.misses.fetch_add(1, Ordering::Relaxed);
        None
    }

    /// Insert a page into cache
    pub fn insert(&self, page_id: PageId, data: Vec<u8>) -> Arc<CachedPage> {
        let page = Arc::new(CachedPage::new(page_id, data));
        
        // Evict if necessary
        self.make_room();

        // Insert into cold ring initially
        {
            let mut cold = self.cold_ring.write();
            if let Some(evicted) = cold.insert(page_id) {
                self.evict(&evicted);
            }
        }

        page.set_state(PageState::Cold);
        self.pages.insert(page_id, page.clone());

        page
    }

    /// Insert with pinning
    pub fn insert_pinned(&self, page_id: PageId, data: Vec<u8>) -> Arc<CachedPage> {
        let page = self.insert(page_id, data);
        page.pin();
        page
    }

    /// Evict a page
    fn evict(&self, page_id: &PageId) {
        if let Some((_, page)) = self.pages.remove(page_id) {
            if page.is_pinned() {
                // Can't evict pinned page, re-insert
                self.pages.insert(*page_id, page);
                return;
            }

            self.stats.evictions.fetch_add(1, Ordering::Relaxed);
            if page.is_dirty() {
                self.stats.dirty_evictions.fetch_add(1, Ordering::Relaxed);
                // Caller should flush before eviction in real implementation
            }

            // Move to test ring
            {
                let mut test = self.test_ring.write();
                test.insert(*page_id);
            }
        }
    }

    /// Promote cold page to hot
    fn promote_to_hot(&self, page_id: &PageId) {
        // Remove from cold ring
        {
            let mut cold = self.cold_ring.write();
            cold.remove(page_id);
        }

        // Add to hot ring
        {
            let mut hot = self.hot_ring.write();
            if let Some(demoted) = hot.insert(*page_id) {
                // Demote a hot page to cold
                self.demote_to_cold(&demoted);
            }
        }

        if let Some(page) = self.pages.get(page_id) {
            page.set_state(PageState::Hot);
        }
    }

    /// Demote hot page to cold
    fn demote_to_cold(&self, page_id: &PageId) {
        {
            let mut cold = self.cold_ring.write();
            cold.insert(*page_id);
        }

        if let Some(page) = self.pages.get(page_id) {
            page.set_state(PageState::Cold);
            page.referenced.store(false, Ordering::Release);
        }
    }

    /// Check if page is in test ring
    fn is_in_test(&self, page_id: &PageId) -> bool {
        let test = self.test_ring.read();
        test.entries.contains(page_id)
    }

    /// Make room for new page
    fn make_room(&self) {
        let total = self.pages.len();
        if total < self.capacity {
            return;
        }

        // Try to evict from cold ring first
        let mut cold = self.cold_ring.write();
        while cold.len() > 0 && self.pages.len() >= self.capacity {
            if let Some(victim) = cold.current() {
                if let Some(page) = self.pages.get(&victim) {
                    if !page.is_pinned() && !page.referenced.load(Ordering::Acquire) {
                        cold.remove(&victim);
                        drop(cold);
                        self.evict(&victim);
                        return;
                    }
                    page.referenced.store(false, Ordering::Release);
                }
            }
            cold.advance_hand();
        }
    }

    /// Adapt: increase hot space
    fn adapt_increase_hot(&self) {
        let current = self.hot_target.load(Ordering::Relaxed);
        let max_hot = (self.capacity as f64 * MAX_HOT_RATIO) as usize;
        
        if current < max_hot {
            let new_hot = (current + 1).min(max_hot);
            self.hot_target.store(new_hot, Ordering::Relaxed);
            
            let mut hot = self.hot_ring.write();
            hot.set_capacity(new_hot);
            
            let mut cold = self.cold_ring.write();
            cold.set_capacity(self.capacity - new_hot);
        }
    }

    /// Adapt: decrease hot space
    #[allow(dead_code)]
    fn adapt_decrease_hot(&self) {
        let current = self.hot_target.load(Ordering::Relaxed);
        let min_hot = (self.capacity as f64 * MIN_HOT_RATIO) as usize;
        
        if current > min_hot {
            let new_hot = current.saturating_sub(1).max(min_hot);
            self.hot_target.store(new_hot, Ordering::Relaxed);
            
            let mut hot = self.hot_ring.write();
            hot.set_capacity(new_hot);
            
            let mut cold = self.cold_ring.write();
            cold.set_capacity(self.capacity - new_hot);
        }
    }

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

    /// Get current size
    pub fn len(&self) -> usize {
        self.pages.len()
    }

    /// Check if empty
    pub fn is_empty(&self) -> bool {
        self.pages.is_empty()
    }

    /// Get capacity
    pub fn capacity(&self) -> usize {
        self.capacity
    }

    /// Get hot ratio
    pub fn hot_ratio(&self) -> f64 {
        self.hot_target.load(Ordering::Relaxed) as f64 / self.capacity as f64
    }

    /// Clear cache
    pub fn clear(&self) {
        self.pages.clear();
        *self.hot_ring.write() = ClockRing::new(self.capacity / 2);
        *self.cold_ring.write() = ClockRing::new(self.capacity / 2);
        *self.test_ring.write() = ClockRing::new(self.capacity);
        self.stats.reset();
    }

    /// Get memory usage in bytes
    pub fn memory_usage(&self) -> usize {
        self.pages.len() * self.page_size
    }
}

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

    #[test]
    fn test_page_cache_basic() {
        let cache = ClockProCache::new(100);
        
        let page_id = PageId::new(1, 0);
        let data = vec![0u8; 1024];
        
        cache.insert(page_id, data.clone());
        
        let page = cache.get(&page_id).unwrap();
        assert_eq!(page.data(), data.as_slice());
    }

    #[test]
    fn test_page_cache_miss() {
        let cache = ClockProCache::new(100);
        
        let page_id = PageId::new(1, 0);
        assert!(cache.get(&page_id).is_none());
        
        assert_eq!(cache.stats().misses.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn test_page_cache_promotion() {
        let cache = ClockProCache::new(100);
        
        let page_id = PageId::new(1, 0);
        cache.insert(page_id, vec![0u8; 1024]);
        
        // First access - cold
        let page = cache.get(&page_id).unwrap();
        assert_eq!(page.state(), PageState::Hot); // Promoted on access
    }

    #[test]
    fn test_page_cache_eviction() {
        let cache = ClockProCache::new(10);
        
        // Fill cache
        for i in 0..15 {
            let page_id = PageId::new(1, i);
            cache.insert(page_id, vec![0u8; 64]);
        }
        
        // Should have evicted some pages
        assert!(cache.len() <= 10);
        assert!(cache.stats().evictions.load(Ordering::Relaxed) > 0);
    }

    #[test]
    fn test_page_cache_pinned() {
        let cache = ClockProCache::new(10);
        
        let page_id = PageId::new(1, 0);
        let page = cache.insert_pinned(page_id, vec![0u8; 64]);
        
        assert!(page.is_pinned());
        
        page.unpin();
        assert!(!page.is_pinned());
    }

    #[test]
    fn test_page_cache_dirty() {
        let cache = ClockProCache::new(10);
        
        let page_id = PageId::new(1, 0);
        cache.insert(page_id, vec![0u8; 64]);
        
        if let Some(page) = cache.get(&page_id) {
            assert!(!page.is_dirty());
        }
    }

    #[test]
    fn test_cache_stats() {
        let cache = ClockProCache::new(100);
        
        let page_id = PageId::new(1, 0);
        cache.insert(page_id, vec![0u8; 1024]);
        
        // Access twice
        cache.get(&page_id);
        cache.get(&page_id);
        
        assert_eq!(cache.stats().hits.load(Ordering::Relaxed), 2);
        assert!(cache.stats().hit_rate() > 0.0);
    }

    #[test]
    fn test_adaptive_hot_ratio() {
        let cache = ClockProCache::new(100);
        
        let initial_ratio = cache.hot_ratio();
        
        // Simulate test hits to increase hot space
        for _ in 0..10 {
            cache.adapt_increase_hot();
        }
        
        assert!(cache.hot_ratio() > initial_ratio);
    }
}