use crate::{
    address::{
        Address, ADDRESS_ENTRY_SIZE, ADDRESS_PAGE_ENTRY_COUNT, ADDRESS_PAGE_EXP, ADDRESS_PAGE_SIZE, FLAG_DELETED,
        FLAG_EXISTS, SEGMENT_DATA_OFFSET, SEGMENT_HASH_OFFSET, SEGMENT_PAGE_DELETE_COUNT_OFFSET,
    },
    allocator::Allocator,
    device::{Page, PageOps},
    error::PERes,
    id::{RecRef, SegmentId},
    io::{ArcSliceRead, InfallibleRead, InfallibleReadFormat, InfallibleWrite, InfallibleWriteFormat},
};
use std::{
    collections::{hash_map::DefaultHasher, HashMap},
    hash::{Hash, Hasher},
    str,
    sync::Arc,
    vec,
};

pub(crate) const SEGMENTS_ROOT_PAGE_VERSION_0: u8 = 0;
pub(crate) const SEGMENTS_ROOT_PAGE_VERSION: u8 = SEGMENTS_ROOT_PAGE_VERSION_0;

pub struct AllocatedSegmentPage {
    pub new_page: u64,
    pub previous_page: u64,
}

pub struct Segment {
    pub first_page: u64,
    pub last_page: u64,
    pub last_pos: u32,
    segment_id: SegmentId,
    name: String,
}

impl Segment {
    pub fn new(first_page: u64, last_page: u64, last_pos: u32, segment_id: SegmentId, name: &str) -> Segment {
        Segment {
            first_page,
            last_page,
            last_pos,
            segment_id,
            name: name.to_string(),
        }
    }

    pub fn read(read: &mut dyn InfallibleRead) -> PERes<Segment> {
        let first_page = read.read_u64();
        let persistent_page = read.read_u64();
        let persistent_pos = read.read_u32();
        let segment_id = SegmentId::read(read);
        let name_size = read.read_u16() as usize;

        let mut slice: Vec<u8> = vec![0; name_size];
        read.read_exact(&mut slice);
        let name: String = str::from_utf8(&slice[0..name_size])?.into();

        Ok(Segment::new(
            first_page,
            persistent_page,
            persistent_pos,
            segment_id,
            &name,
        ))
    }

    pub fn write(&self, write: &mut dyn InfallibleWrite) {
        write.write_u64(self.first_page);
        write.write_u64(self.last_page);
        write.write_u32(self.last_pos);
        self.segment_id.write(write);
        write.write_u16(self.name.len() as u16);
        write.write_all(self.name.as_bytes());
    }

    pub fn allocate_internal(&mut self, allocator: &Allocator) -> PERes<(RecRef, Option<AllocatedSegmentPage>)> {
        let page = self.last_page;
        let pos = self.last_pos;
        let new_pos = pos + ADDRESS_ENTRY_SIZE;
        Ok(if new_pos > ADDRESS_PAGE_SIZE {
            let mut pg = allocator.write_page(page)?;
            // It can happen that the last page has all entry deleted, we never free the last
            // page, so may be there a last page to free, but with no space, in that case is better
            // reuse the page directly and avoid to allocate another page and push the last page in
            // the back leaking it.
            if pg.empty()? {
                let prev = pg.get_prev()?;
                pg.reset()?;
                pg.set_next(0)?;
                pg.set_prev(prev)?;
                pg.set_segment_id(self.segment_id)?;
                allocator.flush_page(pg)?;
                self.last_pos = SEGMENT_DATA_OFFSET + ADDRESS_ENTRY_SIZE;
                (RecRef::new(self.last_page, SEGMENT_DATA_OFFSET), None)
            } else {
                let mut new_pg = allocator.allocate(ADDRESS_PAGE_EXP)?;
                let new_page = new_pg.get_index();
                pg.set_next(new_page)?;
                allocator.flush_page(pg)?;
                new_pg.set_next(0)?;
                new_pg.set_prev(page)?;
                new_pg.set_segment_id(self.segment_id)?;
                allocator.flush_page(new_pg)?;
                self.last_page = new_page;
                self.last_pos = SEGMENT_DATA_OFFSET + ADDRESS_ENTRY_SIZE;
                (
                    RecRef::new(self.last_page, SEGMENT_DATA_OFFSET),
                    Some(AllocatedSegmentPage {
                        new_page,
                        previous_page: page,
                    }),
                )
            }
        } else {
            self.last_pos = new_pos;
            (RecRef::new(page, pos), None)
        })
    }

