disklru 0.3.3

// Copyright (C) 2024 Christian Mauduit <ufoot@ufoot.org>

use crate::error::{Error, Result};
use crate::iter::*;
use crate::serial::*;
use crate::trans;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use sled;
use sled::{Config, Db};
use std::collections::HashMap;
use std::fmt;
use std::hash::Hash;
use std::marker::PhantomData;
use std::path::Path;

const MAX_TO_DISPLAY: usize = 20;

/// LRU store, backed by by [sled](https://crates.io/crates/sled).
///
/// The typical use case is: you want a persistent cache, with an
/// interface that has similarities with a basic HashMap.
/// And, you do not want it to grow too much, and prefer to drop
/// on the floor data which has not been accessed for a long time.
///
/// LRU is normally used for in-memory caches, but here it's implemented
/// on something which is persistent. Implementation is inspired from
/// work on the in-memory cache [HashLRU](https://crates.io/crates/hashlru).
///
/// In most cases this implementation tries to be as close as possible
/// to the [standard collections HashMap](https://doc.rust-lang.org/std/collections/struct.HashMap.html)
/// however there are a few differences:
///
/// - keys and values must be (de)serialiable
/// - keys must implement Eq
/// - no iterator on mutables
/// - as it relies on external data, any operation can fail
/// - inserts take references, gets return owned values
///
/// This latest aspect is really the fundamental difference: as here all
/// the data is serialized and stored within the store, returning a reference
/// does not really make any sense. Conversely, when putting values in the
/// store, you don't need to transfer ownership: the store is going to
/// make a serializable copy anyway.
///
/// # Examples
///
/// ```
/// use disklru::Store;
///
/// let mut store = Store::open_temporary(4).unwrap();
/// store.insert(&1, &10).unwrap();
/// store.insert(&2, &20).unwrap();
/// store.insert(&3, &30).unwrap();
/// store.insert(&4, &40).unwrap();
/// store.insert(&5, &50).unwrap();
/// // key1 has been dropped, size is limited to 4
/// assert_eq!(Some(2), store.lru().unwrap());
/// assert_eq!(Some(20), store.get(&2).unwrap());
/// // getting key2 has made key3 the least recently used item
/// assert_eq!(Some(3), store.lru().unwrap());
/// assert_eq!(Some(40), store.get(&4).unwrap());
/// // getting key4 makes it the most recently used item
/// assert_eq!("[3: 30, 5: 50, 2: 20, 4: 40]", format!("{}", store));
/// store.flush().unwrap(); // commit
/// ```
#[derive(Debug)]
pub struct Store<K, V>
where
    K: Serialize + DeserializeOwned + Eq,
    V: Serialize + DeserializeOwned,
{
    pub(crate) capacity: usize,
    pub(crate) len: usize,
    pub(crate) db: Db,
    pub(crate) phantom_data: PhantomData<(K, V)>,
}

/// Complete dump of a LRU store, easily (de)serializable.
///
/// It reflects the complete current state of the cache, including
/// key/value pairs but also capacity and order. It can be costly
/// to generate as it is O(n) and also it can take a lot of space,
/// but at least you get complete control on cache content.
///
/// Can be used to import/export the store and serialize/unserialize it
/// outside [sled](https://crates.io/crates/sled). There are other ways
/// to import/export keys and values but this one has the advantage
/// to preserve all parameters, including capacity.
///
/// # Examples
///
/// ```
/// use disklru::{Store, Dump};
///
/// let mut store: Store<String, usize> = Store::open_temporary(10).unwrap();;
/// store.insert(&String::from("x"), &1);
/// store.insert(&String::from("y"), &10);
/// store.insert(&String::from("z"), &100);
///
/// let dump: Dump<String, usize> = store.dump().unwrap();
/// let mut restored: Store<String, usize> = Store::open_temporary(0).unwrap();
/// assert_eq!(3, restored.restore(&dump).unwrap());
/// assert_eq!(10, restored.capacity());
/// assert_eq!("[x: 1, y: 10, z: 100]", format!("{}", &restored));
/// ```
#[derive(Debug, Serialize, Deserialize)]
pub struct Dump<K, V> {
    pub capacity: usize,
    pub data: Vec<(K, V)>,
}

impl<K, V> fmt::Display for Store<K, V>
where
    K: Serialize + DeserializeOwned + Eq + fmt::Display,
    V: Serialize + DeserializeOwned + fmt::Display,
{
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        let items = if self.len() <= MAX_TO_DISPLAY {
            self.iter()
                .map(|x| match x {
                    Ok(kv) => format!("{}: {}", kv.0, kv.1),
                    Err(e) => format!("ERROR: {}", e),
                })
                .collect::<Vec<String>>()
        } else {
            let mut acc = self
                .iter()
                .take(2)
                .map(|x| match x {
                    Ok(kv) => format!("{}: {}", kv.0, kv.1),
                    Err(e) => format!("ERROR: {}", e),
                })
                .collect::<Vec<String>>();
            acc.push("...".to_string());
            acc.push(match self.mru() {
                Ok(mk) => match mk {
                    Some(mru_k) => match self.peek_mru() {
                        Ok(mv) => match mv {
                            Some(mru_v) => format!("{}: {}", mru_k, mru_v),
                            None => format!("ERROR: {}", Error::invalid_data("no mru value")),
                        },
                        Err(e) => format!("ERROR: {}", e),
                    },
                    None => format!("ERROR: {}", Error::report_bug("no mru key")),
                },
                Err(e) => format!("ERROR: {}", e),
            });
            acc
        };
        write!(f, "[{}]", items.join(", "))
    }
}

