use std::path::Path;

use super::{
    Access, BatchOp, Decision, ListedEntry, MAX_BATCH_BYTES, MAX_BATCH_OPS, Store, StoredEntry,
    parse_stamp,
};

// One-transaction shorthands, so tests read like the adapters' single ops.
trait Shorthand {
    fn get(&self, keyspace: &str, key: &[u8]) -> std::io::Result<Option<StoredEntry>>;
    fn get_many(&self, gets: &[(String, Vec<u8>)]) -> std::io::Result<Vec<Option<StoredEntry>>>;
    fn set(&self, keyspace: &str, key: &[u8], value: &[u8]) -> std::io::Result<String>;
    fn clear(&self, keyspace: &str) -> std::io::Result<()>;
    fn count(&self, keyspace: &str) -> std::io::Result<u64>;
    fn list_keyspaces(&self) -> std::io::Result<Vec<String>>;
    fn list(
        &self,
        keyspace: &str,
        start: Option<&[u8]>,
        end: Option<&[u8]>,
        prefix: Option<&[u8]>,
        limit: usize,
        reverse: bool,
    ) -> std::io::Result<Vec<ListedEntry>>;
}

impl Shorthand for Store {
    fn get(&self, keyspace: &str, key: &[u8]) -> std::io::Result<Option<StoredEntry>> {
        self.read_tx(|tx| tx.get(keyspace, key))
    }

    fn get_many(&self, gets: &[(String, Vec<u8>)]) -> std::io::Result<Vec<Option<StoredEntry>>> {
        self.read_tx(|tx| gets.iter().map(|(ks, key)| tx.get(ks, key)).collect())
    }

    fn set(&self, keyspace: &str, key: &[u8], value: &[u8]) -> std::io::Result<String> {
        self.write_tx(|tx| tx.put(keyspace, key, value).map(Decision::Commit))
    }

    fn clear(&self, keyspace: &str) -> std::io::Result<()> {
        self.write_tx(|tx| tx.clear(keyspace).map(Decision::Commit))
    }

    fn count(&self, keyspace: &str) -> std::io::Result<u64> {
        self.read_tx(|tx| tx.count(keyspace))
    }

    fn list_keyspaces(&self) -> std::io::Result<Vec<String>> {
        self.read_tx(|tx| tx.list_keyspaces())
    }

    fn list(
        &self,
        keyspace: &str,
        start: Option<&[u8]>,
        end: Option<&[u8]>,
        prefix: Option<&[u8]>,
        limit: usize,
        reverse: bool,
    ) -> std::io::Result<Vec<ListedEntry>> {
        self.read_tx(|tx| tx.list(keyspace, start, end, prefix, limit, reverse))
    }
}

fn open_shared(dir: &Path, name: &str) -> Store {
    Store::open(&dir.join(name), Access::Shared, true).unwrap()
}

fn check(keyspace: &str, key: &[u8], expected: Option<&str>) -> BatchOp {
    BatchOp::Check {
        keyspace: keyspace.into(),
        key: key.to_vec(),
        expected: expected.map(Into::into),
    }
}

fn set(keyspace: &str, key: &[u8], value: &[u8]) -> BatchOp {
    BatchOp::Set {
        keyspace: keyspace.into(),
        key: key.to_vec(),
        value: value.to_vec(),
    }
}

fn delete(keyspace: &str, key: &[u8]) -> BatchOp {
    BatchOp::Delete {
        keyspace: keyspace.into(),
        key: key.to_vec(),
    }
}

fn get(store: &Store, keyspace: &str, key: &[u8]) -> Option<(Vec<u8>, String)> {
    store.get(keyspace, key).unwrap()
}

fn assert_send_sync<T: Send + Sync>() {}

#[test]
fn store_is_send_sync_and_cloneable() {
    assert_send_sync::<Store>();
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let _clone = store.clone();
}

#[test]
fn set_get_roundtrip() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let stamp = store.set("ks", b"key", b"value").unwrap();
    assert_eq!(
        get(&store, "ks", b"key").unwrap(),
        (b"value".to_vec(), stamp)
    );
    assert_eq!(store.get("ks", b"missing").unwrap(), None);
    assert_eq!(store.get("other", b"key").unwrap(), None);
}

