lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0

//! Thin pool management.
//!
//! This module provides the thin pool abstraction that manages multiple
//! thin volumes, tracks overcommit, and handles alerts.

use alloc::collections::BTreeMap;
use alloc::string::{String, ToString};
use alloc::vec::Vec;

use lazy_static::lazy_static;
use spin::Mutex;

use super::alloc::BlockAllocator;
use super::types::{
    Alert, AlertType, OvercommitPolicy, PhysicalBlock, PoolConfig, PoolStats, ThinError,
    ThinResult, ThresholdLevel, Thresholds, VirtualBlock, VolumeConfig,
};
use super::volume::ThinVolume;

// ═══════════════════════════════════════════════════════════════════════════════
// ALERT CALLBACK
// ═══════════════════════════════════════════════════════════════════════════════

/// Callback type for alerts.
pub type AlertCallback = fn(&Alert);

// ═══════════════════════════════════════════════════════════════════════════════
// THIN POOL
// ═══════════════════════════════════════════════════════════════════════════════

/// A thin provisioning pool.
#[derive(Debug)]
pub struct ThinPool {
    /// Pool ID.
    id: u64,
    /// Pool name.
    name: String,
    /// Block allocator.
    allocator: BlockAllocator,
    /// Thin volumes.
    volumes: BTreeMap<u64, ThinVolume>,
    /// Volume names to IDs.
    volume_names: BTreeMap<String, u64>,
    /// Next volume ID.
    next_volume_id: u64,
    /// Configuration.
    config: PoolConfig,
    /// Statistics.
    stats: PoolStats,
    /// Last threshold level.
    last_level: ThresholdLevel,
    /// Alert history.
    alerts: Vec<Alert>,
    /// Maximum alerts to keep.
    max_alerts: usize,
    /// Alert callback.
    alert_callback: Option<AlertCallback>,
}

impl ThinPool {
    /// Create a new thin pool.
    pub fn new(id: u64, config: PoolConfig) -> Self {
        let total_blocks = config.physical_blocks();
        let reserved = (total_blocks * config.metadata_reserve_percent as u64) / 100;
        let allocator = BlockAllocator::new(total_blocks, config.block_size, reserved);

        Self {
            id,
            name: config.name.clone(),
            allocator,
            volumes: BTreeMap::new(),
            volume_names: BTreeMap::new(),
            next_volume_id: 1,
            stats: PoolStats::new(config.capacity),
            config,
            last_level: ThresholdLevel::Normal,
            alerts: Vec::new(),
            max_alerts: 100,
            alert_callback: None,
        }
    }

    /// Get pool ID.
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Get pool name.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Get pool configuration.
    pub fn config(&self) -> &PoolConfig {
        &self.config
    }

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

    /// Set alert callback.
    pub fn set_alert_callback(&mut self, callback: AlertCallback) {
        self.alert_callback = Some(callback);
    }

    /// Get recent alerts.
    pub fn alerts(&self) -> &[Alert] {
        &self.alerts
    }

