featherdb_storage/

page.rs

//! Page format and operations

use bytes::{Buf, BufMut};
use featherdb_core::{constants, Error, PageId, Result};

/// Page type discriminator
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum PageType {
    /// Leaf page containing key-value pairs
    Leaf = 0,
    /// Branch (internal) page containing keys and child pointers
    Branch = 1,
    /// Overflow page for large values
    Overflow = 2,
    /// Free page (in free list)
    Free = 3,
}

impl TryFrom<u8> for PageType {
    type Error = Error;

    fn try_from(value: u8) -> Result<Self> {
        match value {
            0 => Ok(PageType::Leaf),
            1 => Ok(PageType::Branch),
            2 => Ok(PageType::Overflow),
            3 => Ok(PageType::Free),
            _ => Err(Error::InvalidPageType {
                page_id: PageId::INVALID,
                page_type: value,
            }),
        }
    }
}

/// Page flags
#[derive(Debug, Clone, Copy, Default)]
pub struct PageFlags {
    pub dirty: bool,
    pub encrypted: bool,
    pub compressed: bool,
}

impl PageFlags {
    fn to_u8(self) -> u8 {
        let mut flags = 0u8;
        if self.dirty {
            flags |= 0x01;
        }
        if self.encrypted {
            flags |= 0x02;
        }
        if self.compressed {
            flags |= 0x04;
        }
        flags
    }

    fn from_u8(value: u8) -> Self {
        PageFlags {
            dirty: value & 0x01 != 0,
            encrypted: value & 0x02 != 0,
            compressed: value & 0x04 != 0,
        }
    }
}

/// Slot array entry - points to a cell in the page
#[derive(Debug, Clone, Copy)]
pub struct SlotEntry {
    /// Offset from start of page to the cell
    pub offset: u16,
    /// Length of the cell in bytes
    pub len: u16,
    /// Flags (e.g., deleted marker for MVCC)
    pub flags: u8,
    /// Reserved for future use
    pub reserved: u8,
}

impl SlotEntry {
    pub const SIZE: usize = constants::SLOT_SIZE;

    pub fn serialize(&self, buf: &mut impl BufMut) {
        buf.put_u16_le(self.offset);
        buf.put_u16_le(self.len);
        buf.put_u8(self.flags);
        buf.put_u8(self.reserved);
    }

    pub fn deserialize(buf: &mut impl Buf) -> Self {
        SlotEntry {
            offset: buf.get_u16_le(),
            len: buf.get_u16_le(),
            flags: buf.get_u8(),
            reserved: buf.get_u8(),
        }
    }

    pub fn is_deleted(&self) -> bool {
        self.flags & 0x01 != 0
    }
}

/// A database page (4KB by default)
///
/// Layout:
/// - Header (64 bytes)
/// - Slot array (grows down from header)
/// - Free space
/// - Cell data (grows up from bottom)
#[derive(Clone)]
pub struct Page {
    /// Raw page data
    data: Vec<u8>,
    /// Page size
    page_size: usize,
}

impl Page {
    /// Create a new empty page
    pub fn new(page_id: PageId, page_type: PageType, page_size: usize) -> Self {
        let mut data = vec![0u8; page_size];

        // Initialize header
        let mut cursor = &mut data[..];

        // page_id: u64
        cursor.put_u64_le(page_id.0);
        // page_type: u8
        cursor.put_u8(page_type as u8);
        // flags: u8
        cursor.put_u8(0);
        // version: u64 (MVCC version)
        cursor.put_u64_le(0);
        // checksum: u128 (will be computed on write)
        cursor.put_u128_le(0);
        // item_count: u16
        cursor.put_u16_le(0);
        // free_start: u16 (starts right after header)
        cursor.put_u16_le(constants::PAGE_HEADER_SIZE as u16);
        // free_end: u16 (starts at end of page)
        cursor.put_u16_le(page_size as u16);
        // parent: u64
        cursor.put_u64_le(PageId::INVALID.0);
        // right_sibling: u64
        cursor.put_u64_le(PageId::INVALID.0);
        // reserved: 6 bytes
        cursor.put_slice(&[0u8; 6]);

        Page { data, page_size }
    }