#[test]
fn persists_across_reopen_in_shared_mode() {
    let root = tempfile::tempdir().unwrap();
    let path = root.path().join("kv.redb");
    let store = Store::open(&path, Access::Shared, true).unwrap();
    let stamp = store.set("ks", b"key", b"value").unwrap();
    drop(store);
    let store = Store::open(&path, Access::Shared, false).unwrap();
    assert_eq!(
        get(&store, "ks", b"key").unwrap(),
        (b"value".to_vec(), stamp)
    );
}

#[test]
fn exclusive_blocks_second_opener() {
    let root = tempfile::tempdir().unwrap();
    let path = root.path().join("kv.redb");
    let _first = Store::open(&path, Access::Exclusive, true).unwrap();
    let error = Store::open(&path, Access::Exclusive, true).unwrap_err();
    assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists);
    let error = Store::open(&path, Access::Shared, false).unwrap_err();
    assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists);
}

#[test]
fn read_only_does_not_create() {
    let root = tempfile::tempdir().unwrap();
    let path = root.path().join("kv.redb");
    let store = Store::open(&path, Access::ReadOnly, false).unwrap();
    assert!(!path.exists());
    assert_eq!(store.get("ks", b"key").unwrap(), None);
    assert_eq!(store.list_keyspaces().unwrap(), Vec::<String>::new());
    assert_eq!(store.count("ks").unwrap(), 0);
    assert!(
        store
            .list("ks", None, None, None, 10, false)
            .unwrap()
            .is_empty()
    );
    let gets = [
        ("ks".to_owned(), b"a".to_vec()),
        ("ks".to_owned(), b"b".to_vec()),
    ];
    assert_eq!(store.get_many(&gets).unwrap(), vec![None, None]);
    let error = store.set("ks", b"key", b"value").unwrap_err();
    assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
}

#[test]
fn keyspace_isolation_clear_and_list() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    store.set("a", b"k", b"1").unwrap();
    store.set("b", b"k", b"2").unwrap();
    assert_eq!(
        store.list_keyspaces().unwrap(),
        vec!["a".to_owned(), "b".to_owned()]
    );
    store.clear("a").unwrap();
    assert_eq!(store.get("a", b"k").unwrap(), None);
    assert_eq!(get(&store, "b", b"k").unwrap().0, b"2");
    assert_eq!(
        store.list_keyspaces().unwrap(),
        vec!["a".to_owned(), "b".to_owned()]
    );
    store.clear("c").unwrap();
    assert_eq!(
        store.list_keyspaces().unwrap(),
        vec!["a".to_owned(), "b".to_owned(), "c".to_owned()]
    );
}

#[test]
fn atomic_commit_success_across_keyspaces() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let stamp = store.set("a", b"k1", b"v1").unwrap();
    let batch = [
        check("a", b"k1", Some(&stamp)),
        set("a", b"k1", b"v2"),
        set("b", b"k2", b"w2"),
        delete("c", b"ghost"),
    ];
    let (ok, commit_stamp) = store.atomic_commit(&batch).unwrap();
    assert!(ok);
    let commit_stamp = commit_stamp.unwrap();
    assert_eq!(
        get(&store, "a", b"k1").unwrap(),
        (b"v2".to_vec(), commit_stamp.clone())
    );
    assert_eq!(
        get(&store, "b", b"k2").unwrap(),
        (b"w2".to_vec(), commit_stamp)
    );
}

#[test]
fn atomic_check_failure_aborts_everything() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let stamp = store.set("a", b"k1", b"v1").unwrap();
    let batch = [
        check("a", b"k1", Some("255")),
        set("a", b"k1", b"v2"),
        set("b", b"k2", b"w2"),
    ];
    let (ok, commit_stamp) = store.atomic_commit(&batch).unwrap();
    assert!(!ok);
    assert_eq!(commit_stamp, None);
    assert_eq!(get(&store, "a", b"k1").unwrap(), (b"v1".to_vec(), stamp));
    assert_eq!(store.get("b", b"k2").unwrap(), None);
}

#[test]
fn atomic_null_check_means_must_not_exist() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let batch = [check("a", b"k", None), set("a", b"k", b"v")];
    let (ok, _) = store.atomic_commit(&batch).unwrap();
    assert!(ok);
    let batch = [check("a", b"k", None), set("a", b"k", b"other")];
    let (ok, _) = store.atomic_commit(&batch).unwrap();
    assert!(!ok);
    assert_eq!(get(&store, "a", b"k").unwrap().0, b"v");
}

