mvcc_cell 0.1.2

use super::TxnId;
use std::collections::{VecDeque, BTreeSet};

#[derive(Debug)]
pub struct History<T>(VecDeque<(TxnId, T)>);

pub enum Value<T> {
    Current(T),
    Archived(T)
}

impl<T> std::ops::Deref for Value<T> {
    type Target=T;
    fn deref(&self)->&T {
        match self {
            Self::Current(x) => x,
            Self::Archived(x) => x
        }
    }
}

impl<T> History<T> {
    pub fn new(init: T)->Self {
        let mut v = VecDeque::new();
        v.push_front((0,init));
        History(v)
    }

    pub fn push(&mut self, seq: TxnId, val: T) {
        assert!(seq > self.0.front().unwrap().0);
        self.0.push_front((seq, val));
    }

    pub fn lookup(&self, seq: TxnId)->Value<&T> {
        for (i, &(v, ref val)) in self.0.iter().enumerate() {
            if v <= seq {
                return match i {
                    0 => Value::Current(val),
                    _ => Value::Archived(val)
                }
            }
        }
        unreachable!();
    }

    pub fn current(&self)->TxnId {
        self.0.front().unwrap().0
    }

    pub fn vacuum(&mut self, max: TxnId, old: &BTreeSet<TxnId>) {
        let mut next = Some(max);
        let mut keep = old.iter().rev().copied();

        self.0.retain(|&(seq, _)| {
            // Keep all revs newer than snapshot
            if seq > max { return true; }

            // If we're out of revs, discard remaining
            let Some(seek) = next else { return false; };
            if seek >= seq {
                // Found a rev to keep, skip all txns it is good for
                while let Some(seek) = next {
                    if seek < seq { break; }
                    next = keep.next();
                }
                return true;
            } else {
                return false;
            }
        });
    }

    #[allow(dead_code)]
    pub fn len(&self)->usize { self.0.len() }
}

#[cfg(test)]
mod test {
    use proptest::prelude::*;
    use super::*;

    const MAX_ID: TxnId = 1024;

    proptest!{
        #[test] fn test_history_insert(
            vals in prop::collection::btree_map(1..MAX_ID, 1u32..10_000, 0..100),
            t in 0..MAX_ID
        ) {
            let mut hist = History::new(0);
            let mut last = 0;
            let mut correct = u32::MAX;

            for seq in 0..MAX_ID {
                if let Some(&x) = vals.get(&seq) {
                    last = x;
                    hist.push(seq, x);
                }

                if t == seq { correct = last; }
            }

            assert_eq!(correct, **hist.lookup(t));
        }

        #[test] fn test_history_vacuum(
            vals in prop::collection::btree_map(1..MAX_ID, 1u32..10_000, 0..100),
            lo_reads in prop::collection::btree_set(1..(MAX_ID/2), 0..100),
            hi_reads in prop::collection::btree_set((MAX_ID/2)..MAX_ID, 0..100),
        ) {
            let mut hist = History::new(0);
            for (seq, val) in vals {
                hist.push(seq, val);
            }

            let mut correct: BTreeSet<(TxnId, u32)> = BTreeSet::new();
            for &seq in &lo_reads {
                correct.insert((seq, **hist.lookup(seq)));
            }
            for &seq in &hi_reads {
                correct.insert((seq, **hist.lookup(seq)));
            }

            hist.vacuum(MAX_ID/2, &lo_reads);

            let mut actual: BTreeSet<(TxnId, u32)> = BTreeSet::new();
            for &seq in &lo_reads {
                actual.insert((seq, **hist.lookup(seq)));
            }
            for &seq in &hi_reads {
                actual.insert((seq, **hist.lookup(seq)));
            }

            assert_eq!(correct, actual);

            hist.vacuum(MAX_ID + 1, &BTreeSet::new());

            assert_eq!(1, hist.len());
        }
    }
}