shodh-redb 0.3.1

use crate::cdc::types::{CdcEvent, ChangeOp};
use crate::compat::Mutex;
use crate::db::TransactionGuard;
use crate::merge::MergeOperator;
use crate::sealed::Sealed;
use crate::tree_store::{
    AccessGuardMutInPlace, Btree, BtreeExtractIf, BtreeHeader, BtreeMut, BtreeRangeIter,
    MAX_PAIR_LENGTH, MAX_VALUE_LENGTH, PageHint, PageNumber, PageTrackerPolicy, RawBtree,
    RawEntryIter, TransactionalMemory,
};
use crate::types::{Key, MutInPlaceValue, Value};
use crate::{AccessGuard, AccessGuardMut, StorageError, WriteTransaction};
use crate::{Result, TableHandle};
use alloc::string::{String, ToString};
use alloc::sync::Arc;
use alloc::vec::Vec;
use core::borrow::Borrow;
use core::fmt::{Debug, Formatter};
use core::marker::PhantomData;
use core::ops::RangeBounds;

/// Informational storage stats about a table
#[derive(Debug, Clone, Copy)]
pub struct TableStats {
    pub(crate) tree_height: u32,
    pub(crate) leaf_pages: u64,
    pub(crate) branch_pages: u64,
    pub(crate) stored_leaf_bytes: u64,
    pub(crate) metadata_bytes: u64,
    pub(crate) fragmented_bytes: u64,
}

impl TableStats {
    /// Maximum traversal distance to reach the deepest (key, value) pair in the table
    pub fn tree_height(&self) -> u32 {
        self.tree_height
    }

    /// Number of leaf pages that store user data
    pub fn leaf_pages(&self) -> u64 {
        self.leaf_pages
    }

    /// Number of branch pages in the btree that store user data
    pub fn branch_pages(&self) -> u64 {
        self.branch_pages
    }

    /// Number of bytes consumed by keys and values that have been inserted.
    /// Does not include indexing overhead
    pub fn stored_bytes(&self) -> u64 {
        self.stored_leaf_bytes
    }

    /// Number of bytes consumed by keys in internal branch pages, plus other metadata
    pub fn metadata_bytes(&self) -> u64 {
        self.metadata_bytes
    }

    /// Number of bytes consumed by fragmentation, both in data pages and internal metadata tables
    pub fn fragmented_bytes(&self) -> u64 {
        self.fragmented_bytes
    }
}

/// A table containing key-value mappings
pub struct Table<'txn, K: Key + 'static, V: Value + 'static> {
    name: String,
    transaction: &'txn WriteTransaction,
    tree: BtreeMut<'txn, K, V>,
}

impl<K: Key + 'static, V: Value + 'static> TableHandle for Table<'_, K, V> {
    fn name(&self) -> &str {
        &self.name
    }
}

