//! Append-only B-Tree implementation

//!

//! The file format is loosely inspired by

//! [Couchstore](https://github.com/couchbase/couchstore). Nebari is not

//! compatible with Couchstore in any way.

//!

//! ## Numbers and Alignment

//!

//! - All numbers are encoded in big-endian format/network byte order.

//! - All values are tightly packed. There is no padding or alignment that isn't

//!   explicitly included.

//!

//! ## File Organization

//!

//! There is no way to read this file format starting at byte 0 and iterating

//! forward. The contents of any given byte offset are unknown until the file's

//! current root header has been found.

//!

//! When writing data to the file, it will be appended to the end of the file.

//! When a tree is committed, all of the changed nodes will be appended to the

//! end of the file, except for the Root.

//!

//! Before writing the Root, the file is padded to a multiple of

//! [`PAGE_SIZE`]. A 3-byte magic code is written, followed by a byte for the

//! [`PageHeader`].

//!

//! The Root is then serialized and written as a chunk.

//!

//! To locate the most recent header, take the file's length and find the

//! largest multiple of [`PAGE_SIZE`]. Check the first three bytes at that

//! offset for the magic code. If found, attempt to read a chunk. If successful,

//! attempt to deserialize the Root.

//!

//! If any step fails, loop back through the file at each [`PAGE_SIZE`] offset

//! until a valid header is found.

//!

//! ## Chunks

//!

//! Each time a value, B-Tree node, or header is written, it is written as a

//! chunk. If a [`Vault`](crate::Vault) is in-use, each chunk will be

//! pre-processed by the vault before a `CRC-32-BZIP2` checksum is calculated. A

//! chunk is limited to 4 gigabytes of data (2^32).

//!

//! The chunk is written as:

//!

//! - `u32` - Data length, excluding the header.

//! - `u32` - CRC

//! - `[u8]` - Contents

use std::{

    borrow::Cow,

    cell::RefCell,

    collections::HashMap,

    fmt::{Debug, Display},

    hash::BuildHasher,

    io::SeekFrom,

    marker::PhantomData,

    ops::{Bound, Deref, DerefMut, Range, RangeBounds},

    sync::Arc,

};

use byteorder::{BigEndian, ByteOrder, ReadBytesExt, WriteBytesExt};

use crc::{Crc, CRC_32_BZIP2};

use parking_lot::MutexGuard;

use crate::{

    chunk_cache::CacheEntry,

    error::Error,

    io::{

        File, FileManager, FileOp, IntoPathId, ManagedFile, ManagedFileOpener, OpenableFile,

        OperableFile, PathId,

    },

    roots::AbortError,

    transaction::{ManagedTransaction, TransactionManager},

    tree::btree::{BTreeNode, Indexer, KeyOperation, Reducer, ScanArgs},

    vault::AnyVault,

    ArcBytes, ChunkCache, CompareAndSwapError, Context, ErrorKind,

};

/// B+Tree types

pub mod btree;

mod by_id;

mod by_sequence;

mod interior;

mod key_entry;

mod modify;

pub(crate) mod root;

mod serialization;

pub(crate) mod state;

mod unversioned;

mod versioned;

pub(crate) const DEFAULT_MAX_ORDER: usize = 1000;

pub use self::{

    by_id::{ByIdIndexer, ByIdStats, UnversionedByIdIndex, VersionedByIdIndex},

    by_sequence::{BySequenceIndex, BySequenceStats, SequenceId},

    interior::{Interior, Pointer},

    key_entry::{KeyEntry, PositionIndex},

    modify::{CompareSwap, CompareSwapFn, Modification, Operation, PersistenceMode},

    root::{AnyTreeRoot, Root, TreeRoot},

    serialization::BinarySerialization,

    state::{ActiveState, State},

    unversioned::{Unversioned, UnversionedTreeRoot},

    versioned::{KeySequence, SequenceEntry, SequenceIndex, Versioned, VersionedTreeRoot},

};

/// The number of bytes in each page on-disk.

// The memory used by PagedWriter is PAGE_SIZE * PAGED_WRITER_BATCH_COUNT. E.g,

// 4096 * 4 = 16kb

pub const PAGE_SIZE: usize = 256;

const CRC32: Crc<u32> = Crc::<u32>::new(&CRC_32_BZIP2);

/// The header byte for a tree file's page.

pub enum PageHeader {

