lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0
//
// Dataset Management
// Hierarchical filesystem datasets with snapshots, clones, and properties.

use crate::fscore::structs::Blkptr;

use alloc::collections::BTreeMap;
use alloc::format;
use alloc::string::String;
use alloc::vec::Vec;
use lazy_static::lazy_static;
use spin::Mutex;

/// Dataset physical metadata (on-disk structure)
#[repr(C)]
#[derive(Debug, Clone, Copy)]
pub struct DatasetPhys {
    /// Dataset globally unique identifier
    pub guid: u64,
    /// Parent dataset object ID
    pub parent_dataset_obj: u64,
    /// Parent dataset creation TXG
    pub parent_dataset_txg: u64,
    /// Root dnode block pointer
    pub root_dnode_bp: Blkptr,
    /// Root directory object ID
    pub root_dir_obj: u64,
    /// Snapshot list object ID
    pub snapshot_list_obj: u64,
    /// Maximum size in bytes (0 = unlimited)
    pub quota: u64, // Max size in bytes (0 = unlimited)
    /// Minimum guaranteed space
    pub reservation: u64, // Min guaranteed space
    /// Compression algorithm (0=off, 1=lz4, 2=zstd)
    pub compression: u8, // 0=off, 1=lz4, 2=zstd
    /// Checksum algorithm (0=off, 1=fletcher4, 2=blake3)
    pub checksum: u8, // 0=off, 1=fletcher4, 2=blake3
    /// Read-only flag (1 = read-only)
    pub readonly: u8, // 1 = read-only
    /// Deduplication enabled flag
    pub dedup: u8, // 1 = dedup enabled
    /// Number of copies (1-3)
    pub copies: u8, // Number of copies (1-3)
    /// Encryption enabled flag
    pub encryption: u8, // 1 = encrypted
    /// TXG when created
    pub txg_birth: u64, // TXG when created
    /// Current space usage in bytes
    pub used_bytes: u64, // Current space usage
    /// Reserved padding
    pub pad: [u64; 4],
}

impl DatasetPhys {
    /// Create new dataset physical metadata
    pub fn new(guid: u64) -> Self {
        Self {
            guid,
            parent_dataset_obj: 0,
            parent_dataset_txg: 0,
            root_dnode_bp: Blkptr::zero(),
            root_dir_obj: 0,
            snapshot_list_obj: 0,
            quota: 0, // Unlimited
            reservation: 0,
            compression: 1, // LZ4 default
            checksum: 1,    // Fletcher4
            readonly: 0,
            dedup: 0,  // Off by default
            copies: 1, // Single copy
            encryption: 0,
            txg_birth: 1,
            used_bytes: 0,
            pad: [0; 4],
        }
    }
}

/// Dataset properties (runtime configuration)
#[derive(Debug, Clone)]
pub struct DatasetProperties {
    /// Compression algorithm ("off", "lz4", "zstd")
    pub compression: String, // "off", "lz4", "zstd"
    /// Checksum algorithm ("off", "fletcher4", "blake3")
    pub checksum: String, // "off", "fletcher4", "blake3"
    /// Quota in bytes (None = unlimited)
    pub quota: Option<u64>, // Bytes
    /// Reservation in bytes
    pub reservation: Option<u64>,
    /// Read-only flag
    pub readonly: bool,
    /// Deduplication enabled
    pub dedup: bool,
    /// Number of copies (1-3)
    pub copies: u8, // 1-3
    /// Mount point path
    pub mountpoint: String, // Where to mount
}

impl Default for DatasetProperties {
    fn default() -> Self {
        Self {
            compression: "lz4".into(),
            checksum: "fletcher4".into(),
            quota: None,
            reservation: None,
            readonly: false,
            dedup: false,
            copies: 1,
            mountpoint: "/".into(),
        }
    }
}

/// In-memory dataset instance
#[derive(Debug, Clone)]
pub struct Dataset {
    /// Dataset name
    pub name: String,
    /// Physical on-disk metadata
    pub phys: DatasetPhys,
    /// Runtime configuration properties
    pub props: DatasetProperties,
    /// List of snapshot names
    pub snapshots: Vec<String>, // Snapshot names
    /// List of clone names
    pub clones: Vec<String>, // Clone names
}

impl Dataset {
    /// Create a new dataset
    pub fn create(name: String, guid: u64) -> Self {
        crate::lcpfs_println!("[ DATA ] Creating dataset: {}", name);

        Self {
            name,
            phys: DatasetPhys::new(guid),
            props: DatasetProperties::default(),
            snapshots: Vec::new(),
            clones: Vec::new(),
        }
    }