    /// Create a page from raw bytes
    pub fn from_bytes(data: Vec<u8>) -> Result<Self> {
        let page_size = data.len();
        if page_size < constants::PAGE_HEADER_SIZE {
            return Err(Error::Internal("Page too small".into()));
        }
        Ok(Page { data, page_size })
    }

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

    /// Get mutable raw bytes
    pub fn as_bytes_mut(&mut self) -> &mut [u8] {
        &mut self.data
    }

    /// Get page ID
    pub fn page_id(&self) -> PageId {
        PageId(u64::from_le_bytes(self.data[0..8].try_into().unwrap()))
    }

    /// Set page ID
    pub fn set_page_id(&mut self, id: PageId) {
        self.data[0..8].copy_from_slice(&id.0.to_le_bytes());
    }

    /// Get page type
    pub fn page_type(&self) -> PageType {
        PageType::try_from(self.data[8]).unwrap_or(PageType::Free)
    }

    /// Set page type
    pub fn set_page_type(&mut self, page_type: PageType) {
        self.data[8] = page_type as u8;
    }

    /// Get flags
    pub fn flags(&self) -> PageFlags {
        PageFlags::from_u8(self.data[9])
    }

    /// Set flags
    pub fn set_flags(&mut self, flags: PageFlags) {
        self.data[9] = flags.to_u8();
    }

    /// Get MVCC version
    pub fn version(&self) -> u64 {
        u64::from_le_bytes(self.data[10..18].try_into().unwrap())
    }

    /// Set MVCC version
    pub fn set_version(&mut self, version: u64) {
        self.data[10..18].copy_from_slice(&version.to_le_bytes());
    }

    /// Get stored checksum
    pub fn checksum(&self) -> u128 {
        u128::from_le_bytes(self.data[18..34].try_into().unwrap())
    }

    /// Compute and store checksum
    pub fn compute_checksum(&mut self) {
        // Zero out checksum field before computing
        self.data[18..34].copy_from_slice(&[0u8; 16]);

        let hash = xxhash_rust::xxh3::xxh3_128(&self.data);
        self.data[18..34].copy_from_slice(&hash.to_le_bytes());
    }

    /// Verify checksum
    pub fn verify_checksum(&self) -> bool {
        let stored = self.checksum();

        // Zero out checksum field for computation
        let mut temp = self.data.clone();
        temp[18..34].copy_from_slice(&[0u8; 16]);

        let computed = xxhash_rust::xxh3::xxh3_128(&temp);
        stored == computed
    }

    /// Get item count
    pub fn item_count(&self) -> u16 {
        u16::from_le_bytes(self.data[34..36].try_into().unwrap())
    }

    /// Set item count
    fn set_item_count(&mut self, count: u16) {
        self.data[34..36].copy_from_slice(&count.to_le_bytes());
    }

    /// Get free space start offset
    pub fn free_start(&self) -> u16 {
        u16::from_le_bytes(self.data[36..38].try_into().unwrap())
    }

    /// Set free space start offset
    fn set_free_start(&mut self, offset: u16) {
        self.data[36..38].copy_from_slice(&offset.to_le_bytes());
    }

    /// Get free space end offset
    pub fn free_end(&self) -> u16 {
        u16::from_le_bytes(self.data[38..40].try_into().unwrap())
    }

    /// Set free space end offset
    fn set_free_end(&mut self, offset: u16) {
        self.data[38..40].copy_from_slice(&offset.to_le_bytes());
    }

    /// Get parent page ID
    pub fn parent(&self) -> PageId {
        PageId(u64::from_le_bytes(self.data[40..48].try_into().unwrap()))
    }

    /// Set parent page ID
    pub fn set_parent(&mut self, parent: PageId) {
        self.data[40..48].copy_from_slice(&parent.0.to_le_bytes());
    }

    /// Get right sibling page ID
    pub fn right_sibling(&self) -> Option<PageId> {
        let id = PageId(u64::from_le_bytes(self.data[48..56].try_into().unwrap()));
        if id.is_valid() {
            Some(id)
        } else {
            None
        }
    }