    pub fn collect_segment_pages(&self, allocator: &Allocator) -> PERes<Vec<u64>> {
        let mut pages = Vec::new();
        let mut page = self.first_page;
        let last = self.last_page;
        loop {
            let mut pag = allocator.load_page(page)?;
            let next = pag.read_u64();
            let mut pos = SEGMENT_DATA_OFFSET;
            loop {
                pag.seek(pos);
                let data_page = pag.read_u64();
                let flag = pag.read_u8();
                if entry_exits(flag) {
                    pages.push(data_page);
                }
                pos += ADDRESS_ENTRY_SIZE;
                if pos > ADDRESS_PAGE_SIZE - ADDRESS_ENTRY_SIZE {
                    break;
                }
            }
            pages.push(page);
            if page == last {
                break;
            }
            page = next;
        }
        Ok(pages)
    }
}

pub(crate) trait SegmentPageRead: PageOps {
    fn segment_read_entry(&mut self, segment_id: SegmentId, pos: u32) -> Option<(u64, u16)>;
    fn segment_scan_all_entries(&mut self) -> (u64, [(u32, bool); ADDRESS_PAGE_ENTRY_COUNT as usize]);
    fn segment_first_available_pos(&mut self) -> u32;
    fn get_next(&mut self) -> PERes<u64>;
    fn get_prev(&mut self) -> PERes<u64>;
    fn empty(&mut self) -> PERes<bool>;
}
pub(crate) trait SegmentPage: SegmentPageRead {
    fn segment_insert_entry(&mut self, segment_id: SegmentId, pos: u32, record_page: u64);
    fn segment_update_entry(&mut self, segment_id: SegmentId, pos: u32, record_page: u64);
    fn segment_delete_entry(&mut self, segment_id: SegmentId, pos: u32) -> bool;
    fn set_segment_id(&mut self, id: SegmentId) -> PERes<()>;
    fn set_next(&mut self, next: u64) -> PERes<()>;
    fn set_prev(&mut self, prev: u64) -> PERes<()>;
    fn recalc_count(&mut self) -> PERes<()>;
}

fn entry_exits(flags: u8) -> bool {
    flags & FLAG_EXISTS == 1 && flags & FLAG_DELETED == 0
}

impl<T: InfallibleRead + PageOps> SegmentPageRead for T {
    fn segment_read_entry(&mut self, segment_id: SegmentId, pos: u32) -> Option<(u64, u16)> {
        self.seek(SEGMENT_HASH_OFFSET);
        let persistent_id = SegmentId::read(self);
        if persistent_id != segment_id {
            return None;
        }
        self.seek(pos);
        let record = self.read_u64();
        let flag = self.read_u8();
        let version = self.read_u16();
        if !entry_exits(flag) || record == 0 {
            None
        } else {
            Some((record, version))
        }
    }

    fn segment_scan_all_entries(&mut self) -> (u64, [(u32, bool); ADDRESS_PAGE_ENTRY_COUNT as usize]) {
        let next_page = self.read_u64();
        let mut pos = SEGMENT_DATA_OFFSET;
        let mut recs = [(0, false); ADDRESS_PAGE_ENTRY_COUNT as usize];
        let mut iter = 0;
        loop {
            self.seek(pos + 8);
            let flag = self.read_u8();
            recs[iter] = (pos, flag & FLAG_EXISTS == 1);
            pos += ADDRESS_ENTRY_SIZE;
            iter += 1;
            if pos > ADDRESS_PAGE_SIZE - ADDRESS_ENTRY_SIZE {
                break;
            }
        }
        (next_page, recs)
    }

