use crate::{
    error::{PRes, PersyError},
    id::{PersyId, RecRef},
    index::{
        config::{ByteVec, IndexType, IndexTypeId},
        keeper::Container,
        tree::nodes::{Leaf, LeafEntry, Node, NodeRef, Nodes, Value},
    },
    io::{
        InfallibleRead, InfallibleReadFormat, InfallibleReadVarInt, InfallibleWrite, InfallibleWriteFormat,
        InfallibleWriteVarInt,
    },
};
use std::io::Cursor;

macro_rules! impl_index_type {
    ($t:ty, $v:expr,$v1:ident,$w:ty) => {
        impl IndexType for $t {
            type Wrapper = Container<Self>;
            fn get_id() -> u8 {
                $v
            }
            fn get_type_id() -> IndexTypeId {
                IndexTypeId::$v1
            }
        }
    };
}

impl_index_type!(u8, 1, U8, u8);
impl_index_type!(u16, 2, U16, u16);
impl_index_type!(u32, 3, U32, u32);
impl_index_type!(u64, 4, U64, u64);
impl_index_type!(u128, 14, U128, u128);
impl_index_type!(i8, 5, I8, i8);
impl_index_type!(i16, 6, I16, i16);
impl_index_type!(i32, 7, I32, i32);
impl_index_type!(i64, 8, I64, i64);
impl_index_type!(i128, 15, I128, i128);
impl_index_type!(f32, 9, F32W, F32W);
impl_index_type!(f64, 10, F64W, F64W);
impl_index_type!(String, 12, STRING, String);
impl_index_type!(PersyId, 13, PERSYID, PersyId);
impl_index_type!(ByteVec, 16, BYTEVEC, ByteVec);

pub trait IndexSerialization: Sized {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()>;
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self>;
}

pub fn deserialize<K: IndexType, V: IndexType>(value: &[u8]) -> PRes<Node<K, V>> {
    let version = value[0];
    match version {
        0u8 => deserialize_v0(&value[1..]),
        _ => panic!("not compatible disc version"),
    }
}

pub fn deserialize_v0<K: IndexType, V: IndexType>(value: &[u8]) -> PRes<Node<K, V>> {
    let mut reader = Cursor::new(value);
    let t = reader.read_varint_u8();
    match t {
        1 => {
            let prev = if reader.read_varint_u8() == 1 {
                Some(K::deserialize(&mut reader)?)
            } else {
                None
            };
            let size = reader.read_varint_u32();
            let mut entries = Vec::with_capacity(size as usize);
            for _ in 0..size {
                let key = K::deserialize(&mut reader)?;
                let value_type = reader.read_varint_u8();
                if value_type == 1 {
                    let val_size = reader.read_varint_u32();
                    let mut value = Vec::with_capacity(val_size as usize);
                    for _ in 0..val_size {
                        value.push(V::deserialize(&mut reader)?);
                    }
                    entries.push(LeafEntry {
                        key,
                        value: Value::CLUSTER(value),
                    });
                } else {
                    let value = V::deserialize(&mut reader)?;
                    entries.push(LeafEntry {
                        key,
                        value: Value::SINGLE(value),
                    });
                }
            }
            let next = if reader.read_varint_u8() == 1 {
                Some(K::deserialize(&mut reader)?)
            } else {
                None
            };
            Ok(Node::LEAF(Leaf { entries, prev, next }))
        }
        2 => {
            let prev = if reader.read_varint_u8() == 1 {
                Some(K::deserialize(&mut reader)?)
            } else {
                None
            };
            let size = reader.read_varint_u32();
            let mut keys = Vec::with_capacity(size as usize);
            for _ in 0..size {
                let key = K::deserialize(&mut reader)?;
                keys.push(key);
            }
            let size = reader.read_varint_u32();
            let mut pointers = Vec::with_capacity(size as usize);
            for _ in 0..size {
                let page = reader.read_varint_u64();
                let pos = reader.read_varint_u32();
                pointers.push(RecRef::new(page, pos));
            }
            let next = if reader.read_varint_u8() == 1 {
                Some(K::deserialize(&mut reader)?)
            } else {
                None
            };
            Ok(Node::NODE(Nodes {
                keys,
                pointers,
                prev,
                next,
            }))
        }
        _ => panic!("error on index node deserialization"),
    }
}

