pagedb 0.1.0-beta.6

//! Overflow page encode / decode and chain management.
//!
//! An overflow value is a singly-linked chain of pages. Each page body carries a
//! `next` pointer (page id, 0 = end of chain) followed by raw data bytes.
//!
//! - Root page (`PageKind::OverflowRoot`): `refcount[4] || next[8] || data_len[4] || data`.
//!   `release` CoW-copies the root with `refcount - 1` and, when it reaches 0,
//!   frees the entire chain. Snapshot lifetime does not rely on the count:
//!   copy-on-write leaves that still name a chain are protected because commit
//!   tags every freed page and reclamation waits until no retained root or live
//!   reader can reach it. Nothing in this crate raises a count above 1, so the
//!   decrement branch only fires for a root written by some other producer.
//! - Chain page (`PageKind::Overflow`): `next[8] || data_len[4] || data`.
//!
//! The two layouts differ only in the root's 4-byte `refcount` prefix, so the
//! `next` pointer sits at byte 4 in the root and byte 0 in a chain page — any
//! chain walk must account for that offset.

use std::collections::BTreeSet;

use crate::errors::PagedbError;
use crate::pager::Pager;
use crate::pager::format::data_page::ENVELOPE_OVERHEAD;
use crate::pager::format::page_kind::PageKind;
use crate::vfs::Vfs;
use crate::{RealmId, Result};

/// Header length for chain pages (non-root): `next[8] || data_len[4]`.
pub const OVERFLOW_HEADER_LEN: usize = 12;

/// Values longer than this are stored as an overflow chain rather than inline in
/// a leaf. Shared by the `put` path and `bulk_load` so a repack reproduces the
/// same inline/overflow split the original writes produced.
#[must_use]
pub fn inline_value_threshold(page_size: usize) -> usize {
    page_size / 4
}

/// Extra bytes at the start of a root body before the standard header:
/// `refcount[4]`.
const OVERFLOW_ROOT_PREFIX: usize = 4;

/// Header length for a root page: `refcount[4] || next[8] || data_len[4]`.
pub const OVERFLOW_ROOT_HEADER_LEN: usize = OVERFLOW_ROOT_PREFIX + OVERFLOW_HEADER_LEN;

#[must_use]
pub fn overflow_page_capacity(page_size: usize) -> usize {
    page_size - ENVELOPE_OVERHEAD - OVERFLOW_HEADER_LEN
}

/// Capacity of a root page body (4 bytes smaller than a chain page).
#[must_use]
pub fn overflow_root_capacity(page_size: usize) -> usize {
    page_size - ENVELOPE_OVERHEAD - OVERFLOW_ROOT_HEADER_LEN
}

/// Encode a single overflow chain page body (non-root, `PageKind::Overflow`).
/// `data.len()` must be ≤ `overflow_page_capacity(page_size)`.
pub fn encode_overflow(body: &mut [u8], next: u64, data: &[u8]) -> Result<()> {
    let page_size = body.len() + ENVELOPE_OVERHEAD;
    let cap = overflow_page_capacity(page_size);
    if data.len() > cap {
        return Err(PagedbError::PayloadTooLarge);
    }
    for b in body.iter_mut() {
        *b = 0;
    }
    body[0..8].copy_from_slice(&next.to_le_bytes());
    let data_len = u32::try_from(data.len())
        .map_err(|_| PagedbError::Io(std::io::Error::other("overflow data_len overflow")))?;
    body[8..12].copy_from_slice(&data_len.to_le_bytes());
    body[12..12 + data.len()].copy_from_slice(data);
    Ok(())
}

/// Decode an overflow chain page body (non-root). Returns `(next, data_slice)`.
pub fn decode_overflow(body: &[u8]) -> Result<(u64, &[u8])> {
    if body.len() < OVERFLOW_HEADER_LEN {
        return Err(PagedbError::overflow_body_malformed(
            "chain_page.header_length",
        ));
    }
    let mut n = [0u8; 8];
    n.copy_from_slice(&body[0..8]);
    let next = u64::from_le_bytes(n);
    let mut l = [0u8; 4];
    l.copy_from_slice(&body[8..12]);
    let data_len = u32::from_le_bytes(l) as usize;
    if 12 + data_len > body.len() {
        return Err(PagedbError::overflow_body_malformed(
            "chain_page.data_length",
        ));
    }
    Ok((next, &body[12..12 + data_len]))
}