    /// Set right sibling page ID
    pub fn set_right_sibling(&mut self, sibling: Option<PageId>) {
        let id = sibling.unwrap_or(PageId::INVALID);
        self.data[48..56].copy_from_slice(&id.0.to_le_bytes());
    }

    /// Get available free space
    pub fn free_space(&self) -> usize {
        let start = self.free_start() as usize;
        let end = self.free_end() as usize;
        end.saturating_sub(start)
    }

    /// Get usable space (total - header)
    pub fn usable_space(&self) -> usize {
        self.page_size - constants::PAGE_HEADER_SIZE
    }

    /// Check if page needs to be split (> 70% full)
    pub fn needs_split(&self) -> bool {
        let used = self.usable_space() - self.free_space();
        used > (self.usable_space() * 70) / 100
    }

    /// Count non-deleted items by scanning slot flags only.
    /// No cell data is read — this is O(item_count) with minimal work per slot.
    #[inline]
    pub fn count_live_items(&self) -> u16 {
        let count = self.item_count() as usize;
        let mut live = 0u16;
        for i in 0..count {
            // flags byte is at slot_offset + 4 (after offset:u16 + len:u16)
            let flags_offset = constants::PAGE_HEADER_SIZE + i * SlotEntry::SIZE + 4;
            if self.data[flags_offset] & 0x01 == 0 {
                live += 1;
            }
        }
        live
    }

    /// Iterate non-deleted cells, calling `f(cell_data)` for each live cell.
    /// No allocations — reads slot flags inline and yields cell slices directly.
    #[inline]
    pub fn for_each_live_cell<F>(&self, mut f: F)
    where
        F: FnMut(&[u8]),
    {
        let count = self.item_count() as usize;
        for i in 0..count {
            let slot_offset = constants::PAGE_HEADER_SIZE + i * SlotEntry::SIZE;
            let flags = self.data[slot_offset + 4];
            if flags & 0x01 == 0 {
                let offset =
                    u16::from_le_bytes([self.data[slot_offset], self.data[slot_offset + 1]])
                        as usize;
                let len =
                    u16::from_le_bytes([self.data[slot_offset + 2], self.data[slot_offset + 3]])
                        as usize;
                if offset + len <= self.data.len() {
                    f(&self.data[offset..offset + len]);
                }
            }
        }
    }

    /// Get a slot entry by index
    pub fn get_slot(&self, index: usize) -> Option<SlotEntry> {
        if index >= self.item_count() as usize {
            return None;
        }

        let slot_offset = constants::PAGE_HEADER_SIZE + index * SlotEntry::SIZE;
        let mut cursor = &self.data[slot_offset..];
        Some(SlotEntry::deserialize(&mut cursor))
    }

    /// Get cell data by slot index
    pub fn get_cell(&self, index: usize) -> Option<&[u8]> {
        let slot = self.get_slot(index)?;
        let start = slot.offset as usize;
        let end = start + slot.len as usize;

        if end <= self.page_size {
            Some(&self.data[start..end])
        } else {
            None
        }
    }

    /// Insert a cell into the page
    /// Returns the slot index, or None if not enough space
    pub fn insert_cell(&mut self, data: &[u8]) -> Option<usize> {
        let cell_len = data.len();
        let needed = cell_len + SlotEntry::SIZE;

        if self.free_space() < needed {
            return None;
        }

        // Allocate space for cell at the end (growing upward)
        let new_free_end = self.free_end() as usize - cell_len;
        self.data[new_free_end..new_free_end + cell_len].copy_from_slice(data);
        self.set_free_end(new_free_end as u16);

        // Add slot entry (growing downward from header)
        let slot_index = self.item_count() as usize;
        let slot_offset = constants::PAGE_HEADER_SIZE + slot_index * SlotEntry::SIZE;

        let slot = SlotEntry {
            offset: new_free_end as u16,
            len: cell_len as u16,
            flags: 0,
            reserved: 0,
        };

        let mut cursor = &mut self.data[slot_offset..];
        slot.serialize(&mut cursor);

        self.set_item_count((slot_index + 1) as u16);
        self.set_free_start((slot_offset + SlotEntry::SIZE) as u16);

        Some(slot_index)
    }

