use crate::{
    error::{PRes, PersyError},
    id::RecRef,
    index::{
        config::{IndexType, ValueMode},
        keeper::{IndexKeeper, IndexModify},
    },
};
pub type NodeRef = RecRef;
use std::{
    cmp::Ordering,
    iter::{Peekable, Rev},
    ops::Bound,
    rc::Rc,
    vec::IntoIter,
};

#[derive(Clone)]
pub enum Node<K, V> {
    NODE(Nodes<K>),
    LEAF(Leaf<K, V>),
}

impl<K: IndexType, V: IndexType> Node<K, V> {
    pub fn merge_right(&mut self, k: K, node: &mut Node<K, V>) {
        match self {
            Node::NODE(n) => match node {
                Node::NODE(n1) => {
                    n.merge_right(k, n1);
                }
                Node::LEAF(_) => {
                    panic!("impossible merge a leaf to node");
                }
            },
            Node::LEAF(l) => match node {
                Node::NODE(_) => {
                    panic!("impossible merge a node to leaf");
                }
                Node::LEAF(l1) => {
                    l.merge_right(l1);
                }
            },
        }
    }
    pub fn len(&self) -> usize {
        match self {
            Node::NODE(n) => n.len(),
            Node::LEAF(l) => l.len(),
        }
    }
    pub fn split(&mut self, top_limit: usize) -> Vec<(K, Node<K, V>)> {
        match self {
            Node::NODE(n) => n.split(top_limit).into_iter().map(|x| (x.0, Node::NODE(x.1))).collect(),
            Node::LEAF(l) => l.split(top_limit).into_iter().map(|x| (x.0, Node::LEAF(x.1))).collect(),
        }
    }
}

pub(crate) fn compare<T: IndexType>(first: &T, second: &T) -> Ordering {
    first.cmp(second)
}

#[derive(Clone)]
pub struct Nodes<K> {
    pub keys: Vec<K>,
    pub pointers: Vec<NodeRef>,
}

impl<K: IndexType> Nodes<K> {
    pub fn new_from_split(left: NodeRef, values: &[(K, NodeRef)]) -> Nodes<K> {
        let keys = values.iter().map(|z| z.0.clone()).collect();
        let mut pointers: Vec<NodeRef> = values.iter().map(|z| z.1.clone()).collect();
        pointers.insert(0, left);
        Nodes { keys, pointers }
    }

    pub fn add(&mut self, pos: usize, k: &K, node_ref: NodeRef) {
        self.keys.insert(pos, k.clone());
        self.pointers.insert(pos + 1, node_ref);
    }

    pub fn find(&self, k: &K) -> PosRef<K> {
        match self.keys.binary_search_by(|x| compare(x, k)) {
            Ok(index) => PosRef::new(k, index + 1, self.pointers[index + 1].clone()),
            Err(index) => PosRef::new(k, index, self.pointers[index].clone()),
        }
    }

    pub fn get_key(&self, pos: usize) -> K {
        self.keys[pos].clone()
    }

    pub fn get(&self, pos: usize) -> NodeRef {
        self.pointers[pos].clone()
    }

    pub fn insert_after(&mut self, pos: usize, values: &mut Vec<(K, NodeRef)>) {
        values.reverse();
        for val in values.iter() {
            self.add(pos, &val.0, val.1.clone());
        }
    }

    pub fn remove(&mut self, pos: usize) -> Option<NodeRef> {
        if pos < self.pointers.len() {
            self.keys.remove(pos - 1);
            Some(self.pointers.remove(pos))
        } else {
            None
        }
    }

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

    pub fn split(&mut self, max: usize) -> Vec<(K, Nodes<K>)> {
        let mut split_result = Vec::new();
        let size = self.keys.len();
        let n_split = size / max;
        let split_offset = size / (n_split + 1) + 1;
        let mut others = self.keys.split_off(split_offset - 1);
        let mut other_pointers = self.pointers.split_off(split_offset);

        while others.len() > max {
            let new = others.split_off(split_offset);
            let new_pointers = other_pointers.split_off(split_offset);
            let key = others.remove(0);
            let leaf = Nodes {
                keys: others,
                pointers: other_pointers,
            };
            split_result.push((key, leaf));
            others = new;
            other_pointers = new_pointers;
        }

        let key = others.remove(0);
        let leaf = Nodes {
            keys: others,
            pointers: other_pointers,
        };
        split_result.push((key, leaf));
        split_result
    }

    #[allow(dead_code)]
    pub fn merge_left(&mut self, owner: K, nodes: &mut Nodes<K>) {
        let mut keys = std::mem::replace(&mut nodes.keys, Vec::new());
        let mut pointers = std::mem::replace(&mut nodes.pointers, Vec::new());
        keys.push(owner);
        keys.append(&mut self.keys);
        pointers.append(&mut self.pointers);
        self.keys = keys;
        self.pointers = pointers;
    }

    pub fn merge_right(&mut self, owner: K, nodes: &mut Nodes<K>) {
        self.keys.push(owner);
        self.keys.append(&mut nodes.keys);
        self.pointers.append(&mut nodes.pointers);
    }
}

/// The associated value to the index key
#[derive(Clone, PartialEq, Debug)]
pub enum Value<V> {
    /// A cluster of values
    CLUSTER(Vec<V>),
    /// A single value entry
    SINGLE(V),
}

impl<V> IntoIterator for Value<V> {
    type Item = V;
    type IntoIter = IntoIter<V>;

    fn into_iter(self) -> IntoIter<V> {
        match self {
            Value::SINGLE(v) => vec![v].into_iter(),
            Value::CLUSTER(v) => v.into_iter(),
        }
    }
}

pub struct PageIter<K: IndexType, V: IndexType> {
    pub iter: Peekable<IntoIter<LeafEntry<K, V>>>,
    pub next: Option<NodeRef>,
}

pub struct PageIterBack<K: IndexType, V: IndexType> {
    pub iter: Peekable<Rev<IntoIter<LeafEntry<K, V>>>>,
    pub prev: Option<NodeRef>,
}

#[derive(Clone)]
pub struct Leaf<K, V> {
    pub entries: Vec<LeafEntry<K, V>>,
    pub prev: Option<NodeRef>,
    pub next: Option<NodeRef>,
}

#[derive(Clone)]
pub struct LeafEntry<K, V> {
    pub key: K,
    pub value: Value<V>,
}

impl<K: IndexType, V: IndexType> Leaf<K, V> {
    pub fn new() -> Leaf<K, V> {
        Leaf {
            entries: Vec::new(),
            prev: None,
            next: None,
        }
    }