impl<'txn, K: Key + 'static, V: Value + 'static> Table<'txn, K, V> {
    pub(crate) fn new(
        name: &str,
        table_root: Option<BtreeHeader>,
        freed_pages: Arc<Mutex<Vec<PageNumber>>>,
        allocated_pages: Arc<Mutex<PageTrackerPolicy>>,
        mem: Arc<TransactionalMemory>,
        transaction: &'txn WriteTransaction,
    ) -> Table<'txn, K, V> {
        Table {
            name: name.to_string(),
            transaction,
            tree: BtreeMut::new(
                table_root,
                transaction.transaction_guard(),
                mem,
                freed_pages,
                allocated_pages,
            ),
        }
    }

    #[allow(dead_code)]
    #[cfg(feature = "std")]
    pub(crate) fn print_debug(&self, include_values: bool) -> Result {
        self.tree.print_debug(include_values)
    }

    /// Returns an accessor, which allows mutation, to the value corresponding to the given key
    pub fn get_mut<'k>(
        &mut self,
        key: impl Borrow<K::SelfType<'k>>,
    ) -> Result<Option<AccessGuardMut<'_, V>>> {
        self.tree.get_mut(key.borrow())
    }

    /// Removes and returns the first key-value pair in the table
    pub fn pop_first(&mut self) -> Result<Option<(AccessGuard<'_, K>, AccessGuard<'_, V>)>> {
        // TODO: optimize this
        let first = self
            .iter()?
            .next()
            .map(|x| x.map(|(key, _)| K::as_bytes(&key.value()).as_ref().to_vec()));
        if let Some(owned_key) = first {
            let owned_key = owned_key?;
            let key = K::from_bytes(&owned_key);
            let value = self.remove(&key)?.ok_or_else(|| {
                StorageError::Corrupted("key disappeared during pop operation".into())
            })?;
            drop(key);
            Ok(Some((AccessGuard::with_owned_value(owned_key), value)))
        } else {
            Ok(None)
        }
    }

    /// Removes and returns the last key-value pair in the table
    pub fn pop_last(&mut self) -> Result<Option<(AccessGuard<'_, K>, AccessGuard<'_, V>)>> {
        // TODO: optimize this
        let last = self
            .iter()?
            .next_back()
            .map(|x| x.map(|(key, _)| K::as_bytes(&key.value()).as_ref().to_vec()));
        if let Some(owned_key) = last {
            let owned_key = owned_key?;
            let key = K::from_bytes(&owned_key);
            let value = self.remove(&key)?.ok_or_else(|| {
                StorageError::Corrupted("key disappeared during pop operation".into())
            })?;
            drop(key);
            Ok(Some((AccessGuard::with_owned_value(owned_key), value)))
        } else {
            Ok(None)
        }
    }

    /// Applies `predicate` to all key-value pairs. All entries for which
    /// `predicate` evaluates to `true` are returned in an iterator, and those which are read from the iterator are removed
    ///
    /// Note: values not read from the iterator will not be removed
    pub fn extract_if<F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool>(
        &mut self,
        predicate: F,
    ) -> Result<ExtractIf<'_, K, V, F>> {
        self.extract_from_if::<K::SelfType<'_>, F>(.., predicate)
    }

    /// Applies `predicate` to all key-value pairs in the specified range. All entries for which
    /// `predicate` evaluates to `true` are returned in an iterator, and those which are read from the iterator are removed
    ///
    /// Note: values not read from the iterator will not be removed
    pub fn extract_from_if<'a, KR, F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool>(
        &mut self,
        range: impl RangeBounds<KR> + 'a,
        predicate: F,
    ) -> Result<ExtractIf<'_, K, V, F>>
    where
        KR: Borrow<K::SelfType<'a>> + 'a,
    {
        let inner = self.tree.extract_from_if(&range, predicate)?;
        if self.transaction.cdc_log.is_some() {
            Ok(ExtractIf::with_cdc(
                inner,
                self.transaction,
                self.name.clone(),
            ))
        } else {
            Ok(ExtractIf::new(inner))
        }
    }

    /// Applies `predicate` to all key-value pairs. All entries for which
    /// `predicate` evaluates to `false` are removed.
    ///
    pub fn retain<F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool>(
        &mut self,
        predicate: F,
    ) -> Result {
        self.retain_in::<K::SelfType<'_>, F>(.., predicate)
    }

    /// Applies `predicate` to all key-value pairs in the range `start..end`. All entries for which
    /// `predicate` evaluates to `false` are removed.
    ///
    pub fn retain_in<'a, KR, F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool>(
        &mut self,
        range: impl RangeBounds<KR> + 'a,
        mut predicate: F,
    ) -> Result
    where
        KR: Borrow<K::SelfType<'a>> + 'a,
    {
        if self.transaction.cdc_log.is_some() {
            // Route through extract_from_if so CDC events are recorded for each removal
            for result in self.extract_from_if(range, move |k, v| !predicate(k, v))? {
                result?;
            }
            Ok(())
        } else {
            self.tree.retain_in(predicate, range)
        }
    }

    /// Removes all key-value pairs in the given range
    ///
    /// Returns the number of entries removed
    pub fn drain<'a, KR>(&mut self, range: impl RangeBounds<KR> + 'a) -> Result<u64>
    where
        KR: Borrow<K::SelfType<'a>> + 'a,
    {
        let mut count = 0u64;
        for result in self.extract_from_if(range, |_, _| true)? {
            result?;
            count += 1;
        }
        Ok(count)
    }

    /// Removes all key-value pairs from the table
    ///
    /// Returns the number of entries removed
    pub fn drain_all(&mut self) -> Result<u64> {
        self.drain::<K::SelfType<'_>>(..)
    }

    /// Insert mapping of the given key to the given value
    ///
    /// If key is already present it is replaced
    ///
    /// Returns the old value, if the key was present in the table, otherwise None is returned
    pub fn insert<'k, 'v>(
        &mut self,
        key: impl Borrow<K::SelfType<'k>>,
        value: impl Borrow<V::SelfType<'v>>,
    ) -> Result<Option<AccessGuard<'_, V>>> {
        let value_len = V::as_bytes(value.borrow()).as_ref().len();
        if value_len > MAX_VALUE_LENGTH {
            return Err(StorageError::ValueTooLarge(value_len));
        }
        let key_len = K::as_bytes(key.borrow()).as_ref().len();
        if key_len > MAX_VALUE_LENGTH {
            return Err(StorageError::ValueTooLarge(key_len));
        }
        if value_len + key_len > MAX_PAIR_LENGTH {
            return Err(StorageError::ValueTooLarge(value_len + key_len));
        }
        let result = self.tree.insert(key.borrow(), value.borrow())?;
        if self.transaction.cdc_log.is_some() {
            let old_value = result
                .as_ref()
                .map(|g| V::as_bytes(&g.value()).as_ref().to_vec());
            self.transaction.record_cdc(CdcEvent {
                table_name: self.name.clone(),
                op: if old_value.is_some() {
                    ChangeOp::Update
                } else {
                    ChangeOp::Insert
                },
                key: K::as_bytes(key.borrow()).as_ref().to_vec(),
                new_value: Some(V::as_bytes(value.borrow()).as_ref().to_vec()),
                old_value,
            });
        }
        Ok(result)
    }

    /// Removes the given key
    ///
    /// Returns the old value, if the key was present in the table
    pub fn remove<'a>(
        &mut self,
        key: impl Borrow<K::SelfType<'a>>,
    ) -> Result<Option<AccessGuard<'_, V>>> {
        let result = self.tree.remove(key.borrow())?;
        if self.transaction.cdc_log.is_some() && result.is_some() {
            let old_value = result
                .as_ref()
                .map(|g| V::as_bytes(&g.value()).as_ref().to_vec());
            self.transaction.record_cdc(CdcEvent {
                table_name: self.name.clone(),
                op: ChangeOp::Delete,
                key: K::as_bytes(key.borrow()).as_ref().to_vec(),
                new_value: None,
                old_value,
            });
        }
        Ok(result)
    }

    /// Atomically read-modify-write a value using a [`MergeOperator`].
    ///
    /// Reads the current value for `key` (if any), passes the raw bytes to
    /// `operator.merge()` along with `operand`, and writes the result back.
    /// If the operator returns `None`, the key is removed.
    ///
    /// This is equivalent to a get-then-insert but expressed as a single call.
    /// The operation is atomic within the current write transaction.
    pub fn merge<'k>(
        &mut self,
        key: impl Borrow<K::SelfType<'k>>,
        operand: &[u8],
        operator: &dyn MergeOperator,
    ) -> Result<()> {
        let key_ref = key.borrow();

        // Copy key and existing value bytes, then drop the access guard
        // to release the immutable borrow before calling insert/remove.
        // This is the same pattern used by pop_first()/pop_last().
        let key_bytes = K::as_bytes(key_ref).as_ref().to_vec();
        let existing_bytes: Option<Vec<u8>> = {
            let guard = self.get(key_ref)?;
            guard.map(|g| V::as_bytes(&g.value()).as_ref().to_vec())
        };

        let merged = operator.merge(&key_bytes, existing_bytes.as_deref(), operand);

        let key_ref = K::from_bytes(&key_bytes);
        match merged {
            Some(new_bytes) => {
                let new_value = V::from_bytes(&new_bytes);
                self.insert(&key_ref, &new_value)?;
            }
            None => {
                self.remove(&key_ref)?;
            }
        }
        Ok(())
    }

    /// Type-safe variant of [`merge()`](Self::merge) that serializes the operand
    /// via `V::as_bytes()` before passing it to the operator.
    pub fn merge_in<'k, 'v>(
        &mut self,
        key: impl Borrow<K::SelfType<'k>>,
        operand: impl Borrow<V::SelfType<'v>>,
        operator: &dyn MergeOperator,
    ) -> Result<()> {
        let operand_bytes = V::as_bytes(operand.borrow()).as_ref().to_vec();
        self.merge(key, &operand_bytes, operator)
    }
}

