mod helpers;
use helpers::{create_and_drop, create_and_drop_with_config};
use persy::Config;
use persy::SegmentError;
use persy::TxStrategy;

#[test]
fn test_insert_read_commit_record() {
    create_and_drop("irc", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        let id = tx.insert("test", &bytes).unwrap();
        let read_opt = tx.read("test", &id).unwrap();
        assert_eq!(read_opt, Some(bytes));
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let bytes = String::from("something").into_bytes();
        let read_after = persy.read("test", &id).unwrap();
        assert_eq!(read_after, Some(bytes));
    });
}

#[test]
fn test_insert_update_record() {
    create_and_drop("iru", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let rec_data: String = "something".into();
        let bytes = rec_data.into_bytes();
        let id = tx.insert("test", &bytes).unwrap();
        let read_opt = tx.read("test", &id).unwrap();
        assert_eq!(read_opt, Some(bytes));
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let rec_1: String = "newthing".into();
        let bytes_1 = rec_1.into_bytes();
        let mut tx1 = persy.begin().unwrap();
        tx1.update("test", &id, &bytes_1).unwrap();
        let read_after = tx1.read("test", &id).unwrap();
        assert_eq!(read_after, Some(bytes_1));
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();
    });
}

#[test]
fn test_insert_delete_record() {
    create_and_drop("ird", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let rec_data: String = "something".into();
        let bytes = rec_data.into_bytes();
        let id = tx.insert("test", &bytes).unwrap();
        let read_opt = tx.read("test", &id).unwrap();
        assert_eq!(read_opt, Some(bytes));
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx1 = persy.begin().unwrap();
        tx1.delete("test", &id).unwrap();
        let read_after = tx1.read("test", &id).unwrap();
        assert_eq!(read_after, None);
        let finalizer = tx1.prepare().unwrap();
        finalizer.commit().unwrap();
    });
}

#[test]
fn test_insert_scan() {
    create_and_drop("irs", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let rec_data: String = "something".into();
        let bytes = rec_data.into_bytes();
        tx.insert("test", &bytes).unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
        let iter = persy.scan("test").unwrap().into_iter();
        assert_eq!(iter.count(), 1);
        let (_, content) = persy.scan("test").unwrap().into_iter().next().unwrap();
        assert_eq!(bytes, content);
    });
}

#[test]
fn test_insert_update_same_tx() {
    create_and_drop("ius", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        let id = tx.insert("test", &bytes).unwrap();
        let read_opt = tx.read("test", &id).unwrap();
        assert_eq!(read_opt, Some(bytes));
        let bytes = String::from("data1").into_bytes();
        tx.update("test", &id, &bytes).unwrap();
        let read_opt = tx.read("test", &id).unwrap();
        assert_eq!(read_opt, Some(bytes));
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let read_after = persy.read("test", &id).unwrap();
        let bytes = String::from("data1").into_bytes();
        assert_eq!(read_after, Some(bytes));
    });
}

#[test]
fn test_double_update_same_tx() {
    create_and_drop("iuus", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        let id = tx.insert("test", &bytes).unwrap();
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
        let mut tx = persy.begin().unwrap();
        let bytes = String::from("first").into_bytes();
        tx.update("test", &id, &bytes).unwrap();
        let bytes = String::from("second").into_bytes();
        tx.update("test", &id, &bytes).unwrap();
        let read_opt = tx.read("test", &id).unwrap();
        assert_eq!(read_opt, Some(bytes));
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let read_after = persy.read("test", &id).unwrap();
        let bytes = String::from("second").into_bytes();
        assert_eq!(read_after, Some(bytes));
    });
}

#[test]
fn test_insert_update_delete_same_tx() {
    create_and_drop("iuds", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        let id = tx.insert("test", &bytes).unwrap();
        let bytes = String::from("first").into_bytes();
        tx.update("test", &id, &bytes).unwrap();
        tx.delete("test", &id).unwrap();
        let read_opt = tx.read("test", &id).unwrap();
        assert_eq!(read_opt, None);
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let read_after = persy.read("test", &id).unwrap();
        assert_eq!(read_after, None);
    });
}

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

        let mut tx = persy.begin().unwrap();
        let bytes = String::from("first").into_bytes();
        tx.update("test", &id, &bytes).unwrap();
        tx.delete("test", &id).unwrap();
        let read_opt = tx.read("test", &id).unwrap();
        assert_eq!(read_opt, None);
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let read_after = persy.read("test", &id).unwrap();
        assert_eq!(read_after, None);
    });
}