    /// Create a snapshot (copy-on-write)
    pub fn snapshot(&mut self, snap_name: &str, txg: u64) -> Result<Dataset, &'static str> {
        let full_name = format!("{}@{}", self.name, snap_name);
        crate::lcpfs_println!("[ DATA ] Creating snapshot: {}", full_name);

        // Clone physical metadata (copy-on-write)
        let mut snap_phys = self.phys;
        snap_phys.guid += 1; // New GUID for snapshot
        snap_phys.txg_birth = txg;
        snap_phys.readonly = 1; // Snapshots are always read-only

        let snapshot = Dataset {
            name: full_name.clone(),
            phys: snap_phys,
            props: self.props.clone(),
            snapshots: Vec::new(),
            clones: Vec::new(),
        };

        self.snapshots.push(full_name);
        Ok(snapshot)
    }

    /// Clone from this dataset
    pub fn clone(&self, clone_name: &str, txg: u64) -> Result<Dataset, &'static str> {
        crate::lcpfs_println!("[ DATA ] Creating clone: {}", clone_name);

        // Clone metadata (copy-on-write from snapshot)
        let mut clone_phys = self.phys;
        clone_phys.guid += 1000; // New GUID for clone
        clone_phys.parent_dataset_obj = self.phys.guid;
        clone_phys.parent_dataset_txg = txg;
        clone_phys.txg_birth = txg;
        clone_phys.readonly = 0; // Clones are writable

        Ok(Dataset {
            name: clone_name.into(),
            phys: clone_phys,
            props: self.props.clone(),
            snapshots: Vec::new(),
            clones: Vec::new(),
        })
    }

    /// Destroy the dataset
    pub fn destroy(&mut self) -> Result<(), &'static str> {
        if !self.snapshots.is_empty() {
            return Err("Cannot destroy dataset with snapshots");
        }

        if !self.clones.is_empty() {
            return Err("Cannot destroy dataset with clones");
        }

        crate::lcpfs_println!("[ DATA ] Destroying dataset: {}", self.name);
        Ok(())
    }

    /// Rename the dataset
    pub fn rename(&mut self, new_name: &str) -> Result<(), &'static str> {
        crate::lcpfs_println!("[ DATA ] Renaming {} -> {}", self.name, new_name);
        self.name = new_name.into();
        Ok(())
    }

    /// Set property
    pub fn set_property(&mut self, key: &str, value: &str) -> Result<(), &'static str> {
        match key {
            "compression" => {
                self.props.compression = value.into();
                self.phys.compression = match value {
                    "off" => 0,
                    "lz4" => 1,
                    "zstd" => 2,
                    _ => return Err("Invalid compression value"),
                };
            }
            "checksum" => {
                self.props.checksum = value.into();
                self.phys.checksum = match value {
                    "off" => 0,
                    "fletcher4" => 1,
                    "blake3" => 2,
                    _ => return Err("Invalid checksum value"),
                };
            }
            "quota" => {
                let bytes: u64 = value.parse().map_err(|_| "Invalid quota")?;
                self.props.quota = Some(bytes);
                self.phys.quota = bytes;
            }
            "readonly" => {
                self.props.readonly = value == "on";
                self.phys.readonly = if self.props.readonly { 1 } else { 0 };
            }
            "dedup" => {
                self.props.dedup = value == "on";
                self.phys.dedup = if self.props.dedup { 1 } else { 0 };
            }
            "copies" => {
                let copies: u8 = value.parse().map_err(|_| "Invalid copies")?;
                if !(1..=3).contains(&copies) {
                    return Err("Copies must be 1-3");
                }
                self.props.copies = copies;
                self.phys.copies = copies;
            }
            _ => return Err("Unknown property"),
        }

        crate::lcpfs_println!("[ DATA ] Set {}={} on {}", key, value, self.name);
        Ok(())
    }

    /// Get property
    pub fn get_property(&self, key: &str) -> Option<String> {
        match key {
            "compression" => Some(self.props.compression.clone()),
            "checksum" => Some(self.props.checksum.clone()),
            "quota" => self.props.quota.map(|q| format!("{}", q)),
            "reservation" => self.props.reservation.map(|r| format!("{}", r)),
            "readonly" => Some(if self.props.readonly { "on" } else { "off" }.into()),
            "dedup" => Some(if self.props.dedup { "on" } else { "off" }.into()),
            "copies" => Some(format!("{}", self.props.copies)),
            "used" => Some(format!("{}", self.phys.used_bytes)),
            "guid" => Some(format!("{:x}", self.phys.guid)),
            _ => None,
        }
    }

    /// List all properties
    pub fn list_properties(&self) -> BTreeMap<String, String> {
        let mut props = BTreeMap::new();
        props.insert("name".into(), self.name.clone());
        props.insert("guid".into(), format!("{:x}", self.phys.guid));
        props.insert("compression".into(), self.props.compression.clone());
        props.insert("checksum".into(), self.props.checksum.clone());
        props.insert(
            "readonly".into(),
            if self.props.readonly { "on" } else { "off" }.into(),
        );
        props.insert(
            "dedup".into(),
            if self.props.dedup { "on" } else { "off" }.into(),
        );
        props.insert("copies".into(), format!("{}", self.props.copies));
        props.insert("used".into(), format!("{}", self.phys.used_bytes));

        if let Some(quota) = self.props.quota {
            props.insert("quota".into(), format!("{}", quota));
        }

        props
    }

    /// Get space usage
    pub fn get_space_usage(&self) -> (u64, u64) {
        let used = self.phys.used_bytes;
        let available = if let Some(quota) = self.props.quota {
            quota.saturating_sub(used)
        } else {
            u64::MAX // Unlimited
        };
        (used, available)
    }
}