impl<K: Key + 'static, V: MutInPlaceValue + 'static> Table<'_, K, V> {
    /// Reserve space to insert a key-value pair
    ///
    /// If key is already present it is replaced
    ///
    /// The returned reference will have length equal to `value_length`
    pub fn insert_reserve<'a>(
        &mut self,
        key: impl Borrow<K::SelfType<'a>>,
        value_length: usize,
    ) -> Result<AccessGuardMutInPlace<'_, V>> {
        if value_length > MAX_VALUE_LENGTH {
            return Err(StorageError::ValueTooLarge(value_length));
        }
        let key_len = K::as_bytes(key.borrow()).as_ref().len();
        if key_len > MAX_VALUE_LENGTH {
            return Err(StorageError::ValueTooLarge(key_len));
        }
        if value_length + key_len > MAX_PAIR_LENGTH {
            return Err(StorageError::ValueTooLarge(value_length + key_len));
        }
        self.tree.insert_reserve(key.borrow(), value_length)
    }
}

impl<K: Key + 'static, V: Value + 'static> ReadableTableMetadata for Table<'_, K, V> {
    fn stats(&self) -> Result<TableStats> {
        let tree_stats = self.tree.stats()?;

        Ok(TableStats {
            tree_height: tree_stats.tree_height,
            leaf_pages: tree_stats.leaf_pages,
            branch_pages: tree_stats.branch_pages,
            stored_leaf_bytes: tree_stats.stored_leaf_bytes,
            metadata_bytes: tree_stats.metadata_bytes,
            fragmented_bytes: tree_stats.fragmented_bytes,
        })
    }

    fn len(&self) -> Result<u64> {
        self.tree.len()
    }
}

impl<K: Key + 'static, V: Value + 'static> ReadableTable<K, V> for Table<'_, K, V> {
    fn get<'a>(&self, key: impl Borrow<K::SelfType<'a>>) -> Result<Option<AccessGuard<'_, V>>> {
        self.tree.get(key.borrow())
    }

    fn range<'a, KR>(&self, range: impl RangeBounds<KR> + 'a) -> Result<Range<'_, K, V>>
    where
        KR: Borrow<K::SelfType<'a>> + 'a,
    {
        self.tree
            .range(&range)
            .map(|x| Range::new(x, self.transaction.transaction_guard()))
    }

    fn first(&self) -> Result<Option<(AccessGuard<'_, K>, AccessGuard<'_, V>)>> {
        self.tree.first()
    }

    fn last(&self) -> Result<Option<(AccessGuard<'_, K>, AccessGuard<'_, V>)>> {
        self.tree.last()
    }
}