#[test]
fn test_insert_100_same_tx() {
    create_and_drop("i100", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        for _ in [0; 100].iter() {
            tx.insert("test", &bytes).unwrap();
        }
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut count = 0;
        for _ in persy.scan("test").unwrap() {
            count += 1;
        }
        assert_eq!(count, 100);
    });
}

#[test]
fn test_update_100_same_tx() {
    create_and_drop("i100", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        let mut ids = Vec::new();
        for _ in [0; 100].iter() {
            ids.push(tx.insert("test", &bytes).unwrap());
        }
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        for id in ids {
            tx.update("test", &id, &bytes).unwrap();
        }

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

        let mut count = 0;
        for _ in persy.scan("test").unwrap() {
            count += 1;
        }
        assert_eq!(count, 100);
    });
}

#[test]
fn test_create_insert_scan_same_tx() {
    create_and_drop("cistx", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        for _ in [0; 50].iter() {
            tx.insert("test", &bytes).unwrap();
        }
        let mut count = 0;
        for (_, content) in tx.scan("test").unwrap() {
            assert_eq!(content, bytes);
            count += 1;
        }
        assert_eq!(count, 50);
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
    });
}

#[test]
fn test_create_insert_scan_same_tx_access() {
    create_and_drop("cistx", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        tx.create_segment("test1").unwrap();
        let bytes = String::from("something").into_bytes();
        for _ in [0; 50].iter() {
            tx.insert("test", &bytes).unwrap();
        }
        let mut count = 0;
        let mut iter = tx.scan("test").unwrap();
        iter.tx().insert("test1", "bytes".as_bytes()).unwrap();
        let new_bytes = String::from("other").into_bytes();
        while let Some((id, content, tx)) = iter.next_tx() {
            assert_eq!(content, bytes);
            count += 1;
            tx.update("test", &id, &new_bytes).unwrap();
        }
        assert_eq!(count, 50);
        let mut count = 0;
        for (_, content) in tx.scan("test").unwrap() {
            assert_eq!(content, new_bytes);
            count += 1;
        }
        assert_eq!(count, 50);
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
    });
}

#[test]
fn test_insert_100_and_scan_same_tx() {
    create_and_drop("i100tx", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        for _ in [0; 50].iter() {
            tx.insert("test", &bytes).unwrap();
        }
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let mut tx = persy.begin().unwrap();
        for _ in [0; 50].iter() {
            tx.insert("test", &bytes).unwrap();
        }
        let mut count = 0;
        for (_, content) in tx.scan("test").unwrap() {
            assert_eq!(content, bytes);
            count += 1;
        }
        assert_eq!(count, 100);
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
    });
}

#[test]
fn test_insert_100_update_and_scan_same_tx() {
    create_and_drop("i100u_tx", |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("test").unwrap();
        let bytes = String::from("something").into_bytes();
        let mut inserted = Vec::new();
        for _ in [0; 50].iter() {
            inserted.push(tx.insert("test", &bytes).unwrap());
        }
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();

        let bytes_other = String::from("other").into_bytes();
        let mut tx = persy.begin().unwrap();
        for _ in [0; 50].iter() {
            tx.insert("test", &bytes).unwrap();
        }
        for id in inserted.iter() {
            tx.update("test", id, &bytes_other).unwrap();
        }

        let mut count = 0;
        for (id, content) in tx.scan("test").unwrap() {
            if inserted.contains(&id) {
                assert_eq!(content, bytes_other);
            } else {
                assert_eq!(content, bytes);
            }
            count += 1;
        }
        assert_eq!(count, 100);
        let finalizer = tx.prepare().unwrap();
        finalizer.commit().unwrap();
    });
}

#[test]
fn test_scan_tx_segment_not_exist() {
    create_and_drop("sne_tx", |persy| {
        let mut tx = persy.begin().unwrap();
        let res = tx.scan("test");
        if let Err(x) = res {
            match x.error() {
                SegmentError::SegmentNotFound => {}
                _ => assert!(false),
            }
        } else {
            assert!(false);
        }
    });
}

