mod embedded;
pub use embedded::{Den, DocumentChange, Warren};
use ratdb_core::{Change, Database, Error, Result, Storage};
use serde_json::Value;
pub struct Documents<S: Storage>(pub Database<S>);
impl<S: Storage> Documents<S> {
    /// Upsert a JSON object. The external ID is authoritative and stored in `id`.
    pub fn put(&mut self, c: &str, id: &str, value: Value) -> Result<()> {
        self.put_in(DEFAULT_WARREN, c, id, value)
    }
    pub fn get(&mut self, c: &str, id: &str) -> Result<Option<Value>> {
        self.get_in(DEFAULT_WARREN, c, id)
    }
    pub fn delete(&mut self, c: &str, id: &str) -> Result<()> {
        self.delete_in(DEFAULT_WARREN, c, id)
    }
    /// Streams up to `limit` matching documents, examining at most `budget` records.
    /// Returns records examined. Callers should report a budget-limited result as partial.
    pub fn scan(
        &mut self,
        c: &str,
        filter: Option<(&str, &Value)>,
        limit: usize,
        budget: usize,
        mut visit: impl FnMut(Value),
    ) -> Result<usize> {
        self.scan_den(
            &namespace(DEFAULT_WARREN, c)?,
            filter,
            limit,
            budget,
            &mut visit,
        )
        .map(|(examined, _)| examined)
    }
    /// Streams nonempty collection names, once each, without building a collection cache.
    pub fn collections(&mut self, mut visit: impl FnMut(&str)) -> Result<()> {
        let mut previous = String::new();
        let mut error = None;
        self.0.scan_from(&[1], |k, _| {
            let Ok((_, c, _)) = decode(k) else {
                error = Some(Error::Corrupt);
                return false;
            };
            if c != previous {
                visit(c);
                previous = c.into();
            }
            true
        })?;
        error.map_or(Ok(()), Err)
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    use ratdb_core::{Config, MemoryStorage};
    use serde_json::json;
    #[test]
    fn catalog_reopen_and_guarded_deletion() {
        let cfg = Config {
            bank_size: 8192,
            ..Config::default()
        };
        let mut db = Documents(Database::open(MemoryStorage::default(), cfg).unwrap());
        db.create_namespace("garage", Some("cars")).unwrap();
        let stats = db.0.stats();
        db.create_namespace("garage", Some("cars")).unwrap();
        assert_eq!(db.0.stats(), stats);
        let mut db = Documents(Database::open(db.0.into_storage(), cfg).unwrap());
        assert_eq!(db.names(None, None).unwrap().0, ["garage"]);
        assert_eq!(db.names(Some("garage"), None).unwrap().0, ["cars"]);
        assert!(db.names(Some("garage"), Some("cars")).unwrap().0.is_empty());
        assert_eq!(
            db.drop_empty_namespace("garage", None).unwrap(),
            DropNamespace::NotEmpty
        );
        db.put_in("garage", "cars", "x", json!({})).unwrap();
        assert_eq!(
            db.drop_empty_namespace("garage", Some("cars")).unwrap(),
            DropNamespace::NotEmpty
        );
        db.delete_in("garage", "cars", "x").unwrap();
        assert_eq!(
            db.drop_empty_namespace("garage", Some("cars")).unwrap(),
            DropNamespace::Dropped
        );
        assert_eq!(
            db.drop_empty_namespace("garage", None).unwrap(),
            DropNamespace::Dropped
        );
        assert_eq!(
            db.drop_empty_namespace("garage", None).unwrap(),
            DropNamespace::Missing
        );
        assert!(db.create_namespace("bad\0name", None).is_err());
        db.put("legacy", "id", json!({})).unwrap();
        assert_eq!(
            db.drop_empty_namespace(DEFAULT_WARREN, None).unwrap(),
            DropNamespace::NotEmpty
        );
        db.0.put(b"\0C1\0bad", b"unexpected").unwrap();
        assert_eq!(db.create_namespace("bad", None), Err(Error::Corrupt));
        assert_eq!(db.names(None, None), Err(Error::Corrupt));
    }

    #[test]
    fn page_and_drop_guards_work_past_ten_thousand() {
        let cfg = Config {
            bank_size: 1024 * 1024,
            ..Config::default()
        };
        let mut db = Documents(Database::open(MemoryStorage::default(), cfg).unwrap());
        for first in (0..10001).step_by(1000) {
            let records: Vec<_> = (first..(first + 1000).min(10001))
                .map(|n| {
                    let id = format!("{n:05}");
                    let key = shiny_key("a", "bulk", &id).unwrap();
                    (
                        key,
                        serde_json::to_vec(&json!({"id":id,"match":n==10000})).unwrap(),
                    )
                })
                .collect();
            let changes: Vec<_> = records
                .iter()
                .map(|(k, v)| Change {
                    key: k,
                    value: Some(v),
                })
                .collect();
            db.0.apply(&changes).unwrap();
        }
        let first = db
            .page_in("a", "bulk", None, 1, Some(("match", &json!(true))))
            .unwrap();
        assert_eq!(first.examined, 10000);
        assert!(first.documents.is_empty());
        assert_eq!(first.next_after.as_deref(), Some("09999"));
        let last = db
            .page_in(
                "a",
                "bulk",
                first.next_after.as_deref(),
                1,
                Some(("match", &json!(true))),
            )
            .unwrap();
        assert_eq!(last.documents[0]["id"], "10000");
        assert!(last.next_after.is_none());
        db.put_in("z", "tail", "only", json!({})).unwrap();
        let mut ids = Vec::new();
        assert_eq!(
            db.scan_in("z", "tail", None, |v| ids.push(v["id"].clone()))
                .unwrap(),
            (1, false)
        );
        assert_eq!(ids, [json!("only")]);
        assert_eq!(
            db.names(Some("z"), Some("tail")).unwrap(),
            (vec!["only".into()], false)
        );
        assert!(db.names(Some("z"), None).unwrap().1);
        assert_eq!(
            db.drop_empty_namespace("z", None).unwrap(),
            DropNamespace::NotEmpty
        );
        assert_eq!(
            db.drop_empty_namespace("z", Some("tail")).unwrap(),
            DropNamespace::NotEmpty
        );
        assert!(db.get_in("z", "tail", "only").unwrap().is_some());
    }
    #[test]
    fn documents() {
        let mut d = Documents(
            Database::open(
                MemoryStorage::default(),
                Config {
                    bank_size: 4096,
                    ..Config::default()
                },
            )
            .unwrap(),
        );
        d.put("cars", "1", json!({"make":"Ford"})).unwrap();
        d.put("users", "1", json!({})).unwrap();
        assert_eq!(d.get("cars", "1").unwrap().unwrap()["make"], "Ford");
        assert!(d.put("cars", "2", json!([])).is_err());
        assert!(d.put("cars", "2", json!({"id":"3"})).is_err());
        let mut n = 0;
        d.scan("cars", Some(("make", &json!("Ford"))), 10, 100, |_| n += 1)
            .unwrap();
        assert_eq!(n, 1);
        let mut names = Vec::new();
        d.collections(|c| names.push(c.to_owned())).unwrap();
        assert_eq!(names, vec!["cars", "users"]);
        assert_eq!(d.scan("cars", None, 10, 0, |_| panic!()).unwrap(), 0);
        d.put("cars", "1", json!({"year":2021})).unwrap();
        assert!(d.get("cars", "1").unwrap().unwrap().get("make").is_none());
        d.delete("cars", "1").unwrap();
        assert!(d.get("cars", "1").unwrap().is_none());
        assert!(d.get("users", "1").unwrap().is_some());
    }
    #[test]
    fn document_check_rejects_checksum_valid_semantic_damage() {
        let cfg = Config {
            bank_size: 8192,
            ..Config::default()
        };
        let mut db = Documents(Database::open(MemoryStorage::default(), cfg).unwrap());
        db.create_namespace("country", Some("town")).unwrap();
        db.put_in("country", "other", "1", json!({})).unwrap();
        db.put("legacy", "1", json!({})).unwrap();
        db.check().unwrap();
        for (key, value) in [
            (&b"cars\0wrong"[..], &br#"{"id":"right"}"#[..]),
            (b"cars\0malformed", b"["),
            (b"cars\0array", b"[]"),
            (b"badkey", br#"{"id":"badkey"}"#),
            (b"\0W2\0w\0d\0id", br#"{"id":"id"}"#),
            (b"\0C1\0missing-parent\0den", b""),
            (b"\0C1\0bad-value", b"1"),
            (b"\0C1\0extra\0den\0part", b""),
        ] {
            db.0.put(key, value).unwrap();
            db.0.scan(|_, _| true).unwrap();
            assert_eq!(db.check(), Err(Error::Corrupt));
            db.0.delete(key).unwrap();
            db.check().unwrap();
        }
    }
}

pub const DEFAULT_WARREN: &str = "default";
const CATALOG: &[u8] = b"\0C1\0";

#[derive(Debug, PartialEq, Eq)]
pub enum DropNamespace {
    Dropped,
    Missing,
    NotEmpty,
}

fn catalog_key(w: &str, d: Option<&str>) -> Result<Vec<u8>> {
    if [Some(w), d]
        .into_iter()
        .flatten()
        .any(|n| n.is_empty() || n.len() > 256 || n.contains('\0'))
    {
        return Err(Error::Invalid);
    }
    let mut key = CATALOG.to_vec();
    key.extend_from_slice(w.as_bytes());
    if let Some(d) = d {
        key.push(0);
        key.extend_from_slice(d.as_bytes());
    }
    Ok(key)
}

fn catalog_entry<'a>(suffix: &'a [u8], value: &[u8]) -> Result<(&'a str, Option<&'a str>)> {
    let suffix = std::str::from_utf8(suffix).map_err(|_| Error::Corrupt)?;
    let (w, d) = suffix
        .split_once('\0')
        .map_or((suffix, None), |(w, d)| (w, Some(d)));
    if !value.is_empty()
        || [Some(w), d]
            .into_iter()
            .flatten()
            .any(|n| n.is_empty() || n.len() > 256 || n.contains('\0'))
    {
        return Err(Error::Corrupt);
    }
    Ok((w, d))
}
pub const MAX_PAGE_RESULTS: usize = 100;
const PAGE_BYTES: usize = 512 * 1024;
const PAGE_EXAMINED: usize = 10_000;

pub struct Page {
    pub documents: Vec<Value>,
    /// Exclusive lower bound to pass to the next request.
    pub next_after: Option<String>,
    /// Candidate records in this den, excluding any work locating the start key.
    pub examined: usize,
}
fn namespace(w: &str, d: &str) -> Result<Vec<u8>> {
    for name in [w, d] {
        if name.is_empty() || name.contains('\0') || (w != DEFAULT_WARREN && name.len() > 256) {
            return Err(Error::Invalid);
        }
    }
    if w == DEFAULT_WARREN {
        return Ok(format!("{d}\0").into_bytes());
    }
    // Leading NUL cannot occur in any valid legacy document key. Every component
    // excludes NUL, so delimiters are unambiguous (including arbitrary UTF-8).
    Ok(format!("\0W1\0{w}\0{d}\0").into_bytes())
}
fn shiny_key(w: &str, d: &str, id: &str) -> Result<Vec<u8>> {
    if id.is_empty() || id.contains('\0') {
        return Err(Error::Invalid);
    }
    let mut k = namespace(w, d)?;
    k.extend_from_slice(id.as_bytes());
    Ok(k)
}
fn decode(k: &[u8]) -> Result<(&str, &str, &str)> {
    let text = std::str::from_utf8(k).map_err(|_| Error::Corrupt)?;
    let mut parts = text.split('\0');
    let den = parts.next().ok_or(Error::Corrupt)?;
    if den.is_empty() {
        if parts.next() != Some("W1") {
            return Err(Error::Corrupt);
        }
        let w = parts.next().ok_or(Error::Corrupt)?;
        let d = parts.next().ok_or(Error::Corrupt)?;
        let id = parts.next().ok_or(Error::Corrupt)?;
        if w.is_empty()
            || w.len() > 256
            || w == DEFAULT_WARREN
            || d.is_empty()
            || d.len() > 256
            || id.is_empty()
            || parts.next().is_some()
        {
            return Err(Error::Corrupt);
        }
        Ok((w, d, id))
    } else {
        let id = parts.next().ok_or(Error::Corrupt)?;
        if id.is_empty() || parts.next().is_some() {
            return Err(Error::Corrupt);
        }
        Ok((DEFAULT_WARREN, den, id))
    }
}
impl<S: Storage> Documents<S> {
    pub fn put_in(&mut self, w: &str, d: &str, id: &str, mut value: Value) -> Result<()> {
        let k = shiny_key(w, d, id)?;
        let obj = value.as_object_mut().ok_or(Error::Invalid)?;
        if obj.get("id").is_some_and(|v| v.as_str() != Some(id)) {
            return Err(Error::Invalid);
        }
        obj.insert("id".into(), Value::String(id.into()));
        self.0
            .put(&k, &serde_json::to_vec(&value).map_err(|_| Error::Invalid)?)
    }
    pub fn get_in(&mut self, w: &str, d: &str, id: &str) -> Result<Option<Value>> {
        self.0
            .get(&shiny_key(w, d, id)?)?
            .map(|b| stored_document(&b, id))
            .transpose()
    }
    /// Returns false for an existing ID. Each successful insertion commits once.
    pub fn insert_in(&mut self, w: &str, d: &str, id: &str, value: Value) -> Result<bool> {
        if self.get_in(w, d, id)?.is_some() {
            return Ok(false);
        }
        self.put_in(w, d, id, value)?;
        Ok(true)
    }

    /// Sets top-level fields on an existing document with one commit. Null is a value.
    pub fn update_in(&mut self, w: &str, d: &str, id: &str, fields: Value) -> Result<bool> {
        let Value::Object(fields) = fields else {
            return Err(Error::Invalid);
        };
        if fields.is_empty() || fields.contains_key("id") {
            return Err(Error::Invalid);
        }
        let Some(mut value) = self.get_in(w, d, id)? else {
            return Ok(false);
        };
        value.as_object_mut().ok_or(Error::Corrupt)?.extend(fields);
        self.put_in(w, d, id, value)?;
        Ok(true)
    }

    /// Returns a bounded page in bytewise ID order. Resume with the same den and filter.
    pub fn page_in(
        &mut self,
        w: &str,
        d: &str,
        after: Option<&str>,
        limit: usize,
        filter: Option<(&str, &Value)>,
    ) -> Result<Page> {
        if !(1..=MAX_PAGE_RESULTS).contains(&limit) {
            return Err(Error::Invalid);
        }
        let prefix = namespace(w, d)?;
        let start = match after {
            Some(id) => shiny_key(w, d, id)?,
            None => prefix.clone(),
        };
        let mut page = Page {
            documents: Vec::new(),
            next_after: None,
            examined: 0,
        };
        let mut last = None;
        let mut bytes = 0;
        let mut error = None;
        self.0.scan_from(&start, |key, raw| {
            if after.is_some() && key == start {
                return true;
            }
            if !key.starts_with(&prefix) {
                return false;
            }
            if page.documents.len() == limit || page.examined == PAGE_EXAMINED {
                page.next_after = last.take();
                return false;
            }
            let result = (|| {
                let (_, _, id) = decode(key)?;
                let value = stored_document(raw, id)?;
                page.examined += 1;
                if filter.is_none_or(|(field, expected)| value.get(field) == Some(expected)) {
                    let size = serde_json::to_vec(&value)
                        .map_err(|_| Error::Corrupt)?
                        .len();
                    if size > PAGE_BYTES {
                        return Err(Error::Full);
                    }
                    if size > PAGE_BYTES - bytes {
                        page.next_after = last.take();
                        return Ok(false);
                    }
                    bytes += size;
                    page.documents.push(value);
                }
                last = Some(id.to_owned());
                Ok(true)
            })();
            match result {
                Ok(proceed) => proceed,
                Err(e) => {
                    error = Some(e);
                    false
                }
            }
        })?;
        error.map_or(Ok(page), Err)
    }
    pub fn delete_in(&mut self, w: &str, d: &str, id: &str) -> Result<()> {
        self.0.delete(&shiny_key(w, d, id)?)
    }
    /// Persist a warren, or a den and its parent together. Repeated creation is idempotent.
    pub fn create_namespace(&mut self, w: &str, d: Option<&str>) -> Result<()> {
        let wkey = catalog_key(w, None)?;
        let dkey = d.map(|den| catalog_key(w, Some(den))).transpose()?;
        let mut changes = Vec::new();
        for key in std::iter::once(&wkey).chain(dkey.as_ref()) {
            match self.0.get(key)? {
                None => changes.push(Change {
                    key,
                    value: Some(&[]),
                }),
                Some(value) if value.is_empty() => {}
                Some(_) => return Err(Error::Corrupt),
            }
        }
        self.0.apply(&changes)
    }
    fn has_prefix(&mut self, prefix: &[u8]) -> Result<bool> {
        let mut found = false;
        self.0.scan_from(prefix, |key, _| {
            found = key.starts_with(prefix);
            false
        })?;
        Ok(found)
    }
    /// Remove only the named catalog marker. Refuse both explicit children and implicit data.
    pub fn drop_empty_namespace(&mut self, w: &str, d: Option<&str>) -> Result<DropNamespace> {
        let marker = catalog_key(w, d)?;
        let has_children = if let Some(den) = d {
            self.has_prefix(&namespace(w, den)?)?
        } else {
            let mut child_prefix = marker.clone();
            child_prefix.push(0);
            if self.has_prefix(&child_prefix)? {
                return Ok(DropNamespace::NotEmpty);
            }
            if w == DEFAULT_WARREN {
                // Every legacy den starts with a non-NUL UTF-8 byte.
                let mut found = false;
                self.0.scan_from(&[1], |_, _| {
                    found = true;
                    false
                })?;
                found
            } else {
                self.has_prefix(format!("\0W1\0{w}\0").as_bytes())?
            }
        };
        if has_children {
            return Ok(DropNamespace::NotEmpty);
        }
        match self.0.get(&marker)? {
            None => Ok(DropNamespace::Missing),
            Some(v) if !v.is_empty() => Err(Error::Corrupt),
            Some(_) => {
                self.0.delete(&marker)?;
                Ok(DropNamespace::Dropped)
            }
        }
    }
    pub fn scan_in(
        &mut self,
        w: &str,
        d: &str,
        filter: Option<(&str, &Value)>,
        mut visit: impl FnMut(Value),
    ) -> Result<(usize, bool)> {
        self.scan_den(
            &namespace(w, d)?,
            filter,
            MAX_PAGE_RESULTS,
            PAGE_EXAMINED,
            &mut visit,
        )
    }
    fn scan_den(
        &mut self,
        prefix: &[u8],
        filter: Option<(&str, &Value)>,
        limit: usize,
        budget: usize,
        visit: &mut impl FnMut(Value),
    ) -> Result<(usize, bool)> {
        let (mut examined, mut matches) = (0, 0);
        let mut limited = false;
        let mut error = None;
        self.0.scan_from(prefix, |k, raw| {
            if !k.starts_with(prefix) {
                return false;
            }
            if examined >= budget || matches >= limit {
                limited = true;
                return false;
            }
            examined += 1;
            match decode(k).and_then(|(_, _, id)| stored_document(raw, id)) {
                Ok(value) => {
                    if filter.is_none_or(|(field, wanted)| value.get(field) == Some(wanted)) {
                        matches += 1;
                        visit(value);
                    }
                    true
                }
                Err(e) => {
                    error = Some(e);
                    false
                }
            }
        })?;
        error.map_or(Ok((examined, limited)), Err)
    }

    /// Validate the selected snapshot, document key/value pairs, and explicit namespace markers.
    pub fn check(&mut self) -> Result<()> {
        let mut parent = String::new();
        let mut error = None;
        self.0.scan(|k, raw| {
            let valid = if let Some(suffix) = k.strip_prefix(CATALOG) {
                catalog_entry(suffix, raw).and_then(|(w, d)| {
                    if d.is_none() {
                        parent.clear();
                        parent.push_str(w);
                    } else if parent != w {
                        return Err(Error::Corrupt);
                    }
                    Ok(())
                })
            } else {
                decode(k)
                    .and_then(|(_, _, id)| stored_document(raw, id))
                    .map(drop)
            };
            error = valid.err();
            error.is_none()
        })?;
        error.map_or(Ok(()), Err)
    }
    /// Bounded namespace/ID listing. None lists warrens; Some(w),None lists dens.
    /// Examines at most 10,000 records and retains at most 100 distinct names.
    pub fn names(&mut self, w: Option<&str>, d: Option<&str>) -> Result<(Vec<String>, bool)> {
        if let (Some(w), Some(d)) = (w, d) {
            let prefix = namespace(w, d)?;
            let mut ids = Vec::new();
            let mut limited = false;
            let mut error = None;
            self.0.scan_from(&prefix, |k, _| {
                if !k.starts_with(&prefix) {
                    return false;
                }
                if ids.len() == MAX_PAGE_RESULTS {
                    limited = true;
                    return false;
                }
                match decode(k) {
                    Ok((_, _, id)) => {
                        ids.push(id.to_owned());
                        true
                    }
                    Err(e) => {
                        error = Some(e);
                        false
                    }
                }
            })?;
            return error.map_or(Ok((ids, limited)), Err);
        }
        let mut names = std::collections::BTreeSet::new();
        let mut count = 0;
        let mut limited = false;
        let mut error = None;
        self.0.scan(|k, value| {
            if count == 10000 {
                limited = true;
                return false;
            }
            count += 1;
            let entry = if let Some(suffix) = k.strip_prefix(CATALOG) {
                catalog_entry(suffix, value)
            } else {
                decode(k).map(|(w, d, _)| (w, Some(d)))
            };
            match entry {
                Ok((kw, kd)) => {
                    let name = match (w, d) {
                        (None, None) => Some(kw),
                        (Some(w), None) if w == kw => kd,
                        _ => None,
                    };
                    if let Some(n) = name {
                        if !names.contains(n) && names.len() == 100 {
                            limited = true;
                            return false;
                        }
                        names.insert(n.to_owned());
                    }
                }
                Err(e) => {
                    error = Some(e);
                    return false;
                }
            }
            true
        })?;
        if let Some(e) = error {
            return Err(e);
        }
        Ok((names.into_iter().collect(), limited))
    }
}
fn stored_document(raw: &[u8], id: &str) -> Result<Value> {
    let value: Value = serde_json::from_slice(raw).map_err(|_| Error::Corrupt)?;
    if value.get("id").and_then(Value::as_str) != Some(id) {
        return Err(Error::Corrupt);
    }
    Ok(value)
}
#[cfg(test)]
mod namespace_tests {
    use super::*;
    use ratdb_core::{Config, MemoryStorage};
    use serde_json::json;
    #[test]
    fn atomic_update_reopen_and_create_only() {
        let cfg = Config {
            bank_size: 8192,
            max_value: 256,
            ..Config::default()
        };
        let mut db = Documents(Database::open(MemoryStorage::default(), cfg).unwrap());
        assert!(db
            .insert_in(
                "app",
                "users",
                "a",
                json!({"name":"Ann","config":{"old":true}})
            )
            .unwrap());
        let before = db.0.stats();
        assert!(!db
            .insert_in("app", "users", "a", json!({"name":"Bob"}))
            .unwrap());
        for fields in [
            json!({}),
            json!([]),
            json!({"id":"a"}),
            json!({"too_long":"x".repeat(300)}),
        ] {
            assert!(db.update_in("app", "users", "a", fields).is_err());
        }
        assert!(!db
            .update_in("app", "users", "absent", json!({"name":"B"}))
            .unwrap());
        assert_eq!(db.0.stats(), before);
        assert!(db
            .update_in(
                "app",
                "users",
                "a",
                json!({"config":{"new":true},"comment":null})
            )
            .unwrap());
        assert_eq!(db.0.stats().2, before.2 + 1);
        let mut db = Documents(Database::open(db.0.into_storage(), cfg).unwrap());
        assert_eq!(
            db.get_in("app", "users", "a").unwrap(),
            Some(json!({"id":"a","name":"Ann","config":{"new":true},"comment":null}))
        );
    }
    #[test]
    fn pages_resume_at_byte_limit_and_validate_namespace() {
        let cfg = Config {
            bank_size: 1024 * 1024,
            ..Config::default()
        };
        let mut db = Documents(Database::open(MemoryStorage::default(), cfg).unwrap());
        db.put_in("app", "other", "outside", json!({})).unwrap();
        for id in ["c", "a", "b"] {
            db.put_in("app", "people", id, json!({"pad":"z".repeat(200*1024)}))
                .unwrap();
        }
        let page = db.page_in("app", "people", None, 100, None).unwrap();
        assert_eq!(page.documents.len(), 2);
        assert_eq!(page.next_after.as_deref(), Some("b"));
        db.delete_in("app", "people", "b").unwrap();
        db.put_in("app", "people", "aa", json!({})).unwrap();
        let page = db
            .page_in("app", "people", page.next_after.as_deref(), 100, None)
            .unwrap();
        assert_eq!(page.documents.len(), 1);
        assert_eq!(page.documents[0]["id"], "c");
        assert!(page.next_after.is_none());
        for limit in [0, 101] {
            assert!(db.page_in("app", "people", None, limit, None).is_err());
        }
        assert!(db.page_in("app", "people", Some("a\0b"), 1, None).is_err());
        assert!(db
            .page_in("app", "empty", None, 1, None)
            .unwrap()
            .documents
            .is_empty());
    }
    #[test]
    fn isolation_legacy_and_reopen() {
        let config = Config {
            bank_size: 65536,
            ..Config::default()
        };
        let mut db = Documents(Database::open(MemoryStorage::default(), config).unwrap());
        db.put("cars", "1", json!({"old":true})).unwrap();
        let long_den = "d".repeat(300);
        db.put(&long_den, "legacy", json!({})).unwrap();
        assert!(db.get_in("default", &long_den, "legacy").unwrap().is_some());
        db.delete(&long_den, "legacy").unwrap();
        for (w, d, n) in [
            ("garage", "cars", 1),
            ("garage", "trucks", 2),
            ("work", "cars", 3),
            ("a/b", "c", 4),
            ("a", "b/c", 5),
        ] {
            db.put_in(w, d, "1", json!({"n":n})).unwrap();
        }
        let mut db = Documents(Database::open(db.0.into_storage(), config).unwrap());
        for (w, d, n) in [
            ("garage", "cars", 1),
            ("garage", "trucks", 2),
            ("work", "cars", 3),
            ("a/b", "c", 4),
            ("a", "b/c", 5),
        ] {
            assert_eq!(db.get_in(w, d, "1").unwrap().unwrap()["n"], n);
        }
        assert_eq!(
            db.get_in("default", "cars", "1").unwrap().unwrap()["old"],
            true
        );
        let mut legacy = vec![];
        db.collections(|c| legacy.push(c.to_owned())).unwrap();
        assert_eq!(legacy, vec!["cars"]);
        assert_eq!(
            db.names(Some("garage"), None).unwrap().0,
            vec!["cars", "trucks"]
        );
        db.delete_in("garage", "cars", "1").unwrap();
        assert!(db.get_in("work", "cars", "1").unwrap().is_some());
        assert!(db.put_in("a\0b", "c", "1", json!({})).is_err());
    }
    #[test]
    fn bounded_ids() {
        let mut db = Documents(
            Database::open(
                MemoryStorage::default(),
                Config {
                    bank_size: 65536,
                    ..Config::default()
                },
            )
            .unwrap(),
        );
        for n in 0..105 {
            db.put_in("w", "d", &n.to_string(), json!({})).unwrap();
        }
        let (ids, limited) = db.names(Some("w"), Some("d")).unwrap();
        assert_eq!(ids.len(), 100);
        assert!(limited);
    }
}