impl<K: Key, V: Value> Sealed for Table<'_, K, V> {}

impl<K: Key + 'static, V: Value + 'static> Drop for Table<'_, K, V> {
    fn drop(&mut self) {
        self.transaction.close_table(
            &self.name,
            &self.tree,
            self.tree.get_root().map(|x| x.length).unwrap_or_default(),
        );
    }
}

fn debug_helper<K: Key + 'static, V: Value + 'static>(
    f: &mut Formatter<'_>,
    name: &str,
    len: Result<u64>,
    first: Result<Option<(AccessGuard<K>, AccessGuard<V>)>>,
    last: Result<Option<(AccessGuard<K>, AccessGuard<V>)>>,
) -> core::fmt::Result {
    write!(f, "Table [ name: \"{name}\", ")?;
    if let Ok(len) = len {
        if len == 0 {
            write!(f, "No entries")?;
        } else if len == 1 {
            if let Ok(first) = first {
                let (key, value) = first.as_ref().unwrap();
                write!(f, "One key-value: {:?} = {:?}", key.value(), value.value())?;
            } else {
                write!(f, "I/O Error accessing table!")?;
            }
        } else {
            if let Ok(first) = first {
                let (key, value) = first.as_ref().unwrap();
                write!(f, "first: {:?} = {:?}, ", key.value(), value.value())?;
            } else {
                write!(f, "I/O Error accessing table!")?;
            }
            if len > 2 {
                write!(f, "...{} more entries..., ", len - 2)?;
            }
            if let Ok(last) = last {
                let (key, value) = last.as_ref().unwrap();
                write!(f, "last: {:?} = {:?}", key.value(), value.value())?;
            } else {
                write!(f, "I/O Error accessing table!")?;
            }
        }
    } else {
        write!(f, "I/O Error accessing table!")?;
    }
    write!(f, " ]")?;

    Ok(())
}

impl<K: Key + 'static, V: Value + 'static> Debug for Table<'_, K, V> {
    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
        debug_helper(f, &self.name, self.len(), self.first(), self.last())
    }
}

pub trait ReadableTableMetadata {
    /// Retrieves information about storage usage for the table
    fn stats(&self) -> Result<TableStats>;

    /// Returns the number of entries in the table
    fn len(&self) -> Result<u64>;

    /// Returns `true` if the table is empty
    fn is_empty(&self) -> Result<bool> {
        Ok(self.len()? == 0)
    }
}

pub trait ReadableTable<K: Key + 'static, V: Value + 'static>: ReadableTableMetadata {
    /// Returns the value corresponding to the given key
    fn get<'a>(&self, key: impl Borrow<K::SelfType<'a>>) -> Result<Option<AccessGuard<'_, V>>>;

    /// Returns a double-ended iterator over a range of elements in the table
    ///
    /// # Examples
    ///
    /// Usage:
    /// ```rust
    /// use shodh_redb::*;
    /// # use tempfile::NamedTempFile;
    /// const TABLE: TableDefinition<&str, u64> = TableDefinition::new("my_data");
    ///
    /// # fn main() -> Result<(), Error> {
    /// # #[cfg(not(target_os = "wasi"))]
    /// # let tmpfile = NamedTempFile::new().unwrap();
    /// # #[cfg(target_os = "wasi")]
    /// # let tmpfile = NamedTempFile::new_in("/tmp").unwrap();
    /// # let filename = tmpfile.path();
    /// let db = Database::create(filename)?;
    /// let write_txn = db.begin_write()?;
    /// {
    ///     let mut table = write_txn.open_table(TABLE)?;
    ///     table.insert("a", &0)?;
    ///     table.insert("b", &1)?;
    ///     table.insert("c", &2)?;
    /// }
    /// write_txn.commit()?;
    ///
    /// let read_txn = db.begin_read()?;
    /// let table = read_txn.open_table(TABLE)?;
    /// let mut iter = table.range("a".."c")?;
    /// let (key, value) = iter.next().unwrap()?;
    /// assert_eq!("a", key.value());
    /// assert_eq!(0, value.value());
    /// # Ok(())
    /// # }
    /// ```
    fn range<'a, KR>(&self, range: impl RangeBounds<KR> + 'a) -> Result<Range<'_, K, V>>
    where
        KR: Borrow<K::SelfType<'a>> + 'a;

    /// Returns the first key-value pair in the table, if it exists
    fn first(&self) -> Result<Option<(AccessGuard<'_, K>, AccessGuard<'_, V>)>>;

    /// Returns the last key-value pair in the table, if it exists
    fn last(&self) -> Result<Option<(AccessGuard<'_, K>, AccessGuard<'_, V>)>>;

    /// Returns a double-ended iterator over all elements in the table
    fn iter(&self) -> Result<Range<'_, K, V>> {
        self.range::<K::SelfType<'_>>(..)
    }
}

/// A read-only untyped table
pub struct ReadOnlyUntypedTable {
    tree: RawBtree,
}

impl Sealed for ReadOnlyUntypedTable {}

