use std::io;
use std::sync::Arc;

use avril_caps::{Capabilities, Category, Initial};
use avril_kv::{BatchOp, Decision, StoredEntry};

use super::registry::Registry;

// The ops' shapes: one re-authorized store operation per call.
trait Shorthand {
    fn get(&self, db: u64, keyspace: &str, key: &[u8]) -> io::Result<Option<StoredEntry>>;
    fn set(&self, db: u64, keyspace: &str, key: &[u8], value: &[u8]) -> io::Result<String>;
    fn list_keyspaces(&self, db: u64) -> io::Result<Vec<String>>;
    fn atomic_commit(&self, db: u64, ops: Vec<BatchOp>) -> io::Result<(bool, Option<String>)>;
}

impl Shorthand for Registry {
    fn get(&self, db: u64, keyspace: &str, key: &[u8]) -> io::Result<Option<StoredEntry>> {
        self.with_entry(db, false, |store| store.read_tx(|tx| tx.get(keyspace, key)))
    }

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

    fn list_keyspaces(&self, db: u64) -> io::Result<Vec<String>> {
        self.with_entry(db, false, |store| store.read_tx(|tx| tx.list_keyspaces()))
    }

    fn atomic_commit(&self, db: u64, ops: Vec<BatchOp>) -> io::Result<(bool, Option<String>)> {
        self.with_entry(db, true, |store| store.atomic_commit(&ops))
    }
}

fn caps(dir: &std::path::Path) -> Arc<Capabilities> {
    Capabilities::new(Initial::new(false), dir.to_owned())
}

fn open_shared(registry: &Registry, caps: &Arc<Capabilities>, name: &str) -> u64 {
    registry.open(caps, name, "shared", true).unwrap()
}

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

fn get<'a>(
    registry: &Registry,
    db: u64,
    keyspace: &'a str,
    key: &'a [u8],
) -> Option<(Vec<u8>, String)> {
    registry.get(db, keyspace, key).unwrap()
}

#[test]
fn atomic_commit_spans_keyspaces() {
    let root = tempfile::tempdir().unwrap();
    let registry = Registry::new();
    let db = open_shared(&registry, &caps(root.path()), "kv.redb");
    let (ok, stamp) = registry
        .atomic_commit(
            db,
            vec![set_op("a", b"k1", b"v1"), set_op("b", b"k2", b"v2")],
        )
        .unwrap();
    assert!(ok);
    let stamp = stamp.unwrap();
    assert_eq!(
        get(&registry, db, "a", b"k1").unwrap(),
        (b"v1".to_vec(), stamp.clone())
    );
    assert_eq!(
        get(&registry, db, "b", b"k2").unwrap(),
        (b"v2".to_vec(), stamp)
    );
}

#[test]
fn atomic_check_failure_writes_nothing() {
    let root = tempfile::tempdir().unwrap();
    let registry = Registry::new();
    let db = open_shared(&registry, &caps(root.path()), "kv.redb");
    let stamp = registry.set(db, "a", b"k", b"v").unwrap();
    let check = BatchOp::Check {
        keyspace: "a".into(),
        key: b"k".to_vec(),
        expected: None,
    };
    let (ok, commit) = registry
        .atomic_commit(
            db,
            vec![check, set_op("a", b"k", b"v2"), set_op("b", b"k", b"w")],
        )
        .unwrap();
    assert!(!ok);
    assert_eq!(commit, None);
    assert_eq!(
        get(&registry, db, "a", b"k").unwrap(),
        (b"v".to_vec(), stamp)
    );
    assert_eq!(get(&registry, db, "b", b"k"), None);
}

#[test]
fn closed_and_unknown_handles() {
    let root = tempfile::tempdir().unwrap();
    let registry = Registry::new();
    let db = open_shared(&registry, &caps(root.path()), "kv.redb");
    registry.close(db).unwrap();
    // Close is idempotent.
    registry.close(db).unwrap();
    registry.close(999_999).unwrap();
    for error in [
        registry.get(db, "ks", b"k").unwrap_err(),
        registry.set(db, "ks", b"k", b"v").map(|_| ()).unwrap_err(),
        registry.list_keyspaces(db).map(|_| ()).unwrap_err(),
        registry
            .atomic_commit(db, vec![set_op("ks", b"k", b"v")])
            .map(|_| ())
            .unwrap_err(),
    ] {
        assert_eq!(error.kind(), std::io::ErrorKind::NotConnected);
    }
    assert_eq!(
        registry.get(999_999, "ks", b"k").unwrap_err().kind(),
        std::io::ErrorKind::NotConnected
    );
}