/// Encode an overflow root page body (`PageKind::OverflowRoot`).
/// `data.len()` must be ≤ `overflow_root_capacity(page_size)`.
fn encode_overflow_root(body: &mut [u8], refcount: u32, next: u64, data: &[u8]) -> Result<()> {
    let page_size = body.len() + ENVELOPE_OVERHEAD;
    let cap = overflow_root_capacity(page_size);
    if data.len() > cap {
        return Err(PagedbError::PayloadTooLarge);
    }
    for b in body.iter_mut() {
        *b = 0;
    }
    body[0..4].copy_from_slice(&refcount.to_le_bytes());
    body[4..12].copy_from_slice(&next.to_le_bytes());
    let data_len = u32::try_from(data.len())
        .map_err(|_| PagedbError::Io(std::io::Error::other("overflow root data_len overflow")))?;
    body[12..16].copy_from_slice(&data_len.to_le_bytes());
    body[16..16 + data.len()].copy_from_slice(data);
    Ok(())
}

/// Decode an overflow root page body. Returns `(refcount, next, data_slice)`.
fn decode_overflow_root(body: &[u8]) -> Result<(u32, u64, &[u8])> {
    if body.len() < OVERFLOW_ROOT_HEADER_LEN {
        return Err(PagedbError::overflow_body_malformed("root.header_length"));
    }
    let mut r = [0u8; 4];
    r.copy_from_slice(&body[0..4]);
    let refcount = u32::from_le_bytes(r);
    let mut n = [0u8; 8];
    n.copy_from_slice(&body[4..12]);
    let next = u64::from_le_bytes(n);
    let mut l = [0u8; 4];
    l.copy_from_slice(&body[12..16]);
    let data_len = u32::from_le_bytes(l) as usize;
    if 16 + data_len > body.len() {
        return Err(PagedbError::overflow_body_malformed("root.data_length"));
    }
    Ok((refcount, next, &body[16..16 + data_len]))
}

/// Decoded contents of an overflow root page.
pub struct RootPageInfo {
    /// Reference count (shared-chain owner count).
    pub refcount: u32,
    /// `next` page id in the chain (0 = end).
    pub next: u64,
    /// Data bytes stored in the root page.
    pub root_data: Vec<u8>,
}

/// Read the root page of an overflow chain (`PageKind::OverflowRoot`).
pub async fn read_root_page<V: Vfs>(
    pager: &Pager<V>,
    realm_id: RealmId,
    root_page_id: u64,
) -> Result<RootPageInfo> {
    let guard = pager
        .read_main_page(root_page_id, realm_id, PageKind::OverflowRoot)
        .await?;
    let body = guard.body();
    let (refcount, next, data) = decode_overflow_root(&body)?;
    Ok(RootPageInfo {
        refcount,
        next,
        root_data: data.to_vec(),
    })
}