    /// Clear alerts.
    pub fn clear_alerts(&mut self) {
        self.alerts.clear();
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // VOLUME MANAGEMENT
    // ═══════════════════════════════════════════════════════════════════════════

    /// Create a new thin volume.
    pub fn create_volume(&mut self, config: VolumeConfig) -> ThinResult<u64> {
        // Check if name already exists
        if self.volume_names.contains_key(&config.name) {
            return Err(ThinError::VolumeExists(config.name));
        }

        // Check reservation
        if config.reservation > 0 {
            let available = self.allocator.free_bytes();
            if config.reservation > available {
                return Err(ThinError::AllocationFailed);
            }
        }

        let id = self.next_volume_id;
        self.next_volume_id += 1;

        let name = config.name.clone();
        let virtual_size = config.virtual_size;
        let volume = ThinVolume::new(id, config);

        self.volumes.insert(id, volume);
        self.volume_names.insert(name, id);

        // Update stats
        self.stats.volume_count += 1;
        self.stats.total_virtual += virtual_size;
        self.update_stats();

        Ok(id)
    }

    /// Delete a thin volume.
    pub fn delete_volume(&mut self, id: u64) -> ThinResult<()> {
        let volume = self
            .volumes
            .remove(&id)
            .ok_or(ThinError::VolumeNotFound(id.to_string()))?;

        // Check for snapshots
        if !volume.snapshots().is_empty() {
            // Put volume back
            self.volumes.insert(id, volume);
            return Err(ThinError::NotPermitted);
        }

        // Free all allocated blocks
        // Note: In a full implementation, we'd need to handle shared blocks
        self.volume_names.remove(volume.name());

        // Update stats
        self.stats.volume_count = self.stats.volume_count.saturating_sub(1);
        self.stats.total_virtual = self
            .stats
            .total_virtual
            .saturating_sub(volume.virtual_size());
        self.stats.physical_used = self
            .stats
            .physical_used
            .saturating_sub(volume.physical_used());
        self.update_stats();

        Ok(())
    }

    /// Get a volume by ID.
    pub fn get_volume(&self, id: u64) -> Option<&ThinVolume> {
        self.volumes.get(&id)
    }

    /// Get a mutable volume by ID.
    pub fn get_volume_mut(&mut self, id: u64) -> Option<&mut ThinVolume> {
        self.volumes.get_mut(&id)
    }

    /// Get a volume by name.
    pub fn get_volume_by_name(&self, name: &str) -> Option<&ThinVolume> {
        self.volume_names
            .get(name)
            .and_then(|id| self.volumes.get(id))
    }

    /// Get a mutable volume by name.
    pub fn get_volume_by_name_mut(&mut self, name: &str) -> Option<&mut ThinVolume> {
        let id = self.volume_names.get(name).copied();
        id.and_then(move |id| self.volumes.get_mut(&id))
    }

    /// List all volume IDs.
    pub fn volume_ids(&self) -> Vec<u64> {
        self.volumes.keys().copied().collect()
    }

    /// List all volume names.
    pub fn volume_names(&self) -> Vec<String> {
        self.volume_names.keys().cloned().collect()
    }

    /// Get volume count.
    pub fn volume_count(&self) -> usize {
        self.volumes.len()
    }

    /// Create a snapshot of a volume.
    pub fn create_snapshot(&mut self, volume_id: u64, name: String) -> ThinResult<u64> {
        if self.volume_names.contains_key(&name) {
            return Err(ThinError::VolumeExists(name));
        }

        let snapshot_id = self.next_volume_id;
        self.next_volume_id += 1;

        let volume = self
            .volumes
            .get_mut(&volume_id)
            .ok_or(ThinError::VolumeNotFound(volume_id.to_string()))?;

        let snapshot = volume.create_snapshot(snapshot_id, name.clone());
        let virtual_size = snapshot.virtual_size();

        self.volumes.insert(snapshot_id, snapshot);
        self.volume_names.insert(name, snapshot_id);

        // Update stats
        self.stats.snapshot_count += 1;
        self.stats.total_virtual += virtual_size;
        self.update_stats();

        Ok(snapshot_id)
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // BLOCK ALLOCATION
    // ═══════════════════════════════════════════════════════════════════════════

    /// Allocate a block for a volume.
    pub fn allocate_block(
        &mut self,
        volume_id: u64,
        vblock: VirtualBlock,
    ) -> ThinResult<PhysicalBlock> {
        // Check threshold policy
        self.check_allocation_allowed()?;

        let pblock = self.allocator.allocate()?;

        if let Some(volume) = self.volumes.get_mut(&volume_id) {
            volume.map_block(vblock, pblock);
        }

        // Update stats
        self.stats.physical_used += self.config.block_size;
        self.stats.physical_free = self
            .stats
            .physical_free
            .saturating_sub(self.config.block_size);
        self.update_stats();

        Ok(pblock)
    }

    /// Free a block from a volume.
    pub fn free_block(&mut self, volume_id: u64, vblock: VirtualBlock) -> Option<PhysicalBlock> {
        let pblock = if let Some(volume) = self.volumes.get_mut(&volume_id) {
            volume.unmap_block(vblock)
        } else {
            None
        };

        if let Some(pb) = pblock {
            self.allocator.free(pb);

            // Update stats
            self.stats.physical_used = self
                .stats
                .physical_used
                .saturating_sub(self.config.block_size);
            self.stats.physical_free += self.config.block_size;
            self.update_stats();
        }

        pblock
    }

    /// Check if allocation is allowed based on policy.
    fn check_allocation_allowed(&mut self) -> ThinResult<()> {
        let level = self.current_threshold_level();

        if !self.config.overcommit_policy.allows_write(level) {
            self.raise_alert(
                AlertType::PoolFull,
                level,
                "Pool threshold exceeded, writes blocked",
            );
            return Err(ThinError::PoolFull);
        }

        Ok(())
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // OVERCOMMIT TRACKING
    // ═══════════════════════════════════════════════════════════════════════════

    /// Get current threshold level.
    pub fn current_threshold_level(&self) -> ThresholdLevel {
        let percent = self.allocator.usage_percent();
        self.config.thresholds.level(percent)
    }

    /// Get overcommit ratio.
    pub fn overcommit_ratio(&self) -> f64 {
        if self.stats.total_capacity == 0 {
            return 0.0;
        }
        self.stats.total_virtual as f64 / self.stats.total_capacity as f64
    }

    /// Get physical space usage percentage.
    pub fn usage_percent(&self) -> u8 {
        self.allocator.usage_percent()
    }

    /// Get free physical space in bytes.
    pub fn free_bytes(&self) -> u64 {
        self.allocator.free_bytes()
    }

    /// Get free physical blocks.
    pub fn free_blocks(&self) -> u64 {
        self.allocator.free_blocks()
    }

    /// Update statistics and check thresholds.
    fn update_stats(&mut self) {
        self.stats.physical_used = (self.allocator.allocated_blocks()) * self.config.block_size;
        self.stats.physical_free = self.allocator.free_bytes();
        self.stats.update_overcommit();

        // Check threshold crossing
        let current_level = self.current_threshold_level();
        if current_level != self.last_level {
            if current_level > self.last_level {
                self.raise_threshold_alert(current_level);
            }
            self.last_level = current_level;
            self.stats.threshold_level = current_level;
        }
    }

    /// Raise a threshold crossing alert.
    fn raise_threshold_alert(&mut self, level: ThresholdLevel) {
        let percent = self.usage_percent();
        let msg = alloc::format!("Pool usage at {}%, entering {} level", percent, level);
        self.raise_alert(AlertType::ThresholdCrossed, level, &msg);
    }

    /// Raise an alert.
    fn raise_alert(&mut self, alert_type: AlertType, level: ThresholdLevel, message: &str) {
        let alert =
            Alert::new(alert_type, level, &self.name, message).with_usage(self.usage_percent());

        // Call callback if set
        if let Some(callback) = self.alert_callback {
            callback(&alert);
        }

        // Store alert
        self.alerts.push(alert);
        if self.alerts.len() > self.max_alerts {
            self.alerts.remove(0);
        }
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // CAPACITY MANAGEMENT
    // ═══════════════════════════════════════════════════════════════════════════

    /// Check if pool can accommodate a new volume.
    pub fn can_create_volume(&self, virtual_size: u64, reservation: u64) -> bool {
        if reservation > 0 {
            reservation <= self.allocator.free_bytes()
        } else {
            // Without reservation, always allow (thin provisioning)
            true
        }
    }

    /// Get capacity summary.
    pub fn capacity_summary(&self) -> CapacitySummary {
        CapacitySummary {
            total_physical: self.stats.total_capacity,
            used_physical: self.stats.physical_used,
            free_physical: self.stats.physical_free,
            total_virtual: self.stats.total_virtual,
            overcommit_ratio: self.overcommit_ratio(),
            usage_percent: self.usage_percent(),
            threshold_level: self.current_threshold_level(),
        }
    }

    /// Trim unused blocks (for SSD TRIM support).
    pub fn trim(&mut self) -> u64 {
        // In a full implementation, this would send TRIM commands
        // For now, just return potential TRIM blocks
        self.allocator.free_blocks()
    }
}

/// Summary of pool capacity.
#[derive(Debug, Clone)]
pub struct CapacitySummary {
    /// Total physical capacity.
    pub total_physical: u64,
    /// Used physical space.
    pub used_physical: u64,
    /// Free physical space.
    pub free_physical: u64,
    /// Total virtual space.
    pub total_virtual: u64,
    /// Overcommit ratio.
    pub overcommit_ratio: f64,
    /// Usage percentage.
    pub usage_percent: u8,
    /// Current threshold level.
    pub threshold_level: ThresholdLevel,
}

// ═══════════════════════════════════════════════════════════════════════════════
// GLOBAL POOL REGISTRY
// ═══════════════════════════════════════════════════════════════════════════════

lazy_static! {
    /// Global pool registry.
    static ref POOLS: Mutex<PoolRegistry> = Mutex::new(PoolRegistry::new());
}

/// Registry of thin pools.
#[derive(Debug)]
struct PoolRegistry {
    /// Pools by ID.
    pools: BTreeMap<u64, ThinPool>,
    /// Pool names to IDs.
    names: BTreeMap<String, u64>,
    /// Next pool ID.
    next_id: u64,
}

impl PoolRegistry {
    fn new() -> Self {
        Self {
            pools: BTreeMap::new(),
            names: BTreeMap::new(),
            next_id: 1,
        }
    }
}

/// Create a new thin pool.
pub fn create_pool(config: PoolConfig) -> ThinResult<u64> {
    let mut reg = POOLS.lock();

    if reg.names.contains_key(&config.name) {
        return Err(ThinError::InvalidConfig("Pool name already exists".into()));
    }

    let id = reg.next_id;
    reg.next_id += 1;

    let name = config.name.clone();
    let pool = ThinPool::new(id, config);

    reg.pools.insert(id, pool);
    reg.names.insert(name, id);

    Ok(id)
}

/// Delete a thin pool.
pub fn delete_pool(id: u64) -> ThinResult<()> {
    let mut reg = POOLS.lock();

    let pool = reg
        .pools
        .remove(&id)
        .ok_or(ThinError::PoolNotFound(id.to_string()))?;

    if pool.volume_count() > 0 {
        reg.pools.insert(id, pool);
        return Err(ThinError::NotPermitted);
    }

    reg.names.remove(pool.name());
    Ok(())
}

/// Check if a pool exists.
pub fn pool_exists(id: u64) -> bool {
    POOLS.lock().pools.contains_key(&id)
}

/// Check if a pool exists by name.
pub fn pool_exists_by_name(name: &str) -> bool {
    POOLS.lock().names.contains_key(name)
}

/// Get pool name by ID.
pub fn get_pool_name(id: u64) -> Option<String> {
    POOLS.lock().pools.get(&id).map(|p| p.name().to_string())
}

/// Get pool ID by name.
pub fn get_pool_id(name: &str) -> Option<u64> {
    POOLS.lock().names.get(name).copied()
}

/// Execute a function on a pool.
pub fn with_pool<F, R>(id: u64, f: F) -> ThinResult<R>
where
    F: FnOnce(&mut ThinPool) -> R,
{
    let mut reg = POOLS.lock();
    let pool = reg
        .pools
        .get_mut(&id)
        .ok_or(ThinError::PoolNotFound(id.to_string()))?;
    Ok(f(pool))
}

/// List all pool IDs.
pub fn list_pools() -> Vec<u64> {
    POOLS.lock().pools.keys().copied().collect()
}

/// Get pool count.
pub fn pool_count() -> usize {
    POOLS.lock().pools.len()
}

/// Clear all pools (for testing).
pub fn clear_pools() {
    let mut reg = POOLS.lock();
    reg.pools.clear();
    reg.names.clear();
}

// ═══════════════════════════════════════════════════════════════════════════════
// TESTS
// ═══════════════════════════════════════════════════════════════════════════════

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

    fn setup() {
        clear_pools();
    }

    fn create_test_pool() -> ThinPool {
        let config = PoolConfig::new("test-pool", 100 * 1024 * 1024 * 1024); // 100GB
        ThinPool::new(1, config)
    }

    #[test]
    fn test_pool_creation() {
        let pool = create_test_pool();

        assert_eq!(pool.id(), 1);
        assert_eq!(pool.name(), "test-pool");
        assert_eq!(pool.volume_count(), 0);
    }

    #[test]
    fn test_volume_creation() {
        let mut pool = create_test_pool();

        let config = VolumeConfig::new("vol1", 10 * 1024 * 1024 * 1024); // 10GB
        let id = pool.create_volume(config).unwrap();

        assert_eq!(pool.volume_count(), 1);
        assert!(pool.get_volume(id).is_some());
        assert!(pool.get_volume_by_name("vol1").is_some());
    }

    #[test]
    fn test_volume_creation_duplicate_name() {
        let mut pool = create_test_pool();

        let config1 = VolumeConfig::new("vol1", 10 * 1024 * 1024 * 1024);
        pool.create_volume(config1).unwrap();

        let config2 = VolumeConfig::new("vol1", 5 * 1024 * 1024 * 1024);
        let result = pool.create_volume(config2);

        assert!(matches!(result, Err(ThinError::VolumeExists(_))));
    }

    #[test]
    fn test_volume_deletion() {
        let mut pool = create_test_pool();

        let config = VolumeConfig::new("vol1", 10 * 1024 * 1024 * 1024);
        let id = pool.create_volume(config).unwrap();

        pool.delete_volume(id).unwrap();
        assert_eq!(pool.volume_count(), 0);
    }

    #[test]
    fn test_snapshot_creation() {
        let mut pool = create_test_pool();

        let config = VolumeConfig::new("vol1", 10 * 1024 * 1024 * 1024);
        let vol_id = pool.create_volume(config).unwrap();

        let snap_id = pool.create_snapshot(vol_id, "snap1".into()).unwrap();

        assert_eq!(pool.volume_count(), 2);
        assert!(pool.get_volume(snap_id).unwrap().is_snapshot());
        assert_eq!(pool.stats().snapshot_count, 1);
    }

    #[test]
    fn test_block_allocation() {
        let mut pool = create_test_pool();

        let config = VolumeConfig::new("vol1", 10 * 1024 * 1024 * 1024);
        let vol_id = pool.create_volume(config).unwrap();

        let vblock = VirtualBlock::new(100);
        let pblock = pool.allocate_block(vol_id, vblock).unwrap();

        assert!(pblock.0 > 0);
        assert!(pool.get_volume(vol_id).unwrap().is_allocated(vblock));
    }

    #[test]
    fn test_block_freeing() {
        let mut pool = create_test_pool();

        let config = VolumeConfig::new("vol1", 10 * 1024 * 1024 * 1024);
        let vol_id = pool.create_volume(config).unwrap();

        let vblock = VirtualBlock::new(100);
        pool.allocate_block(vol_id, vblock).unwrap();

        let initial_free = pool.free_blocks();
        pool.free_block(vol_id, vblock);

        assert_eq!(pool.free_blocks(), initial_free + 1);
    }

    #[test]
    fn test_overcommit_ratio() {
        let mut pool = create_test_pool();

        // Create volumes totaling 500GB virtual on 100GB pool
        for i in 0..5 {
            let config = VolumeConfig::new(alloc::format!("vol{}", i), 100 * 1024 * 1024 * 1024);
            pool.create_volume(config).unwrap();
        }

        // Overcommit should be ~5x
        let ratio = pool.overcommit_ratio();
        assert!(ratio > 4.0 && ratio < 6.0);
    }

    #[test]
    fn test_threshold_levels() {
        let config = PoolConfig::new("test", 1000 * 128 * 1024) // 1000 blocks
            .with_thresholds(Thresholds::new(80, 90, 95));
        let mut pool = ThinPool::new(1, config);

        // Initially normal
        assert_eq!(pool.current_threshold_level(), ThresholdLevel::Normal);

        let vol_config = VolumeConfig::new("vol1", 1024 * 1024 * 1024);
        let vol_id = pool.create_volume(vol_config).unwrap();

        // Allocate until warning
        for i in 0..800 {
            let _ = pool.allocate_block(vol_id, VirtualBlock::new(i));
        }

        let level = pool.current_threshold_level();
        assert!(level >= ThresholdLevel::Warning);
    }

    #[test]
    fn test_capacity_summary() {
        let mut pool = create_test_pool();

        let config = VolumeConfig::new("vol1", 50 * 1024 * 1024 * 1024);
        pool.create_volume(config).unwrap();

        let summary = pool.capacity_summary();

        assert!(summary.total_physical > 0);
        assert!(summary.total_virtual > 0);
        assert!(summary.overcommit_ratio > 0.0);
    }

    #[test]
    fn test_global_pool_registry() {
        setup();

        let config = PoolConfig::new("global-test", 10 * 1024 * 1024 * 1024);
        let id = create_pool(config).unwrap();

        assert!(pool_exists(id));
        assert!(pool_exists_by_name("global-test"));
        assert_eq!(pool_count(), 1);

        delete_pool(id).unwrap();
        assert_eq!(pool_count(), 0);
    }

    #[test]
    fn test_with_pool() {
        setup();

        let config = PoolConfig::new("with-pool-test", 10 * 1024 * 1024 * 1024);
        let id = create_pool(config).unwrap();

        let result = with_pool(id, |pool| pool.volume_count()).unwrap();

        assert_eq!(result, 0);
    }
}