impl ReadableTableMetadata for ReadOnlyUntypedTable {
    /// Retrieves information about storage usage for the table
    fn stats(&self) -> Result<TableStats> {
        let tree_stats = self.tree.stats()?;

        Ok(TableStats {
            tree_height: tree_stats.tree_height,
            leaf_pages: tree_stats.leaf_pages,
            branch_pages: tree_stats.branch_pages,
            stored_leaf_bytes: tree_stats.stored_leaf_bytes,
            metadata_bytes: tree_stats.metadata_bytes,
            fragmented_bytes: tree_stats.fragmented_bytes,
        })
    }

    fn len(&self) -> Result<u64> {
        self.tree.len()
    }
}

impl ReadOnlyUntypedTable {
    pub(crate) fn new(
        root_page: Option<BtreeHeader>,
        fixed_key_size: Option<usize>,
        fixed_value_size: Option<usize>,
        mem: Arc<TransactionalMemory>,
    ) -> Self {
        Self {
            tree: RawBtree::new(root_page, fixed_key_size, fixed_value_size, mem),
        }
    }

    /// Iterate all entries as raw key/value byte slices.
    /// Useful for CLI tools, debugging, and migration without knowing types at compile time.
    pub fn iter_raw(&self) -> Result<RawEntryIter> {
        self.tree.raw_iter()
    }
}

/// A read-only table
pub struct ReadOnlyTable<K: Key + 'static, V: Value + 'static> {
    name: String,
    tree: Btree<K, V>,
    transaction_guard: Arc<TransactionGuard>,
}

impl<K: Key + 'static, V: Value + 'static> TableHandle for ReadOnlyTable<K, V> {
    fn name(&self) -> &str {
        &self.name
    }
}

impl<K: Key + 'static, V: Value + 'static> ReadOnlyTable<K, V> {
    pub(crate) fn new(
        name: String,
        root_page: Option<BtreeHeader>,
        hint: PageHint,
        guard: Arc<TransactionGuard>,
        mem: Arc<TransactionalMemory>,
    ) -> Result<ReadOnlyTable<K, V>> {
        Ok(ReadOnlyTable {
            name,
            tree: Btree::new(root_page, hint, guard.clone(), mem)?,
            transaction_guard: guard,
        })
    }

    /// Create a `ReadOnlyTable` that never applies value-level decompression.
    /// Used for system/internal tables whose values are always stored uncompressed.
    pub(crate) fn new_uncompressed(
        name: String,
        root_page: Option<BtreeHeader>,
        hint: PageHint,
        guard: Arc<TransactionGuard>,
        mem: Arc<TransactionalMemory>,
    ) -> Result<ReadOnlyTable<K, V>> {
        Ok(ReadOnlyTable {
            name,
            tree: Btree::new_uncompressed(root_page, hint, guard.clone(), mem)?,
            transaction_guard: guard,
        })
    }

    /// This method is like [`ReadableTable::get()`], but the [`AccessGuard`] is reference counted
    /// and keeps the transaction alive until it is dropped.
    pub fn get<'a>(
        &self,
        key: impl Borrow<K::SelfType<'a>>,
    ) -> Result<Option<AccessGuard<'static, V>>> {
        self.tree.get(key.borrow())
    }

    /// This method is like [`ReadableTable::range()`], but the iterator is reference counted and keeps the transaction
    /// alive until it is dropped.
    pub fn range<'a, KR>(&self, range: impl RangeBounds<KR>) -> Result<Range<'static, K, V>>
    where
        KR: Borrow<K::SelfType<'a>>,
    {
        self.tree
            .range(&range)
            .map(|x| Range::new(x, self.transaction_guard.clone()))
    }
}

impl<K: Key + 'static, V: Value + 'static> ReadableTableMetadata for ReadOnlyTable<K, V> {
    fn stats(&self) -> Result<TableStats> {
        let tree_stats = self.tree.stats()?;

        Ok(TableStats {
            tree_height: tree_stats.tree_height,
            leaf_pages: tree_stats.leaf_pages,
            branch_pages: tree_stats.branch_pages,
            stored_leaf_bytes: tree_stats.stored_leaf_bytes,
            metadata_bytes: tree_stats.metadata_bytes,
            fragmented_bytes: tree_stats.fragmented_bytes,
        })
    }

    fn len(&self) -> Result<u64> {
        self.tree.len()
    }
}

impl<K: Key + 'static, V: Value + 'static> ReadableTable<K, V> for ReadOnlyTable<K, V> {
    fn get<'a>(&self, key: impl Borrow<K::SelfType<'a>>) -> Result<Option<AccessGuard<'_, V>>> {
        self.tree.get(key.borrow())
    }

    fn range<'a, KR>(&self, range: impl RangeBounds<KR> + 'a) -> Result<Range<'_, K, V>>
    where
        KR: Borrow<K::SelfType<'a>> + 'a,
    {
        self.tree
            .range(&range)
            .map(|x| Range::new(x, self.transaction_guard.clone()))
    }

    fn first(&self) -> Result<Option<(AccessGuard<'_, K>, AccessGuard<'_, V>)>> {
        self.tree.first()
    }

    fn last(&self) -> Result<Option<(AccessGuard<'_, K>, AccessGuard<'_, V>)>> {
        self.tree.last()
    }
}

impl<K: Key, V: Value> Sealed for ReadOnlyTable<K, V> {}

