persy 1.5.2

use crate::{
    id::{RecRef, SegmentId},
    journal::{records::FreedPage, JournalId},
    snapshots::{SnapshotId, Snapshots},
};

use std::{
    collections::HashMap,
    sync::{Arc, Weak},
};

#[derive(Clone, Debug)]
pub struct SnapshotEntry {
    id: RecRef,
    case: EntryCase,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Change {
    pub pos: u64,
    pub version: u16,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum EntryCase {
    Change(Change),
    Insert,
}

impl SnapshotEntry {
    pub fn change(id: &RecRef, pos: u64, version: u16) -> SnapshotEntry {
        SnapshotEntry {
            id: *id,
            case: EntryCase::Change(Change { pos, version }),
        }
    }
    pub fn insert(id: &RecRef) -> SnapshotEntry {
        SnapshotEntry {
            id: *id,
            case: EntryCase::Insert,
        }
    }
    pub fn rec_id(&self) -> &RecRef {
        &self.id
    }
    pub(crate) fn case(&self) -> &EntryCase {
        &self.case
    }
}

#[derive(Clone, Debug)]
pub struct SegmentSnapshot {
    name: String,
    id: SegmentId,
    first_page: u64,
}
impl SegmentSnapshot {
    pub fn new(name: &str, id: SegmentId, first_page: u64) -> SegmentSnapshot {
        SegmentSnapshot {
            name: name.to_string(),
            id,
            first_page,
        }
    }
    pub(crate) fn name(&self) -> &str {
        &self.name
    }
    pub(crate) fn segment_id(&self) -> SegmentId {
        self.id
    }
    pub(crate) fn first_page(&self) -> u64 {
        self.first_page
    }
}
#[derive(Clone, Debug)]
pub struct CleanInfo {
    freed_pages: Vec<u64>,
    remove_page_address: Vec<(SegmentId, u64)>,
}
impl CleanInfo {
    pub(crate) fn new(freed_pages: Vec<FreedPage>, remove_page_address: Vec<(SegmentId, u64)>) -> Self {
        Self {
            freed_pages: freed_pages.iter().map(|x| x.page).collect(),
            remove_page_address,
        }
    }
    pub(crate) fn freed_pages(&self) -> &[u64] {
        &self.freed_pages
    }
    pub(crate) fn segment_pages(&self) -> &[(SegmentId, u64)] {
        &self.remove_page_address
    }
}

#[derive(Clone)]
pub(crate) struct PendingClean {
    ops: Weak<Snapshots>,
    journal_id: Option<JournalId>,
    freed_pages: Option<CleanInfo>,
}

impl PendingClean {
    fn new(ops: &Arc<Snapshots>, journal_id: Option<JournalId>, freed_pages: Option<CleanInfo>) -> Self {
        Self {
            ops: Arc::downgrade(ops),
            journal_id,
            freed_pages,
        }
    }
}
impl Drop for PendingClean {
    fn drop(&mut self) {
        if let Some(ops) = self.ops.upgrade() {
            ops.free_resources(&self.journal_id, &self.freed_pages).unwrap();
        }
    }
}

#[derive(Clone)]
pub struct SnapshotData {
    snapshot_id: SnapshotId,
    entries: Option<Vec<SnapshotEntry>>,
    journal_id: Option<JournalId>,
    freed_pages: Option<CleanInfo>,
    pending_clean: Option<Arc<PendingClean>>,
    segments: Option<HashMap<SegmentId, SegmentSnapshot>>,
    segments_name: Option<HashMap<String, SegmentSnapshot>>,
    reference_count: u32,
}

impl SnapshotData {
    pub(crate) fn new(id: SnapshotId, reference_count: u32) -> Self {
        Self {
            snapshot_id: id,
            entries: None,
            journal_id: None,
            freed_pages: None,
            pending_clean: None,
            segments: None,
            segments_name: None,
            reference_count,
        }
    }
    pub(crate) fn fill(
        &mut self,
        segs_id: HashMap<SegmentId, SegmentSnapshot>,
        segs_name: HashMap<String, SegmentSnapshot>,
    ) {
        self.segments = Some(segs_id);
        self.segments_name = Some(segs_name);
    }
    pub(crate) fn fill_records(&mut self, journal_id: JournalId, entries: Vec<SnapshotEntry>) {
        self.entries = Some(entries);
        self.journal_id = Some(journal_id);
    }

    pub(crate) fn fill_clean_info(&mut self, clean: CleanInfo) {
        self.freed_pages = Some(clean);
    }

    pub(crate) fn pending_clean(&mut self, snapshots: &Arc<Snapshots>) -> Arc<PendingClean> {
        if let Some(pc) = &mut self.pending_clean {
            pc.clone()
        } else {
            let pc = Arc::new(PendingClean::new(
                snapshots,
                self.journal_id.take(),
                self.freed_pages.take(),
            ));
            self.pending_clean = Some(pc.clone());
            pc
        }
    }

    pub fn id(&self) -> SnapshotId {
        self.snapshot_id
    }
    pub(crate) fn segments_list(&self) -> Option<Vec<(String, SegmentId)>> {
        self.segments.as_ref().map(|segs| {
            segs.values()
                .map(|data| (data.name().to_owned(), data.segment_id()))
                .collect::<Vec<_>>()
        })
    }
    pub(crate) fn find_segment(&self, segment: &str) -> Option<SegmentId> {
        self.segments_name
            .as_ref()
            .and_then(|segs| segs.get(segment).map(|sd| sd.segment_id()))
    }
    pub(crate) fn find_segment_snapsthot(&self, segment_id: SegmentId) -> Option<&SegmentSnapshot> {
        self.segments.as_ref().and_then(|segs| segs.get(&segment_id))
    }
    pub(crate) fn release(&mut self) -> bool {
        self.reference_count -= 1;
        self.reference_count == 0
    }
    pub(crate) fn acquire(&mut self) {
        self.reference_count += 1;
    }
    pub(crate) fn take_entries(&mut self) -> Option<Vec<SnapshotEntry>> {
        self.entries.take()
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RecordVersion {
    snapshot_id: SnapshotId,
    case: EntryCase,
}

impl RecordVersion {
    pub(crate) fn new(snapshot_id: SnapshotId, case: EntryCase) -> Self {
        Self { snapshot_id, case }
    }
    pub(crate) fn id(&self) -> SnapshotId {
        self.snapshot_id
    }
    pub(crate) fn case(&self) -> &EntryCase {
        &self.case
    }
}