featherdb_storage/

file.rs

//! File I/O abstraction

use crate::compression::{
    CompressionStats, CompressionType as StorageCompressionType, PageCompressor,
};
use featherdb_core::{
    constants, CompressionConfig, CompressionType, Config, EncryptionConfig, Error, PageId, Result,
};
use featherdb_crypto::{Cipher, KeyDerivation, ENCRYPTION_OVERHEAD, SALT_SIZE};
use parking_lot::RwLock;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};

/// Superblock - first 512 bytes of the database file
#[derive(Debug, Clone)]
pub struct Superblock {
    /// Magic bytes
    pub magic: [u8; 10],
    /// Format version
    pub format_version: u32,
    /// Page size
    pub page_size: u32,
    /// Total number of pages
    pub total_pages: u64,
    /// Free list head page
    pub free_list_head: u64,
    /// Schema root page
    pub schema_root: u64,
    /// Last checkpoint LSN
    pub last_checkpoint_lsn: u64,
    /// God byte for commit slot selection (0 = A, 1 = B)
    pub god_byte: u8,
    /// Encryption flags (0 = none, 1 = encrypted)
    pub encryption_flags: u8,
    /// Salt for password-based key derivation (32 bytes)
    pub encryption_salt: [u8; SALT_SIZE],
}

impl Superblock {
    pub fn new(page_size: u32, encrypted: bool, salt: [u8; SALT_SIZE]) -> Self {
        let mut magic = [0u8; 10];
        magic.copy_from_slice(constants::MAGIC);

        Superblock {
            magic,
            format_version: constants::FORMAT_VERSION,
            page_size,
            total_pages: constants::FIRST_DATA_PAGE,
            free_list_head: 0,
            schema_root: constants::SCHEMA_ROOT_PAGE,
            last_checkpoint_lsn: 0,
            god_byte: 0,
            encryption_flags: if encrypted { 1 } else { 0 },
            encryption_salt: salt,
        }
    }

    pub fn is_encrypted(&self) -> bool {
        self.encryption_flags != 0
    }

    pub fn serialize(&self) -> [u8; constants::SUPERBLOCK_SIZE] {
        let mut buf = [0u8; constants::SUPERBLOCK_SIZE];
        let mut offset = 0;

        buf[offset..offset + 10].copy_from_slice(&self.magic);
        offset += 10;

        buf[offset..offset + 4].copy_from_slice(&self.format_version.to_le_bytes());
        offset += 4;

        buf[offset..offset + 4].copy_from_slice(&self.page_size.to_le_bytes());
        offset += 4;

        buf[offset..offset + 8].copy_from_slice(&self.total_pages.to_le_bytes());
        offset += 8;

        buf[offset..offset + 8].copy_from_slice(&self.free_list_head.to_le_bytes());
        offset += 8;

        buf[offset..offset + 8].copy_from_slice(&self.schema_root.to_le_bytes());
        offset += 8;

        buf[offset..offset + 8].copy_from_slice(&self.last_checkpoint_lsn.to_le_bytes());
        offset += 8;

        buf[offset] = self.god_byte;
        offset += 1;

        buf[offset] = self.encryption_flags;
        offset += 1;

        buf[offset..offset + SALT_SIZE].copy_from_slice(&self.encryption_salt);

        buf
    }

    pub fn deserialize(data: &[u8]) -> Result<Self> {
        if data.len() < constants::SUPERBLOCK_SIZE {
            return Err(Error::InvalidDatabaseFile {
                message: "Superblock too small".into(),
            });
        }

        let mut magic = [0u8; 10];
        magic.copy_from_slice(&data[0..10]);

        if &magic != constants::MAGIC {
            return Err(Error::InvalidDatabaseFile {
                message: "Invalid magic bytes".into(),
            });
        }

        let format_version = u32::from_le_bytes([data[10], data[11], data[12], data[13]]);
        if format_version != constants::FORMAT_VERSION {
            return Err(Error::VersionMismatch {
                file_version: format_version,
                expected: constants::FORMAT_VERSION,
            });
        }

        let mut encryption_salt = [0u8; SALT_SIZE];
        encryption_salt.copy_from_slice(&data[52..52 + SALT_SIZE]);

        Ok(Superblock {
            magic,
            format_version,
            page_size: u32::from_le_bytes([data[14], data[15], data[16], data[17]]),
            total_pages: u64::from_le_bytes([
                data[18], data[19], data[20], data[21], data[22], data[23], data[24], data[25],
            ]),
            free_list_head: u64::from_le_bytes([
                data[26], data[27], data[28], data[29], data[30], data[31], data[32], data[33],
            ]),
            schema_root: u64::from_le_bytes([
                data[34], data[35], data[36], data[37], data[38], data[39], data[40], data[41],
            ]),
            last_checkpoint_lsn: u64::from_le_bytes([
                data[42], data[43], data[44], data[45], data[46], data[47], data[48], data[49],
            ]),
            god_byte: data[50],
            encryption_flags: data[51],
            encryption_salt,
        })
    }