impl<K: Key + 'static, V: Value + 'static> Debug for ReadOnlyTable<K, V> {
    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
        debug_helper(f, &self.name, self.len(), self.first(), self.last())
    }
}

pub struct ExtractIf<
    'a,
    K: Key + 'static,
    V: Value + 'static,
    F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool,
> {
    inner: BtreeExtractIf<'a, K, V, F>,
    cdc: Option<(&'a WriteTransaction, String)>,
}

impl<
    'a,
    K: Key + 'static,
    V: Value + 'static,
    F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool,
> ExtractIf<'a, K, V, F>
{
    pub(crate) fn new(inner: BtreeExtractIf<'a, K, V, F>) -> Self {
        Self { inner, cdc: None }
    }

    pub(crate) fn with_cdc(
        inner: BtreeExtractIf<'a, K, V, F>,
        transaction: &'a WriteTransaction,
        table_name: String,
    ) -> Self {
        Self {
            inner,
            cdc: Some((transaction, table_name)),
        }
    }
}

impl<
    'a,
    K: Key + 'static,
    V: Value + 'static,
    F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool,
> Iterator for ExtractIf<'a, K, V, F>
{
    type Item = Result<(AccessGuard<'a, K>, AccessGuard<'a, V>)>;

    fn next(&mut self) -> Option<Self::Item> {
        let entry = self.inner.next()?;
        Some(entry.map(|entry| {
            let (page, key_range, value_range, decompressed_value) = entry.into_raw();
            let key = AccessGuard::with_page(page.clone(), key_range);
            let value = if let Some(bytes) = decompressed_value {
                AccessGuard::with_owned_value(bytes)
            } else {
                AccessGuard::with_page(page, value_range)
            };
            if let Some((txn, table_name)) = &self.cdc {
                txn.record_cdc(CdcEvent {
                    table_name: table_name.clone(),
                    op: ChangeOp::Delete,
                    key: K::as_bytes(&key.value()).as_ref().to_vec(),
                    new_value: None,
                    old_value: Some(V::as_bytes(&value.value()).as_ref().to_vec()),
                });
            }
            (key, value)
        }))
    }
}

impl<
    K: Key + 'static,
    V: Value + 'static,
    F: for<'f> FnMut(K::SelfType<'f>, V::SelfType<'f>) -> bool,
> DoubleEndedIterator for ExtractIf<'_, K, V, F>
{
    fn next_back(&mut self) -> Option<Self::Item> {
        let entry = self.inner.next_back()?;
        Some(entry.map(|entry| {
            let (page, key_range, value_range, decompressed_value) = entry.into_raw();
            let key = AccessGuard::with_page(page.clone(), key_range);
            let value = if let Some(bytes) = decompressed_value {
                AccessGuard::with_owned_value(bytes)
            } else {
                AccessGuard::with_page(page, value_range)
            };
            if let Some((txn, table_name)) = &self.cdc {
                txn.record_cdc(CdcEvent {
                    table_name: table_name.clone(),
                    op: ChangeOp::Delete,
                    key: K::as_bytes(&key.value()).as_ref().to_vec(),
                    new_value: None,
                    old_value: Some(V::as_bytes(&value.value()).as_ref().to_vec()),
                });
            }
            (key, value)
        }))
    }
}

#[derive(Clone)]
pub struct Range<'a, K: Key + 'static, V: Value + 'static> {
    inner: BtreeRangeIter<K, V>,
    _transaction_guard: Arc<TransactionGuard>,
    // This lifetime is here so that `&` can be held on `Table` preventing concurrent mutation
    _lifetime: PhantomData<&'a ()>,
}

impl<K: Key + 'static, V: Value + 'static> Range<'_, K, V> {
    pub(super) fn new(inner: BtreeRangeIter<K, V>, guard: Arc<TransactionGuard>) -> Self {
        Self {
            inner,
            _transaction_guard: guard,
            _lifetime: Default::default(),
        }
    }
}

impl<'a, K: Key + 'static, V: Value + 'static> Iterator for Range<'a, K, V> {
    type Item = Result<(AccessGuard<'a, K>, AccessGuard<'a, V>)>;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(|x| {
            x.map(|entry| {
                let (page, key_range, value_range, decompressed_value) = entry.into_raw();
                let key = AccessGuard::with_page(page.clone(), key_range);
                let value = if let Some(bytes) = decompressed_value {
                    AccessGuard::with_owned_value(bytes)
                } else {
                    AccessGuard::with_page(page, value_range)
                };
                (key, value)
            })
        })
    }
}

impl<K: Key + 'static, V: Value + 'static> DoubleEndedIterator for Range<'_, K, V> {
    fn next_back(&mut self) -> Option<Self::Item> {
        self.inner.next_back().map(|x| {
            x.map(|entry| {
                let (page, key_range, value_range, decompressed_value) = entry.into_raw();
                let key = AccessGuard::with_page(page.clone(), key_range);
                let value = if let Some(bytes) = decompressed_value {
                    AccessGuard::with_owned_value(bytes)
                } else {
                    AccessGuard::with_page(page, value_range)
                };
                (key, value)
            })
        })
    }
}