lazy_static! {
    /// Global dataset registry
    pub static ref DATASET_REGISTRY: Mutex<BTreeMap<String, Dataset>> = Mutex::new(BTreeMap::new());
}

/// Dataset management API for creating and managing datasets
pub struct DatasetManager;

impl DatasetManager {
    /// Create a dataset
    pub fn create(name: &str) -> Result<(), &'static str> {
        let mut registry = DATASET_REGISTRY.lock();

        if registry.contains_key(name) {
            return Err("Dataset already exists");
        }

        let guid = registry.len() as u64 + 1;
        let dataset = Dataset::create(name.into(), guid);
        registry.insert(name.into(), dataset);

        Ok(())
    }

    /// Destroy a dataset
    pub fn destroy(name: &str) -> Result<(), &'static str> {
        let mut registry = DATASET_REGISTRY.lock();

        let dataset = registry.get_mut(name).ok_or("Dataset not found")?;
        dataset.destroy()?;

        registry.remove(name);
        Ok(())
    }

    /// Snapshot a dataset
    pub fn snapshot(dataset_name: &str, snap_name: &str, txg: u64) -> Result<(), &'static str> {
        let mut registry = DATASET_REGISTRY.lock();

        let dataset = registry.get_mut(dataset_name).ok_or("Dataset not found")?;
        let snapshot = dataset.snapshot(snap_name, txg)?;

        let full_name = snapshot.name.clone();
        registry.insert(full_name, snapshot);

        Ok(())
    }

    /// Clone a dataset
    pub fn clone(source: &str, target: &str, txg: u64) -> Result<(), &'static str> {
        let mut registry = DATASET_REGISTRY.lock();

        let source_ds = registry.get(source).ok_or("Source dataset not found")?;
        let clone = source_ds.clone(target, txg)?;

        registry.insert(target.into(), clone);
        Ok(())
    }

    /// Set property
    pub fn set_property(dataset_name: &str, key: &str, value: &str) -> Result<(), &'static str> {
        let mut registry = DATASET_REGISTRY.lock();
        let dataset = registry.get_mut(dataset_name).ok_or("Dataset not found")?;
        dataset.set_property(key, value)
    }

    /// Get property
    pub fn get_property(dataset_name: &str, key: &str) -> Result<String, &'static str> {
        let registry = DATASET_REGISTRY.lock();
        let dataset = registry.get(dataset_name).ok_or("Dataset not found")?;
        dataset.get_property(key).ok_or("Property not found")
    }

    /// List all datasets
    pub fn list() -> Vec<String> {
        let registry = DATASET_REGISTRY.lock();
        registry.keys().cloned().collect()
    }

    /// Get dataset info
    pub fn info(dataset_name: &str) -> Result<BTreeMap<String, String>, &'static str> {
        let registry = DATASET_REGISTRY.lock();
        let dataset = registry.get(dataset_name).ok_or("Dataset not found")?;
        Ok(dataset.list_properties())
    }
}