    /// A [`VersionedTreeRoot`] header.

    VersionedHeader = 2,

    /// An [`UnversionedTreeRoot`] header.

    UnversionedHeader = 3,

}

impl TryFrom<u8> for PageHeader {

    type Error = ErrorKind;

        }

}

/// An append-only tree file.

///

/// ## Generics

/// - `File`: An [`ManagedFile`] implementor.

pub struct TreeFile<Root: root::Root, File: ManagedFile> {

    /// The file handle the tree is stored within.

    pub file: <File::Manager as FileManager>::FileHandle,

    /// The state of the file.

    pub state: State<Root>,

    /// The vault used to encrypt/decrypt chunks.

    pub vault: Option<Arc<dyn AnyVault>>,

    /// The cache used to cache chunks from the file.

    pub cache: Option<ChunkCache>,

    scratch: Vec<u8>,

}

impl<Root: root::Root, File: ManagedFile> Deref for TreeFile<Root, File> {

    type Target = <File::Manager as FileManager>::FileHandle;

}

impl<Root: root::Root, File: ManagedFile> DerefMut for TreeFile<Root, File> {

}

impl<Root: root::Root, File: ManagedFile> TreeFile<Root, File> {

    /// Returns a tree as contained in `file`.

    ///

    /// `state` should already be initialized using [`Self::initialize_state`] if the file exists.

    /// Opens a tree file with read-only permissions.

    /// Opens a tree file with the ability to read and write.

    /// Attempts to load the last saved state of this tree into `state`.

        // Scan back block by block until we find a header page.