// ---------------------------------------------------------------------------
// storage_traits::WriteTable implementation for legacy Table
// ---------------------------------------------------------------------------

use crate::storage_traits::OwnedKv;

/// Iterator adapter that converts `(AccessGuard<K>, AccessGuard<V>)` pairs
/// into `(OwnedKv<K>, OwnedKv<V>)` pairs for the storage trait interface.
pub struct LegacyRangeIter<'a, K: Key + 'static, V: Value + 'static> {
    inner: Range<'a, K, V>,
}

impl<K: Key + 'static, V: Value + 'static> Iterator for LegacyRangeIter<'_, K, V> {
    type Item = crate::Result<(OwnedKv<K>, OwnedKv<V>)>;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(|result| {
            result.map(|(k_guard, v_guard)| {
                let k_bytes = K::as_bytes(&k_guard.value()).as_ref().to_vec();
                let v_bytes = V::as_bytes(&v_guard.value()).as_ref().to_vec();
                (OwnedKv::new(k_bytes), OwnedKv::new(v_bytes))
            })
        })
    }
}

impl<K: Key + 'static, V: Value + 'static> crate::storage_traits::WriteTable<K, V>
    for Table<'_, K, V>
{
    type RangeIter<'a>
        = LegacyRangeIter<'a, K, V>
    where
        Self: 'a;

    fn st_get(&self, key: &K::SelfType<'_>) -> crate::Result<Option<OwnedKv<V>>> {
        ReadableTable::get(self, key).map(|opt| {
            opt.map(|guard| {
                let bytes = V::as_bytes(&guard.value()).as_ref().to_vec();
                OwnedKv::new(bytes)
            })
        })
    }

    fn st_insert(
        &mut self,
        key: &K::SelfType<'_>,
        value: &V::SelfType<'_>,
    ) -> crate::Result<Option<OwnedKv<V>>> {
        self.insert(key, value).map(|opt| {
            opt.map(|guard| {
                let bytes = V::as_bytes(&guard.value()).as_ref().to_vec();
                OwnedKv::new(bytes)
            })
        })
    }

    fn st_remove(&mut self, key: &K::SelfType<'_>) -> crate::Result<Option<OwnedKv<V>>> {
        self.remove(key).map(|opt| {
            opt.map(|guard| {
                let bytes = V::as_bytes(&guard.value()).as_ref().to_vec();
                OwnedKv::new(bytes)
            })
        })
    }

    fn st_range<'a>(
        &'a self,
        start: Option<&K::SelfType<'_>>,
        end: Option<&K::SelfType<'_>>,
        start_inclusive: bool,
        end_inclusive: bool,
    ) -> crate::Result<Self::RangeIter<'a>> {
        let inner =
            legacy_build_range::<K, V, Self>(self, start, end, start_inclusive, end_inclusive)?;
        Ok(LegacyRangeIter { inner })
    }

    fn st_drain_all(&mut self) -> crate::Result<u64> {
        self.drain_all()
    }
}

// ---------------------------------------------------------------------------
// storage_traits::ReadTable for Table (writable tables are also readable)
// ---------------------------------------------------------------------------

impl<K: Key + 'static, V: Value + 'static> crate::storage_traits::ReadTable<K, V>
    for Table<'_, K, V>
{
    type RangeIter<'a>
        = LegacyRangeIter<'a, K, V>
    where
        Self: 'a;

    fn st_get(&self, key: &K::SelfType<'_>) -> crate::Result<Option<OwnedKv<V>>> {
        ReadableTable::get(self, key).map(|opt| {
            opt.map(|guard| {
                let bytes = V::as_bytes(&guard.value()).as_ref().to_vec();
                OwnedKv::new(bytes)
            })
        })
    }

    fn st_range<'a>(
        &'a self,
        start: Option<&K::SelfType<'_>>,
        end: Option<&K::SelfType<'_>>,
        start_inclusive: bool,
        end_inclusive: bool,
    ) -> crate::Result<Self::RangeIter<'a>> {
        let inner =
            legacy_build_range::<K, V, Self>(self, start, end, start_inclusive, end_inclusive)?;
        Ok(LegacyRangeIter { inner })
    }
}

// ---------------------------------------------------------------------------
// storage_traits::ReadTable for ReadOnlyTable
// ---------------------------------------------------------------------------

/// Iterator adapter for read-only table range scans.
pub struct LegacyReadOnlyRangeIter<'a, K: Key + 'static, V: Value + 'static> {
    inner: Range<'a, K, V>,
}

impl<K: Key + 'static, V: Value + 'static> Iterator for LegacyReadOnlyRangeIter<'_, K, V> {
    type Item = crate::Result<(OwnedKv<K>, OwnedKv<V>)>;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(|result| {
            result.map(|(k_guard, v_guard)| {
                let k_bytes = K::as_bytes(&k_guard.value()).as_ref().to_vec();
                let v_bytes = V::as_bytes(&v_guard.value()).as_ref().to_vec();
                (OwnedKv::new(k_bytes), OwnedKv::new(v_bytes))
            })
        })
    }
}