    pub fn is_valid(&self) -> bool {
        &self.magic == constants::MAGIC && self.format_version == constants::FORMAT_VERSION
    }
}

/// Manages file I/O for the database
pub struct FileManager {
    file: RwLock<File>,
    superblock: RwLock<Superblock>,
    page_size: usize,
    /// Optional cipher for page encryption
    cipher: Option<Cipher>,
    /// Optional page compressor
    compressor: Option<PageCompressor>,
    /// Storage limits configuration
    storage_limits: featherdb_core::StorageLimitsConfig,
}

impl FileManager {
    /// Open or create a database file
    pub fn open(config: &Config) -> Result<Self> {
        let exists = config.path.exists();

        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(config.create_if_missing)
            .open(&config.path)?;

        let page_size = config.page_size;

        // Determine encryption salt and flags
        let (superblock, cipher) = if exists {
            // Read existing superblock
            let mut file_ref = &file;
            file_ref.seek(SeekFrom::Start(0))?;

            let mut buf = [0u8; constants::SUPERBLOCK_SIZE];
            file_ref.read_exact(&mut buf)?;

            let sb = Superblock::deserialize(&buf)?;

            if sb.page_size as usize != page_size {
                return Err(Error::InvalidDatabaseFile {
                    message: format!(
                        "Page size mismatch: file has {}, config has {}",
                        sb.page_size, page_size
                    ),
                });
            }

            // Check encryption configuration matches file
            if sb.is_encrypted() && !config.is_encrypted() {
                return Err(Error::InvalidDatabaseFile {
                    message: "Database is encrypted but no password/key provided".into(),
                });
            }
            if !sb.is_encrypted() && config.is_encrypted() {
                return Err(Error::InvalidDatabaseFile {
                    message: "Database is not encrypted but password/key was provided".into(),
                });
            }

            // Derive cipher from config and stored salt
            let cipher = Self::create_cipher(config, &sb.encryption_salt)?;

            (sb, cipher)
        } else {
            // New database - generate salt if encrypted
            let (encrypted, salt) = match &config.encryption {
                EncryptionConfig::None => (false, [0u8; SALT_SIZE]),
                EncryptionConfig::Password(_) => (true, KeyDerivation::generate_salt()),
                EncryptionConfig::Key(_) => (true, [0u8; SALT_SIZE]), // No salt needed for raw key
            };

            let sb = Superblock::new(page_size as u32, encrypted, salt);
            let cipher = Self::create_cipher(config, &salt)?;

            (sb, cipher)
        };

        // Create compressor if compression is enabled
        let compressor = Self::create_compressor(&config.compression);

        Ok(FileManager {
            file: RwLock::new(file),
            superblock: RwLock::new(superblock),
            page_size,
            cipher,
            compressor,
            storage_limits: config.storage_limits.clone(),
        })
    }

    /// Create compressor from compression config
    fn create_compressor(config: &CompressionConfig) -> Option<PageCompressor> {
        if !config.is_enabled() {
            return None;
        }

        let storage_type = match config.compression_type {
            CompressionType::None => StorageCompressionType::None,
            CompressionType::Lz4 => StorageCompressionType::Lz4,
            CompressionType::Zstd { level } => StorageCompressionType::Zstd { level },
        };

        let compressor = PageCompressor::new(storage_type, config.threshold);
        Some(compressor)
    }

    /// Create cipher from encryption config and salt
    fn create_cipher(config: &Config, salt: &[u8; SALT_SIZE]) -> Result<Option<Cipher>> {
        match &config.encryption {
            EncryptionConfig::None => Ok(None),
            EncryptionConfig::Password(password) => {
                let key = KeyDerivation::derive_key(password.as_bytes(), salt)?;
                Ok(Some(Cipher::new(&key)))
            }
            EncryptionConfig::Key(key) => Ok(Some(Cipher::new(key))),
        }
    }