        loop {

            // Read the page header

            #[allow(clippy::match_on_vec_items)]

            match (

                    };

                        {

                            // The transaction wasn't written successfully, so

                            // we cannot trust the data present.

                                // No data was ever fully written.

                }

                (_, Ok(_) | Err(_)) => {

                }

            }

        }

    /// Sets a key/value pair. Replaces any previous value if set. If you wish

    /// to retrieve the previously stored value, use

    /// [`replace()`](Self::replace) instead.

    ///

    /// Returns the new/updated index for this key.

    /// Executes a modification. Returns a list of modified keys and their

    /// updated indexes, if the keys are still present.

    /// Compares the value of `key` against `old`. If the values match, key will

    /// be set to the new value if `new` is `Some` or removed if `new` is

    /// `None`.

                                                                     _index,

                                                                     value: Option<

                Root::Value,

                    }

                } else {

                }

    /// Removes `key` and returns the existing value and index, if present.

                    } else {

                    };

    /// Sets `key` to `value`. Returns a tuple containing two elements:

    ///

    /// - The previously stored value, if a value was already present.

    /// - The new/updated index for this key.

    #[allow(clippy::missing_panics_doc)]

    /// Gets the value stored for `key`.

    pub fn get(&mut self, key: &[u8], in_transaction: bool) -> Result<Option<Root::Value>, Error> {

        let mut buffer = None;

        self.file.execute(TreeGetter {

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            keys: KeyRange::new(std::iter::once(key)),

        })?;

        Ok(buffer)

    }

    /// Gets the index stored for `key`.

    pub fn get_index(

        &mut self,

        key: &[u8],

        in_transaction: bool,

    ) -> Result<Option<Root::Index>, Error> {

        let mut found_index = None;

        self.file.execute(TreeGetter {

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            keys: KeyRange::new(std::iter::once(key)),

        })?;

        Ok(found_index)

    }

    /// Gets the value and index stored for `key`.

    pub fn get_with_index(

        &mut self,

        key: &[u8],

        in_transaction: bool,

    ) -> Result<Option<TreeValueIndex<Root>>, Error> {

        let mut buffer = None;

        let mut found_index = None;

        self.file.execute(TreeGetter {

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            keys: KeyRange::new(std::iter::once(key)),

        })?;

        if let (Some(value), Some(index)) = (buffer, found_index) {

            Ok(Some(ValueIndex { value, index }))

        } else {

            Ok(None)

        }

    }

    /// Gets the values stored in `keys`. Does not error if a key is missing.

    /// Returns key/value pairs in an unspecified order. Keys are required to be

    /// pre-sorted.

    pub fn get_multiple<'keys, KeysIntoIter, KeysIter>(

        &mut self,

        keys: KeysIntoIter,

        in_transaction: bool,

    ) -> Result<Vec<(ArcBytes<'static>, Root::Value)>, Error>

    where

        KeysIntoIter: IntoIterator<Item = &'keys [u8], IntoIter = KeysIter>,

        KeysIter: Iterator<Item = &'keys [u8]> + ExactSizeIterator,

    {

        let keys = keys.into_iter();

        let mut buffers = Vec::with_capacity(keys.len());

        self.file.execute(TreeGetter {

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            keys: KeyRange::new(keys),

        })?;

        Ok(buffers)

    }

    /// Gets the indexes stored in `keys`. Does not error if a key is missing.

    /// Returns key/value pairs in an unspecified order. Keys are required to be

    /// pre-sorted.

    pub fn get_multiple_indexes<'keys, KeysIntoIter, KeysIter>(

        &mut self,

        keys: KeysIntoIter,

        in_transaction: bool,

    ) -> Result<Vec<(ArcBytes<'static>, Root::Index)>, Error>

    where

        KeysIntoIter: IntoIterator<Item = &'keys [u8], IntoIter = KeysIter>,

        KeysIter: Iterator<Item = &'keys [u8]> + ExactSizeIterator,

    {

        let keys = keys.into_iter();

        let mut buffers = Vec::with_capacity(keys.len());

        self.file.execute(TreeGetter {

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            keys: KeyRange::new(keys),

        })?;

        Ok(buffers)

    }

    /// Gets the values and indexes stored in `keys`. Does not error if a key is

    /// missing. Returns key/value pairs in an unspecified order. Keys are

    /// required to be pre-sorted.

    pub fn get_multiple_with_indexes<'keys, KeysIntoIter, KeysIter>(

        &mut self,

        keys: KeysIntoIter,

        in_transaction: bool,

    ) -> Result<Vec<TreeEntry<Root>>, Error>

    where

        KeysIntoIter: IntoIterator<Item = &'keys [u8], IntoIter = KeysIter>,

        KeysIter: Iterator<Item = &'keys [u8]> + ExactSizeIterator,

    {

        let keys = keys.into_iter();

        let mut buffers = Vec::with_capacity(keys.len());

        self.file.execute(TreeGetter {

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            keys: KeyRange::new(keys),

        })?;

        Ok(buffers)

    }

    /// Retrieves all keys and values for keys that are contained by `range`.

    pub fn get_range<'keys, KeyRangeBounds>(

        &mut self,

        range: &'keys KeyRangeBounds,

        in_transaction: bool,

    ) -> Result<Vec<(ArcBytes<'static>, Root::Value)>, Error>

    where

        KeyRangeBounds: RangeBounds<&'keys [u8]> + Debug + ?Sized,

    {

        let mut results = Vec::new();

        self.scan(

            range,

            true,

            in_transaction,

        )?;

        Ok(results)

    }

    /// Retrieves all keys and indexes for keys that are contained by `range`.

    pub fn get_range_indexes<'keys, KeyRangeBounds>(

        &mut self,

        range: &'keys KeyRangeBounds,

        in_transaction: bool,

    ) -> Result<Vec<(ArcBytes<'static>, Root::Index)>, Error>

    where

        KeyRangeBounds: RangeBounds<&'keys [u8]> + Debug + ?Sized,

    {

        let mut results = Vec::new();

        self.scan(

            range,

            true,

            in_transaction,

        )?;

        Ok(results)

    }

    /// Retrieves all keys and values and indexes for keys that are contained by

    /// `range`.

    pub fn get_range_with_indexes<'keys, KeyRangeBounds>(

        &mut self,

        range: &'keys KeyRangeBounds,

        in_transaction: bool,

    ) -> Result<Vec<TreeEntry<Root>>, Error>

    where

        KeyRangeBounds: RangeBounds<&'keys [u8]> + Debug + ?Sized,

    {

        let mut results = Vec::new();

        self.scan(

            range,

            true,

            in_transaction,

        )?;

        Ok(results)

    }

    /// Scans the tree across all nodes that might contain nodes within `range`.

    ///

    /// If `forwards` is true, the tree is scanned in ascending order.

    /// Otherwise, the tree is scanned in descending order.

    ///

    /// `node_evaluator` is invoked for each [`Interior`] node to determine if

    /// the node should be traversed. The parameters to the callback are:

    ///

    /// - `&ArcBytes<'static>`: The maximum key stored within the all children

    ///   nodes.

    /// - `&Root::ReducedIndex`: The reduced index value stored within the node.

    /// - `usize`: The depth of the node. The root nodes are depth 0.

    ///

    /// The result of the callback is a [`ScanEvaluation`]. To read children

    /// nodes, return [`ScanEvaluation::ReadData`].

    ///

    /// `key_evaluator` is invoked for each key encountered that is contained

    /// within `range`. For all [`ScanEvaluation::ReadData`] results returned,

    /// `callback` will be invoked with the key and values. `callback` may not

    /// be invoked in the same order as the keys are scanned.

    #[cfg_attr(

        feature = "tracing",

    )]

    pub fn scan<'keys, CallerError, KeyRangeBounds, NodeEvaluator, KeyEvaluator, DataCallback>(

        &mut self,

        range: &'keys KeyRangeBounds,

        forwards: bool,

        in_transaction: bool,

        node_evaluator: NodeEvaluator,

        key_evaluator: KeyEvaluator,

        key_reader: DataCallback,

    ) -> Result<(), AbortError<CallerError>>

    where

        KeyRangeBounds: RangeBounds<&'keys [u8]> + Debug + ?Sized,

        NodeEvaluator: FnMut(&ArcBytes<'static>, &Root::ReducedIndex, usize) -> ScanEvaluation,

        KeyEvaluator: FnMut(&ArcBytes<'static>, &Root::Index) -> ScanEvaluation,

        DataCallback: FnMut(

            ArcBytes<'static>,

            &Root::Index,

            Root::Value,

        ) -> Result<(), AbortError<CallerError>>,

        CallerError: Display + Debug,

    {

        self.file.execute(TreeScanner {

            forwards,

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            range,

            node_evaluator,

            key_reader,

            key_evaluator,

            _phantom: PhantomData,

        })?;

        Ok(())

    }

    /// Returns the reduced index over the provided range. This is an

    /// aggregation function that builds atop the `scan()` operation which calls

    /// [`Reducer::reduce()`] and [`Reducer::rereduce()`] on all matching

    /// indexes stored within the nodes of this tree, producing a single

    /// aggregated [`Root::ReducedIndex`] value.

    ///

    /// If no keys match, the returned result is what [`Reducer::rereduce()`]

    /// returns when an empty slice is provided.

    pub fn reduce<'keys, KeyRangeBounds>(

        &mut self,

        range: &'keys KeyRangeBounds,

        in_transaction: bool,

    ) -> Result<Option<Root::ReducedIndex>, Error>

    where

        KeyRangeBounds: RangeBounds<&'keys [u8]> + Debug + ?Sized,

        Root::Index: Clone,

    {

        let reducer = {

            let state = self.state.lock();

            state.root.reducer().clone()

        };

        let reduce_state = RefCell::new(ReduceState::new(reducer));

        self.file.execute(TreeScanner {

            forwards: true,

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            range,

                };

                };

                };

                    // We are beyond the end, we can stop scanning.

                    // The node is fully included. Copy the index to the

                    // stack and skip all the children.

                } else {

                    // This node is partially contained.

                }

            _phantom: PhantomData,

        })?;

        let reduce_state = reduce_state.into_inner();

        Ok(reduce_state.finish())

    }

    /// Returns the first key of the tree.

    /// Returns the first key and value of the tree.

                } else {

                }

    /// Returns the last key of the tree.

    /// Returns the last key and value of the tree.

                } else {

                }

    /// Commits the tree. This is only needed if writes were done with a

    /// transaction id. This will fully flush the tree and publish the

    /// transactional state to be available to readers.

    /// Rewrites the database, removing all unused data in the process. For a

    /// `VersionedTreeRoot`, this will remove old version information.

    ///

    /// This process is done atomically by creating a new file containing the

    /// active data. Once the new file has all the current file's data, the file

    /// contents are swapped using atomic file operations.

}