/// Write a value's overflow chain via the Pager. The root page is written as
/// `PageKind::OverflowRoot` with `refcount = 1`; chain pages use
/// `PageKind::Overflow`. Returns the root page's `page_id`.
pub async fn write_chain<V: Vfs>(
    pager: &Pager<V>,
    realm_id: RealmId,
    value: &[u8],
    page_size: usize,
    allocate_page: &mut (dyn FnMut() -> u64 + Send),
) -> Result<u64> {
    let root_cap = overflow_root_capacity(page_size);
    let chain_cap = overflow_page_capacity(page_size);

    // Collect chunk boundaries. The first chunk goes into the root page
    // (smaller capacity); subsequent chunks go into chain pages.
    let mut offsets: Vec<usize> = Vec::new();
    let mut o = 0usize;
    loop {
        offsets.push(o);
        let cap = if offsets.len() == 1 {
            root_cap
        } else {
            chain_cap
        };
        o += cap;
        if o >= value.len() {
            break;
        }
    }
    // Always have at least one page (root), even for zero-byte values.

    let page_ids: Vec<u64> = offsets.iter().map(|_| allocate_page()).collect();

    for (i, &start) in offsets.iter().enumerate() {
        let is_root = i == 0;
        let cap = if is_root { root_cap } else { chain_cap };
        let end = (start + cap).min(value.len());
        let next = if i + 1 < page_ids.len() {
            page_ids[i + 1]
        } else {
            0
        };
        let chunk = &value[start..end];
        let mut body = vec![0u8; page_size - ENVELOPE_OVERHEAD];
        if is_root {
            encode_overflow_root(&mut body, 1, next, chunk)?;
            pager
                .write_main_page(page_ids[i], realm_id, PageKind::OverflowRoot, &body)
                .await?;
        } else {
            encode_overflow(&mut body, next, chunk)?;
            pager
                .write_main_page(page_ids[i], realm_id, PageKind::Overflow, &body)
                .await?;
        }
    }
    Ok(page_ids[0])
}

/// The result of a `release` call.
pub enum ReleaseResult {
    /// Refcount decremented; the decremented root was written to the caller's
    /// `new_page_id`. The caller must free the old root page.
    Decremented,
    /// Refcount reached 0; all chain pages are listed in `freed_pages`
    /// (including the original root). The caller must free them all.
    Freed { freed_pages: Vec<u64> },
}

/// Decrement the reference count of an overflow root page (`CoW`).
///
/// - If `refcount > 1`: writes new root at `new_page_id` with `refcount - 1`
///   and returns `ReleaseResult::Decremented`. The caller must free the old
///   root page.
/// - If `refcount == 1`: collects all chain page ids and returns
///   `ReleaseResult::Freed`. The caller must free them all.
///   `new_page_id` is unused in this case.
pub async fn release<V: Vfs>(
    pager: &Pager<V>,
    realm_id: RealmId,
    root_page_id: u64,
    new_page_id: u64,
) -> Result<ReleaseResult> {
    let page_size = pager.page_size();
    let info = read_root_page(pager, realm_id, root_page_id).await?;

    if info.refcount <= 1 {
        // Collect entire chain.
        let mut freed = vec![root_page_id];
        let mut seen = BTreeSet::from([root_page_id]);
        let mut cur = info.next;
        while cur != 0 {
            if !seen.insert(cur) {
                return Err(PagedbError::overflow_chain_cycle(root_page_id, cur));
            }
            let guard = pager
                .read_main_page(cur, realm_id, PageKind::Overflow)
                .await?;
            let body = guard.body();
            let (n, _) = decode_overflow(&body)?;
            freed.push(cur);
            cur = n;
        }
        return Ok(ReleaseResult::Freed { freed_pages: freed });
    }

    let new_refcount = info.refcount - 1;
    let mut body = vec![0u8; page_size - ENVELOPE_OVERHEAD];
    encode_overflow_root(&mut body, new_refcount, info.next, &info.root_data)?;
    pager
        .write_main_page(new_page_id, realm_id, PageKind::OverflowRoot, &body)
        .await?;
    Ok(ReleaseResult::Decremented)
}