    /// Initialize the database file if it's new
    pub fn init_if_needed(&self, _config: &Config) -> Result<()> {
        let superblock = self.superblock.read();

        // Check if file is already initialized
        let mut file = self.file.write();
        let file_len = file.seek(SeekFrom::End(0))?;

        if file_len == 0 {
            // New file - write superblock and initial pages
            file.seek(SeekFrom::Start(0))?;
            file.write_all(&superblock.serialize())?;

            // Pad to first page boundary (using disk_page_size for encrypted DBs)
            let disk_page = self.disk_page_size();
            let padding = vec![0u8; disk_page - constants::SUPERBLOCK_SIZE];
            file.write_all(&padding)?;
            drop(superblock); // Release lock before calling write_page

            // Write empty schema catalog page (page 1) with valid checksum
            let mut schema_page = crate::Page::new(
                PageId(constants::SCHEMA_ROOT_PAGE),
                crate::PageType::Leaf,
                self.page_size,
            );
            schema_page.compute_checksum();
            drop(file); // Release lock before calling write_page
            self.write_page(PageId(constants::SCHEMA_ROOT_PAGE), schema_page.as_bytes())?;

            // Write empty freelist page (page 2) with valid checksum
            let mut freelist_page = crate::Page::new(
                PageId(constants::FREELIST_ROOT_PAGE),
                crate::PageType::Free,
                self.page_size,
            );
            freelist_page.compute_checksum();
            self.write_page(
                PageId(constants::FREELIST_ROOT_PAGE),
                freelist_page.as_bytes(),
            )?;

            self.file.write().sync_all()?;
        }

        Ok(())
    }

    /// Read a page from disk
    ///
    /// The read order is: read from disk -> decrypt (if encrypted) -> decompress (if compressed)
    pub fn read_page(&self, page_id: PageId) -> Result<Vec<u8>> {
        let offset = self.page_offset(page_id);

        // Step 1: Read raw data from disk
        let raw_data = if let Some(ref cipher) = self.cipher {
            // Read encrypted data (always full page + overhead when encrypted)
            let encrypted_size = self.page_size + ENCRYPTION_OVERHEAD;
            let mut buf = vec![0u8; encrypted_size];

            let mut file = self.file.write();
            file.seek(SeekFrom::Start(offset))?;
            file.read_exact(&mut buf)?;

            // Decrypt
            cipher.decrypt(&buf, page_id.0)?
        } else {
            // No encryption - read raw page
            let mut buf = vec![0u8; self.page_size];

            let mut file = self.file.write();
            file.seek(SeekFrom::Start(offset))?;
            file.read_exact(&mut buf)?;

            buf
        };

        // Step 2: Decompress if we have a compressor
        if let Some(ref compressor) = self.compressor {
            compressor.decompress_page(&raw_data, self.page_size)
        } else {
            Ok(raw_data)
        }
    }

    /// Write a page to disk
    ///
    /// The write order is: compress (if enabled) -> encrypt (if enabled) -> write to disk
    pub fn write_page(&self, page_id: PageId, data: &[u8]) -> Result<()> {
        if data.len() != self.page_size {
            return Err(Error::Internal(format!(
                "Page data size mismatch: expected {}, got {}",
                self.page_size,
                data.len()
            )));
        }

        // Step 1: Compress if we have a compressor
        let processed_data = if let Some(ref compressor) = self.compressor {
            let compressed = compressor.compress_page(data)?;
            // Pad to page size if compressed is smaller (needed for fixed-size page storage)
            if compressed.len() < self.page_size {
                let mut padded = compressed;
                padded.resize(self.page_size, 0);
                padded
            } else {
                compressed
            }
        } else {
            data.to_vec()
        };

        let offset = self.page_offset(page_id);
        let mut file = self.file.write();
        file.seek(SeekFrom::Start(offset))?;

        // Step 2: Encrypt if we have a cipher
        if let Some(ref cipher) = self.cipher {
            // Encrypt and write
            let encrypted = cipher.encrypt(&processed_data, page_id.0)?;
            file.write_all(&encrypted)?;
        } else {
            // No encryption
            file.write_all(&processed_data)?;
        }

        Ok(())
    }

    /// Sync all changes to disk
    pub fn sync(&self) -> Result<()> {
        self.file.write().sync_all()?;
        Ok(())
    }

    /// Get the current total number of pages
    pub fn total_pages(&self) -> u64 {
        self.superblock.read().total_pages
    }