    fn segment_first_available_pos(&mut self) -> u32 {
        let elements = (ADDRESS_PAGE_SIZE - SEGMENT_DATA_OFFSET) / ADDRESS_ENTRY_SIZE;
        let mut pos = SEGMENT_DATA_OFFSET + (elements - 1) * ADDRESS_ENTRY_SIZE;
        loop {
            self.seek(pos + 8);
            let flag = self.read_u8();
            if entry_exits(flag) {
                pos += ADDRESS_ENTRY_SIZE;
                break;
            }
            if pos == SEGMENT_DATA_OFFSET {
                break;
            }
            pos -= ADDRESS_ENTRY_SIZE;
            debug_assert!(pos >= SEGMENT_DATA_OFFSET);
        }
        pos
    }

    fn get_next(&mut self) -> PERes<u64> {
        self.seek(0);
        Ok(self.read_u64())
    }

    fn get_prev(&mut self) -> PERes<u64> {
        self.seek(8);
        Ok(self.read_u64())
    }
    fn empty(&mut self) -> PERes<bool> {
        self.seek(SEGMENT_PAGE_DELETE_COUNT_OFFSET);
        Ok(self.read_u16() as u32 == ADDRESS_PAGE_ENTRY_COUNT)
    }
}

fn inc_version(version: u16) -> u16 {
    if version == u16::MAX {
        1
    } else {
        version + 1
    }
}

impl<T: InfallibleRead + InfallibleWrite + PageOps> SegmentPage for T {
    fn segment_insert_entry(&mut self, segment_id: SegmentId, pos: u32, record_page: u64) {
        debug_assert!(pos >= SEGMENT_DATA_OFFSET, "invalid page position {}", pos);
        self.seek(SEGMENT_HASH_OFFSET);
        let persistent_id = SegmentId::read(self);
        debug_assert!(persistent_id == segment_id);
        self.seek(pos);
        self.write_u64(record_page);
        self.write_u8(FLAG_EXISTS);
        self.write_u16(1);
    }

    fn segment_update_entry(&mut self, segment_id: SegmentId, pos: u32, record_page: u64) {
        debug_assert!(pos >= SEGMENT_DATA_OFFSET);
        self.seek(SEGMENT_HASH_OFFSET);
        let persistent_id = SegmentId::read(self);
        debug_assert!(persistent_id == segment_id);
        self.seek(pos + 9);
        let version = self.read_u16();
        self.seek(pos);
        self.write_u64(record_page);
        self.seek(pos + 9);
        self.write_u16(inc_version(version));
    }

    fn segment_delete_entry(&mut self, segment_id: SegmentId, pos: u32) -> bool {
        debug_assert!(pos >= SEGMENT_DATA_OFFSET);
        self.seek(SEGMENT_HASH_OFFSET);
        let persistent_id = SegmentId::read(self);
        debug_assert!(persistent_id == segment_id);
        self.seek(SEGMENT_PAGE_DELETE_COUNT_OFFSET);
        let count = self.read_u16() + 1;
        self.seek(SEGMENT_PAGE_DELETE_COUNT_OFFSET);
        self.write_u16(count);
        self.seek(pos + 8);
        let flag = self.read_u8();
        self.seek(pos + 8);
        self.write_u8(flag | FLAG_DELETED);
        count as u32 == ADDRESS_PAGE_ENTRY_COUNT
    }

    fn set_segment_id(&mut self, id: SegmentId) -> PERes<()> {
        self.seek(SEGMENT_HASH_OFFSET);
        id.write(self);
        Ok(())
    }

    fn set_next(&mut self, next: u64) -> PERes<()> {
        self.seek(0);
        self.write_u64(next);
        Ok(())
    }

    fn set_prev(&mut self, prev: u64) -> PERes<()> {
        self.seek(8);
        self.write_u64(prev);
        Ok(())
    }