    pub fn add(&mut self, pos: usize, k: &K, v: &V, _value_mode: ValueMode) {
        self.entries.insert(
            pos,
            LeafEntry {
                key: k.clone(),
                value: Value::SINGLE(v.clone()),
            },
        );
    }

    pub fn find<'a>(&'a self, k: &K) -> Result<(K, Value<V>), usize> {
        self.entries
            .binary_search_by(|n| compare(&n.key, k))
            .map(|index| (self.entries[index].key.clone(), self.entries[index].value.clone()))
    }

    pub fn iter(&self) -> IntoIter<LeafEntry<K, V>> {
        self.entries.clone().into_iter()
    }

    pub fn iter_from(&self, bound: Bound<&K>) -> IntoIter<LeafEntry<K, V>> {
        let index = match bound {
            Bound::Included(k) => match self.entries.binary_search_by(|n| compare(&n.key, k)) {
                Ok(index) => index,
                Err(index) => index,
            },
            Bound::Excluded(k) => match self.entries.binary_search_by(|n| compare(&n.key, k)) {
                Ok(index) => index + 1,
                Err(index) => index,
            },
            Bound::Unbounded => 0,
        };
        self.entries[index..].to_vec().into_iter()
    }

    pub fn back_iter_from(&self, bound: Bound<&K>) -> Rev<IntoIter<LeafEntry<K, V>>> {
        let index = match bound {
            Bound::Included(k) => match self.entries.binary_search_by(|n| compare(&n.key, k)) {
                Ok(index) => index + 1,
                Err(index) => index,
            },
            Bound::Excluded(k) => match self.entries.binary_search_by(|n| compare(&n.key, k)) {
                Ok(index) => index,
                Err(index) => index,
            },
            Bound::Unbounded => 0,
        };
        self.entries[..index].to_vec().into_iter().rev()
    }

    pub fn insert_or_update(&mut self, k: &K, v: &V, value_mode: ValueMode, index_name: &str) -> PRes<()> {
        match self.entries.binary_search_by(|n| compare(&n.key, k)) {
            Ok(index) => {
                let entry = &mut self.entries[index];
                match value_mode {
                    ValueMode::REPLACE => {
                        entry.value = Value::SINGLE(v.clone());
                    }
                    ValueMode::EXCLUSIVE => match entry.value {
                        Value::SINGLE(ref ev) => {
                            if compare(ev, v) != Ordering::Equal {
                                return Err(PersyError::IndexDuplicateKey(index_name.to_string(), format!("{}", k)));
                            }
                        }
                        _ => unreachable!("Exclusive leafs never have cluster values"),
                    },
                    ValueMode::CLUSTER => {
                        let mut new_value = None;
                        match entry.value {
                            Value::SINGLE(ref ev) => {
                                if compare(ev, v) != Ordering::Equal {
                                    new_value = Some(Value::CLUSTER(vec![ev.clone(), v.clone()]));
                                }
                            }
                            Value::CLUSTER(ref mut cl) => {
                                if let Err(index) = cl.binary_search_by(|x| compare(x, v)) {
                                    cl.insert(index, v.clone());
                                }
                            }
                        }
                        if let Some(v) = new_value {
                            entry.value = v;
                        }
                    }
                }
            }
            Err(index) => self.add(index, k, v, value_mode),
        }
        Ok(())
    }

    pub fn remove(&mut self, k: &K, v: &Option<V>) -> bool {
        match self.entries.binary_search_by(|n| compare(&n.key, k)) {
            Ok(index) => {
                if let Some(rv) = v {
                    let mut removed = false;
                    let remove_entry = {
                        let mut new_value = None;
                        let entry = &mut self.entries[index];
                        let remove_entry = match &mut entry.value {
                            Value::SINGLE(val) => {
                                if compare(val, rv) == Ordering::Equal {
                                    removed = true;
                                    true
                                } else {
                                    false
                                }
                            }
                            Value::CLUSTER(ref mut cl) => {
                                if let Ok(index) = cl.binary_search_by(|x| compare(x, rv)) {
                                    removed = true;
                                    cl.remove(index);
                                }
                                if cl.len() == 1 {
                                    new_value = Some(Value::SINGLE(cl.pop().unwrap()));
                                    false
                                } else {
                                    cl.is_empty()
                                }
                            }
                        };
                        if let Some(new) = new_value {
                            entry.value = new;
                        }
                        remove_entry
                    };
                    if remove_entry {
                        self.entries.remove(index);
                    }
                    removed
                } else {
                    self.entries.remove(index);
                    true
                }
            }
            Err(_) => false,
        }
    }

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

    pub fn split(&mut self, max: usize) -> Vec<(K, Leaf<K, V>)> {
        let mut split_result = Vec::new();
        let size = self.entries.len();
        let n_split = size / max;
        let split_offset = size / (n_split + 1) + 1;
        let mut others = self.entries.split_off(split_offset);

        while others.len() > max {
            let new = others.split_off(split_offset);
            let key = others[0].key.clone();
            let leaf = Leaf {
                entries: others,
                prev: None,
                next: None,
            };
            split_result.push((key, leaf));
            others = new;
        }

        let key = others[0].key.clone();
        let leaf = Leaf {
            entries: others,
            prev: None,
            next: None,
        };
        split_result.push((key, leaf));
        split_result
    }

    pub fn set_prev(&mut self, prev: Option<NodeRef>) {
        self.prev = prev;
    }

    pub fn set_next(&mut self, next: Option<NodeRef>) {
        self.next = next;
    }

    #[allow(dead_code)]
    pub fn merge_left(&mut self, leaf: &mut Leaf<K, V>) {
        let mut entries = std::mem::replace(&mut leaf.entries, Vec::new());
        entries.append(&mut self.entries);
        self.entries = entries;
    }

    pub fn merge_right(&mut self, leaf: &mut Leaf<K, V>) {
        self.entries.append(&mut leaf.entries);
    }
}

pub enum ValueChange<V> {
    ADD(V),
    REMOVE(Option<V>),
}

pub struct KeyChanges<K, V> {
    pub k: K,
    changes: Vec<ValueChange<V>>,
}

impl<K: IndexType, V: IndexType> KeyChanges<K, V> {
    #[allow(dead_code)]
    pub fn single_add(k: K, v: V) -> KeyChanges<K, V> {
        KeyChanges {
            k,
            changes: vec![ValueChange::ADD(v)],
        }
    }

    #[allow(dead_code)]
    fn single_delete(k: K, v: Option<V>) -> KeyChanges<K, V> {
        KeyChanges {
            k,
            changes: vec![ValueChange::REMOVE(v)],
        }
    }