#[test]
fn versionstamp_changes_on_rewrite_and_is_shared_within_a_commit() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let first = store.set("ks", b"k", b"v1").unwrap();
    let second = store.set("ks", b"k", b"v2").unwrap();
    assert_ne!(first, second);
    let batch = [set("a", b"x", b"1"), set("b", b"y", b"2")];
    let (ok, stamp) = store.atomic_commit(&batch).unwrap();
    assert!(ok);
    let stamp = stamp.unwrap();
    assert_eq!(get(&store, "a", b"x").unwrap().1, stamp);
    assert_eq!(get(&store, "b", b"y").unwrap().1, stamp);
    assert_ne!(stamp, second);
}

#[test]
fn list_ordering_prefix_limit_and_reverse() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    for key in [b"c".as_slice(), b"a", b"ab", b"aa", b"b"] {
        store.set("ks", key, key).unwrap();
    }
    let keys = |entries: Vec<(Vec<u8>, Vec<u8>, String)>| {
        entries
            .into_iter()
            .map(|(key, _, _)| key)
            .collect::<Vec<_>>()
    };
    assert_eq!(
        keys(store.list("ks", None, None, None, 100, false).unwrap()),
        vec![
            b"a".to_vec(),
            b"aa".to_vec(),
            b"ab".to_vec(),
            b"b".to_vec(),
            b"c".to_vec()
        ]
    );
    assert_eq!(
        keys(
            store
                .list("ks", None, None, Some(b"a"), 100, false)
                .unwrap()
        ),
        vec![b"a".to_vec(), b"aa".to_vec(), b"ab".to_vec()]
    );
    assert_eq!(
        keys(store.list("ks", None, None, Some(b"a"), 2, false).unwrap()),
        vec![b"a".to_vec(), b"aa".to_vec()]
    );
    assert_eq!(
        keys(store.list("ks", None, None, Some(b"a"), 100, true).unwrap()),
        vec![b"ab".to_vec(), b"aa".to_vec(), b"a".to_vec()]
    );
    assert_eq!(
        keys(
            store
                .list("ks", Some(b"aa"), Some(b"c"), None, 100, false)
                .unwrap()
        ),
        vec![b"aa".to_vec(), b"ab".to_vec(), b"b".to_vec()]
    );
    assert!(
        store
            .list("ks", None, None, Some(b"zzz"), 100, false)
            .unwrap()
            .is_empty()
    );
    assert!(
        store
            .list("missing", None, None, None, 100, false)
            .unwrap()
            .is_empty()
    );
}

#[test]
fn get_many_reads_one_snapshot() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let first = store.set("a", b"k1", b"v1").unwrap();
    let gets = vec![
        ("a".to_owned(), b"k1".to_vec()),
        ("a".to_owned(), b"absent".to_vec()),
        ("b".to_owned(), b"k1".to_vec()),
    ];
    assert_eq!(
        store.get_many(&gets).unwrap(),
        vec![Some((b"v1".to_vec(), first)), None, None]
    );
}

#[test]
fn failed_atomic_changes_nothing_including_metadata() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let batch = [
        check("a", b"k1", Some("1")),
        set("a", b"k1", b"v1"),
        set("b", b"k2", b"v2"),
    ];
    let (ok, stamp) = store.atomic_commit(&batch).unwrap();
    assert!(!ok);
    assert_eq!(stamp, None);
    assert!(store.list_keyspaces().unwrap().is_empty());
    assert_eq!(store.get("a", b"k1").unwrap(), None);
    assert_eq!(store.get("b", b"k2").unwrap(), None);
}

#[test]
fn check_only_batch_burns_no_versionstamp() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let first = store.set("ks", b"k", b"v1").unwrap();
    let batch = [check("ks", b"k", Some(&first))];
    let (ok, stamp) = store.atomic_commit(&batch).unwrap();
    assert!(ok);
    assert_eq!(stamp, None);
    let second = store.set("ks", b"k", b"v2").unwrap();
    assert_eq!(
        parse_stamp(&second).unwrap(),
        parse_stamp(&first).unwrap() + 1
    );
}