/// Read a value's overflow chain via the Pager. Follows `next` pointers until 0.
pub async fn read_chain<V: Vfs>(
    pager: &Pager<V>,
    realm_id: RealmId,
    root_page_id: u64,
    total_len: u64,
) -> Result<Vec<u8>> {
    let total_len = usize::try_from(total_len)
        .ok()
        .filter(|len| isize::try_from(*len).is_ok())
        .ok_or_else(|| PagedbError::overflow_body_malformed("chain.total_length"))?;
    // Durable metadata must not choose an arbitrarily large allocation before
    // any chain page has been authenticated. Grow only as bytes are verified.
    let mut out: Vec<u8> = Vec::with_capacity(total_len.min(pager.page_size()));

    let info = read_root_page(pager, realm_id, root_page_id).await?;
    if info.root_data.len() > total_len {
        return Err(PagedbError::overflow_body_malformed(
            "chain.assembled_length",
        ));
    }
    out.extend_from_slice(&info.root_data);

    let mut seen = BTreeSet::from([root_page_id]);
    let mut next = info.next;
    while next != 0 {
        if !seen.insert(next) {
            return Err(PagedbError::overflow_chain_cycle(root_page_id, next));
        }
        let guard = pager
            .read_main_page(next, realm_id, PageKind::Overflow)
            .await?;
        let body = guard.body();
        let (n, data) = decode_overflow(&body)?;
        if out
            .len()
            .checked_add(data.len())
            .is_none_or(|assembled| assembled > total_len)
        {
            return Err(PagedbError::overflow_body_malformed(
                "chain.assembled_length",
            ));
        }
        out.extend_from_slice(data);
        next = n;
    }
    if out.len() != total_len {
        return Err(PagedbError::overflow_body_malformed(
            "chain.assembled_length",
        ));
    }
    Ok(out)
}