#[test]
fn test_concurrency_version_on_write() {
    let mut config = Config::new();
    config.change_tx_strategy(TxStrategy::VersionOnWrite);
    create_and_drop_with_config("cvow", config, |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("seg").unwrap();
        let bytes = String::from("aaa").into_bytes();
        let id = tx.insert("seg", &bytes).unwrap();
        let prep = tx.prepare().unwrap();
        prep.commit().unwrap();

        let bytes2 = String::from("bbb").into_bytes();
        let mut tx = persy.begin().unwrap();
        tx.update("seg", &id, &bytes2).unwrap();

        let bytes3 = String::from("cccc").into_bytes();
        let mut tx1 = persy.begin().unwrap();
        tx1.update("seg", &id, &bytes3).unwrap();

        let prep = tx1.prepare().unwrap();
        prep.commit().unwrap();
        let to_fail = tx.prepare();
        assert!(to_fail.is_err());
        let val = persy.read("seg", &id).unwrap().unwrap();
        assert_eq!(val, bytes3);
    });
}

#[test]
fn test_concurrency_last_win() {
    let mut config = Config::new();
    config.change_tx_strategy(TxStrategy::LastWin);
    create_and_drop_with_config("cvol", config, |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("seg").unwrap();
        let bytes = String::from("aaa").into_bytes();
        let id = tx.insert("seg", &bytes).unwrap();
        let prep = tx.prepare().unwrap();
        prep.commit().unwrap();

        let bytes2 = String::from("bbb").into_bytes();
        let mut tx = persy.begin().unwrap();
        tx.update("seg", &id, &bytes2).unwrap();

        let bytes3 = String::from("cccc").into_bytes();
        let mut tx1 = persy.begin().unwrap();
        tx1.update("seg", &id, &bytes3).unwrap();

        let prep = tx1.prepare().unwrap();
        prep.commit().unwrap();

        let prep = tx.prepare().unwrap();
        prep.commit().unwrap();

        let val = persy.read("seg", &id).unwrap().unwrap();
        assert_eq!(val, bytes2);
    });
}

#[test]
fn test_concurrency_version_on_read() {
    let mut config = Config::new();
    config.change_tx_strategy(TxStrategy::VersionOnRead);
    create_and_drop_with_config("cvor", config, |persy| {
        let mut tx = persy.begin().unwrap();
        tx.create_segment("seg").unwrap();
        let bytes = String::from("aaa").into_bytes();
        let id = tx.insert("seg", &bytes).unwrap();
        let prep = tx.prepare().unwrap();
        prep.commit().unwrap();

        let bytes2 = String::from("bbb").into_bytes();
        let mut tx = persy.begin().unwrap();
        let _ = tx.read("seg", &id).unwrap();

        let bytes3 = String::from("cccc").into_bytes();

        let mut tx1 = persy.begin().unwrap();
        tx1.update("seg", &id, &bytes3).unwrap();
        let prep = tx1.prepare().unwrap();
        prep.commit().unwrap();

        tx.update("seg", &id, &bytes2).unwrap();
        let to_fail = tx.prepare();
        assert!(to_fail.is_err());

        let val = persy.read("seg", &id).unwrap().unwrap();
        assert_eq!(val, bytes3);
    });
}

#[test]
fn test_scan_segment_not_exist() {
    create_and_drop("sne", |persy| {
        let res = persy.scan("test");
        if let Err(x) = res {
            match x.error() {
                SegmentError::SegmentNotFound => {}
                _ => assert!(false),
            }
        } else {
            assert!(false);
        }
    });
}

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

        let mut tx = persy.begin().unwrap();
        for _i in 0..5000 {
            tx.insert("one", "two".as_bytes()).unwrap();
        }
        tx.prepare().unwrap().commit().unwrap();
        let mut tx = persy.begin().unwrap();
        for (id, _rec) in persy.scan("one").unwrap() {
            tx.delete("one", &id).unwrap();
        }
        tx.prepare().unwrap().commit().unwrap();
        assert!(persy.scan("one").unwrap().next().is_none());
    });
}