    fn recalc_count(&mut self) -> PERes<()> {
        let mut pos = SEGMENT_DATA_OFFSET;
        let mut count = 0;
        loop {
            self.seek(pos + 8);
            let flag = self.read_u8();
            if flag & FLAG_DELETED == FLAG_DELETED {
                count += 1;
            }
            pos += ADDRESS_ENTRY_SIZE;
            if pos > ADDRESS_PAGE_SIZE - ADDRESS_ENTRY_SIZE {
                break;
            }
        }
        self.seek(SEGMENT_PAGE_DELETE_COUNT_OFFSET);
        self.write_u8(count as u8);
        Ok(())
    }
}

pub struct Segments {
    pub root_page: u64,
    pub segments: HashMap<SegmentId, Segment>,
    pub segments_id: HashMap<String, SegmentId>,
    pub temp_segments: HashMap<SegmentId, Segment>,
    pub temp_segments_id: HashMap<String, SegmentId>,
}

pub fn segment_hash(segment: &str) -> u64 {
    let mut val: u64;
    let hasher = &mut DefaultHasher::new();
    segment.hash(hasher);
    val = hasher.finish();
    val <<= 32;
    val |= u64::from(rand::random::<u32>());
    val
}

impl Segments {
    pub fn new(root_page: u64, allocator: &Arc<Allocator>) -> PERes<Segments> {
        let read_buffer = allocator.read_address_buffer(root_page)?;
        let mut segments = HashMap::new();
        let mut segments_id = HashMap::new();
        if let Some(buffer) = read_buffer {
            let mut page = ArcSliceRead::new_vec(buffer);
            loop {
                let flag = page.read_u8();
                if flag == 1 {
                    let segment = Segment::read(&mut page)?;

                    let name = segment.name.clone();
                    let pers_hash = segment.segment_id;
                    segments.insert(pers_hash, segment);
                    segments_id.insert(name, pers_hash);
                } else {
                    break;
                }
            }
        }
        Ok(Segments {
            root_page,
            segments,
            segments_id,
            temp_segments: HashMap::new(),
            temp_segments_id: HashMap::new(),
        })
    }

    pub fn init(root: Page, allocator: &Allocator) -> PERes<()> {
        allocator.create_address_root(root)?;
        Ok(())
    }

    pub fn segment_id(&self, segment: &str) -> Option<SegmentId> {
        if let Some(id) = self.segments_id.get(segment) {
            self.segments.get(id).map(|x| x.segment_id)
        } else {
            None
        }
    }

    pub fn segment_by_id(&self, id: SegmentId) -> Option<&Segment> {
        self.segments.get(&id)
    }

    pub fn segment_name_by_id(&self, id: SegmentId) -> Option<String> {
        self.segments.get(&id).map(|s| s.name.clone())
    }

    pub fn segment_by_id_temp(&self, id: SegmentId) -> Option<&Segment> {
        self.temp_segments.get(&id)
    }

    pub fn has_segment(&self, segment: &str) -> bool {
        self.segments_id.contains_key(segment)
    }
    pub fn has_segment_by_id(&self, segment: &SegmentId) -> bool {
        self.segments.contains_key(segment)
    }

    pub fn create_temp_segment(&mut self, allocator: &Allocator, segment: &str) -> PERes<(SegmentId, u64)> {
        let mut allocated = allocator.allocate(ADDRESS_PAGE_EXP)?;
        let allocated_id = allocated.get_index();
        let segment_id = SegmentId::new(segment_hash(segment));
        let seg = Segment::new(allocated_id, allocated_id, SEGMENT_DATA_OFFSET, segment_id, segment);
        self.temp_segments.insert(segment_id, seg);
        self.temp_segments_id.insert(segment.to_string(), segment_id);
        allocated.write_u64(0);
        allocated.write_u64(0);
        segment_id.write(&mut allocated);
        allocator.flush_page(allocated)?;
        Ok((segment_id, allocated_id))
    }