struct ReduceState<R, I, RI> {

    depths: Vec<DepthState<I, RI>>,

    lowest_key: ArcBytes<'static>,

    reducer: R,

}

impl<R, I, RI> ReduceState<R, I, RI>

where

    I: Clone,

    RI: Clone,

    R: Reducer<I, RI>,

{

        }

}

struct DepthState<I, RI> {

    reduced: Vec<RI>,

    indexes: Vec<I>,

}

impl<I, RI> Default for DepthState<I, RI> {

}

impl<I, RI> DepthState<I, RI>

where

    I: Clone,

    RI: Clone,

{

}

impl<File: ManagedFile, Index> TreeFile<VersionedTreeRoot<Index>, File>

where

    Index: EmbeddedIndex<ArcBytes<'static>> + Clone + Debug + 'static,

{

    /// Scans the tree for keys that are contained within `range`. If `forwards`

    /// is true, scanning starts at the lowest sort-order key and scans forward.

    /// Otherwise, scanning starts at the highest sort-order key and scans

    /// backwards. `key_evaluator` is invoked for each key as it is encountered.

    /// For all [`ScanEvaluation::ReadData`] results returned, `callback` will be

    /// invoked with the key and values. The callback may not be invoked in the

    /// same order as the keys are scanned.

    #[cfg_attr(

        feature = "tracing",

    )]

    pub fn scan_sequences<CallerError, Range, KeyEvaluator, DataCallback>(

        &mut self,

        range: Range,

        forwards: bool,

        in_transaction: bool,

        mut key_evaluator: KeyEvaluator,

        data_callback: DataCallback,

    ) -> Result<(), AbortError<CallerError>>

    where

        Range: RangeBounds<SequenceId> + Debug + 'static,

        KeyEvaluator: FnMut(KeySequence<Index>) -> ScanEvaluation,

        DataCallback:

            FnMut(KeySequence<Index>, ArcBytes<'static>) -> Result<(), AbortError<CallerError>>,

        CallerError: Display + Debug,

    {

        self.file.execute(TreeSequenceScanner {

            forwards,

            from_transaction: in_transaction,

            state: &self.state,

            vault: self.vault.as_deref(),

            cache: self.cache.as_ref(),

            range: &U64Range::new(range).borrow_as_bytes(),

            data_callback,

        })?;

        Ok(())

    }

    /// Retrieves the keys and values associated with one or more `sequences`.

    /// The value retrieved is the value of the key at the given [`SequenceId`].

    /// If a sequence is not found, it will not appear in the result map. If

    /// the value was removed, None is returned for the value.

    /// Retrieves the keys and indexes associated with one or more `sequences`.

    /// The value retrieved is the value of the key at the given [`SequenceId`].

    /// If a sequence is not found, it will not appear in the result list.

    /// Retrieves the keys, values, and indexes associated with one or more

    /// `sequences`. The value retrieved is the value of the key at the given

    /// [`SequenceId`]. If a sequence is not found, it will not appear in the

    /// result list.

}

