mod helpers;

use helpers::{create_and_drop, CountDown};
use persy::{PrepareError, UpdateError, PE::PE};
use std::sync::Arc;
use std::thread;

#[test]
fn test_create_drop_segment() {
    create_and_drop("cds", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        assert!(persy.exists_segment("test").unwrap());
        let mut tx = persy.begin().unwrap();
        tx.drop_segment("test").unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
        assert!(!persy.exists_segment("test").unwrap());
    });
}

#[test]
fn test_create_drop_double_segments_and_insert_data() {
    create_and_drop("cdss", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        tx.create_segment("test1").unwrap();
        let bytes = String::from("some").into_bytes();
        tx.insert("test", &bytes).unwrap();

        let bytes = String::from("some").into_bytes();
        tx.insert("test1", &bytes).unwrap();
        {
            let scanner = tx.scan("test").unwrap();
            assert_eq!(1, scanner.into_iter().count());
        }

        let scanner = tx.scan("test1").unwrap();
        assert_eq!(1, scanner.into_iter().count());
    });
}

#[test]
fn test_create_drop_segment_same_tx() {
    create_and_drop("cdss", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        assert!(tx.exists_segment("test").unwrap());
        let err = tx.drop_segment("test");
        assert!(err.is_err());
    });
}

#[test]
fn test_create_drop_recreate_segment_same_tx() {
    create_and_drop("cdcs", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        assert!(tx.exists_segment("test").unwrap());
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        tx.drop_segment("test").unwrap();
        assert!(!tx.exists_segment("test").unwrap());
        tx.create_segment("test").unwrap();
        assert!(tx.exists_segment("test").unwrap());
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
        assert!(persy.exists_segment("test").unwrap());
    });
}

#[test]
fn test_update_record_of_dropped_segment_other_tx() {
    create_and_drop("urds", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("some").into_bytes();
        let rec = tx.insert("test", &bytes).unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        tx.update("test", &rec, &bytes).unwrap();

        let mut tx1 = persy.begin().unwrap();
        tx1.drop_segment("test").unwrap();
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();

        let finalizer = tx.prepare();
        assert!(finalizer.is_err());
    });
}

#[test]
fn test_update_record_of_dropped_recreated_segment_other_tx() {
    create_and_drop("urdcs", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("some").into_bytes();
        let rec = tx.insert("test", &bytes).unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        tx.update("test", &rec, &bytes).unwrap();

        let mut tx1 = persy.begin().unwrap();
        tx1.drop_segment("test").unwrap();
        tx1.create_segment("test").unwrap();
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();

        let finalizer = tx.prepare();
        assert!(finalizer.is_err());
    });
}

#[test]
fn test_delete_record_of_dropped_segment_other_tx() {
    create_and_drop("drds", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("some").into_bytes();
        let rec = tx.insert("test", &bytes).unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        tx.delete("test", &rec).unwrap();

        let mut tx1 = persy.begin().unwrap();
        tx1.drop_segment("test").unwrap();
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();

        let finalizer = tx.prepare();
        assert!(finalizer.is_err());
    });
}

#[test]
fn test_delete_record_of_dropped_created_segment_other_tx() {
    create_and_drop("drdcs", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("some").into_bytes();
        let rec = tx.insert("test", &bytes).unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        tx.delete("test", &rec).unwrap();

        let mut tx1 = persy.begin().unwrap();
        tx1.drop_segment("test").unwrap();
        tx1.create_segment("test").unwrap();
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();

        let finalizer = tx.prepare();
        assert!(finalizer.is_err());
    });
}

#[test]
fn test_insert_record_of_dropped_recreated_segment_other_tx() {
    create_and_drop("irdcs", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("some").into_bytes();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        tx.insert("test", &bytes).unwrap();

        let mut tx1 = persy.begin().unwrap();
        tx1.drop_segment("test").unwrap();
        tx1.create_segment("test").unwrap();
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();

        let finalizer = tx.prepare();
        assert!(finalizer.is_err());
    });
}

#[test]
fn test_insert_record_of_dropped_segment_other_tx() {
    create_and_drop("irds", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("some").into_bytes();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        tx.insert("test", &bytes).unwrap();

        let mut tx1 = persy.begin().unwrap();
        tx1.drop_segment("test").unwrap();
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();

        let finalizer = tx.prepare();
        assert!(finalizer.is_err());
    });
}