/// Collect every `page_id` in an overflow chain. Does not modify any pages.
/// Used when refcount tracking is handled externally.
pub async fn collect_chain<V: Vfs>(
    pager: &Pager<V>,
    realm_id: RealmId,
    root_page_id: u64,
) -> Result<Vec<u64>> {
    let mut out = vec![root_page_id];
    let mut seen = BTreeSet::from([root_page_id]);
    let info = read_root_page(pager, realm_id, root_page_id).await?;
    let mut next = info.next;
    while next != 0 {
        if !seen.insert(next) {
            return Err(PagedbError::overflow_chain_cycle(root_page_id, next));
        }
        let guard = pager
            .read_main_page(next, realm_id, PageKind::Overflow)
            .await?;
        let body = guard.body();
        let (n, _) = decode_overflow(&body)?;
        out.push(next);
        next = n;
    }
    Ok(out)
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::time::Duration;

    use crate::crypto::CipherId;
    use crate::crypto::kdf::derive_mk;
    use crate::errors::CorruptionDetail;
    use crate::pager::PagerConfig;
    use crate::vfs::memory::MemVfs;

    use super::*;

    const TEST_PAGE_SIZE: usize = 4096;
    const TEST_REALM: RealmId = RealmId::new([0xA4; 16]);

    async fn test_pager() -> Arc<Pager<MemVfs>> {
        let mk = derive_mk(&[0xA5; 32], &[0u8; 16], 0).unwrap();
        let cfg = PagerConfig {
            page_size: TEST_PAGE_SIZE,
            buffer_pool_pages: 16,
            segment_cache_pages: 16,
            cipher_id: CipherId::Aes256Gcm,
            mk_epoch: 0,
            main_db_file_id: [0xB4; 16],
            main_db_path: "/main.db".into(),
            anchor_budget: 1_000_000,
            dek_lru_capacity: 16,
            observer_retry_count: 0,
            metrics_enabled: true,
        };
        Arc::new(Pager::open(MemVfs::new(), mk, cfg).await.unwrap())
    }

    #[test]
    fn round_trip_chain_page() {
        let mut body = vec![0u8; 4096 - ENVELOPE_OVERHEAD];
        encode_overflow(&mut body, 7, b"hello").unwrap();
        let (n, d) = decode_overflow(&body).unwrap();
        assert_eq!(n, 7);
        assert_eq!(d, b"hello");
    }

    #[test]
    fn round_trip_root() {
        let mut body = vec![0u8; 4096 - ENVELOPE_OVERHEAD];
        encode_overflow_root(&mut body, 3, 99, b"world").unwrap();
        let (rc, n, d) = decode_overflow_root(&body).unwrap();
        assert_eq!(rc, 3);
        assert_eq!(n, 99);
        assert_eq!(d, b"world");
    }

    #[test]
    fn capacity_4k_page() {
        // chain: 4096 - 40 - 12 = 4044
        assert_eq!(overflow_page_capacity(4096), 4044);
        // root: 4096 - 40 - 16 = 4040
        assert_eq!(overflow_root_capacity(4096), 4040);
    }

    async fn cyclic_chain(pager: &Pager<MemVfs>, root_page_id: u64, chain_page_id: u64) {
        let mut root_body = vec![0u8; TEST_PAGE_SIZE - ENVELOPE_OVERHEAD];
        encode_overflow_root(&mut root_body, 1, chain_page_id, b"").unwrap();
        pager
            .write_main_page(root_page_id, TEST_REALM, PageKind::OverflowRoot, &root_body)
            .await
            .unwrap();

        let mut chain_body = vec![0u8; TEST_PAGE_SIZE - ENVELOPE_OVERHEAD];
        encode_overflow(&mut chain_body, chain_page_id, b"").unwrap();
        pager
            .write_main_page(chain_page_id, TEST_REALM, PageKind::Overflow, &chain_body)
            .await
            .unwrap();
    }

    #[tokio::test(flavor = "current_thread")]
    async fn read_chain_rejects_absurd_total_len_without_allocation_panic() {
        let pager = test_pager().await;
        let root_page_id = 42;
        let mut body = vec![0u8; TEST_PAGE_SIZE - ENVELOPE_OVERHEAD];
        encode_overflow_root(&mut body, 1, 0, b"small").unwrap();
        pager
            .write_main_page(root_page_id, TEST_REALM, PageKind::OverflowRoot, &body)
            .await
            .unwrap();

        let error = read_chain(&pager, TEST_REALM, root_page_id, u64::MAX)
            .await
            .unwrap_err();
        assert!(matches!(
            error,
            PagedbError::Corruption(CorruptionDetail::OverflowBodyMalformed { .. })
        ));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn read_chain_rejects_more_data_than_declared() {
        let pager = test_pager().await;
        let root_page_id = 43;
        let mut body = vec![0u8; TEST_PAGE_SIZE - ENVELOPE_OVERHEAD];
        encode_overflow_root(&mut body, 1, 0, b"too-long").unwrap();
        pager
            .write_main_page(root_page_id, TEST_REALM, PageKind::OverflowRoot, &body)
            .await
            .unwrap();

        let error = read_chain(&pager, TEST_REALM, root_page_id, 1)
            .await
            .unwrap_err();
        assert!(matches!(
            error,
            PagedbError::Corruption(CorruptionDetail::OverflowBodyMalformed { .. })
        ));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn read_chain_rejects_cycle_without_hanging() {
        let pager = test_pager().await;
        cyclic_chain(&pager, 44, 45).await;
        let error = tokio::time::timeout(
            Duration::from_secs(1),
            read_chain(&pager, TEST_REALM, 44, 0),
        )
        .await
        .expect("cycle detection should return before the timeout")
        .unwrap_err();
        assert!(matches!(
            error,
            PagedbError::Corruption(CorruptionDetail::OverflowChainCycle { .. })
        ));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn release_rejects_cycle_without_hanging() {
        let pager = test_pager().await;
        cyclic_chain(&pager, 46, 47).await;
        let result =
            tokio::time::timeout(Duration::from_secs(1), release(&pager, TEST_REALM, 46, 48))
                .await
                .expect("cycle detection should return before the timeout");
        let Err(error) = result else {
            panic!("overflow release cycles must not be accepted");
        };
        assert!(matches!(
            error,
            PagedbError::Corruption(CorruptionDetail::OverflowChainCycle { .. })
        ));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn collect_chain_rejects_cycle_without_hanging() {
        let pager = test_pager().await;
        cyclic_chain(&pager, 49, 50).await;
        let error = tokio::time::timeout(
            Duration::from_secs(1),
            collect_chain(&pager, TEST_REALM, 49),
        )
        .await
        .expect("cycle detection should return before the timeout")
        .expect_err("overflow collect cycles must not be accepted");
        assert!(matches!(
            error,
            PagedbError::Corruption(CorruptionDetail::OverflowChainCycle { .. })
        ));
    }
}