impl<K, V> PartialEq<Store<K, V>> for Store<K, V>
where
    K: Serialize + DeserializeOwned + Eq,
    V: Serialize + DeserializeOwned + Eq,
{
    fn eq(&self, other: &Self) -> bool {
        match self.try_eq(other) {
            Ok(eq) => eq,
            Err(e) => panic!("unable to compare stores: {}", e),
        }
    }
}

impl<K, V> Eq for Store<K, V>
where
    K: Serialize + DeserializeOwned + Eq,
    V: Serialize + DeserializeOwned + Eq,
{
}

impl<K, V> Store<K, V>
where
    K: Serialize + DeserializeOwned + Eq,
    V: Serialize + DeserializeOwned + Eq,
{
    fn try_eq(&self, other: &Self) -> Result<bool> {
        if self.capacity != other.capacity {
            return Ok(false);
        }
        if self.len != other.len {
            return Ok(false);
        }

        let self_head = self.find_head()?;
        let other_head = other.find_head()?;
        if self_head != other_head {
            return Ok(false);
        }

        let iter_self = self.iter();
        let mut iter_other = other.iter();
        for self_next in iter_self {
            let self_item = self_next?;
            let other_next = iter_other.next();
            match other_next {
                Some(other_next) => {
                    let other_item = other_next?;
                    if self_item != other_item {
                        return Ok(false);
                    }
                }
                None => return Err(Error::report_bug("found self but not other")),
            }
        }

        Ok(true)
    }
}

