use ratdb_core::{Config, Database, Error, MemoryStorage, Result, Storage};
use ratdb_document::{DocumentChange, Documents};
use serde_json::{json, Value};
use std::{cell::Cell, rc::Rc};

fn db() -> Documents<MemoryStorage> {
    Documents(Database::create_indexed(MemoryStorage::default(), Config::default()).unwrap())
}
fn change<'a>(id: &'a str, v: Value) -> DocumentChange<'a> {
    DocumentChange {
        warren: "web",
        den: "users",
        id,
        document: Some(v),
    }
}
#[test]
fn sparse_unique_lifecycle_batch_and_types() {
    let mut db = db();
    for (id, v) in [
        ("a", json!("A")),
        ("b", json!("B")),
        ("bool", json!(true)),
        ("i", json!(1)),
        ("f", json!(1.0)),
        ("null", Value::Null),
    ] {
        db.put_in("web", "users", id, json!({"u":v})).unwrap();
    }
    db.put_in("web", "users", "missing", json!({})).unwrap();
    db.create_index("web", "users", "u", true).unwrap();
    assert_eq!(db.indexes("web", "users").unwrap(), [("u".into(), true)]);
    let g = db.0.stats();
    assert_eq!(
        db.insert_in("web", "users", "c", json!({"u":"A"})),
        Err(Error::Unique)
    );
    assert_eq!(
        db.put_in("web", "users", "b", json!({"u":"A"})),
        Err(Error::Unique)
    );
    assert_eq!(
        db.update_in("web", "users", "b", json!({"u":"A"})),
        Err(Error::Unique)
    );
    assert_eq!(
        db.write_batch(vec![
            change("a", json!({"u":"C"})),
            change("b", json!({"u":"C"}))
        ]),
        Err(Error::Unique)
    );
    assert_eq!(db.0.stats(), g);
    assert_eq!(db.get_in("web", "users", "b").unwrap().unwrap()["u"], "B");
    db.write_batch(vec![
        change("a", json!({"u":"B"})),
        change("b", json!({"u":"A"})),
    ])
    .unwrap();
    for (v, id) in [
        (json!(true), "bool"),
        (json!(1), "i"),
        (json!(1.0), "f"),
        (json!("A"), "b"),
        (Value::Null, "null"),
    ] {
        assert_eq!(
            db.page_in("web", "users", None, 9, Some(("u", &v)))
                .unwrap()
                .documents[0]["id"],
            id
        );
    }
    for v in [json!({}), json!([])] {
        assert_eq!(
            db.put_in("web", "users", "c", json!({"u":v})),
            Err(Error::Invalid)
        );
    }
    db.put_in("web", "users", "second-null", json!({"u":null}))
        .unwrap();
    for (w, d) in [("web", "other"), ("other", "users")] {
        db.create_index(w, d, "u", true).unwrap();
        db.put_in(w, d, "c", json!({"u":"A"})).unwrap();
    }
    db.delete_in("web", "users", "b").unwrap();
    db.put_in("web", "users", "reused", json!({"u":"A"}))
        .unwrap();
    db = Documents(Database::open(db.0.into_storage(), Config::default()).unwrap());
    db.check().unwrap();
    assert_eq!(
        db.explain("web", "users", "u", &json!("A")).unwrap(),
        "secondary_index"
    );
    db.drop_index("web", "users", "u").unwrap();
    assert_eq!(
        db.explain("web", "users", "u", &json!("A")).unwrap(),
        "bounded_scan"
    );
    db.put_in("web", "users", "duplicate", json!({"u":"A"}))
        .unwrap();
    assert_eq!(
        db.create_index("web", "users", "u", true),
        Err(Error::Unique)
    );
    assert!(db.indexes("web", "users").unwrap().is_empty());
    db.create_index("web", "users", "u", false).unwrap();
    let first = db
        .page_in("web", "users", None, 1, Some(("u", &json!("A"))))
        .unwrap();
    assert_eq!(first.documents[0]["id"], "duplicate");
    assert_eq!(
        db.page_in(
            "web",
            "users",
            first.next_after.as_deref(),
            1,
            Some(("u", &json!("A")))
        )
        .unwrap()
        .documents[0]["id"],
        "reused"
    );
    db.check().unwrap();
}
struct Fault {
    active: MemoryStorage,
    durable: MemoryStorage,
    remaining: Rc<Cell<usize>>,
    tear: bool,
}
impl Fault {
    fn fail(&self) -> bool {
        let left = self.remaining.get();
        self.remaining.set(left.saturating_sub(1));
        left == 1
    }
}
impl Storage for Fault {
    fn len(&mut self) -> Result<u64> {
        self.active.len()
    }
    fn read(&mut self, p: u64, b: &mut [u8]) -> Result<()> {
        self.active.read(p, b)
    }
    fn resize(&mut self, n: u64) -> Result<()> {
        if self.fail() {
            return Err(Error::Io);
        }
        self.active.resize(n)
    }
    fn write(&mut self, p: u64, b: &[u8]) -> Result<()> {
        if self.fail() {
            if self.tear {
                self.active.write(p, &b[..b.len() / 2])?;
                self.durable = self.active.clone();
            }
            return Err(Error::Io);
        }
        self.active.write(p, b)
    }
    fn sync(&mut self) -> Result<()> {
        let fail = self.fail();
        if !fail || self.tear {
            self.durable = self.active.clone()
        }
        if fail {
            Err(Error::Io)
        } else {
            Ok(())
        }
    }
}
#[test]
fn every_mutating_adapter_boundary_preserves_index_agreement() {
    for op in ["create", "drop", "replace", "delete", "insert", "swap"] {
        let mut db = db();
        db.put_in("web", "users", "a", json!({"u":"A"})).unwrap();
        db.put_in("web", "users", "b", json!({"u":"B"})).unwrap();
        if op != "create" {
            db.create_index("web", "users", "u", true).unwrap();
        }
        let old = db.0.stats().2;
        let base = db.0.into_storage();
        for tear in [false, true] {
            let mut complete = false;
            for step in 1..500 {
                let left = Rc::new(Cell::new(step));
                let storage = Fault {
                    active: base.clone(),
                    durable: base.clone(),
                    remaining: left.clone(),
                    tear,
                };
                let mut db = Documents(Database::open(storage, Config::default()).unwrap());
                let r = match op {
                    "create" => db.create_index("web", "users", "u", true),
                    "drop" => db.drop_index("web", "users", "u"),
                    "replace" => db
                        .update_in("web", "users", "a", json!({"u":"C"}))
                        .map(drop),
                    "delete" => db.delete_in("web", "users", "b"),
                    "insert" => db
                        .insert_in("web", "users", "c", json!({"u":"C"}))
                        .map(drop),
                    _ => db.write_batch(vec![
                        change("a", json!({"u":"B"})),
                        change("b", json!({"u":"A"})),
                    ]),
                };
                let mut recovered = Documents(
                    Database::open(db.0.into_storage().durable, Config::default()).unwrap(),
                );
                recovered.check().unwrap();
                let gen = recovered.0.stats().2;
                assert!(gen == old || gen == old + 1);
                let new = gen == old + 1;
                assert_eq!(
                    !recovered.indexes("web", "users").unwrap().is_empty(),
                    match op {
                        "create" => new,
                        "drop" => !new,
                        _ => true,
                    }
                );
                let get = |db: &mut Documents<MemoryStorage>, id| {
                    db.get_in("web", "users", id)
                        .unwrap()
                        .map(|v| v["u"].clone())
                };
                assert_eq!(
                    get(&mut recovered, "a"),
                    Some(json!(match (op, new) {
                        ("replace", true) => "C",
                        ("swap", true) => "B",
                        _ => "A",
                    }))
                );
                assert_eq!(
                    get(&mut recovered, "b"),
                    if op == "delete" && new {
                        None
                    } else {
                        Some(json!(if op == "swap" && new { "A" } else { "B" }))
                    }
                );
                assert_eq!(
                    get(&mut recovered, "c"),
                    (op == "insert" && new).then(|| json!("C"))
                );
                if r.is_ok() {
                    assert!(new);
                    complete = true;
                    break;
                }
            }
            assert!(complete);
        }
    }
}
#[test]
fn corrupted_entry_and_missing_claim_are_detected() {
    for missing in [false, true] {
        let mut db = db();
        db.put("users", "a", json!({"u":"name"})).unwrap();
        db.create_index("default", "users", "u", true).unwrap();
        let mut key = Vec::new();
        db.0.scan_from(b"\0K3\0", |k, _| {
            key = k.to_vec();
            false
        })
        .unwrap();
        if missing {
            db.0.delete(&key).unwrap()
        } else {
            db.0.put(&key, b"wrong-owner").unwrap()
        }
        assert_eq!(db.check(), Err(Error::Corrupt));
    }
}