#[test]
fn close_releases_the_database_file() {
    let root = tempfile::tempdir().unwrap();
    let registry = Registry::new();
    let db = registry
        .open(&caps(root.path()), "kv.redb", "exclusive", true)
        .unwrap();
    registry.set(db, "ks", b"k", b"v").unwrap();
    registry.close(db).unwrap();
    // Reopening exclusively works only if close physically released the file.
    let db = registry
        .open(&caps(root.path()), "kv.redb", "exclusive", false)
        .unwrap();
    registry.set(db, "ks", b"k", b"v2").unwrap();
    assert_eq!(get(&registry, db, "ks", b"k").unwrap().0, b"v2");
    registry.close(db).unwrap();
}

#[test]
fn capability_denial() {
    let root = tempfile::tempdir().unwrap();
    let locked = Capabilities::new(Initial::new(true), root.path().to_owned());
    let registry = Registry::new();
    let error = registry
        .open(&locked, "kv.redb", "shared", true)
        .unwrap_err();
    assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
    assert!(!root.path().join("kv.redb").exists());
}

#[test]
fn pledge_narrowing_applies_to_open_handles() {
    let root = tempfile::tempdir().unwrap();
    let live = caps(root.path());
    let registry = Registry::new();
    let db = open_shared(&registry, &live, "kv.redb");
    registry.set(db, "ks", b"k", b"v").unwrap();
    assert!(registry.get(db, "ks", b"k").unwrap().is_some());
    live.pledge(vec![
        (Category::Read, Some(vec![]), vec![]),
        (Category::Write, Some(vec![]), vec![]),
    ])
    .unwrap();
    assert_eq!(
        registry.get(db, "ks", b"k").unwrap_err().kind(),
        std::io::ErrorKind::PermissionDenied
    );
    assert_eq!(
        registry.set(db, "ks", b"k", b"v2").unwrap_err().kind(),
        std::io::ErrorKind::PermissionDenied
    );
    assert_eq!(
        registry.list_keyspaces(db).unwrap_err().kind(),
        std::io::ErrorKind::PermissionDenied
    );
    assert_eq!(
        registry
            .atomic_commit(db, vec![set_op("ks", b"k", b"v3")])
            .unwrap_err()
            .kind(),
        std::io::ErrorKind::PermissionDenied
    );
}

#[test]
fn two_read_only_handles_coexist_on_the_same_database() {
    let root = tempfile::tempdir().unwrap();
    let registry = Registry::new();
    let db = open_shared(&registry, &caps(root.path()), "kv.redb");
    registry.set(db, "ks", b"k", b"v").unwrap();
    registry.close(db).unwrap();
    let first = registry
        .open(&caps(root.path()), "kv.redb", "read-only", false)
        .unwrap();
    let second = registry
        .open(&caps(root.path()), "kv.redb", "read-only", false)
        .unwrap();
    assert_eq!(get(&registry, first, "ks", b"k").unwrap().0, b"v");
    assert_eq!(get(&registry, second, "ks", b"k").unwrap().0, b"v");
    registry.close(first).unwrap();
    registry.close(second).unwrap();
}

#[test]
fn read_only_handle_on_a_missing_file_does_not_block_exclusive_create() {
    let root = tempfile::tempdir().unwrap();
    let registry = Registry::new();
    let read = registry
        .open(&caps(root.path()), "kv.redb", "read-only", false)
        .unwrap();
    assert!(!root.path().join("kv.redb").exists());
    let write = registry
        .open(&caps(root.path()), "kv.redb", "exclusive", true)
        .unwrap();
    registry.set(write, "ks", b"k", b"v").unwrap();
    // The read-only handle snapshotted the missing file at open; it still
    // reads empty rather than observing the later creation.
    assert_eq!(registry.get(read, "ks", b"k").unwrap(), None);
    registry.close(read).unwrap();
    registry.close(write).unwrap();
}