#[test]
fn malformed_versionstamp_is_invalid_input() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    store.set("ks", b"k", b"v").unwrap();
    for bad in [
        "not-a-stamp",
        "",
        "-1",
        "+1",
        "0x10",
        "1.0",
        "18446744073709551616",
    ] {
        let batch = [check("ks", b"k", Some(bad)), set("ks", b"k", b"other")];
        assert_eq!(
            store.atomic_commit(&batch).unwrap_err().kind(),
            std::io::ErrorKind::InvalidInput,
            "stamp {bad:?} must be rejected",
        );
    }
    assert_eq!(get(&store, "ks", b"k").unwrap().0, b"v");
}

#[test]
fn missing_file_without_create_fails() {
    let root = tempfile::tempdir().unwrap();
    for access in [Access::Shared, Access::Exclusive] {
        let error = Store::open(&root.path().join("missing.redb"), access, false).unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::NotFound);
    }
    assert!(!root.path().join("missing.redb").exists());
}

#[test]
fn invalid_inputs_are_rejected() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    assert_eq!(
        Access::parse("bogus").unwrap_err().kind(),
        std::io::ErrorKind::InvalidInput
    );
    assert_eq!(
        store.set("", b"k", b"v").unwrap_err().kind(),
        std::io::ErrorKind::InvalidInput
    );
    assert_eq!(
        store
            .atomic_commit(&[set("", b"k", b"v")])
            .unwrap_err()
            .kind(),
        std::io::ErrorKind::InvalidInput
    );
    assert_eq!(
        Store::open(&root.path().join("ro.redb"), Access::ReadOnly, true)
            .unwrap_err()
            .kind(),
        std::io::ErrorKind::InvalidInput
    );
}

#[test]
fn write_tx_commit_applies_and_abort_changes_nothing() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let stamp = store
        .write_tx(|tx| {
            tx.put("ks", b"k", b"v")?;
            Ok(Decision::Commit(tx.stamp().unwrap()))
        })
        .unwrap();
    assert_eq!(get(&store, "ks", b"k").unwrap(), (b"v".to_vec(), stamp));
    let before = store.list_keyspaces().unwrap();
    let (aborted, missing) = store
        .write_tx(|tx| {
            tx.put("fresh", b"k", b"v")?;
            tx.put_meta("adapter/counter", 7)?;
            assert_eq!(tx.get("fresh", b"k")?.unwrap().0, b"v");
            assert_eq!(tx.get_meta("adapter/counter")?, Some(7));
            Ok(Decision::Abort((false, true)))
        })
        .unwrap();
    assert!(!aborted && missing);
    assert_eq!(store.get("fresh", b"k").unwrap(), None);
    assert_eq!(store.list_keyspaces().unwrap(), before);
    store
        .read_tx(|tx| {
            assert_eq!(tx.get_meta("adapter/counter")?, None);
            Ok(())
        })
        .unwrap();
}

#[test]
fn write_tx_sees_its_own_writes() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    store
        .write_tx(|tx| {
            tx.put("ks", b"k", b"one")?;
            let first = tx.get("ks", b"k")?.unwrap();
            tx.put("ks", b"k", b"two")?;
            let second = tx.get("ks", b"k")?.unwrap();
            assert_eq!(first.1, second.1, "puts in one transaction share the stamp");
            assert!(tx.delete("ks", b"k")?);
            assert!(!tx.delete("ks", b"k")?, "second delete reports absence");
            assert_eq!(tx.get("ks", b"k")?, None);
            tx.clear("ks")?;
            assert!(tx.list_keyspaces()?.contains(&"ks".to_owned()));
            Ok(Decision::Commit(()))
        })
        .unwrap();
    assert_eq!(store.get("ks", b"k").unwrap(), None);
}

#[test]
fn metadata_roundtrips_and_the_counter_slot_is_rejected() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    store
        .write_tx(|tx| {
            tx.put_meta("webstorage/bytes_used", 4096)?;
            assert_eq!(tx.get_meta("webstorage/bytes_used")?, Some(4096));
            tx.put_meta("webstorage/bytes_used", 8192)?;
            Ok(Decision::Commit(()))
        })
        .unwrap();
    store
        .read_tx(|tx| {
            assert_eq!(tx.get_meta("webstorage/bytes_used")?, Some(8192));
            assert_eq!(tx.get_meta("webstorage/other")?, None);
            Ok(())
        })
        .unwrap();
    // "" is the commit counter's slot.
    assert_eq!(
        store
            .write_tx(|tx| {
                tx.put_meta("", 1)?;
                Ok(Decision::Commit(()))
            })
            .unwrap_err()
            .kind(),
        std::io::ErrorKind::InvalidInput,
    );
    assert_eq!(
        store.read_tx(|tx| tx.get_meta("")).unwrap_err().kind(),
        std::io::ErrorKind::InvalidInput,
    );
    // Adapter metadata coexists with backend state without interference.
    store.set("ks", b"k", b"v").unwrap();
    assert_eq!(
        store
            .read_tx(|tx| tx.get_meta("webstorage/bytes_used"))
            .unwrap(),
        Some(8192)
    );
    assert_eq!(get(&store, "ks", b"k").unwrap().0, b"v");
}