    /// Extend the file by one page, return the new page ID
    pub fn extend(&self) -> Result<PageId> {
        let mut superblock = self.superblock.write();
        let new_page_id = PageId(superblock.total_pages);

        // Calculate how many bytes the extension will add
        let new_total = superblock.total_pages + 1;
        let new_file_size = self.disk_page_size() as u64 * (new_total + 1);
        let current_size = {
            let file = self.file.read();
            file.metadata()?.len()
        };
        let additional_bytes = new_file_size.saturating_sub(current_size);

        // Check storage limit before extending
        if self.would_exceed_limit(additional_bytes) {
            let limit = self.storage_limits.effective_database_limit().unwrap_or(0);
            return Err(Error::StorageLimitExceeded {
                current_bytes: current_size,
                limit_bytes: limit,
                operation_bytes: additional_bytes,
            });
        }

        // Check disk space availability
        self.check_disk_space(additional_bytes)?;

        // Extend file - calculate new file size based on disk page size
        let mut file = self.file.write();
        file.set_len(new_file_size)?;

        superblock.total_pages = new_total;

        // Write updated superblock
        file.seek(SeekFrom::Start(0))?;
        file.write_all(&superblock.serialize())?;

        Ok(new_page_id)
    }

    /// Update superblock on disk
    pub fn write_superblock(&self) -> Result<()> {
        let superblock = self.superblock.read();
        let mut file = self.file.write();

        file.seek(SeekFrom::Start(0))?;
        file.write_all(&superblock.serialize())?;

        Ok(())
    }

    /// Get superblock
    pub fn superblock(&self) -> Superblock {
        self.superblock.read().clone()
    }

    /// Update superblock
    pub fn update_superblock<F>(&self, f: F) -> Result<()>
    where
        F: FnOnce(&mut Superblock),
    {
        let mut superblock = self.superblock.write();
        f(&mut superblock);
        drop(superblock);
        self.write_superblock()
    }

    /// Get page size
    pub fn page_size(&self) -> usize {
        self.page_size
    }

    /// Check if encryption is enabled
    pub fn is_encrypted(&self) -> bool {
        self.cipher.is_some()
    }

    /// Check if compression is enabled
    pub fn is_compressed(&self) -> bool {
        self.compressor.is_some()
    }

    /// Get compression statistics
    /// Returns the stats from the compressor if compression is enabled,
    /// otherwise returns the empty stats.
    pub fn compression_stats(&self) -> &CompressionStats {
        if let Some(ref compressor) = self.compressor {
            compressor.stats()
        } else {
            // Leak the Arc to get a static reference for the empty case.
            // This is safe because the Arc is owned by FileManager.
            // We use Box::leak on a clone to avoid lifetime issues.
            static EMPTY_STATS: std::sync::OnceLock<CompressionStats> = std::sync::OnceLock::new();
            EMPTY_STATS.get_or_init(CompressionStats::default)
        }
    }

    /// Get the compression type (if any)
    pub fn compression_type(&self) -> Option<StorageCompressionType> {
        self.compressor.as_ref().map(|c| c.compression_type())
    }

    /// Get current database file size in bytes
    pub fn current_size(&self) -> u64 {
        let file = self.file.read();
        file.metadata().map(|m| m.len()).unwrap_or(0)
    }

    /// Get configured maximum size (None if unlimited)
    pub fn max_size(&self) -> Option<u64> {
        self.storage_limits.max_database_size
    }

    /// Get remaining capacity before hitting limit (None if unlimited)
    pub fn remaining_capacity(&self) -> Option<u64> {
        if let Some(limit) = self.storage_limits.effective_database_limit() {
            let current = self.current_size();
            Some(limit.saturating_sub(current))
        } else {
            None
        }
    }

    /// Check if an operation of given size would exceed the limit
    pub fn would_exceed_limit(&self, additional_bytes: u64) -> bool {
        if let Some(limit) = self.storage_limits.effective_database_limit() {
            let current = self.current_size();
            current.saturating_add(additional_bytes) > limit
        } else {
            false
        }
    }

    /// Check if sufficient disk space is available for an operation
    ///
    /// Returns Ok(()) if enough space, or Err(InsufficientDiskSpace) if not.
    /// This performs a best-effort check. On platforms where we can't determine
    /// available space, we optimistically allow the operation (it will fail with
    /// I/O error if there's actually insufficient space).
    pub fn check_disk_space(&self, _required_bytes: u64) -> Result<()> {
        // Note: A proper implementation would use platform-specific APIs to check
        // available disk space (statvfs on Unix, GetDiskFreeSpaceEx on Windows).
        // For now, we do a best-effort check by attempting the operation and
        // letting the OS return an error if there's insufficient space.
        //
        // Future improvement: Use the fs2 crate or platform-specific syscalls
        // to check available space before attempting the write.
        Ok(())
    }

