use ratdb_core::{Config, Database, Error, MemoryStorage};
use ratdb_document::{Documents, PageBudget};
use serde_json::{json, Value};

#[test]
fn exact_generations_include_index_definitions_and_postings() {
    let mut db =
        Documents(Database::create_indexed(MemoryStorage::default(), Config::default()).unwrap());
    for (id, n) in [("a", 1), ("b", 2), ("c", 3), ("d", 4)] {
        db.put_in(
            "crm",
            "contacts",
            id,
            json!({"team":"old","email":id,"n":n}),
        )
        .unwrap();
    }
    db.create_index("crm", "contacts", "team", false).unwrap();
    db.create_index("crm", "contacts", "email", true).unwrap();
    let mut all = db
        .begin_snapshot("crm", "contacts", None, PageBudget::ordinary(1), 0)
        .unwrap();
    let mut filtered = db
        .begin_snapshot(
            "crm",
            "contacts",
            Some(("team".into(), json!("old"))),
            PageBudget::ordinary(2),
            0,
        )
        .unwrap();
    let mut unique = db
        .begin_snapshot(
            "crm",
            "contacts",
            Some(("email".into(), json!("b"))),
            PageBudget::ordinary(1),
            0,
        )
        .unwrap();
    let mut primary = db
        .begin_snapshot(
            "crm",
            "contacts",
            Some(("id".into(), json!("b"))),
            PageBudget::ordinary(1),
            0,
        )
        .unwrap();
    assert_eq!(all.next_page(&mut db, 1, 1).unwrap().documents[0]["n"], 1);
    db.delete_in("crm", "contacts", "c").unwrap();
    db.put_in("crm", "contacts", "aa", json!({"team":"new","email":"c"}))
        .unwrap();
    for i in 0..10 {
        db.update_in(
            "crm",
            "contacts",
            "b",
            json!({"team":"new","n":i+10,"email":"changed"}),
        )
        .unwrap();
    }
    assert_eq!(
        db.put_in("crm", "contacts", "bad", json!({"email":"changed"})),
        Err(Error::Unique)
    );
    db.drop_index("crm", "contacts", "team").unwrap();
    let mut values = Vec::new();
    loop {
        let p = all.next_page(&mut db, 2, 1).unwrap();
        values.extend(p.documents);
        if !p.has_more {
            break;
        }
    }
    assert_eq!(
        values
            .iter()
            .map(|d| d["n"].as_i64().unwrap())
            .collect::<Vec<_>>(),
        [2, 3, 4]
    );
    for s in [&mut unique, &mut primary] {
        let p = s.next_page(&mut db, 2, 1).unwrap();
        assert_eq!(p.documents[0]["n"], 2);
        assert!(!p.has_more);
    }
    let a = filtered.next_page(&mut db, 3, 2).unwrap();
    let b = filtered.next_page(&mut db, 4, 2).unwrap();
    assert_eq!(a.documents.len() + b.documents.len(), 4);
    assert_eq!(b.documents[0]["id"], "c");
    assert_eq!(
        db.get_in("crm", "contacts", "b").unwrap().unwrap()["email"],
        "changed"
    );
    assert!(matches!(
        all.next_page(&mut db, 4, 1),
        Err(Error::SnapshotInvalid)
    ));
    db.check().unwrap();
}
#[test]
fn sparse_scan_empty_pages_and_byte_limits() {
    let mut db =
        Documents(Database::create_indexed(MemoryStorage::default(), Config::default()).unwrap());
    let budget = PageBudget {
        rows: 3,
        examined: 2,
        bytes: 60,
    };
    let mut empty = db.begin_snapshot("w", "d", None, budget, 0).unwrap();
    for i in 0..6 {
        db.put_in("w", "d", &i.to_string(), json!({"v":i})).unwrap();
    }
    assert!(empty.next_page(&mut db, 0, 3).unwrap().documents.is_empty());
    let mut s = db
        .begin_snapshot("w", "d", Some(("v".into(), json!(5))), budget, 0)
        .unwrap();
    db.delete_in("w", "d", "5").unwrap();
    for _ in 0..2 {
        let p = s.next_page(&mut db, 0, 3).unwrap();
        assert!(p.documents.is_empty());
        assert!(p.has_more);
        assert_eq!(p.examined, 2);
    }
    let p = s.next_page(&mut db, 0, 3).unwrap();
    assert_eq!(p.documents[0]["v"], 5);
    assert!(!p.has_more);
    let mut small = db
        .begin_snapshot("w", "d", None, PageBudget { bytes: 1, ..budget }, 0)
        .unwrap();
    assert!(matches!(small.next_page(&mut db, 0, 3), Err(Error::Full)));
    assert!(matches!(
        small.next_page(&mut db, 0, 4),
        Err(Error::Invalid)
    ));
    assert!(matches!(
        db.begin_snapshot(
            "w",
            "d",
            Some(("v".into(), Value::String("x".repeat(4097)))),
            budget,
            0
        ),
        Err(Error::Invalid)
    ));
}