    pub fn new(k: K, changes: Vec<ValueChange<V>>) -> KeyChanges<K, V> {
        KeyChanges { k, changes }
    }
    fn apply(&self, leaf: &mut Leaf<K, V>, value_mode: ValueMode, index: &str) -> PRes<bool> {
        let mut update = false;
        for vc in &self.changes {
            update |= match vc {
                ValueChange::ADD(v) => {
                    if leaf.len() > 0 {
                        leaf.insert_or_update(&self.k, &v, value_mode.clone(), index)?;
                    } else {
                        leaf.add(0, &self.k, v, value_mode.clone());
                    }
                    true
                }
                ValueChange::REMOVE(r) => leaf.remove(&self.k, &r),
            };
        }
        Ok(update)
    }
}

pub trait Index<K: IndexType, V: IndexType> {
    fn get(&mut self, k: &K) -> PRes<Option<Value<V>>>;
    fn iter_node(&mut self, node: &NodeRef) -> PRes<PageIter<K, V>>;
    fn iter_from(&mut self, first: Bound<&K>) -> PRes<PageIter<K, V>>;
    fn back_iter_from(&mut self, first: Bound<&K>) -> PRes<PageIterBack<K, V>>;
    fn back_iter_node(&mut self, node: &NodeRef) -> PRes<PageIterBack<K, V>>;
}

pub trait IndexApply<K: IndexType, V: IndexType>: Index<K, V> {
    fn apply(&mut self, adds: &[KeyChanges<K, V>]) -> PRes<()>;
}

#[derive(PartialEq, Clone, Debug)]
pub struct PosRef<K: IndexType> {
    pub k: K,
    pub pos: usize,
    pub node_ref: NodeRef,
    pub version: u16,
}

impl<K: IndexType> PosRef<K> {
    fn new(k: &K, pos: usize, node_ref: NodeRef) -> PosRef<K> {
        PosRef {
            k: k.clone(),
            pos,
            node_ref: node_ref.clone(),
            version: 0,
        }
    }
}

struct ParentNodeChanged<K: IndexType> {
    path: Vec<PosRef<K>>,
    children: Vec<PosRef<K>>,
}

fn owned<K: IndexType, V: IndexType>(mut node: Rc<Node<K, V>>) -> Node<K, V> {
    Rc::make_mut(&mut node);
    Rc::try_unwrap(node).ok().unwrap()
}

fn split_link_save<K: IndexType, V: IndexType, I: IndexModify<K, V>>(
    index: &mut I,
    mut prev_id: NodeRef,
    mut first: Node<K, V>,
    cur_version: u16,
) -> PRes<Vec<(K, NodeRef)>> {
    let new_nodes = first.split(index.top_limit());
    let mut ids = Vec::new();
    for (k, new_node) in new_nodes {
        let saved_id = index.insert(new_node)?;
        ids.push((k, saved_id));
    }
    if let Node::LEAF(leaf) = first {
        let first_next = leaf.next.clone();
        let mut prev_leaf = leaf;
        let mut prev_version = cur_version;
        for (_, id) in &ids {
            let (sibling_node, version) = index.load_modify(&id)?;
            if let Node::LEAF(mut sibling) = owned(sibling_node) {
                sibling.set_prev(Some(prev_id.clone()));
                prev_leaf.set_next(Some(id.clone()));
                index.update(&prev_id, Node::LEAF(prev_leaf), prev_version)?;
                prev_id = id.clone();
                prev_leaf = sibling;
                prev_version = version;
            }
        }
        prev_leaf.set_next(first_next);
        index.update(&prev_id, Node::LEAF(prev_leaf), prev_version)?;
    } else {
        index.update(&prev_id, first, cur_version)?;
    }
    Ok(ids)
}

fn group_by_parent<K: IndexType>(updates: Vec<Vec<PosRef<K>>>) -> Vec<ParentNodeChanged<K>> {
    let mut parent_updates = Vec::new();
    let mut parent_node: Option<NodeRef> = None;
    let mut new_update: Option<ParentNodeChanged<K>> = None;
    for mut update in updates {
        if let Some(last) = update.pop() {
            if parent_node == update.last().map(|x| x.node_ref.clone()) {
                if let Some(p) = &mut new_update {
                    p.children.push(last);
                }
            } else {
                if let Some(p) = new_update {
                    parent_updates.push(p);
                }
                parent_node = update.last().map(|x| x.node_ref.clone());
                let mut children = Vec::new();
                children.push(last);
                new_update = Some(ParentNodeChanged { path: update, children });
            }
        }
    }
    if let Some(p) = new_update {
        parent_updates.push(p);
    }
    parent_updates
}
fn lock_parents<K: IndexType, V: IndexType, I: IndexModify<K, V>>(
    index: &mut I,
    mut updates: Vec<Vec<PosRef<K>>>,
) -> PRes<Vec<Vec<PosRef<K>>>> {
    for i in 0..updates.len() {
        loop {
            debug_assert!(
                updates[i].len() >= 2,
                "never lock a path that shorter than 2 because that would be the root"
            );
            let parent = {
                let update = &updates[i];
                update[update.len() - 2].clone()
            };
            if index.lock(&parent.node_ref, parent.version)? {
                break;
            } else {
                let last = updates[i].last().unwrap().clone();
                if let Some(ref node) = index.get_root()? {
                    let mut path = Vec::new();
                    let mut cur_node = PosRef::new(&parent.k, 0, node.clone());
                    while cur_node.node_ref == last.node_ref {
                        let (node_modify, version) = index.load_modify(&cur_node.node_ref)?;
                        cur_node.version = version;
                        path.push(cur_node.clone());
                        if let Node::NODE(n) = &*node_modify {
                            cur_node = n.find(&parent.k);
                        } else {
                            panic!("cannot ever get a leaf")
                        }
                    }
                    updates[i] = path;
                }
            }
        }
    }
    Ok(updates)
}

fn merge_and_save<K: IndexType, V: IndexType, I: IndexModify<K, V>>(
    index: &mut I,
    parent: &mut Nodes<K>,
    pos: usize,
    mut cur: Node<K, V>,
    version: u16,
) -> PRes<bool> {
    Ok(if pos > 0 {
        let node_ref = parent.get(pos - 1);
        let (dest_node, dest_version) = index.load_modify(&node_ref)?;
        let mut dest_merge = owned(dest_node);
        dest_merge.merge_right(parent.get_key(pos - 1), &mut cur);
        index.update(&node_ref, dest_merge, dest_version)?;
        index.delete(&parent.remove(pos).unwrap(), version)?;
        false
    } else {
        let node_ref = parent.get(pos + 1);
        let (source_node, source_version) = index.load_modify(&node_ref)?;
        let mut source_merge = owned(source_node);
        cur.merge_right(parent.get_key(pos), &mut source_merge);
        index.delete(&parent.remove(pos + 1).unwrap(), source_version)?;
        index.update(&parent.get(pos), cur, version)?;
        true
    })
}