pub fn serialize<K: IndexType, V: IndexType>(node: &Node<K, V>) -> PRes<Vec<u8>> {
    let mut dest = Vec::new();
    dest.push(0u8);
    serialize_v0(node, &mut dest)?;
    Ok(dest)
}
pub fn serialize_v0<K: IndexType, V: IndexType>(node: &Node<K, V>, dest: &mut dyn InfallibleWrite) -> PRes<()> {
    let write_page_and_pos = |dest: &mut dyn InfallibleWrite, page, pos| -> PRes<()> {
        dest.write_varint_u64(page);
        dest.write_varint_u32(pos);
        Ok(())
    };

    let write_opt_leafptr = |dest: &mut dyn InfallibleWrite, x: Option<&NodeRef>| -> PRes<()> {
        if let Some(y) = x {
            write_page_and_pos(dest, y.page, y.pos)?;
        } else {
            write_page_and_pos(dest, 0, 0)?;
        }
        Ok(())
    };

    match node {
        Node::LEAF(leaf) => {
            dest.write_varint_u8(1);
            if let Some(pk) = &leaf.prev {
                dest.write_varint_u8(1);
                pk.serialize(dest)?;
            } else {
                dest.write_varint_u8(0);
            }
            dest.write_varint_u32(leaf.entries.len() as u32);
            for entry in &leaf.entries {
                entry.key.serialize(dest)?;
                match &entry.value {
                    Value::CLUSTER(cluster) => {
                        dest.write_varint_u8(1);
                        dest.write_varint_u32(cluster.len() as u32);
                        for val in cluster {
                            val.serialize(dest)?;
                        }
                    }
                    Value::SINGLE(val) => {
                        dest.write_varint_u8(2);
                        val.serialize(dest)?;
                    }
                }
            }
            if let Some(pk) = &leaf.next {
                dest.write_varint_u8(1);
                pk.serialize(dest)?;
            } else {
                dest.write_varint_u8(0);
            }
        }
        Node::NODE(node) => {
            dest.write_varint_u8(2);
            if let Some(pk) = &node.prev {
                dest.write_varint_u8(1);
                pk.serialize(dest)?;
            } else {
                dest.write_varint_u8(0);
            }
            dest.write_varint_u32(node.keys.len() as u32);
            for k in &node.keys {
                k.serialize(dest)?;
            }
            dest.write_varint_u32(node.pointers.len() as u32);
            for p in &node.pointers {
                write_opt_leafptr(dest, Some(p))?;
            }
            if let Some(pk) = &node.next {
                dest.write_varint_u8(1);
                pk.serialize(dest)?;
            } else {
                dest.write_varint_u8(0);
            }
        }
    }
    Ok(())
}

impl IndexSerialization for u8 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_u8(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_u8())
    }
}

impl IndexSerialization for u16 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_u16(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_u16())
    }
}

impl IndexSerialization for u32 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_u32(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_u32())
    }
}

impl IndexSerialization for u64 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_u64(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_u64())
    }
}

impl IndexSerialization for u128 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_u128(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_u128())
    }
}

impl IndexSerialization for i8 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_i8(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_i8())
    }
}

impl IndexSerialization for i16 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_i16(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_i16())
    }
}

impl IndexSerialization for i32 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_i32(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_i32())
    }
}

impl IndexSerialization for i64 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_i64(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_i64())
    }
}

impl IndexSerialization for i128 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_i128(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_varint_i128())
    }
}

impl IndexSerialization for f32 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_f32(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_f32())
    }
}

impl IndexSerialization for f64 {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_f64(*self);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        Ok(value.read_f64())
    }
}

impl IndexSerialization for PersyId {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_u64(self.0.page);
        buffer.write_varint_u32(self.0.pos);
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        let page = value.read_varint_u64();
        let pos = value.read_varint_u32();
        Ok(PersyId(RecRef::new(page, pos)))
    }
}

impl IndexSerialization for String {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_u32(self.len() as u32);
        buffer.write_all(self.as_bytes());
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        let string_size = value.read_varint_u32();
        let mut buff = vec![0; string_size as usize];
        value.read_exact(&mut buff);
        let string = String::from_utf8(buff).map_err(|e| PersyError::DecodingUtf8(e.utf8_error()))?;
        Ok(string)
    }
}

impl IndexSerialization for ByteVec {
    fn serialize(&self, buffer: &mut dyn InfallibleWrite) -> PRes<()> {
        buffer.write_varint_u32(self.0.len() as u32);
        buffer.write_all(self.0.as_slice());
        Ok(())
    }
    fn deserialize(value: &mut dyn InfallibleRead) -> PRes<Self> {
        let vec_size = value.read_varint_u32();
        let mut slice: Vec<u8> = vec![0; vec_size as usize];
        value.read_exact(&mut slice);
        Ok(ByteVec(slice))
    }
}