    /// Delete a cell by marking it as deleted
    pub fn delete_cell(&mut self, index: usize) -> bool {
        if index >= self.item_count() as usize {
            return false;
        }

        let slot_offset = constants::PAGE_HEADER_SIZE + index * SlotEntry::SIZE;
        // Set deleted flag
        self.data[slot_offset + 4] |= 0x01;
        true
    }

    /// Update a cell in place (must fit in existing space)
    pub fn update_cell(&mut self, index: usize, data: &[u8]) -> bool {
        let slot = match self.get_slot(index) {
            Some(s) => s,
            None => return false,
        };

        if data.len() > slot.len as usize {
            return false; // Doesn't fit
        }

        let start = slot.offset as usize;
        self.data[start..start + data.len()].copy_from_slice(data);

        // Update length in slot if smaller
        if data.len() < slot.len as usize {
            let slot_offset = constants::PAGE_HEADER_SIZE + index * SlotEntry::SIZE;
            self.data[slot_offset + 2..slot_offset + 4]
                .copy_from_slice(&(data.len() as u16).to_le_bytes());
        }

        true
    }

    /// Compact the page, removing deleted cells and reclaiming space
    pub fn compact(&mut self) {
        let mut live_cells: Vec<(usize, Vec<u8>)> = Vec::new();

        // Collect non-deleted cells
        for i in 0..self.item_count() as usize {
            let slot = self.get_slot(i).unwrap();
            if !slot.is_deleted() {
                if let Some(cell) = self.get_cell(i) {
                    live_cells.push((i, cell.to_vec()));
                }
            }
        }

        // Reset page
        let page_id = self.page_id();
        let page_type = self.page_type();
        let parent = self.parent();
        let sibling = self.right_sibling();
        let version = self.version();

        *self = Page::new(page_id, page_type, self.page_size);
        self.set_parent(parent);
        self.set_right_sibling(sibling);
        self.set_version(version);

        // Re-insert live cells
        for (_, cell) in live_cells {
            self.insert_cell(&cell);
        }
    }

    /// Apply a redo operation from WAL
    pub fn apply_redo(&mut self, data: &[u8]) -> Result<()> {
        if data.len() != self.data.len() {
            return Err(Error::Internal("Redo data size mismatch".into()));
        }
        self.data.copy_from_slice(data);
        Ok(())
    }
}

impl std::fmt::Debug for Page {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Page")
            .field("page_id", &self.page_id())
            .field("page_type", &self.page_type())
            .field("item_count", &self.item_count())
            .field("free_space", &self.free_space())
            .finish()
    }
}

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

    #[test]
    fn test_page_creation() {
        let page = Page::new(PageId(1), PageType::Leaf, 4096);

        assert_eq!(page.page_id(), PageId(1));
        assert_eq!(page.page_type(), PageType::Leaf);
        assert_eq!(page.item_count(), 0);
        assert!(page.free_space() > 4000);
    }

    #[test]
    fn test_cell_operations() {
        let mut page = Page::new(PageId(1), PageType::Leaf, 4096);

        // Insert cells
        let idx1 = page.insert_cell(b"hello").unwrap();
        let idx2 = page.insert_cell(b"world").unwrap();

        assert_eq!(idx1, 0);
        assert_eq!(idx2, 1);
        assert_eq!(page.item_count(), 2);

        // Read cells
        assert_eq!(page.get_cell(0), Some(b"hello".as_slice()));
        assert_eq!(page.get_cell(1), Some(b"world".as_slice()));
    }

    #[test]
    fn test_checksum() {
        let mut page = Page::new(PageId(1), PageType::Leaf, 4096);
        page.insert_cell(b"test data");

        page.compute_checksum();
        assert!(page.verify_checksum());

        // Corrupt the page
        page.data[100] ^= 0xFF;
        assert!(!page.verify_checksum());
    }
}