impl<K, V> Store<K, V>
where
    K: Serialize + DeserializeOwned + Eq,
    V: Serialize + DeserializeOwned,
{
    pub fn open_with_config(config: Config, capacity: usize) -> Result<Self> {
        let db = match config.open() {
            Ok(db) => db,
            Err(e) => return Err(Error::from(e)),
        };
        let mut store = Store {
            capacity,
            len: 0,
            db,
            phantom_data: PhantomData,
        };
        // We've initialized the store with some capacity passed
        // as an arg, but we still read it from db, if it already
        // existed. Also we store it back within the db, this both:
        // - initializes it if it did not exist yet
        // - ensures the store can be read/write, this way we fail early
        store.set_capacity_from_db()?;
        store.update_db_capacity()?;

        // Set let from value stored in DB if available.
        // Calculating it dynamically on big stores can *REALLY* be slow.
        store.set_len_from_db()?;
        store.update_db_len()?;

        // Set the head of the store to none, having a head, even
        // undefined, is really an invariant.
        store.init_head_if_not_exists()?;

        // A bit of cleanup, trim extra data, having unclean stuff
        // here would really mess up things. We do not error on this,
        // there could be a workable state where all things are good
        // but there is just a capacity mismatch.
        store.remove_extra(capacity)?;

        Ok(store)
    }

    fn init_head_if_not_exists(&self) -> Result<()> {
        let db = self.db.clone();
        db.transaction(|db| trans::init_head_if_not_exists::<K>(db))?;
        Ok(())
    }

    pub fn open_with_path<P: AsRef<Path>>(path: P, capacity: usize) -> Result<Self> {
        Self::open_with_config(Config::default().path(path), capacity)
    }

    pub fn open_temporary(capacity: usize) -> Result<Self> {
        Self::open_with_config(Config::default().temporary(true), capacity)
    }

    pub fn capacity(&self) -> usize {
        self.capacity
    }

    fn set_len_from_db(&mut self) -> Result<()> {
        let db = self.db.clone();
        let trans_ret = db.transaction(|db| trans::get_len(db))?;
        match trans_ret {
            Some(len) => {
                self.len = len;
            }
            None => (),
        }
        Ok(())
    }

    fn set_capacity_from_db(&mut self) -> Result<()> {
        let db = self.db.clone();
        let trans_ret = db.transaction(|db| trans::get_capacity(db))?;
        match trans_ret {
            Some(capacity) => {
                self.capacity = capacity;
            }
            None => (),
        }
        Ok(())
    }

    fn update_db_capacity(&mut self) -> Result<()> {
        let db = self.db.clone();
        let capacity = self.capacity;
        db.transaction(|db| trans::set_capacity(db, capacity))?;
        Ok(())
    }

    pub fn len(&self) -> usize {
        self.len
    }

    fn update_db_len(&mut self) -> Result<()> {
        let db = self.db.clone();
        let len = self.len;
        db.transaction(|db| trans::set_len(db, len))?;
        Ok(())
    }

    pub fn resize(&mut self, capacity: usize) -> Result<usize> {
        self.capacity = capacity;
        let removed = self.remove_extra(self.capacity)?;
        self.update_db_capacity()?;
        Ok(removed)
    }

    pub fn clear(&mut self) -> Result<()> {
        self.remove_extra(0)?;
        Ok(())
    }

    pub(crate) fn find_head(&self) -> Result<Option<K>> {
        let db = self.db.clone();
        let trans_ret = db.transaction(|db| trans::find_head(db))?;
        Ok(trans_ret)
    }

    pub(crate) fn find_tail(&self) -> Result<Option<K>> {
        let db = self.db.clone();
        let trans_ret = db.transaction(|db| trans::find_tail(db))?;
        Ok(trans_ret)
    }

    pub fn flush(&self) -> Result<usize> {
        match self.db.flush() {
            Ok(s) => Ok(s),
            Err(e) => Err(Error::from(e)),
        }
    }

    pub async fn flush_async(&self) -> Result<usize> {
        match self.db.flush_async().await {
            Ok(s) => Ok(s),
            Err(e) => Err(Error::from(e)),
        }
    }

    pub fn mru(&self) -> Result<Option<K>> {
        self.find_head()
    }

    pub fn lru(&self) -> Result<Option<K>> {
        self.find_tail()
    }

    fn remove_extra(&mut self, limit: usize) -> Result<usize> {
        let mut removed: usize = 0;
        if self.len > limit {
            let db = self.db.clone();
            while self.len > limit {
                let len = self.len;
                self.len = db.transaction(|db| trans::forget_tail_must_exist::<K>(db, len))?;
                removed += 1;
            }
        }
        Ok(removed)
    }

    pub fn insert(&mut self, k: &K, v: &V) -> Result<Option<V>> {
        if self.capacity == 0 {
            return Ok(None);
        }

        let db = self.db.clone();
        let len = self.len;

        let trans_ret = db.transaction(|db| trans::insert(db, k, v, len))?;
        self.len = trans_ret.1;

        self.remove_extra(self.capacity)?;

        Ok(trans_ret.0)
    }

    pub fn push(&mut self, k: &K, v: &V) -> Result<Option<(K, V)>> {
        if self.capacity == 0 {
            return Ok(None);
        }

        self.remove_extra(self.capacity)?;

        let db = self.db.clone();
        let len = self.len;
        let capacity = self.capacity;

        let trans_ret = db.transaction(|db| trans::push(db, k, v, len, capacity))?;
        self.len = trans_ret.1;

        Ok(trans_ret.0)
    }

    pub fn remove(&mut self, k: &K) -> Result<Option<V>> {
        let db = self.db.clone();
        let len = self.len;

        let trans_ret = db.transaction(|db| trans::remove(db, k, len))?;
        self.len = trans_ret.1;

        Ok(trans_ret.0)
    }

    pub fn pop(&mut self, k: &K) -> Result<Option<(K, V)>> {
        let db = self.db.clone();
        let len = self.len;

        let trans_ret = db.transaction(|db| trans::pop(db, k, len))?;
        self.len = trans_ret.1;

        Ok(trans_ret.0)
    }

    pub fn pop_lru(&mut self) -> Result<Option<(K, V)>> {
        let db = self.db.clone();

        let ret = match self.len {
            0 => None,
            len => {
                // Using a custom transaction as popping the LRU is something
                // very common, and also it may have an optimized code path.
                let trans_ret = db.transaction(|db| trans::pop_tail_must_exist(db, len))?;
                self.len = trans_ret.1;
                Some(trans_ret.0)
            }
        };
        Ok(ret)
    }

    pub fn pop_mru(&mut self) -> Result<Option<(K, V)>> {
        let mru = self.mru()?;

        match mru {
            Some(mru) => self.pop(&mru),
            None => Ok(None),
        }
    }

    pub fn forget(&mut self, k: &K) -> Result<()> {
        let db = self.db.clone();
        let len = self.len;

        let trans_ret = db.transaction(|db| trans::forget(db, k, len))?;
        self.len = trans_ret;

        Ok(())
    }

    pub fn forget_mru(&mut self) -> Result<()> {
        let mru = self.mru()?;
        match mru {
            Some(mru) => self.forget(&mru),
            None => Ok(()),
        }
    }

    pub fn forget_lru(&mut self) -> Result<()> {
        let lru = self.lru()?;
        match lru {
            Some(lru) => self.forget(&lru),
            None => Ok(()),
        }
    }

    pub fn contains_key(&mut self, k: &K) -> Result<bool> {
        let db = self.db.clone();

        let contains = db.transaction(|db| trans::contains_key(db, k))?;

        Ok(contains)
    }

    pub fn bump(&mut self, k: &K) -> Result<()> {
        let db = self.db.clone();

        db.transaction(|db| trans::bump(db, k))?;

        Ok(())
    }

    pub fn get(&mut self, k: &K) -> Result<Option<V>> {
        let db = self.db.clone();

        let trans_ret = db.transaction(|db| trans::get_data(db, k))?;
        match trans_ret {
            Some(data) => Ok(Some(data.value)),
            None => Ok(None),
        }
    }

    pub fn get_key_value(&mut self, k: &K) -> Result<Option<(K, V)>> {
        let db = self.db.clone();

        let trans_ret = db.transaction(|db| trans::get_data(db, k))?;
        match trans_ret {
            Some(data) => Ok(Some((data.key, data.value))),
            None => Ok(None),
        }
    }

    pub fn get_lru(&mut self) -> Result<Option<V>> {
        let lru = self.lru()?;
        match lru {
            Some(lru) => {
                let kv = self.get(&lru)?;
                Ok(kv)
            }
            None => Ok(None),
        }
    }

    pub fn peek(&self, k: &K) -> Result<Option<V>> {
        let db = self.db.clone();

        let trans_ret = db.transaction(|db| trans::peek_data(db, k))?;
        match trans_ret {
            Some(data) => Ok(Some(data.value)),
            None => Ok(None),
        }
    }

    pub fn peek_mru(&self) -> Result<Option<V>> {
        let mru = self.mru()?;
        match mru {
            Some(mru) => self.peek(&mru),
            None => Ok(None),
        }
    }

    pub fn peek_lru(&self) -> Result<Option<V>> {
        let lru = self.lru()?;
        match lru {
            Some(lru) => self.peek(&lru),
            None => Ok(None),
        }
    }

    pub fn peek_key_value(&self, k: &K) -> Result<Option<(K, V)>> {
        let db = self.db.clone();

        let trans_ret = db.transaction(|db| trans::peek_data(db, k))?;
        match trans_ret {
            Some(data) => Ok(Some((data.key, data.value))),
            None => Ok(None),
        }
    }

    pub(crate) fn peek_node(&self, k: &K) -> Result<Option<Node<K, V>>> {
        let db = self.db.clone();

        let trans_ret = db.transaction(|db| trans::peek_node(db, k))?;
        match trans_ret {
            Some(node) => Ok(Some(node)),
            None => Ok(None),
        }
    }

    pub fn export(&self) -> Export<'_, K, V> {
        Export {
            pos: None,
            done: 0,
            store: &self,
        }
    }

    pub fn iter(&self) -> Iter<'_, K, V> {
        Iter {
            pos: None,
            done: 0,
            store: &self,
        }
    }

    pub fn to_vec(&self) -> Result<Vec<(K, V)>> {
        let mut iter = self.export();
        let mut ret: Vec<(K, V)> = Vec::with_capacity(self.len);
        loop {
            let next_item = iter.try_next()?;
            match next_item {
                Some(kv) => ret.push((kv.0, kv.1)),
                None => break,
            }
        }
        Ok(ret)
    }

    pub fn import<I>(&mut self, iter: I) -> Result<usize>
    where
        I: Iterator<Item = (K, V)>,
    {
        let mut imported = 0;
        for kv in iter {
            let old = self.insert(&kv.0, &kv.1)?;
            match old {
                Some(_) => {
                    return Err(Error::invalid_data(
                        "import is overwritting existing values",
                    ))
                }
                None => imported += 1,
            }
        }
        Ok(imported)
    }

    pub fn import_map_iter<'a, I>(&mut self, iter: I) -> Result<usize>
    where
        K: Serialize + DeserializeOwned + Eq + 'a,
        V: Serialize + DeserializeOwned + 'a,
        I: Iterator<Item = (&'a K, &'a V)>,
    {
        let mut imported = 0;
        for kv in iter {
            let old = self.insert(kv.0, kv.1)?;
            match old {
                Some(_) => {
                    return Err(Error::invalid_data(
                        "import is overwritting existing values",
                    ))
                }
                None => imported += 1,
            }
        }
        Ok(imported)
    }

    pub fn import_vec_iter<'a, I>(&mut self, iter: I) -> Result<usize>
    where
        K: Serialize + DeserializeOwned + Eq + 'a,
        V: Serialize + DeserializeOwned + 'a,
        I: Iterator<Item = &'a (K, V)>,
    {
        let mut imported = 0;
        for kv in iter {
            let old = self.insert(&kv.0, &kv.1)?;
            match old {
                Some(_) => {
                    return Err(Error::invalid_data(
                        "import is overwritting existing values",
                    ))
                }
                None => imported += 1,
            }
        }
        Ok(imported)
    }

    pub fn keys(&self) -> Keys<'_, K, V> {
        Keys { iter: self.iter() }
    }

    pub fn values(&self) -> Values<'_, K, V> {
        Values { iter: self.iter() }
    }

    pub fn export_keys(&self) -> ExportKeys<'_, K, V> {
        ExportKeys {
            export: self.export(),
        }
    }

    pub fn export_values(&self) -> ExportValues<'_, K, V> {
        ExportValues {
            export: self.export(),
        }
    }

    pub fn dump(&self) -> Result<Dump<K, V>> {
        let mut data: Vec<(K, V)> = Vec::with_capacity(self.len());
        for i in self.iter() {
            let i = i?;
            data.push((i.0, i.1));
        }
        Ok(Dump {
            capacity: self.capacity(),
            data,
        })
    }

    pub fn restore(&mut self, dump: &Dump<K, V>) -> Result<usize> {
        self.clear()?;
        self.resize(dump.capacity)?;
        for i in dump.data.iter() {
            self.insert(&i.0, &i.1)?;
        }
        Ok(self.len)
    }
}