#[test]
fn entries_and_count_cover_uncommitted_and_committed_state() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    assert_eq!(store.count("missing").unwrap(), 0);
    store
        .write_tx(|tx| {
            assert_eq!(tx.count("ks").unwrap(), 0);
            tx.put("ks", b"b", b"2")?;
            tx.put("ks", b"a", b"1")?;
            // Full-scan entries in byte order, including uncommitted writes.
            let keys: Vec<Vec<u8>> = tx
                .entries("ks")?
                .into_iter()
                .map(|(key, _, _)| key)
                .collect();
            assert_eq!(keys, vec![b"a".to_vec(), b"b".to_vec()]);
            assert_eq!(tx.count("ks").unwrap(), 2);
            Ok(Decision::Commit(()))
        })
        .unwrap();
    assert_eq!(store.count("ks").unwrap(), 2);
    store
        .read_tx(|tx| {
            assert_eq!(tx.entries("missing").unwrap(), Vec::new());
            Ok(())
        })
        .unwrap();
    store.clear("ks").unwrap();
    assert_eq!(store.count("ks").unwrap(), 0);
}

#[test]
fn oversized_batches_are_rejected() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    let too_large = [set("ks", b"k", &vec![0; MAX_BATCH_BYTES])];
    let too_many = vec![delete("ks", b"k"); MAX_BATCH_OPS + 1];
    for batch in [too_large.as_slice(), &too_many] {
        assert_eq!(
            store.atomic_commit(batch).unwrap_err().kind(),
            std::io::ErrorKind::InvalidInput
        );
    }
    assert_eq!(store.get("ks", b"k").unwrap(), None);
}

#[test]
fn open_read_only_repairing_recovers_unclean_shutdown() {
    // Child mode: commit a value, then exit without dropping the store,
    // leaving the file marked as needing repair.
    if let Ok(cache) = std::env::var("AVRIL_KV_UNCLEAN_CACHE") {
        let path = Path::new(&cache).join("kv.redb");
        let store = Store::open(&path, Access::Exclusive, true).unwrap();
        store.set("ks", b"key", b"unclean").unwrap();
        std::process::exit(0);
    }
    let dir = tempfile::tempdir().unwrap();
    let status = std::process::Command::new(std::env::current_exe().unwrap())
        .args([
            "--exact",
            "tests::open_read_only_repairing_recovers_unclean_shutdown",
        ])
        .env("AVRIL_KV_UNCLEAN_CACHE", dir.path())
        .stdout(std::process::Stdio::null())
        .status()
        .unwrap();
    assert!(status.success());
    let store = Store::open_read_only_repairing(&dir.path().join("kv.redb")).unwrap();
    let entry = store
        .get("ks", b"key")
        .unwrap()
        .expect("repaired value readable");
    assert_eq!(entry.0, b"unclean".to_vec());
    parse_stamp(&entry.1).unwrap();
    // A missing file opens as empty without creating anything.
    let missing = dir.path().join("absent.redb");
    let store = Store::open_read_only_repairing(&missing).unwrap();
    assert_eq!(store.get("ks", b"key").unwrap(), None);
    assert!(!missing.exists());
}

#[test]
fn write_tx_reads_and_deletes_materialize_nothing() {
    let root = tempfile::tempdir().unwrap();
    let store = open_shared(root.path(), "kv.redb");
    store
        .write_tx(|tx| {
            assert_eq!(tx.get("ghost", b"k")?, None);
            assert!(!tx.delete("ghost", b"k")?);
            assert_eq!(tx.count("ghost")?, 0);
            assert_eq!(tx.get_meta("adapter")?, None);
            Ok(Decision::Commit(()))
        })
        .unwrap();
    assert!(store.list_keyspaces().unwrap().is_empty());
    store.clear("empty").unwrap();
    assert_eq!(store.list_keyspaces().unwrap(), vec!["empty".to_owned()]);
}