impl<K: IndexType, V: IndexType, T> IndexApply<K, V> for T
where
    T: IndexModify<K, V>,
    T: Index<K, V>,
{
    fn apply(&mut self, adds: &[KeyChanges<K, V>]) -> PRes<()> {
        let mut root = self.get_root()?;
        let mut modified_root = false;
        let mut updates = Vec::new();
        let mut prev_leaf_id = None;
        for add in adds {
            if let Some(ref node) = root {
                let mut path = Vec::new();
                let mut cur_node = PosRef::new(&add.k, 0, node.clone());
                loop {
                    let (node_modify, version) = self.load_modify(&cur_node.node_ref)?;
                    cur_node.version = version;
                    path.push(cur_node.clone());
                    if let Node::NODE(n) = &*node_modify {
                        cur_node = n.find(&add.k);
                    } else if let Node::LEAF(mut leaf) = owned(node_modify) {
                        let node_ref = &cur_node.node_ref;
                        if self.lock(node_ref, version)? {
                            if add.apply(&mut leaf, self.value_mode(), self.index_name())? {
                                self.update(node_ref, Node::LEAF(leaf), version)?;
                                if Some(node_ref.clone()) != prev_leaf_id {
                                    updates.push(path);
                                }
                                prev_leaf_id = Some(node_ref.clone())
                            }
                            break;
                        } else {
                            // Could not lock different version from the read version
                            // restart scanning from root for this key
                            path = Vec::new();
                            cur_node = PosRef::new(&add.k, 0, node.clone());
                        }
                    }
                }
            } else {
                let mut leaf = Leaf::new();
                add.apply(&mut leaf, self.value_mode(), self.index_name())?;
                let leaf_ref = self.insert(Node::LEAF(leaf))?;
                root = Some(leaf_ref);
                modified_root = true;
            }
        }
        while updates.len() > 1 || (updates.len() == 1 && updates[0].len() > 1) {
            updates = lock_parents(self, updates)?;
            let parent_updates = group_by_parent(updates);
            updates = Vec::new();
            for update in parent_updates {
                let parent_id = update.path.last().unwrap().node_ref.clone();
                let (read_node, n_version) = self.load_modify(&parent_id)?;
                if let Node::NODE(mut n) = owned(read_node) {
                    let mut save = false;
                    let mut flags = vec![false; n.len()];
                    for ch in update.children {
                        let pos = n.find(&ch.k);
                        flags[pos.pos] = true;
                    }
                    let mut i = 0;
                    while i < n.len() {
                        let pos = i;
                        i += 1;
                        if flags[pos] && n.len() > 1 {
                            let (cur, version) = self.load_modify(&n.get(pos))?;
                            if cur.len() < self.bottom_limit() {
                                let mut new_pos = pos;
                                if merge_and_save(self, &mut n, pos, owned(cur), version)? {
                                    new_pos += 1;
                                }
                                flags.remove(new_pos);
                                flags[new_pos - 1] = true;
                                i = new_pos - 1;
                                save = true;
                            }
                        }
                    }
                    for pos in 0..n.len() {
                        if flags[pos] {
                            let (cur, cur_version) = self.load_modify(&n.get(pos))?;
                            if cur.len() > self.top_limit() {
                                let mut ids = split_link_save(self, n.get(pos), owned(cur), cur_version)?;
                                for _ in 0..ids.len() {
                                    flags.insert(pos, false);
                                }
                                n.insert_after(pos, &mut ids);
                                save = true;
                            }
                        }
                    }
                    if save {
                        if !update.path.is_empty() {
                            updates.push(update.path);
                            self.update(&parent_id, Node::NODE(n), n_version)?;
                        } else {
                            break;
                        }
                    }
                }
            }
        }
        if updates.len() == 1 && updates[0].len() == 1 {
            while let Some(r) = root.clone() {
                let (n, n_version) = self.load_modify(&r)?;
                if n.len() > self.top_limit() {
                    let ids = split_link_save(self, r.clone(), owned(n), n_version)?;
                    let node = Node::NODE(Nodes::new_from_split(r, &ids));
                    let cur_id = self.insert(node)?;
                    root = Some(cur_id);
                } else if n.len() == 1 {
                    if let Node::NODE(cn) = &*n {
                        self.delete(&r, n_version)?;
                        root = Some(cn.get(0));
                    } else {
                        break;
                    }
                } else if n.len() == 0 {
                    root = None;
                } else {
                    break;
                }
                modified_root = true;
            }
        }
        if modified_root {
            self.set_root(root)?;
        }

        Ok(())
    }
}