#[test]
fn test_record_of_drop_segment_same_tx() {
    create_and_drop("rds", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        let bytes = String::from("some").into_bytes();
        let id = tx.insert("test", &bytes).expect("insert record works");
        tx.drop_segment("test").unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        assert!(match tx.update("test", &id, &String::from("none").into_bytes()) {
            Err(PE(UpdateError::RecordNotFound(ref id2))) if *id2 == id => true,
            _ => false,
        });
        assert_eq!(tx.read("test", &id).ok(), Some(None));

        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
    });
}

#[test]
pub fn test_councurrent_double_create() {
    create_and_drop("concurrent_segment_double_create.", |persy| {
        let both_create = Arc::new(CountDown::new(2));
        let end = Arc::new(CountDown::new(2));
        for _ in [0; 2].iter() {
            let both_create_moved = both_create.clone();
            let end_moved = end.clone();
            let persy = persy.clone();
            thread::spawn(move || {
                let mut tx = persy.begin().expect("error on transaction begin");
                tx.create_segment("def").expect("error on segment creation");
                both_create_moved.count_down().expect("lock not panic");
                both_create_moved.wait().expect("thread wait the other");
                if let Ok(fin) = tx.prepare() {
                    fin.commit().expect("error on commit");
                }
                end_moved.count_down().expect("lock not panic");
            });
        }

        end.wait().expect("threas finisced");
        assert!(persy.exists_segment("def").unwrap());
    });
}

#[test]
pub fn test_councurrent_double_drop() {
    create_and_drop("concurrent_segment_double_drop.", |persy| {
        let mut tx = persy.begin().expect("error on transaction begin");
        tx.create_segment("def").expect("error on segment creation");
        let fin = tx.prepare().expect("error on commit prepare");
        fin.commit().expect("error on commit");
        let both_create = Arc::new(CountDown::new(2));
        let end = Arc::new(CountDown::new(2));
        for _ in [0; 2].iter() {
            let both_create_moved = both_create.clone();
            let end_moved = end.clone();
            let persy = persy.clone();
            thread::spawn(move || {
                let mut tx = persy.begin().expect("error on transaction begin");
                tx.drop_segment("def").expect("error on segment creation");
                both_create_moved.count_down().expect("lock not panic");
                both_create_moved.wait().expect("thread wait the other");
                let fin = tx.prepare().expect("error on commit prepare");
                fin.commit().expect("error on commit");
                end_moved.count_down().expect("lock not panic");
            });
        }

        end.wait().expect("threas finisced");
        assert!(!persy.exists_segment("def").unwrap());
    });
}

#[test]
pub fn test_list_segments() {
    create_and_drop("test_list_segments", |persy| {
        let mut tx = persy.begin().expect("error on transaction begin");
        tx.create_segment("def").expect("error on segment creation");
        tx.create_segment("two").expect("error on segment creation");
        let fin = tx.prepare().expect("error on commit prepare");
        fin.commit().expect("error on commit");
        let segments = persy.list_segments().expect("list segments works as expected");
        assert_eq!(segments.len(), 2);
        let names = segments.into_iter().map(|(name, _id)| name).collect::<Vec<String>>();
        assert!(names.contains(&"def".to_string()));
        assert!(names.contains(&"two".to_string()));
    });
}

#[test]
pub fn test_list_segments_tx() {
    create_and_drop("test_list_segments", |persy| {
        let mut tx = persy.begin().expect("error on transaction begin");
        tx.create_segment("def").expect("error on segment creation");
        tx.create_segment("two").expect("error on segment creation");
        let fin = tx.prepare().expect("error on commit prepare");
        fin.commit().expect("error on commit");

        let mut tx = persy.begin().expect("error on transaction begin");
        tx.drop_segment("two").expect("error on segment drop");
        tx.create_segment("three").expect("error on segment creation");
        let segments = tx.list_segments().expect("list segments works as expected");
        assert_eq!(segments.len(), 2);
        let names = segments.into_iter().map(|x| x.0).collect::<Vec<String>>();
        assert!(names.contains(&"def".to_string()));
        assert!(names.contains(&"three".to_string()));
    });
}

#[test]
fn test_double_create_concurrent() {
    create_and_drop("cdcs", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        assert!(tx.exists_segment("test").unwrap());

        let mut tx1 = persy.begin().unwrap();
        assert!(!tx1.exists_segment("test").unwrap());
        tx1.create_segment("test").unwrap();
        assert!(tx1.exists_segment("test").unwrap());
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();
        match tx.prepare() {
            Err(persy::PE::PE(PrepareError::SegmentAlreadyExists)) => {}
            _ => panic!("Expect fail for duplicate segment"),
        }
        assert!(persy.exists_segment("test").unwrap());
    });
}