/// A compaction process that runs in concert with a transaction manager.

pub struct TransactableCompaction<'a, Manager: FileManager> {

    /// The name of the tree being compacted.

    pub name: &'a str,

    /// The transaction manager.

    pub manager: &'a TransactionManager<Manager>,

}

struct TreeCompactor<'a, Root: root::Root, Manager: FileManager> {

    manager: &'a Manager,

    state: &'a State<Root>,

    vault: Option<&'a dyn AnyVault>,

    transactions: Option<TransactableCompaction<'a, Manager>>,

    scratch: &'a mut Vec<u8>,

}

impl<'a, Root, Manager>

    FileOp<Result<(Manager::File, TreeCompactionFinisher<'a, Root, Manager>), Error>>

    for TreeCompactor<'a, Root, Manager>

where

    Root: root::Root,

    Manager: FileManager,

{

        // Use the read state to list all the currently live chunks

        // Now, do the same with the write state, which should be very fast,

        // since only nodes that have changed will need to be visited.

        // Close any existing handles to the file. This ensures that once we

        // save the tree, new requests to the file manager will point to the new

        // file.

}

struct TreeCompactionFinisher<'a, Root: root::Root, Manager: FileManager> {

    state: &'a State<Root>,

    write_state: MutexGuard<'a, ActiveState<Root>>,

    _transaction: Option<ManagedTransaction<Manager>>,

}

impl<'a, Root: root::Root, Manager: FileManager> TreeCompactionFinisher<'a, Root, Manager> {

}

struct TreeWriter<'a, Root: root::Root> {

    state: &'a State<Root>,

    vault: Option<&'a dyn AnyVault>,

    cache: Option<&'a ChunkCache>,

    scratch: &'a mut Vec<u8>,

}

impl<'a, Root> FileOp<Result<(), Error>> for TreeWriter<'a, Root>

where

    Root: root::Root,

{