impl<K, V> Store<K, V>
where
    K: Serialize + DeserializeOwned + Eq + Hash,
    V: Serialize + DeserializeOwned,
{
    pub fn to_map(&self) -> Result<HashMap<K, V>> {
        let mut iter = self.export();
        let mut ret: HashMap<K, V> = HashMap::with_capacity(self.len);
        loop {
            let next_item = iter.try_next()?;
            match next_item {
                Some(kv) => {
                    ret.insert(kv.0, kv.1);
                }
                None => break,
            }
        }
        Ok(ret)
    }
}

impl<K, V> Store<K, V>
where
    K: Serialize + DeserializeOwned + Eq + Clone + std::fmt::Debug,
    V: Serialize + DeserializeOwned,
{
    pub fn audit(&self) -> Result<()> {
        let db = self.db.clone();

        let trans_len = db.transaction(|db| trans::get_len(db))?;
        match trans_len {
            Some(db_len) => {
                if self.len != db_len {
                    return Err(Error::invalid_data(&format!(
                        "len mismatch, expected {} but db contains {}",
                        self.len, db_len
                    )));
                }
            }
            None => return Err(Error::invalid_data("no len in db")),
        }

        let trans_capacity = db.transaction(|db| trans::get_capacity(db))?;
        match trans_capacity {
            Some(db_capacity) => {
                if self.capacity != db_capacity {
                    return Err(Error::invalid_data(&format!(
                        "capacity mismatch, expected {} but db contains {}",
                        self.capacity, db_capacity
                    )));
                }
            }
            None => return Err(Error::invalid_data("no capacity in db")),
        }

        let audit_len = db.transaction(|db| trans::audit::<K, V>(db))?;
        if self.len != audit_len {
            return Err(Error::invalid_data(&format!(
                "len mismatch, expected {} but audit returned {}",
                self.len, audit_len
            )));
        }

        let mut ptr_count = 0;
        let mut size_count = 0;
        let mut link_count = 0;
        let mut data_count = 0;
        for key in self.db.iter().keys() {
            let key = key?;
            let key_details: (Option<K>, KeyType) = deserialize_key(key)?;
            match key_details.1 {
                KeyType::Ptr => ptr_count += 1,
                KeyType::Size => size_count += 1,
                KeyType::Data => data_count += 1,
                KeyType::Link => link_count += 1,
            }
        }
        if ptr_count != 1 {
            return Err(Error::invalid_data(&format!(
                "expected 1 pointer, got {}",
                ptr_count
            )));
        }
        if size_count != 2 {
            return Err(Error::invalid_data(&format!(
                "expected 2 sizes, got {}",
                size_count
            )));
        }
        if link_count != self.len {
            return Err(Error::invalid_data(&format!(
                "expected {} links, got {}",
                self.len, link_count
            )));
        }
        if data_count != self.len {
            return Err(Error::invalid_data(&format!(
                "expected {} datas, got {}",
                self.len, data_count
            )));
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sled::Mode;
    use rand::Rng;
    use std::collections::HashMap;
    use tempfile::TempDir;

    #[test]
    fn test_open_new() {
        let tmp_dir = TempDir::new().unwrap();
        let file_path = tmp_dir.path().join("test.db");
        let store: Store<String, String> = Store::open_with_path(file_path, 10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        assert_eq!(None, store.mru().unwrap());
        store.audit().unwrap();
    }

    #[test]
    fn test_simple_insert_peek_1() {
        let tmp_dir = TempDir::new().unwrap();
        let file_path = tmp_dir.path().join("test.db");
        let mut store: Store<usize, usize> = Store::open_with_path(file_path, 10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        assert_eq!(None, store.mru().unwrap());
        assert_eq!(None, store.peek(&1).unwrap());
        assert_eq!(None, store.insert(&1, &100).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(1, store.len());
        store.audit().unwrap();
        assert_eq!(Some(100), store.peek(&1).unwrap());
        assert_eq!(Some(1), store.mru().unwrap());
        assert_eq!(Some(1), store.lru().unwrap());
        store.audit().unwrap();
    }

    #[test]
    fn test_simple_insert_peek_remove_1() {
        let tmp_dir = TempDir::new().unwrap();
        let file_path = tmp_dir.path().join("test.db");
        let mut store: Store<usize, usize> = Store::open_with_path(file_path, 10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        store.audit().unwrap();
        assert_eq!(None, store.mru().unwrap());
        assert_eq!(None, store.peek(&1).unwrap());
        assert_eq!(None, store.remove(&1).unwrap());
        assert_eq!(None, store.insert(&1, &100).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(1, store.len());
        store.audit().unwrap();
        assert_eq!(Some(100), store.peek(&1).unwrap());
        assert_eq!(Some(1), store.mru().unwrap());
        assert_eq!(None, store.remove(&2).unwrap());
        assert_eq!(Some(100), store.remove(&1).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        store.audit().unwrap();
    }

    #[test]
    fn test_simple_insert_peek_remove_3() {
        let tmp_dir = TempDir::new().unwrap();
        let file_path = tmp_dir.path().join("test.db");
        let mut store: Store<usize, usize> = Store::open_with_path(file_path, 10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        store.audit().unwrap();
        assert_eq!(None, store.mru().unwrap());
        assert_eq!(None, store.peek(&1).unwrap());
        assert_eq!(None, store.remove(&1).unwrap());
        assert_eq!(None, store.insert(&1, &100).unwrap());
        store.audit().unwrap();
        assert_eq!(None, store.insert(&2, &200).unwrap());
        store.audit().unwrap();
        assert_eq!(None, store.insert(&3, &300).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(3, store.len());
        store.audit().unwrap();
        assert_eq!(Some(100), store.peek(&1).unwrap());
        assert_eq!(Some(3), store.mru().unwrap());
        assert_eq!(Some(1), store.lru().unwrap());
        assert_eq!(None, store.remove(&4).unwrap());
        assert_eq!(Some(100), store.remove(&1).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(200), store.remove(&2).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(300), store.remove(&3).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        store.audit().unwrap();
    }

    #[test]
    fn test_insert_respects_capacity() {
        let mut store: Store<usize, usize> = Store::open_temporary(10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        for i in 0..100 as usize {
            store.insert(&i, &i).unwrap();
        }
        assert_eq!(10, store.capacity());
        assert_eq!(10, store.len());
        assert_eq!(Some(99), store.mru().unwrap());
        assert_eq!(Some(90), store.lru().unwrap());
        store.audit().unwrap();
    }

    #[test]
    fn test_get_reorders_store() {
        let mut store: Store<usize, usize> = Store::open_temporary(10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        assert_eq!(None, store.insert(&1, &10).unwrap());
        assert_eq!(None, store.insert(&2, &20).unwrap());
        assert_eq!(None, store.insert(&3, &30).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(3, store.len());
        store.audit().unwrap();
        assert_eq!(Some(10), store.get(&1).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(3, store.len());
        store.audit().unwrap();
        assert_eq!(Some(1), store.mru().unwrap());
        assert_eq!(Some(2), store.lru().unwrap());
        store.audit().unwrap();
    }

    #[test]
    fn test_get_in_detail() {
        let mut store: Store<usize, usize> = Store::open_temporary(10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());

        assert_eq!(None, store.get(&1).unwrap());
        store.audit().unwrap();

        assert_eq!(None, store.insert(&1, &10).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(10), store.get(&1).unwrap());
        store.audit().unwrap();
        assert_eq!(None, store.get(&2).unwrap());
        store.audit().unwrap();

        assert_eq!(None, store.insert(&2, &20).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(10), store.get(&1).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(10), store.get(&1).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(20), store.get(&2).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(20), store.get(&2).unwrap());
        store.audit().unwrap();

        assert_eq!(None, store.insert(&3, &30).unwrap());
        store.audit().unwrap();
        let mut gets: Vec<usize> = vec![
            1, 1, 2, 2, 3, 3, 1, 2, 3, 3, 2, 1, 2, 3, 1, 2, 1, 3, 5, 3, 3, 2, 2, 1, 1, 2, 3, 2, 3,
            1, 3, 2, 5, 3, 2, 1,
        ];
        for i in 0..gets.len() {
            let get = gets[i];
            println!("len {}, step {}, getting {}", store.len(), i, get);
            if i >= 1 && i <= 3 {
                assert_eq!(Some(10 * i), store.get(&i).unwrap());
            } else {
                assert_eq!(None, store.get(&i).unwrap());
            }
        }
        store.audit().unwrap();

        assert_eq!(None, store.insert(&4, &40).unwrap());
        store.audit().unwrap();
        gets = vec![
            1, 1, 2, 2, 3, 3, 4, 4, 1, 2, 3, 4, 4, 3, 2, 1, 1, 3, 2, 4, 2, 4, 1, 3, 2, 4, 3, 1, 5,
            3, 1, 5, 4, 2,
        ];
        for i in 0..gets.len() {
            let get = gets[i];
            println!("len {}, step {}, getting {}", store.len(), i, get);
            if i >= 1 && i <= 4 {
                assert_eq!(Some(10 * i), store.get(&i).unwrap());
            } else {
                assert_eq!(None, store.get(&i).unwrap());
            }
        }
        store.audit().unwrap();

        assert_eq!(None, store.insert(&5, &50).unwrap());
        store.audit().unwrap();
        gets = vec![
            1, 2, 3, 4, 5, 1, 1, 2, 3, 4, 5, 2, 1, 2, 3, 4, 5, 3, 1, 2, 3, 4, 5, 4, 1, 2, 3, 4, 5,
            5, 1, 2, 3, 4, 5, 6, 1, 2, 3, 4, 5,
        ];
        for i in 0..gets.len() {
            let get = gets[i];
            println!("len {}, step {}, getting {}", store.len(), i, get);
            if i >= 1 && i <= 5 {
                assert_eq!(Some(10 * i), store.get(&i).unwrap());
            } else {
                assert_eq!(None, store.get(&i).unwrap());
            }
        }
        store.audit().unwrap();

        // BEGIN bug-fix on gets
        // What happens -> the *FIRST* successful get somewhere in the
        // middle of the store fails.
        store.clear().unwrap();
        store.audit().unwrap();
        for i in 0..10 {
            assert_eq!(None, store.insert(&i, &(i * 10)).unwrap());
            store.audit().unwrap();
        }
        assert_eq!(Some(50), store.peek(&5).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(50), store.get(&5).unwrap());
        store.audit().unwrap();
        // END bug-fix on gets
    }

    #[test]
    fn test_push() {
        let mut store: Store<usize, usize> = Store::open_temporary(3).unwrap();
        assert_eq!(3, store.capacity());
        assert_eq!(0, store.len());
        assert_eq!(None, store.push(&1, &10).unwrap());
        store.audit().unwrap();
        assert_eq!(None, store.push(&2, &20).unwrap());
        store.audit().unwrap();
        assert_eq!(None, store.push(&3, &30).unwrap());
        store.audit().unwrap();
        assert_eq!(Some((1, 10)), store.push(&4, &40).unwrap());
        store.audit().unwrap();
        assert_eq!(Some((2, 20)), store.push(&5, &50).unwrap());
        assert_eq!(3, store.capacity());
        assert_eq!(3, store.len());
        store.audit().unwrap();
    }

    #[test]
    fn test_pop_lru() {
        let mut store: Store<usize, usize> = Store::open_temporary(10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        assert_eq!(None, store.insert(&1, &10).unwrap());
        assert_eq!(None, store.insert(&2, &20).unwrap());
        assert_eq!(None, store.insert(&3, &30).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(3, store.len());
        store.audit().unwrap();
        assert_eq!(Some((1, 10)), store.pop_lru().unwrap());
        store.audit().unwrap();
        assert_eq!(Some((2, 20)), store.pop_lru().unwrap());
        store.audit().unwrap();
        assert_eq!(Some((3, 30)), store.pop_lru().unwrap());
        store.audit().unwrap();
        assert_eq!(None, store.pop_lru().unwrap());
        store.audit().unwrap();
    }

    #[test]
    fn test_pop() {
        let mut store: Store<usize, usize> = Store::open_temporary(10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        assert_eq!(None, store.insert(&1, &10).unwrap());
        assert_eq!(None, store.insert(&2, &20).unwrap());
        assert_eq!(None, store.insert(&3, &30).unwrap());
        assert_eq!(10, store.capacity());
        assert_eq!(3, store.len());
        store.audit().unwrap();
        assert_eq!(Some((2, 20)), store.pop(&2).unwrap());
        store.audit().unwrap();
        assert_eq!(None, store.pop(&2).unwrap());
        store.audit().unwrap();
        assert_eq!(Some(1), store.lru().unwrap());
        assert_eq!(Some(3), store.mru().unwrap());
        assert_eq!(Some((1, 10)), store.pop(&1).unwrap());
        store.audit().unwrap();
        assert_eq!(Some((3, 30)), store.pop(&3).unwrap());
        assert_eq!(0, store.len());
        store.audit().unwrap();
        assert_eq!(None, store.pop(&3).unwrap());
        store.audit().unwrap();
    }

    #[test]
    fn test_capacity_is_stored_in_db() {
        let tmp_dir = TempDir::new().unwrap();
        let file_path = tmp_dir.path().join("test.db");
        {
            let mut store: Store<usize, usize> =
                Store::open_with_path(file_path.clone(), 10).unwrap();
            assert_eq!(10, store.capacity());
            for i in 0..20 as usize {
                store.push(&i, &(i * 10)).unwrap();
            }
            store.resize(5).unwrap();
            store.resize(15).unwrap();
            store.flush().unwrap();
        }
        {
            let mut store: Store<usize, usize> =
                Store::open_with_path(file_path.clone(), 10).unwrap();
            assert_eq!(15, store.capacity());
            assert_eq!(5, store.len());
            assert_eq!(Some((15, 150)), store.pop_lru().unwrap());
            store.flush().unwrap();
        }
    }

    #[test]
    fn test_import_export() {
        let mut src: Store<usize, usize> = Store::open_temporary(10).unwrap();
        let mut dst: Store<usize, usize> = Store::open_temporary(10).unwrap();

        for i in 0..5 as usize {
            src.push(&i, &(i * 10)).unwrap();
        }
        let export = src.export();
        dst.import(export).unwrap();
        assert_eq!(&src, &dst);
    }

    #[test]
    fn test_import_map_iter() {
        let mut src: HashMap<usize, usize> = HashMap::new();
        let mut dst: Store<usize, usize> = Store::open_temporary(10).unwrap();

        for i in 0..5 as usize {
            src.insert(i, i * 10);
        }
        let export = src.iter();
        dst.import_map_iter(export).unwrap();
    }

    #[test]
    fn test_import_vec_iter() {
        let mut src: Vec<(usize, usize)> = Vec::new();
        let mut dst: Store<usize, usize> = Store::open_temporary(10).unwrap();

        for i in 0..5 as usize {
            src.push((i, i * 10));
        }
        let export = src.iter();
        dst.import_vec_iter(export).unwrap();
    }

    #[test]
    fn test_dump_restore() {
        let mut src: Store<usize, usize> = Store::open_temporary(10).unwrap();
        let mut dst: Store<usize, usize> = Store::open_temporary(10).unwrap();

        for i in 0..5 as usize {
            src.push(&i, &(i * 10)).unwrap();
            dst.push(&(i * 100), &(i * 1000)).unwrap();
        }
        dst.resize(2).unwrap();
        let dump = src.dump().unwrap();
        assert_eq!(5, dst.restore(&dump).unwrap());
        assert_eq!(src, dst);
    }

    #[test]
    fn test_to_vec() {
        let mut store: Store<usize, usize> = Store::open_temporary(3).unwrap();

        for i in 0..10 as usize {
            store.push(&i, &(i * 10)).unwrap();
        }
        let vec = store.to_vec().unwrap();
        assert_eq!(vec![(7, 70), (8, 80), (9, 90)], vec);
    }

    #[test]
    fn test_to_map() {
        let mut store: Store<usize, usize> = Store::open_temporary(3).unwrap();

        for i in 0..10 as usize {
            store.push(&i, &(i * 10)).unwrap();
        }
        let mut expected: HashMap<usize, usize> = HashMap::new();
        expected.insert(7, 70);
        expected.insert(8, 80);
        expected.insert(9, 90);
        let map = store.to_map().unwrap();
        assert_eq!(expected, map);
    }

    #[test]
    fn test_display() {
        let mut store: Store<String, String> = Store::open_temporary(10).unwrap();
        assert_eq!(10, store.capacity());
        assert_eq!(0, store.len());
        store.audit().unwrap();
        assert_eq!("[]", format!("{}", &store));
        assert_eq!(None, store.mru().unwrap());
        assert_eq!(
            None,
            store
                .insert(&"hip".to_string(), &"hop".to_string())
                .unwrap()
        );
        assert_eq!("[hip: hop]", format!("{}", &store));
        store.audit().unwrap();
    }

    #[derive(Debug, Eq, PartialEq, Serialize, Deserialize)]
    struct Coord {
        x: i64,
        y: i64,
    }

    impl fmt::Display for Coord {
        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
            write!(f, "({},{})", self.x, self.y)
        }
    }

    #[test]
    fn test_monkey() {
        let tmp_dir = TempDir::new().unwrap();
        let file_path = tmp_dir.path().join("monkey.db");
        let config = Config::default()
            .path(file_path)
            .temporary(false)
            .create_new(true)
            .use_compression(true)
            .compression_factor(10)
            .mode(Mode::LowSpace)
            .flush_every_ms(Some(100));

        let mut store: Store<String, Coord> = Store::open_with_config(config, 10).unwrap();
        let mut rng = rand::thread_rng();
        let mut dummy: i64 = 0;
        for i in 0..200 {
            if i % 19 == 0 {
                println!("flush");
                store.flush().unwrap();
                continue;
            }
            if i % 83 == 0 {
                println!("audit/clear");
                println!("clear");
                store.audit().unwrap();
                store.clear().unwrap();
                continue;
            }
            let dice = rng.gen_range(0..=100);
            if dice < 20 {
                let key = format!("key_{}", i % 15);
                let value = Coord {
                    x: i % 89,
                    y: i % 97,
                };
                println!("insert {:?} -> {:?}", &key, &value);
                store.insert(&key, &value).unwrap();
                continue;
            }
            if dice < 40 {
                let key = format!("key_{}", i % 15);
                let value = Coord {
                    x: i % 89,
                    y: i % 97,
                };
                println!("push {:?} -> {:?}", &key, &value);
                store.push(&key, &value).unwrap();
                continue;
            }
            if dice < 50 {
                let key = format!("key_{}", i % 15);
                println!("get {:?}", &key);
                println!("current state {}", &store);
                store.get(&key).unwrap();
                continue;
            }
            if dice < 60 {
                let key = format!("key_{}", i % 15);
                println!("peek {:?}", &key);
                store.peek(&key).unwrap();
                continue;
            }
            if dice < 70 {
                let key = format!("key_{}", i % 15);
                println!("pop {:?}", &key);
                store.pop(&key).unwrap();
                continue;
            }
            if dice < 80 {
                let key = format!("key_{}", i % 15);
                println!("remove {:?}", &key);
                store.remove(&key).unwrap();
                continue;
            }
            if dice < 81 {
                println!("dump");
                let dump = store.dump().unwrap();
                let mut other: Store<String, Coord> = Store::open_temporary(100).unwrap();
                other.restore(&dump).unwrap();
            }
            if dice < 82 {
                println!("export");
                for kv in store.export() {
                    dummy += kv.1.x * kv.1.y;
                }
            }
            if dice < 83 {
                println!("export_values");
                for v in store.export_values() {
                    dummy += v.x * v.y;
                }
            }
            if dice < 84 {
                println!("iter");
                for item in store.iter() {
                    let kv = item.unwrap();
                    dummy += kv.1.x * kv.1.y;
                }
            }
            if dice < 85 {
                println!("values");
                for item in store.values() {
                    let v = item.unwrap();
                    dummy += v.x * v.y;
                }
            }
        }
        assert_ne!(0, dummy);
    }

    #[test]
    fn test_this_is_not_a_bench() {
        let tmp_dir = TempDir::new().unwrap();
        let file_path = tmp_dir.path().join("monkey.db");
        let config = Config::default()
            .path(file_path)
            .temporary(false)
            .create_new(true)
            .use_compression(false)
            .mode(Mode::HighThroughput)
            .flush_every_ms(Some(3000));

        let mut store: Store<String, String> = Store::open_with_config(config, 100).unwrap();
        for i in 0..100 {
            store
                .insert(&format!("key{}", i % 17), &format!("value{}", i))
                .unwrap();
        }
        store.flush().unwrap();
    }
}