    /// Get the on-disk size of a page (including encryption overhead if any)
    fn disk_page_size(&self) -> usize {
        if self.cipher.is_some() {
            self.page_size + ENCRYPTION_OVERHEAD
        } else {
            self.page_size
        }
    }

    /// Calculate file offset for a page
    fn page_offset(&self, page_id: PageId) -> u64 {
        // Page 0 starts at offset page_size (after superblock + padding)
        // But the superblock takes up less than a page, so:
        // Offset 0-511: Superblock
        // Offset 512 to page_size-1: Padding
        // Offset page_size: Page 1 (schema root)
        // etc.
        // When encrypted, pages are larger on disk
        self.disk_page_size() as u64 * page_id.0
    }
}

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

    #[test]
    fn test_superblock_serialization() {
        let sb = Superblock::new(4096, false, [0u8; SALT_SIZE]);
        let serialized = sb.serialize();
        let deserialized = Superblock::deserialize(&serialized).unwrap();

        assert_eq!(sb.format_version, deserialized.format_version);
        assert_eq!(sb.page_size, deserialized.page_size);
        assert_eq!(sb.total_pages, deserialized.total_pages);
        assert_eq!(sb.encryption_flags, deserialized.encryption_flags);
    }

    #[test]
    fn test_superblock_encryption_flags() {
        let salt = [0x42u8; SALT_SIZE];
        let sb = Superblock::new(4096, true, salt);
        let serialized = sb.serialize();
        let deserialized = Superblock::deserialize(&serialized).unwrap();

        assert!(deserialized.is_encrypted());
        assert_eq!(deserialized.encryption_salt, salt);
    }

    #[test]
    fn test_file_manager() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("test.db");

        let config = Config::new(&path);
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        // Write a page
        let page_data = vec![0xAB; 4096];
        fm.write_page(PageId(3), &page_data).unwrap();

        // Read it back
        let read_data = fm.read_page(PageId(3)).unwrap();
        assert_eq!(page_data, read_data);
    }

    #[test]
    fn test_file_manager_encrypted() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("encrypted.db");

        let config = Config::new(&path).with_password("test_password");
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        assert!(fm.is_encrypted());

        // Write a page
        let page_data = vec![0xAB; 4096];
        fm.write_page(PageId(3), &page_data).unwrap();

        // Read it back
        let read_data = fm.read_page(PageId(3)).unwrap();
        assert_eq!(page_data, read_data);
    }

    #[test]
    fn test_encrypted_db_reopen() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("encrypted_reopen.db");

        // Create encrypted DB
        {
            let config = Config::new(&path).with_password("my_secret");
            let fm = FileManager::open(&config).unwrap();
            fm.init_if_needed(&config).unwrap();

            let page_data = vec![0xCD; 4096];
            fm.write_page(PageId(3), &page_data).unwrap();
        }

        // Reopen with correct password
        {
            let config = Config::new(&path).with_password("my_secret");
            let fm = FileManager::open(&config).unwrap();

            let read_data = fm.read_page(PageId(3)).unwrap();
            assert_eq!(read_data, vec![0xCD; 4096]);
        }
    }

    #[test]
    fn test_encrypted_db_wrong_password() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("encrypted_wrong.db");

        // Create encrypted DB
        {
            let config = Config::new(&path).with_password("correct_password");
            let fm = FileManager::open(&config).unwrap();
            fm.init_if_needed(&config).unwrap();

            let page_data = vec![0xEF; 4096];
            fm.write_page(PageId(3), &page_data).unwrap();
        }

        // Try with wrong password - should fail on decrypt
        {
            let config = Config::new(&path).with_password("wrong_password");
            let fm = FileManager::open(&config).unwrap();

            let result = fm.read_page(PageId(3));
            assert!(result.is_err());
        }
    }

    #[test]
    fn test_file_manager_lz4_compression() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("compressed_lz4.db");

        let config = Config::new(&path).with_lz4_compression();
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        assert!(fm.is_compressed());

        // Write a compressible page (repeated data compresses well)
        let page_data = vec![0xAB; 4096];
        fm.write_page(PageId(3), &page_data).unwrap();

        // Read it back
        let read_data = fm.read_page(PageId(3)).unwrap();
        assert_eq!(page_data, read_data);

        // Check compression stats
        let stats = fm.compression_stats();
        assert!(stats.compression_ratio() < 1.0, "Data should be compressed");
    }

    #[test]
    fn test_file_manager_zstd_compression() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("compressed_zstd.db");

        let config = Config::new(&path).with_zstd_compression();
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        assert!(fm.is_compressed());

        // Write a compressible page
        let page_data = vec![0xCD; 4096];
        fm.write_page(PageId(3), &page_data).unwrap();

        // Read it back
        let read_data = fm.read_page(PageId(3)).unwrap();
        assert_eq!(page_data, read_data);
    }

    #[test]
    fn test_file_manager_compression_with_encryption() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("compressed_encrypted.db");

        let config = Config::new(&path)
            .with_lz4_compression()
            .with_password("test_password");
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        assert!(fm.is_compressed());
        assert!(fm.is_encrypted());

        // Write a compressible page
        let page_data = vec![0xEF; 4096];
        fm.write_page(PageId(3), &page_data).unwrap();

        // Read it back
        let read_data = fm.read_page(PageId(3)).unwrap();
        assert_eq!(page_data, read_data);
    }

    #[test]
    fn test_compression_with_varied_data() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("varied_data.db");

        let config = Config::new(&path).with_lz4_compression();
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        // Write varied data (semi-random pattern)
        let page_data: Vec<u8> = (0..4096).map(|i| (i * 17 + i / 3) as u8).collect();
        fm.write_page(PageId(3), &page_data).unwrap();

        // Read it back
        let read_data = fm.read_page(PageId(3)).unwrap();
        assert_eq!(page_data, read_data);
    }

    #[test]
    fn test_storage_limit_enforcement() {
        use featherdb_core::StorageLimitsConfig;

        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("limited.db");

        // Create a database with a very small limit (5MB)
        let storage_limits = StorageLimitsConfig::new()
            .with_max_database_size_mb(5)
            .with_safety_margin_percent(10);

        let config = Config::new(&path).with_storage_limits(storage_limits);
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        // Check size tracking methods
        let current = fm.current_size();
        assert!(current > 0); // Should have superblock + initial pages

        let max = fm.max_size();
        assert_eq!(max, Some(5 * 1024 * 1024));

        let remaining = fm.remaining_capacity();
        assert!(remaining.is_some());
        assert!(remaining.unwrap() < max.unwrap());

        // Try to extend beyond limit - this should eventually fail
        // Note: We can't easily test hitting the limit without filling the entire 5MB,
        // so we just verify the methods work correctly
        let would_exceed = fm.would_exceed_limit(10 * 1024 * 1024); // 10MB
        assert!(would_exceed);

        let would_fit = fm.would_exceed_limit(100); // 100 bytes
        assert!(!would_fit || fm.remaining_capacity().unwrap() < 100);
    }

    #[test]
    fn test_storage_limit_exceeded_error() {
        use featherdb_core::StorageLimitsConfig;

        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("tiny_limit.db");

        // Create a database with a tiny limit that we can easily exceed
        // Set to 1MB with 5% safety margin = ~950KB effective limit
        let storage_limits = StorageLimitsConfig::new()
            .with_max_database_size_mb(1)
            .with_safety_margin_percent(5);

        let config = Config::new(&path).with_storage_limits(storage_limits);
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        // Try to extend many pages until we hit the limit
        // Each page is 4KB, so we should hit 1MB (~250 pages) eventually
        let mut extended = 0;
        loop {
            match fm.extend() {
                Ok(_) => {
                    extended += 1;
                    // Safety check to avoid infinite loop
                    if extended > 300 {
                        panic!("Should have hit storage limit by now");
                    }
                }
                Err(Error::StorageLimitExceeded { .. }) => {
                    // Expected error
                    break;
                }
                Err(e) => {
                    panic!("Unexpected error: {}", e);
                }
            }
        }

        assert!(extended > 0, "Should have extended at least some pages");
        assert!(extended < 300, "Should have hit limit before 300 pages");
    }

    #[test]
    fn test_unlimited_storage() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("unlimited.db");

        let config = Config::new(&path); // No limits
        let fm = FileManager::open(&config).unwrap();
        fm.init_if_needed(&config).unwrap();

        assert_eq!(fm.max_size(), None);
        assert_eq!(fm.remaining_capacity(), None);
        assert!(!fm.would_exceed_limit(u64::MAX));

        // Should be able to extend without limit checks
        fm.extend().unwrap();
    }
}