impl<K: IndexType, V: IndexType, T> Index<K, V> for T
where
    T: IndexKeeper<K, V>,
{
    fn get(&mut self, k: &K) -> PRes<Option<Value<V>>> {
        Ok(if let Some(node) = self.get_root()? {
            let mut cur_node = node;
            loop {
                match self.load(&cur_node)? {
                    Node::NODE(n) => {
                        cur_node = n.find(k).node_ref;
                    }
                    Node::LEAF(leaf) => {
                        break leaf.find(k).map(|el| el.1).ok();
                    }
                }
            }
        } else {
            None
        })
    }

    fn iter_from(&mut self, first: Bound<&K>) -> PRes<PageIter<K, V>> {
        Ok(if let Some(node) = self.get_root()? {
            let mut cur_node = node;
            loop {
                match self.load(&cur_node)? {
                    Node::NODE(n) => {
                        cur_node = match first {
                            Bound::Included(f) => n.find(f).node_ref,
                            Bound::Excluded(f) => n.find(f).node_ref,
                            Bound::Unbounded => n.get(0),
                        };
                    }
                    Node::LEAF(leaf) => {
                        break PageIter {
                            iter: leaf.iter_from(first).peekable(),
                            next: leaf.next,
                        };
                    }
                }
            }
        } else {
            PageIter {
                iter: Vec::new().into_iter().peekable(),
                next: None,
            }
        })
    }

    fn iter_node(&mut self, node: &NodeRef) -> PRes<PageIter<K, V>> {
        match self.load(node)? {
            Node::NODE(_) => panic!("impossible to iterate a node can just iterate a leaf"),
            Node::LEAF(leaf) => Ok(PageIter {
                iter: leaf.iter().peekable(),
                next: leaf.next,
            }),
        }
    }

    fn back_iter_from(&mut self, last: Bound<&K>) -> PRes<PageIterBack<K, V>> {
        Ok(if let Some(mut cur_node) = self.get_root()? {
            loop {
                match self.load(&cur_node)? {
                    Node::NODE(n) => {
                        cur_node = match last {
                            Bound::Included(f) => n.find(f).node_ref,
                            Bound::Excluded(f) => n.find(f).node_ref,
                            Bound::Unbounded => n.get(n.len() - 1),
                        };
                    }
                    Node::LEAF(leaf) => {
                        break PageIterBack {
                            iter: leaf.back_iter_from(last).peekable(),
                            prev: leaf.next,
                        };
                    }
                }
            }
        } else {
            PageIterBack {
                iter: Vec::new().into_iter().rev().peekable(),
                prev: None,
            }
        })
    }

    fn back_iter_node(&mut self, node: &NodeRef) -> PRes<PageIterBack<K, V>> {
        match self.load(node)? {
            Node::NODE(_) => panic!("impossible to iterate a node can just iterate a leaf"),
            Node::LEAF(leaf) => Ok(PageIterBack {
                iter: leaf.iter().rev().peekable(),
                prev: leaf.prev,
            }),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        Index, IndexApply, IndexKeeper, IndexModify, IndexType, KeyChanges, Leaf, Node, NodeRef, Nodes, PRes, PosRef,
        Value, ValueChange, ValueMode,
    };
    use crate::{error::PersyError, id::RecRef};
    use rand::random;
    use std::{collections::HashMap, fmt, ops::Bound, rc::Rc};

    impl<V> fmt::Display for Value<V>
    where
        V: fmt::Display,
    {
        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
            match self {
                Value::CLUSTER(x) => {
                    write!(f, "{} values: [", x.len())?;
                    for v in x {
                        write!(f, " {}, ", v)?;
                    }
                    write!(f, "]")?;
                }
                Value::SINGLE(v) => {
                    write!(f, "value: {}", v)?;
                }
            }
            Ok(())
        }
    }

    fn print_nodes<K: IndexType, V: IndexType>(
        tree: &mut dyn IndexKeeper<K, V>,
        node: &Nodes<K>,
        depth: usize,
    ) -> PRes<()> {
        let padding = String::from_utf8(vec![b' '; depth]).unwrap();
        for i in 0..node.len() {
            if i == 0 {
                println!("{} __ ", padding);
            } else {
                println!("{} {}  ", padding, node.keys[i - 1]);
            }
            print_node(tree, &node.pointers[i], depth + 1)?;
        }
        Ok(())
    }

    fn print_leaf<K: IndexType, V: IndexType>(
        _tree: &mut dyn IndexKeeper<K, V>,
        leaf: &Leaf<K, V>,
        depth: usize,
    ) -> PRes<()> {
        let padding = String::from_utf8(vec![b' '; depth]).unwrap();
        println!("{} prev: {:?} ", padding, leaf.prev);
        for i in 0..leaf.len() {
            println!("{} {} {} ", padding, leaf.entries[i].key, leaf.entries[i].value);
        }
        println!("{} next: {:?} ", padding, leaf.next);
        Ok(())
    }

    fn print_node<K: IndexType, V: IndexType>(
        tree: &mut dyn IndexKeeper<K, V>,
        node: &NodeRef,
        depth: usize,
    ) -> PRes<()> {
        match tree.load(node)? {
            Node::NODE(n) => {
                print_nodes(tree, &n, depth)?;
            }
            Node::LEAF(l) => {
                print_leaf(tree, &l, depth)?;
            }
        }
        Ok(())
    }

    fn print_tree<K: IndexType, V: IndexType>(tree: &mut dyn IndexKeeper<K, V>) -> PRes<()> {
        let root = tree.get_root()?;
        if let Some(r) = root {
            print_node(tree, &r, 0)?;
        } else {
            println!(" Empty Root");
        }

        Ok(())
    }

    fn random_pointer() -> NodeRef {
        RecRef::new(random::<u64>(), random::<u32>())
    }

    #[test]
    fn single_node_add_test() {
        let val1 = random_pointer();
        let val2 = random_pointer();
        let val3 = random_pointer();
        let val4 = random_pointer();
        let val5 = random_pointer();
        let val6 = random_pointer();
        let mut node = Nodes::new_from_split(val1, &[(0, val2)]);
        let pos = node.find(&2).pos;
        node.add(pos, &2, val3.clone());
        let pos = node.find(&5).pos;
        node.add(pos, &5, val4.clone());
        let pos = node.find(&6).pos;
        node.add(pos, &6, val5);
        let pos = node.find(&4).pos;
        node.add(pos, &4, val6.clone());

        let found = node.find(&4);
        assert_eq!(found.pos, 3);
        //If i search for 4 i get the one on the left of 4 so the value of 2 that is val3
        assert_eq!(found.node_ref, val6);

        let found = node.find(&5);
        assert_eq!(found.pos, 4);
        //If i search for 5 i get the one on the left of 5 so the value of 4 that is val6
        assert_eq!(found.node_ref, val4);

        let found = node.find(&3);
        //If i search for a value that do not exist i get the position of the value at is right
        //that is value 4 position 2
        assert_eq!(found.pos, 2);
        //If i search for 3 i get the value at the left of 4 that is val3
        assert_eq!(found.node_ref, val3);
    }

    #[test]
    fn single_leaf_insert_test() {
        let mut leaf = Leaf::new();
        for n in 0..50 {
            leaf.insert_or_update(&n, &n, ValueMode::REPLACE, "aa")
                .expect("insert is ok");
        }
        let res = leaf.find(&10);
        assert_eq!(Ok((10, Value::SINGLE(10))), res);

        let res = leaf.find(&60);
        assert_eq!(Err(50), res);
    }

    #[test]
    fn single_leaf_cluster_insert_test() {
        let mut leaf = Leaf::new();
        leaf.insert_or_update(&10, &1, ValueMode::CLUSTER, "aa")
            .expect("insert is ok");
        leaf.insert_or_update(&10, &2, ValueMode::CLUSTER, "aa")
            .expect("insert is ok");
        let res = leaf.find(&10);
        assert_eq!(Ok((10, Value::CLUSTER(vec![1, 2]))), res);
    }

    #[test]
    fn leaf_cluster_remove_test() {
        let mut leaf = Leaf::new();
        leaf.insert_or_update(&10, &1, ValueMode::CLUSTER, "aa")
            .expect("insert is ok");
        leaf.insert_or_update(&10, &2, ValueMode::CLUSTER, "aa")
            .expect("insert is ok");
        assert!(leaf.remove(&10, &Some(2)));
        let res = leaf.find(&10);
        assert_eq!(Ok((10, Value::SINGLE(1))), res);
    }

    #[test]
    fn leaf_cluster_remove_not_exist_value_test() {
        let mut leaf = Leaf::new();
        leaf.insert_or_update(&10, &1, ValueMode::CLUSTER, "aa")
            .expect("insert is ok");
        leaf.insert_or_update(&10, &2, ValueMode::CLUSTER, "aa")
            .expect("insert is ok");
        assert!(!leaf.remove(&10, &Some(10)));
        let res = leaf.find(&10);
        assert_eq!(Ok((10, Value::CLUSTER(vec![1, 2]))), res);
    }

    #[test]
    fn leaf_single_delete_not_exist_value_test() {
        let mut leaf = Leaf::new();
        leaf.insert_or_update(&10, &1, ValueMode::EXCLUSIVE, "aa")
            .expect("insert is ok");
        assert!(!leaf.remove(&10, &Some(10)));
        let res = leaf.find(&10);
        assert_eq!(Ok((10, Value::SINGLE(1))), res);
    }

    #[test]
    fn leaf_duplicate_key_test() {
        let mut leaf = Leaf::new();
        leaf.insert_or_update(&10, &1, ValueMode::EXCLUSIVE, "aa")
            .expect("insert is ok");
        let res = leaf.insert_or_update(&10, &2, ValueMode::EXCLUSIVE, "aa");
        assert!(res.is_err());
    }

    #[test]
    fn test_leaf_split() {
        let mut leaf = Leaf::new();

        for n in 0..103 {
            leaf.insert_or_update(&n, &n, ValueMode::REPLACE, "aa")
                .expect("insert is ok");
        }

        let res = leaf.split(21);
        assert_eq!(leaf.len(), 21);
        assert_eq!(res[0].1.len(), 21);
        assert_eq!(res[1].1.len(), 21);
        assert_eq!(res[2].1.len(), 21);
        assert_eq!(res[3].1.len(), 19);
    }

    #[test]
    fn test_node_split() {
        let mut node = Nodes::new_from_split(random_pointer(), &[(0, random_pointer())]);
        for n in 1..103 {
            let pos = node.find(&n).pos;
            node.add(pos, &n, random_pointer());
        }

        let res = node.split(21);
        assert_eq!(node.len(), 21);
        assert_eq!(node.pointers.len(), 21);
        assert_eq!(node.keys.len(), 20);
        assert_eq!(res[0].1.len(), 21);
        assert_eq!(res[0].1.pointers.len(), 21);
        assert_eq!(res[0].1.keys.len(), 20);
        assert_eq!(res[1].1.len(), 21);
        assert_eq!(res[1].1.pointers.len(), 21);
        assert_eq!(res[1].1.keys.len(), 20);
        assert_eq!(res[2].1.len(), 21);
        assert_eq!(res[2].1.pointers.len(), 21);
        assert_eq!(res[2].1.keys.len(), 20);
        assert_eq!(res[3].1.len(), 20);
        assert_eq!(res[3].1.pointers.len(), 20);
        assert_eq!(res[3].1.keys.len(), 19);
    }

    #[test]
    fn test_remove_from_leaf() {
        let mut leaf = Leaf::new();
        for n in 0..50 {
            leaf.insert_or_update(&n, &n, ValueMode::REPLACE, "aa")
                .expect("insert is ok");
        }
        assert!(leaf.remove(&10, &Some(10)));
        assert!(!leaf.remove(&100, &Some(100)));
        assert_eq!(leaf.len(), 49);
        let res = leaf.find(&10);
        assert_eq!(Err(10), res);
    }

    #[test]
    fn test_remove_from_node() {
        //TODO: check why the remove of 10 make to point to 9
        let mut node = Nodes::new_from_split(random_pointer(), &[(0, random_pointer())]);
        let mut keep = None;
        for n in 1..50 {
            let pos = node.find(&n).pos;
            let point = random_pointer();
            if n == 9 {
                keep = Some(point.clone());
            }
            node.add(pos, &n, point);
        }
        let pos = node.find(&10).pos;
        node.remove(pos);
        assert_eq!(node.len(), 50);
        let res = node.find(&10);
        assert_eq!(PosRef::new(&10, 10, keep.unwrap()), res);
    }

    #[test]
    fn test_merge_leaf() {
        let mut leaf = Leaf::new();
        let mut leaf2 = Leaf::new();
        for n in 0..20 {
            leaf.insert_or_update(&n, &n, ValueMode::REPLACE, "aa")
                .expect("insert is ok");
        }

        for n in 20..40 {
            leaf2
                .insert_or_update(&n, &n, ValueMode::REPLACE, "aa")
                .expect("insert is ok");
        }
        leaf.merge_right(&mut leaf2);
        assert_eq!(leaf.len(), 40);
        assert_eq!(leaf2.len(), 0);
        let res = leaf.find(&35);
        assert_eq!(res, Ok((35, Value::SINGLE(35))));

        let mut leaf = Leaf::new();
        let mut leaf2 = Leaf::new();
        for n in 20..40 {
            leaf.insert_or_update(&n, &n, ValueMode::REPLACE, "aa")
                .expect("insert is ok");
        }

        for n in 0..20 {
            leaf2
                .insert_or_update(&n, &n, ValueMode::REPLACE, "aa")
                .expect("insert is ok");
        }
        leaf.merge_left(&mut leaf2);
        assert_eq!(leaf.len(), 40);
        assert_eq!(leaf2.len(), 0);
        let res = leaf.find(&35);
        assert_eq!(res, Ok((35, Value::SINGLE(35))));
    }

    #[test]
    fn test_merge_nodes() {
        let mut node = Nodes::new_from_split(random_pointer(), &[(0, random_pointer())]);
        for n in 1..20 {
            let pos = node.find(&n).pos;
            let point = random_pointer();
            node.add(pos, &n, point);
        }

        let mut node2 = Nodes::new_from_split(random_pointer(), &[(21, random_pointer())]);
        let mut keep = None;
        for n in 22..40 {
            let pos = node2.find(&n).pos;
            let point = random_pointer();
            if n == 26 {
                keep = Some(point.clone());
            }
            node2.add(pos, &n, point);
        }

        node.merge_right(20, &mut node2);
        assert_eq!(node.len(), 41);
        assert_eq!(node2.len(), 0);
        let res = node.find(&26);
        assert_eq!(PosRef::new(&26, 27, keep.unwrap()), res);

        let mut node = Nodes::new_from_split(random_pointer(), &[(21, random_pointer())]);
        let mut keep = None;
        for n in 22..40 {
            let pos = node.find(&n).pos;
            let point = random_pointer();
            if n == 26 {
                keep = Some(point.clone());
            }
            node.add(pos, &n, point);
        }

        let mut node2 = Nodes::new_from_split(random_pointer(), &[(0, random_pointer())]);
        for n in 1..20 {
            let pos = node2.find(&n).pos;
            let point = random_pointer();
            node2.add(pos, &n, point);
        }

        node.merge_left(20, &mut node2);
        assert_eq!(node.len(), 41);
        assert_eq!(node2.len(), 0);
        let res = node.find(&26);
        assert_eq!(PosRef::new(&26, 27, keep.unwrap()), res);
    }

    struct MockIndexKeeper<K: Clone + Ord, V: Clone> {
        store: HashMap<NodeRef, Node<K, V>>,
        root: Option<NodeRef>,
        v: ValueMode,
        name: String,
    }

    impl<K: Clone + Ord, V: Clone> MockIndexKeeper<K, V> {
        fn new() -> MockIndexKeeper<K, V> {
            MockIndexKeeper {
                store: HashMap::new(),
                root: None,
                v: ValueMode::REPLACE,
                name: "test_index".to_string(),
            }
        }

        fn new_mode(v: ValueMode) -> MockIndexKeeper<K, V> {
            MockIndexKeeper {
                store: HashMap::new(),
                root: None,
                v,
                name: "test_index".to_string(),
            }
        }
    }

    impl<K: Clone + Ord, V: Clone> IndexKeeper<K, V> for MockIndexKeeper<K, V> {
        fn load(&self, node: &NodeRef) -> PRes<Node<K, V>> {
            Ok(self.store.get(&node).unwrap().clone())
        }
        fn get_root(&self) -> PRes<Option<NodeRef>> {
            Ok(self.root.clone())
        }

        fn value_mode(&self) -> ValueMode {
            self.v.clone()
        }

        fn index_name(&self) -> &String {
            &self.name
        }
    }

    impl<K: Clone + Ord, V: Clone> IndexModify<K, V> for MockIndexKeeper<K, V> {
        fn load_modify(&self, node: &NodeRef) -> PRes<(Rc<Node<K, V>>, u16)> {
            Ok((Rc::new(self.store.get(&node).unwrap().clone()), 0))
        }

        fn lock(&mut self, _node: &NodeRef, _version: u16) -> PRes<bool> {
            Ok(true)
        }

        fn insert(&mut self, node: Node<K, V>) -> PRes<NodeRef> {
            let node_ref = random_pointer();
            self.store.insert(node_ref.clone(), node.clone());
            Ok(node_ref)
        }

        fn update(&mut self, node_ref: &NodeRef, node: Node<K, V>, _version: u16) -> PRes<()> {
            self.store.insert(node_ref.clone(), node);
            Ok(())
        }

        fn set_root(&mut self, root: Option<NodeRef>) -> PRes<()> {
            Ok(self.root = root)
        }

        fn delete(&mut self, node: &NodeRef, _version: u16) -> PRes<()> {
            self.store.remove(&node);
            Ok(())
        }
        fn bottom_limit(&self) -> usize {
            10
        }
        fn top_limit(&self) -> usize {
            30
        }
    }

    #[test]
    fn test_single_add() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        changes.push(KeyChanges::single_add(1, 1));
        changes.push(KeyChanges::single_add(2, 2));
        changes.push(KeyChanges::single_add(3, 4));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(2))));
    }

    #[test]
    fn test_many_add() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        for i in 0..200 {
            changes.push(KeyChanges::single_add(i, i));
        }
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(2))));
        assert_eq!(keeper.get(&100).ok(), Some(Some(Value::SINGLE(100))));
        assert_eq!(keeper.get(&201).ok(), Some(None));
    }

    #[test]
    fn test_many_add_multiple_times() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        for n in 0..8 {
            for i in 0..20 {
                changes.push(KeyChanges::single_add(i, i));
            }
            keeper.apply(&changes).unwrap();
            assert_eq!(keeper.get(&(n + 2)).ok(), Some(Some(Value::SINGLE(n + 2))));
        }
    }

    #[test]
    fn test_single_add_remove() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        changes.push(KeyChanges::single_add(1, 1));
        changes.push(KeyChanges::single_add(2, 2));
        changes.push(KeyChanges::single_add(3, 4));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(2))));

        let mut changes = Vec::new();
        changes.push(KeyChanges::single_delete(1, Some(1)));
        changes.push(KeyChanges::single_delete(2, Some(2)));
        changes.push(KeyChanges::single_delete(3, Some(4)));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(None));
    }

    #[test]
    fn test_aggregate_add_remove() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(
            2,
            vec![ValueChange::ADD(1), ValueChange::REMOVE(Some(1)), ValueChange::ADD(2)],
        ));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(2))));

        let mut changes = Vec::new();
        changes.push(KeyChanges::new(
            2,
            vec![
                ValueChange::REMOVE(Some(2)),
                ValueChange::ADD(1),
                ValueChange::REMOVE(Some(1)),
            ],
        ));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(None));
    }

    #[test]
    fn test_group_replace_remove() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(2, vec![ValueChange::ADD(1), ValueChange::ADD(2)]));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(2))));

        let mut changes = Vec::new();
        changes.push(KeyChanges::new(2, vec![ValueChange::REMOVE(Some(2))]));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(None));
    }

    #[test]
    fn test_group_values_remove() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new_mode(ValueMode::CLUSTER);
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(
            2,
            vec![ValueChange::ADD(1), ValueChange::ADD(2), ValueChange::ADD(3)],
        ));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::CLUSTER(vec![1, 2, 3]))));

        let mut changes = Vec::new();
        changes.push(KeyChanges::new(
            2,
            vec![ValueChange::REMOVE(Some(1)), ValueChange::REMOVE(Some(2))],
        ));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(3))));
    }

    #[test]
    fn test_group_values_remove_no_order() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new_mode(ValueMode::CLUSTER);
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(
            2,
            vec![ValueChange::ADD(3), ValueChange::ADD(1), ValueChange::ADD(2)],
        ));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::CLUSTER(vec![3, 1, 2]))));

        let mut changes = Vec::new();
        changes.push(KeyChanges::new(
            2,
            vec![ValueChange::REMOVE(Some(1)), ValueChange::REMOVE(Some(2))],
        ));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(3))));
    }

    #[test]
    fn test_add_same_value_twice() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new_mode(ValueMode::CLUSTER);
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(
            2,
            vec![ValueChange::ADD(1), ValueChange::ADD(2), ValueChange::ADD(1)],
        ));

        changes.push(KeyChanges::new(1, vec![ValueChange::ADD(1), ValueChange::ADD(1)]));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::CLUSTER(vec![1, 2]))));
        assert_eq!(keeper.get(&1).ok(), Some(Some(Value::SINGLE(1))));
    }

    #[test]
    fn test_add_to_esclusive() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new_mode(ValueMode::EXCLUSIVE);
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(2, vec![ValueChange::ADD(1)]));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(1))));
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(2, vec![ValueChange::ADD(1)]));

        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(1))));
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(2, vec![ValueChange::ADD(2)]));
        assert!(match keeper.apply(&changes) {
            Err(PersyError::IndexDuplicateKey(ref idxname, ref keyname))
                if (idxname == "test_index" && keyname == "2") =>
            {
                true
            }
            _ => false,
        });
    }

    #[test]
    fn test_group_key_remove() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new_mode(ValueMode::CLUSTER);
        let mut changes = Vec::new();
        changes.push(KeyChanges::new(2, vec![ValueChange::ADD(1), ValueChange::ADD(2)]));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::CLUSTER(vec![1, 2]))));

        let mut changes = Vec::new();
        changes.push(KeyChanges::new(2, vec![ValueChange::REMOVE(None)]));
        keeper.apply(&changes).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(None));
    }

    #[test]
    fn test_many_add_remove() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        for i in 0..200 {
            changes.push(KeyChanges::single_add(i, i));
        }
        changes.sort_by_key(|k| k.k);
        keeper.apply(&changes).unwrap();
        print_tree(&mut keeper).unwrap();
        assert_eq!(keeper.get(&2).ok(), Some(Some(Value::SINGLE(2))));
        assert_eq!(keeper.get(&100).ok(), Some(Some(Value::SINGLE(100))));
        assert_eq!(keeper.get(&201).ok(), Some(None));
        let mut changes = Vec::new();
        for i in 0..200 {
            changes.push(KeyChanges::single_delete(i, Some(i)));
        }
        changes.sort_by_key(|k| k.k);
        keeper.apply(&changes).unwrap();
        print_tree(&mut keeper).unwrap();
        assert_eq!(keeper.get_root().ok(), Some(None));
        assert_eq!(keeper.get(&2).ok(), Some(None));
        assert_eq!(keeper.get(&100).ok(), Some(None));
    }

    #[test]
    fn test_many_add_remove_multiple_times() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        let mut rchanges = Vec::new();
        for n in 0..8 {
            for i in 0..20 {
                changes.push(KeyChanges::single_add(i, i));
                rchanges.push(KeyChanges::single_delete(i, Some(i)));
            }
            changes.sort_by_key(|k| k.k);
            keeper.apply(&changes).unwrap();
            assert_eq!(keeper.get(&(n + 2)).ok(), Some(Some(Value::SINGLE(n + 2))));
            rchanges.sort_by_key(|k| k.k);
            keeper.apply(&rchanges).unwrap();
            assert_eq!(keeper.get(&(n + 2)).ok(), Some(None));
        }
    }

    #[test]
    fn test_iter_from() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        for i in 0..50 {
            changes.push(KeyChanges::single_add(i, i));
        }
        keeper.apply(&changes).unwrap();
        print_tree(&mut keeper).unwrap();
        let mut iter = keeper.iter_from(Bound::Included(&5)).unwrap();
        let next = iter.iter.next();

        assert_eq!(5, next.unwrap().key);
        let next = iter.iter.next();
        assert_eq!(6, next.unwrap().key);
        let mut last_val = None;
        for v in iter.iter {
            last_val = Some(v);
        }
        let mut next_page = keeper.iter_node(&iter.next.unwrap()).unwrap();
        let next = next_page.iter.next();
        assert_eq!(last_val.unwrap().key + 1, next.unwrap().key);
    }

    #[test]
    fn test_iter() {
        let mut keeper = MockIndexKeeper::<u8, u8>::new();
        let mut changes = Vec::new();
        for i in 0..50 {
            changes.push(KeyChanges::single_add(i, i));
        }
        keeper.apply(&changes).unwrap();
        print_tree(&mut keeper).unwrap();
        let mut iter = keeper.iter_from(Bound::Unbounded).unwrap();
        let next = iter.iter.next();

        assert_eq!(0, next.unwrap().key);
        let mut last_val = None;
        for v in iter.iter {
            last_val = Some(v);
        }
        let mut next_page = keeper.iter_node(&iter.next.unwrap()).unwrap();
        let next = next_page.iter.next();
        assert_eq!(last_val.unwrap().key + 1, next.unwrap().key);
        let mut last_val = None;
        for v in next_page.iter {
            last_val = Some(v);
        }
        assert_eq!(last_val.unwrap().key, 49);
    }

    #[test]
    fn test_a_lot_add_remove_multiple_times() {
        let mut keeper = MockIndexKeeper::<u32, u32>::new();
        let mut changes = Vec::new();
        let mut remove = Vec::new();
        for n in 1..30 {
            for i in 1..200 {
                changes.push(KeyChanges::single_add(i * n, i * n));
                if i % 2 == 0 {
                    remove.push(KeyChanges::single_delete(i * n, Some(i * n)));
                }
            }
            changes.sort_by_key(|k| k.k);
            keeper.apply(&changes).unwrap();
            assert_eq!(keeper.get(&(2 * n)).ok(), Some(Some(Value::SINGLE(2 * n))));
            assert_eq!(keeper.get(&(100 * n)).ok(), Some(Some(Value::SINGLE(100 * n))));
            assert_eq!(keeper.get(&20001).ok(), Some(None));
            remove.sort_by_key(|k| k.k);
            keeper.apply(&remove).unwrap();
            assert_eq!(keeper.get(&(2 * n)).ok(), Some(None));
            assert_eq!(keeper.get(&(100 * n)).ok(), Some(None));
        }
    }

    #[test]
    fn test_big_tree() {
        let mut keeper = MockIndexKeeper::<u32, u32>::new();
        for n in 1..20 {
            let mut changes = Vec::new();
            let mut remove = Vec::new();
            for i in 1..301 {
                changes.push(KeyChanges::single_add(i * n, i * n));
                if i % 2 == 0 {
                    remove.push(KeyChanges::single_delete(i * n, Some(i * n)));
                }
            }
            keeper.apply(&changes).unwrap();
            assert_eq!(keeper.get(&(2 * n)).ok(), Some(Some(Value::SINGLE(2 * n))));
            assert_eq!(keeper.get(&(100 * n)).ok(), Some(Some(Value::SINGLE(100 * n))));
            assert_eq!(keeper.get(&20001).ok(), Some(None));

            keeper.apply(&remove).unwrap();
            assert_eq!(keeper.get(&(2 * n)).ok(), Some(None));
            assert_eq!(keeper.get(&(100 * n)).ok(), Some(None));
        }
    }
}