    pub fn get_temp_segment_mut(&mut self, segment: SegmentId) -> Option<&mut Segment> {
        self.temp_segments.get_mut(&segment)
    }

    pub fn drop_temp_segment(&mut self, allocator: &Allocator, segment: SegmentId) -> PERes<()> {
        if let Some(segment) = self.temp_segments.remove(&segment) {
            self.temp_segments_id.remove(&segment.name);
            //TODO: to review this, may cause disc leaks in case of crash on rollback of segment
            // creation
            let pages = segment.collect_segment_pages(allocator)?;
            for page in pages {
                allocator.free(page)?;
            }
        }
        Ok(())
    }

    pub fn exists_real_or_temp(&self, segment: SegmentId) -> bool {
        self.segments.contains_key(&segment) || self.temp_segments.contains_key(&segment)
    }

    pub fn create_segment(&mut self, segment: SegmentId, first_page: u64) {
        if let Some(mut s) = self.temp_segments.remove(&segment) {
            // This is needed mainly for recover
            s.first_page = first_page;
            self.temp_segments_id.remove(&s.name);
            self.segments_id.insert(s.name.clone(), s.segment_id);
            self.segments.insert(s.segment_id, s);
        }
    }

    pub fn drop_segment(&mut self, segment: &str) {
        if let Some(seg) = self.segments_id.remove(segment) {
            self.segments.remove(&seg);
        }
    }

    pub fn collect_segment_pages(&self, allocator: &Allocator, segment: SegmentId) -> PERes<Vec<u64>> {
        if let Some(seg) = self.segments.get(&segment) {
            seg.collect_segment_pages(allocator)
        } else {
            Ok(Vec::new())
        }
    }

    pub fn set_first_page(&mut self, segment: SegmentId, first_page: u64) -> PERes<()> {
        if let Some(seg) = self.segments.get_mut(&segment) {
            seg.first_page = first_page;
        }
        Ok(())
    }

    pub fn confirm_allocations(&mut self, segments: &[SegmentId], allocator: &Allocator, recover: bool) -> PERes<()> {
        for seg in segments {
            // If happen that a segment is not found something went wrong and is better crash
            let segment = if let Some(s) = self.segments.get_mut(seg) {
                s
            } else if let Some(s) = self.temp_segments.get_mut(seg) {
                s
            } else {
                panic!("segment operation when segment do not exists");
            };
            if recover {
                // In case of recover recalculate the last available position
                let mut page = allocator.load_page(segment.last_page)?;
                segment.last_pos = page.segment_first_available_pos();
            }
        }
        Ok(())
    }

    pub fn recompute_last_pages(&mut self, allocator: &Allocator) -> PERes<()> {
        for (_, segment) in self.segments.iter_mut() {
            loop {
                let mut page = allocator.load_page(segment.last_page)?;
                let n_page = page.get_next()?;
                if n_page == 0 {
                    break;
                }
                segment.last_page = n_page;
            }
        }
        Ok(())
    }

    pub fn flush_segments(&mut self, allocator: &Allocator) -> PERes<()> {
        let mut buffer = Vec::<u8>::new();
        for segment in self.segments.values() {
            buffer.write_u8(1);
            segment.write(&mut buffer);
        }
        buffer.write_u8(0);
        allocator.write_address_root(self.root_page, &mut buffer, SEGMENTS_ROOT_PAGE_VERSION)?;
        Ok(())
    }

    pub fn list(&self) -> Vec<(String, SegmentId)> {
        self.segments_id.iter().map(|(n, id)| (n.clone(), *id)).collect()
    }
    pub fn snapshot_list(&self) -> Vec<(String, SegmentId, u64)> {
        self.segments
            .iter()
            .map(|(id, seg)| (seg.name.clone(), *id, seg.first_page))
            .collect()
    }
}

pub struct SegmentPageIterator {
    cur_page: u64,
    next_page: u64,
    per_page_iterator: [(u32, bool); ADDRESS_PAGE_ENTRY_COUNT as usize],
    iter_pos: usize,
    include_deleted: bool,
}