#[cfg(test)]
mod tests {
    use super::{deserialize, serialize};
    use crate::id::{PersyId, RecRef};
    use crate::index::{
        config::{ByteVec, IndexType, ValueMode},
        tree::nodes::{compare, Leaf, Node, NodeRef, Nodes, Value},
    };
    use rand::random;
    use std::{cmp::Ordering, fmt::Debug};

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

    #[test]
    fn test_serialization_deserialization_nodes() {
        let val1 = random_pointer();
        let val2 = random_pointer();
        let val3 = 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 value = serialize::<u8, u8>(&Node::NODE(node)).expect("serialization works");
        let read = deserialize::<u8, u8>(&value).expect("deserialzie successfully");
        match read {
            Node::NODE(n) => {
                assert_eq!(n.keys.len(), 2);
                assert_eq!(n.pointers.len(), 3);
            }
            _ => panic!("expected a node"),
        }
    }

    fn single_type_leaf_test<K: IndexType + Debug, V: IndexType + Debug>(key: K, value: V, value1: V) {
        let mut leaf = Leaf::new();
        leaf.insert_or_update(&key, &value, ValueMode::REPLACE, "deserialization error")
            .expect("insert work");
        let binary = serialize::<K, V>(&Node::LEAF(leaf)).expect("serialization works");
        let read = deserialize::<K, V>(&binary).expect("deserialize successfully");
        match read {
            Node::LEAF(n) => {
                assert_eq!(n.entries.len(), 1);
                match n.entries[0].value {
                    Value::SINGLE(ref iv) => assert_eq!(compare(iv, &value), Ordering::Equal),
                    _ => panic!("expected SINGLE"),
                }
            }
            _ => panic!("expected a leaf"),
        }
        let mut leaf_many = Leaf::new();
        leaf_many
            .insert_or_update(&key, &value, ValueMode::CLUSTER, "deserialization error")
            .expect("insert work");
        leaf_many
            .insert_or_update(&key, &value1, ValueMode::CLUSTER, "deserialization error")
            .expect("insert work");
        let binary = serialize::<K, V>(&Node::LEAF(leaf_many)).expect("serialization works");
        let read = deserialize::<K, V>(&binary).expect("deserialize successfully");
        match read {
            Node::LEAF(n) => {
                assert_eq!(n.entries.len(), 1);
                match n.entries[0].value {
                    Value::CLUSTER(ref iv) => {
                        assert_eq!(compare(&iv[0], &value), Ordering::Equal);
                        assert_eq!(compare(&iv[1], &value1), Ordering::Equal);
                    }
                    _ => panic!("expected CLUSTER"),
                }
            }
            _ => panic!("expected a leaf"),
        }
    }

    #[test]
    fn test_serialization_deserialization_leafs() {
        single_type_leaf_test::<u8, u8>(20, 10, 20);
        single_type_leaf_test::<u16, u16>(20, 10, 20);
        single_type_leaf_test::<u32, u32>(20, 10, 20);
        single_type_leaf_test::<u64, u64>(20, 10, 20);
        single_type_leaf_test::<u128, u128>(20, 10, 20);
        single_type_leaf_test::<i8, i8>(20, 10, 20);
        single_type_leaf_test::<i16, i16>(20, 10, 20);
        single_type_leaf_test::<i32, i32>(20, 10, 20);
        single_type_leaf_test::<i64, i64>(20, 10, 20);
        single_type_leaf_test::<i128, i128>(20, 10, 20);
        single_type_leaf_test::<f32, f32>(20.0, 10.0, 20.0);
        single_type_leaf_test::<f64, f64>(20.0, 10.0, 20.0);
        single_type_leaf_test::<String, String>("o".to_string(), "a".to_string(), "b".to_string());
        single_type_leaf_test::<i32, String>(10, "a".to_string(), "b".to_string());
        single_type_leaf_test::<String, i32>("a".to_string(), 10, 20);
        single_type_leaf_test::<String, ByteVec>("a".to_string(), vec![0, 1].into(), vec![2, 10].into());
        let id = PersyId(RecRef::new(10, 20));
        let id1 = PersyId(RecRef::new(20, 20));
        let id2 = PersyId(RecRef::new(30, 20));
        single_type_leaf_test::<PersyId, PersyId>(id, id1, id2);
    }
}