impl<K: Key + 'static, V: Value + 'static> crate::storage_traits::ReadTable<K, V>
    for ReadOnlyTable<K, V>
{
    type RangeIter<'a>
        = LegacyReadOnlyRangeIter<'a, K, V>
    where
        Self: 'a;

    fn st_get(&self, key: &K::SelfType<'_>) -> crate::Result<Option<OwnedKv<V>>> {
        self.get(key).map(|opt| {
            opt.map(|guard| {
                let bytes = V::as_bytes(&guard.value()).as_ref().to_vec();
                OwnedKv::new(bytes)
            })
        })
    }

    fn st_range<'a>(
        &'a self,
        start: Option<&K::SelfType<'_>>,
        end: Option<&K::SelfType<'_>>,
        start_inclusive: bool,
        end_inclusive: bool,
    ) -> crate::Result<Self::RangeIter<'a>> {
        let inner =
            legacy_build_range_ro::<K, V>(self, start, end, start_inclusive, end_inclusive)?;
        Ok(LegacyReadOnlyRangeIter { inner })
    }
}

// ---------------------------------------------------------------------------
// Range construction helpers (shared logic for building Range from bounds)
// ---------------------------------------------------------------------------

/// Build a `Range` from optional start/end keys for types implementing `ReadableTable`.
fn legacy_build_range<'a, K: Key + 'static, V: Value + 'static, T: ReadableTable<K, V>>(
    table: &'a T,
    start: Option<&K::SelfType<'_>>,
    end: Option<&K::SelfType<'_>>,
    start_inclusive: bool,
    end_inclusive: bool,
) -> crate::Result<Range<'a, K, V>> {
    match (start, end) {
        (None, None) => table.range::<K::SelfType<'_>>(..),
        (Some(s), None) => {
            let s_bytes = K::as_bytes(s).as_ref().to_vec();
            let s_val = K::from_bytes(&s_bytes);
            if start_inclusive {
                table.range::<K::SelfType<'_>>(s_val..)
            } else {
                // Exclusive start bound: start from inclusive, caller filters
                table.range::<K::SelfType<'_>>(s_val..)
            }
        }
        (None, Some(e)) => {
            let e_bytes = K::as_bytes(e).as_ref().to_vec();
            let e_val = K::from_bytes(&e_bytes);
            if end_inclusive {
                table.range::<K::SelfType<'_>>(..=e_val)
            } else {
                table.range::<K::SelfType<'_>>(..e_val)
            }
        }
        (Some(s), Some(e)) => {
            let s_bytes = K::as_bytes(s).as_ref().to_vec();
            let e_bytes = K::as_bytes(e).as_ref().to_vec();
            let s_val = K::from_bytes(&s_bytes);
            let e_val = K::from_bytes(&e_bytes);
            if start_inclusive && end_inclusive {
                table.range::<K::SelfType<'_>>(s_val..=e_val)
            } else if start_inclusive {
                table.range::<K::SelfType<'_>>(s_val..e_val)
            } else if end_inclusive {
                table.range::<K::SelfType<'_>>(s_val..=e_val)
            } else {
                table.range::<K::SelfType<'_>>(s_val..e_val)
            }
        }
    }
}

/// Build a `Range` from optional start/end keys for `ReadOnlyTable`.
///
/// `ReadOnlyTable` has its own `range()` method (not via `ReadableTable` trait)
/// with a slightly different signature (returns `Range<'static, K, V>`).
fn legacy_build_range_ro<'a, K: Key + 'static, V: Value + 'static>(
    table: &'a ReadOnlyTable<K, V>,
    start: Option<&K::SelfType<'_>>,
    end: Option<&K::SelfType<'_>>,
    start_inclusive: bool,
    end_inclusive: bool,
) -> crate::Result<Range<'a, K, V>> {
    match (start, end) {
        (None, None) => table.range::<K::SelfType<'_>>(..),
        (Some(s), None) => {
            let s_bytes = K::as_bytes(s).as_ref().to_vec();
            let s_val = K::from_bytes(&s_bytes);
            // Note: RangeFrom is always inclusive in Rust's std. Exclusive-start
            // semantics are handled by the caller via iterator-level filtering.
            table.range::<K::SelfType<'_>>(s_val..)
        }
        (None, Some(e)) => {
            let e_bytes = K::as_bytes(e).as_ref().to_vec();
            let e_val = K::from_bytes(&e_bytes);
            if end_inclusive {
                table.range::<K::SelfType<'_>>(..=e_val)
            } else {
                table.range::<K::SelfType<'_>>(..e_val)
            }
        }
        (Some(s), Some(e)) => {
            let s_bytes = K::as_bytes(s).as_ref().to_vec();
            let e_bytes = K::as_bytes(e).as_ref().to_vec();
            let s_val = K::from_bytes(&s_bytes);
            let e_val = K::from_bytes(&e_bytes);
            if start_inclusive && end_inclusive {
                table.range::<K::SelfType<'_>>(s_val..=e_val)
            } else if start_inclusive {
                table.range::<K::SelfType<'_>>(s_val..e_val)
            } else if end_inclusive {
                table.range::<K::SelfType<'_>>(s_val..=e_val)
            } else {
                table.range::<K::SelfType<'_>>(s_val..e_val)
            }
        }
    }
}