impl SegmentPageIterator {
    pub fn new(first_page: u64) -> SegmentPageIterator {
        SegmentPageIterator {
            cur_page: first_page,
            next_page: first_page,
            per_page_iterator: [(0, false); ADDRESS_PAGE_ENTRY_COUNT as usize],
            iter_pos: 0,
            include_deleted: false,
        }
    }
    pub fn snapshot(first_page: u64) -> SegmentPageIterator {
        SegmentPageIterator {
            cur_page: first_page,
            next_page: first_page,
            per_page_iterator: [(0, false); ADDRESS_PAGE_ENTRY_COUNT as usize],
            iter_pos: 0,
            include_deleted: true,
        }
    }

    pub fn next(&mut self, address: &Address) -> Option<RecRef> {
        // This loop is needed because some pages may be empty
        loop {
            if self.iter_pos < self.per_page_iterator.len() {
                let (pos, exists) = self.per_page_iterator[self.iter_pos];
                self.iter_pos += 1;
                if exists || self.include_deleted {
                    break Some(RecRef::new(self.cur_page, pos));
                } else {
                    continue;
                }
            } else if self.next_page != 0 {
                self.cur_page = self.next_page;
                if let Ok((next_page, elements)) = address.scan_page_all(self.cur_page) {
                    self.next_page = next_page;
                    self.per_page_iterator = elements;
                    self.iter_pos = 0;
                }
            } else {
                break None;
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{segment_hash, SegmentPage, SegmentPageRead, Segments};
    use crate::{
        address::ADDRESS_PAGE_EXP,
        allocator::Allocator,
        config::Config,
        device::{FileDevice, Page, PageOps},
        id::SegmentId,
    };
    use std::sync::Arc;
    use tempfile::Builder;

    fn create_allocator(file_name: &str) -> Allocator {
        let file = Builder::new()
            .prefix(file_name)
            .suffix(".persy")
            .tempfile()
            .unwrap()
            .reopen()
            .unwrap();
        let config = Arc::new(Config::new());
        let disc = Box::new(FileDevice::new(file).unwrap());
        let (_, allocator) = Allocator::init(disc, &config).unwrap();
        allocator.allocate(5).unwrap(); //Just to be sure it not start from 0, it cannot happen in not test cases.

        allocator
    }

    #[test]
    fn test_create_drop_segment() {
        let allocator = create_allocator("./raw_segment_create_delete.persy");
        let all = Arc::new(allocator);
        let root = all.allocate(ADDRESS_PAGE_EXP).unwrap();
        let root_index = root.get_index();
        Segments::init(root, &all).unwrap();
        all.disc_sync().unwrap();
        let mut segments = Segments::new(root_index, &all).unwrap();

        let (id, fp) = segments.create_temp_segment(&all, "some").unwrap();
        segments.create_segment(id, fp);
        segments.flush_segments(&all).unwrap();
        assert!(segments.segments_id.contains_key("some"));
        assert!(segments.segments.contains_key(&id));
        segments.drop_segment("some");
        segments.flush_segments(&all).unwrap();
        assert!(!segments.segments_id.contains_key("some"));
        assert!(!segments.segments.contains_key(&id));
    }

    #[test]
    fn test_create_drop_temp_segment() {
        let allocator = create_allocator("./segment_create_delete.persy");
        let all = Arc::new(allocator);
        let root = all.allocate(ADDRESS_PAGE_EXP).unwrap();
        let root_index = root.get_index();
        Segments::init(root, &all).unwrap();
        all.disc_sync().unwrap();
        let mut segments = Segments::new(root_index, &all).unwrap();

        let (id, _) = segments.create_temp_segment(&all, "some").unwrap();
        assert!(segments.temp_segments.contains_key(&id));
        assert!(segments.temp_segments_id.contains_key("some"));
        segments.drop_temp_segment(&all, id).unwrap();
        assert!(!segments.temp_segments.contains_key(&id));
        assert!(!segments.temp_segments_id.contains_key("some"));
    }

    #[test]
    fn test_create_close_drop_close_segment() {
        let allocator = create_allocator("./segment_pers_create_delete.persy");
        let all = Arc::new(allocator);
        let root = all.allocate(ADDRESS_PAGE_EXP).unwrap();
        let root_index = root.get_index();
        Segments::init(root, &all).unwrap();
        all.disc_sync().unwrap();
        let id;
        {
            let mut segments = Segments::new(root_index, &all).unwrap();
            let (c_id, fp) = segments.create_temp_segment(&all, "some").unwrap();
            id = c_id;
            segments.create_segment(id, fp);
            segments.flush_segments(&all).unwrap();
            all.disc_sync().unwrap();
        }
        {
            let mut segments = Segments::new(root_index, &all).unwrap();
            assert_eq!(segments.segments.len(), 1);
            assert!(segments.segments_id.contains_key("some"));
            assert!(segments.segments.contains_key(&id));
            segments.drop_segment("some");
            segments.flush_segments(&all).unwrap();
            all.disc_sync().unwrap();
        }
        {
            let segments = Segments::new(root_index, &all).unwrap();
            assert!(!segments.segments_id.contains_key("some"));
            assert!(!segments.segments.contains_key(&id));
        }
    }

    #[test]
    fn test_create_close_drop_close_segment_off_page() {
        let allocator = create_allocator("./segment_pers_create_delete_off_page.persy");
        let all = Arc::new(allocator);
        let root = all.allocate(ADDRESS_PAGE_EXP).unwrap();
        let root_index = root.get_index();
        Segments::init(root, &all).unwrap();
        all.disc_sync().unwrap();
        {
            let mut segments = Segments::new(root_index, &all).unwrap();
            for i in 0..100 {
                let (id, fp) = segments.create_temp_segment(&all, &format!("some{}", i)).unwrap();
                segments.create_segment(id, fp);
            }
            segments.flush_segments(&all).unwrap();
            all.disc_sync().unwrap();
        }
        {
            let mut segments = Segments::new(root_index, &all).unwrap();
            for i in 0..100 {
                assert!(segments.segments_id.contains_key(&format!("some{}", i)));
                segments.drop_segment(&format!("some{}", i));
            }
            segments.flush_segments(&all).unwrap();
            all.disc_sync().unwrap();
        }
        {
            let segments = Segments::new(root_index, &all).unwrap();
            for i in 0..100 {
                assert!(!segments.segments_id.contains_key(&format!("some{}", i)));
            }
        }
    }

    #[test]
    fn test_seg_insert_read_pointer() {
        let mut page = Page::new(vec![0; 1024], 0, 0, 10);
        let segment_id = SegmentId::new(0);
        page.segment_insert_entry(segment_id, 30, 10);
        let read = page.segment_read_entry(segment_id, 30);
        match read {
            Some(val) => assert_eq!(val.0, 10),
            None => assert!(false),
        }
    }

    #[test]
    fn test_seg_insert_update_read_pointer() {
        let mut page = Page::new(vec![0; 1024], 0, 0, 10);
        let segment_id = SegmentId::new(0);
        page.segment_insert_entry(segment_id, 30, 10);
        page.segment_update_entry(segment_id, 30, 15);
        let read = page.segment_read_entry(segment_id, 30);
        match read {
            Some(val) => assert_eq!(val.0, 15),
            None => assert!(false),
        }
    }

    #[test]
    fn test_seg_insert_delete_read_pointer() {
        let mut page = Page::new(vec![0; 1024], 0, 0, 10);
        let segment_id = SegmentId::new(0);
        page.segment_insert_entry(segment_id, 30, 10);
        page.segment_delete_entry(segment_id, 30);
        let read = page.segment_read_entry(segment_id, 30);
        match read {
            Some(_) => assert!(false),
            None => assert!(true),
        }
    }

    #[test]
    fn test_hash_id_generator() {
        assert!(0 != segment_hash("some"));
    }
}