haematite-cli 0.21.7

//! GX1: a filtered `graph.export`, through the service.

use std::error::Error;
use std::num::NonZeroU32;

use haematite::db::{DatabaseConfig, ReadOnlyDatabase};
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json, value::RawValue};

use super::VALUE_BYTES_MOST;
use crate::serve::access::{Standalone, TestSource, request};
use crate::serve::protocol::{
    Cardinality, Change, CollectionWrite, DocumentImport, DocumentWrite, Edge, EdgeImport,
    GraphPage, ImportedDocument, Mutation, Node, Operation, Relation, Selection, Unrelated,
    Version,
};
use crate::serve::service::Service;
use crate::serve::storage::edges::visible_export;
use crate::serve::storage::reads::PageLimits;
use crate::serve::storage::{Captured, Failure, Storage};

type TestResult = Result<(), Box<dyn Error>>;

fn head(storage: &Storage) -> Result<Version, Failure> {
    Ok(storage
        .capture(&Selection {
            branch: "main".to_owned(),
            version: None,
        })?
        .version)
}

fn apply(storage: &mut Storage, id: &str, change: Change) -> Result<(), Failure> {
    let expected = head(storage)?;
    let max_page = NonZeroU32::new(2).ok_or_else(|| Failure::new("page_limit", "zero"))?;
    storage.mutate_bounded(
        &Mutation {
            request_id: id.to_owned(),
            branch: "main".to_owned(),
            expected,
            change,
        },
        max_page,
    )?;
    Ok(())
}

/// The fixture, with what today's export reads through kept beside it.
struct Fixture {
    _directory: tempfile::TempDir,
    service: Service,
    reader: ReadOnlyDatabase,
    at: Captured,
}

fn create() -> Result<(tempfile::TempDir, Storage), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 2,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    Ok((directory, storage))
}

fn serve(
    directory: tempfile::TempDir,
    storage: Storage,
    max_page: u32,
) -> Result<Fixture, Box<dyn Error>> {
    let reader = storage.reader()?;
    let at = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    let service = Service::new(storage, NonZeroU32::new(max_page).ok_or("not zero")?)?;
    Ok(Fixture {
        _directory: directory,
        service,
        reader,
        at,
    })
}

fn node(collection: &str, id: &str) -> Node {
    Node {
        collection: collection.to_owned(),
        document_id: id.to_owned(),
    }
}

/// messages m1, m2 (m1 holds a 1e2, an 18-digit number and a long body);
/// people p1, p2; a room r1. `sent_by` messages -> people, `posted_in`
/// messages -> rooms, `knows` people -> people. Edges: m1 `sent_by` p1, m2
/// `sent_by` p2, m1 `posted_in` r1, p1 `knows` p2.
fn service() -> Result<Fixture, Box<dyn Error>> {
    let (directory, mut storage) = create()?;
    for collection in ["messages", "people", "rooms"] {
        apply(
            &mut storage,
            collection,
            Change::Collection(CollectionWrite {
                collection: collection.to_owned(),
                schema: None,
            }),
        )?;
    }
    let long = "x".repeat(VALUE_BYTES_MOST);
    let documents = [
        (
            "messages",
            "m1",
            format!(r#"{{"weight": 1e2, "big": 123456789012345678, "body": "{long}"}}"#),
        ),
        ("messages", "m2", r#"{"weight": 7}"#.to_owned()),
        (
            "people",
            "p1",
            r#"{"name": "Ann", "kind": "nurse"}"#.to_owned(),
        ),
        ("people", "p2", r#"{"name": "Bo"}"#.to_owned()),
        ("rooms", "r1", r#"{"name": "Ward 3"}"#.to_owned()),
    ];
    for (collection, id, document) in documents {
        apply(
            &mut storage,
            id,
            Change::Put(DocumentWrite {
                collection: collection.to_owned(),
                document_id: id.to_owned(),
                document: RawValue::from_string(document)?,
            }),
        )?;
    }
    for (relation, from, to) in [
        ("sent_by", "messages", "people"),
        ("posted_in", "messages", "rooms"),
        ("knows", "people", "people"),
    ] {
        apply(
            &mut storage,
            relation,
            Change::RelationPut(Relation {
                relation: relation.to_owned(),
                from_collection: from.to_owned(),
                to_collection: to.to_owned(),
                cardinality: Cardinality::ManyToMany,
            }),
        )?;
    }
    for (relation, from, to) in [
        ("sent_by", node("messages", "m1"), node("people", "p1")),
        ("sent_by", node("messages", "m2"), node("people", "p2")),
        ("posted_in", node("messages", "m1"), node("rooms", "r1")),
        ("knows", node("people", "p1"), node("people", "p2")),
    ] {
        apply(
            &mut storage,
            &format!("{relation}-{}-{}", from.document_id, to.document_id),
            Change::EdgePut(Edge {
                relation: relation.to_owned(),
                from,
                to,
                properties: json!({}),
            }),
        )?;
    }
    serve(directory, storage, 100)
}

fn page(limit: u32) -> Result<GraphPage, Box<dyn Error>> {
    Ok(GraphPage {
        selection: Selection {
            branch: "main".to_owned(),
            version: None,
        },
        after: None,
        limit: NonZeroU32::new(limit).ok_or("not zero")?,
        relations: None,
        collections: None,
        members: None,
        records: None,
        unrelated: None,
    })
}

fn names(list: &[&str]) -> Vec<String> {
    list.iter().map(|name| (*name).to_owned()).collect()
}

fn export(service: &Service, page: GraphPage) -> Result<Value, Box<dyn Error>> {
    Ok(service.dispatch(Operation::GraphExport(page))?)
}

/// (relation, from id, to id) of every edge answered.
fn edges(answer: &Value) -> Vec<(String, String, String)> {
    answer["edges"]
        .as_array()
        .map(|edges| {
            edges
                .iter()
                .map(|edge| {
                    (
                        edge["relation"].as_str().unwrap_or_default().to_owned(),
                        edge["from"]["document_id"]
                            .as_str()
                            .unwrap_or_default()
                            .to_owned(),
                        edge["to"]["document_id"]
                            .as_str()
                            .unwrap_or_default()
                            .to_owned(),
                    )
                })
                .collect()
        })
        .unwrap_or_default()
}

fn edge(relation: &str, from: &str, to: &str) -> (String, String, String) {
    (relation.to_owned(), from.to_owned(), to.to_owned())
}

/// GX1 red: one relation named, only its edges.
#[test]
fn filter_by_relation() -> TestResult {
    let fixture = service()?;
    let service = &fixture.service;
    let mut asked = page(10)?;
    asked.relations = Some(names(&["sent_by"]));
    let answer = export(service, asked)?;
    assert_eq!(
        edges(&answer),
        vec![edge("sent_by", "m1", "p1"), edge("sent_by", "m2", "p2")]
    );
    assert_eq!(answer["next_after"], Value::Null);
    Ok(())
}

/// GX1 red: collections need both ends; with messages and people named,
/// messages -> rooms is left out.
#[test]
fn filter_by_collection_needs_both_ends() -> TestResult {
    let fixture = service()?;
    let service = &fixture.service;
    let mut asked = page(10)?;
    asked.collections = Some(names(&["messages", "people"]));
    let answer = export(service, asked)?;
    assert_eq!(
        edges(&answer),
        vec![
            edge("sent_by", "m1", "p1"),
            edge("sent_by", "m2", "p2"),
            edge("knows", "p1", "p2"),
        ]
    );
    Ok(())
}

/// GX1 red: members are their stored text: 1e2 and an 18-digit number
/// unchanged; an absent member is absent; an over-bound value is named
/// with its length and not carried.
#[test]
fn members_are_carried_raw_absent_and_over_bound_named() -> TestResult {
    let fixture = service()?;
    let service = &fixture.service;
    let mut asked = page(10)?;
    asked.relations = Some(names(&["sent_by"]));
    asked.members = Some(
        [
            ("messages".to_owned(), names(&["weight", "big", "body"])),
            ("people".to_owned(), names(&["name", "kind"])),
        ]
        .into_iter()
        .collect(),
    );
    let answer = export(service, asked)?;
    let nodes = answer["nodes"].as_array().ok_or("no nodes")?;
    let find = |collection: &str, id: &str| {
        nodes
            .iter()
            .find(|node| node["collection"] == collection && node["document_id"] == id)
            .cloned()
            .unwrap_or(Value::Null)
    };
    let m1 = find("messages", "m1");
    assert_eq!(m1["members"]["weight"], "1e2");
    assert_eq!(m1["members"]["big"], "123456789012345678");
    assert_eq!(m1["members"].get("body"), None);
    assert_eq!(m1["over_bound"][0]["member"], "body");
    assert_eq!(m1["over_bound"][0]["bytes"], VALUE_BYTES_MOST + 2);
    let p2 = find("people", "p2");
    assert_eq!(p2["members"]["name"], r#""Bo""#);
    assert_eq!(p2["members"].get("kind"), None);
    assert_eq!(nodes.len(), 4);
    Ok(())
}

/// GX1 red: a continuation under another filter is refused by name.
#[test]
fn a_continuation_with_a_changed_filter_is_refused() -> TestResult {
    let fixture = service()?;
    let service = &fixture.service;
    let mut asked = page(1)?;
    asked.relations = Some(names(&["sent_by"]));
    let first = export(service, asked)?;
    let after = first["next_after"]
        .as_str()
        .ok_or("no continuation")?
        .to_owned();
    let version: Version = serde_json::from_value(first["selection"]["version"].clone())?;
    let mut next = page(1)?;
    next.selection.version = Some(version);
    next.after = Some(after.clone());
    next.relations = Some(names(&["sent_by"]));
    assert_eq!(
        edges(&export(service, next)?),
        vec![edge("sent_by", "m2", "p2")]
    );
    let mut changed = page(1)?;
    changed.selection.version = Some(version);
    changed.after = Some(after);
    changed.relations = Some(names(&["posted_in"]));
    let refused = export(service, changed).err().ok_or("answered")?;
    assert!(refused.to_string().starts_with("graph_filter_changed"));
    Ok(())
}

/// GX1: an unknown relation or collection is refused by name.
#[test]
fn unknown_names_are_refused() -> TestResult {
    let fixture = service()?;
    let service = &fixture.service;
    let mut asked = page(10)?;
    asked.relations = Some(names(&["sent_by", "nobody"]));
    let refused = export(service, asked).err().ok_or("answered")?;
    assert!(refused.to_string().starts_with("graph_relation_unknown"));
    let mut asked = page(10)?;
    asked.collections = Some(names(&["nowhere"]));
    let refused = export(service, asked).err().ok_or("answered")?;
    assert!(refused.to_string().starts_with("graph_collection_unknown"));
    let mut asked = page(10)?;
    asked.relations = Some(Vec::new());
    assert_eq!(edges(&export(service, asked)?), Vec::new());
    Ok(())
}

/// With no filter, the export has the same members, ordered edges and
/// continuation as the unfiltered reader. JSON object member order is immaterial.
#[test]
fn absent_params_answer_as_today() -> TestResult {
    let fixture = service()?;
    let answer = export(&fixture.service, page(2)?)?;
    let keys: std::collections::BTreeSet<&str> = answer
        .as_object()
        .ok_or("not an object")?
        .keys()
        .map(String::as_str)
        .collect();
    assert_eq!(
        keys,
        ["edges", "next_after", "selection"].into_iter().collect()
    );
    let today = visible_export(
        &fixture.reader,
        &fixture.at,
        &request(&Standalone)?,
        None,
        PageLimits {
            rows: std::num::NonZeroUsize::new(2).ok_or("not zero")?,
            bytes: std::num::NonZeroUsize::MAX,
        },
    )?;
    assert_eq!(answer["edges"], serde_json::to_value(&today.entries)?);
    assert_eq!(answer["next_after"], json!(today.next_after));
    Ok(())
}

/// GX1 (Apollo's condition): 70 `posted_in` edges leave messages before
/// the one `sent_by` edge. At `max_page` 1 a page examines 64 keys, so the
/// first page of `sent_by` is empty, says the examination ended, and
/// carries a continuation; the second page finds the edge and ends.
#[test]
fn an_empty_page_at_the_examination_bound_says_so_and_continues() -> TestResult {
    let (directory, mut storage) = create()?;
    for collection in ["messages", "people", "rooms"] {
        apply(
            &mut storage,
            collection,
            Change::Collection(CollectionWrite {
                collection: collection.to_owned(),
                schema: None,
            }),
        )?;
    }
    let mut documents = (0..70)
        .map(|index| {
            Ok(ImportedDocument {
                document_id: format!("a{index:02}"),
                document: RawValue::from_string("{}".to_owned())?,
            })
        })
        .collect::<Result<Vec<_>, serde_json::Error>>()?;
    documents.push(ImportedDocument {
        document_id: "z".to_owned(),
        document: RawValue::from_string("{}".to_owned())?,
    });
    apply(
        &mut storage,
        "messages-documents",
        Change::PutMany(DocumentImport {
            collection: "messages".to_owned(),
            documents,
        }),
    )?;
    for (collection, id) in [("people", "p1"), ("rooms", "r1")] {
        apply(
            &mut storage,
            id,
            Change::Put(DocumentWrite {
                collection: collection.to_owned(),
                document_id: id.to_owned(),
                document: RawValue::from_string("{}".to_owned())?,
            }),
        )?;
    }
    for (relation, to) in [("sent_by", "people"), ("posted_in", "rooms")] {
        apply(
            &mut storage,
            relation,
            Change::RelationPut(Relation {
                relation: relation.to_owned(),
                from_collection: "messages".to_owned(),
                to_collection: to.to_owned(),
                cardinality: Cardinality::ManyToMany,
            }),
        )?;
    }
    let mut edges_put: Vec<Edge> = (0..70)
        .map(|index| Edge {
            relation: "posted_in".to_owned(),
            from: node("messages", &format!("a{index:02}")),
            to: node("rooms", "r1"),
            properties: json!({}),
        })
        .collect();
    edges_put.push(Edge {
        relation: "sent_by".to_owned(),
        from: node("messages", "z"),
        to: node("people", "p1"),
        properties: json!({}),
    });
    apply(
        &mut storage,
        "edges",
        Change::EdgePutMany(EdgeImport { edges: edges_put }),
    )?;
    let fixture = serve(directory, storage, 1)?;
    let mut asked = page(1)?;
    asked.relations = Some(names(&["sent_by"]));
    let first = export(&fixture.service, asked)?;
    assert_eq!(edges(&first), Vec::new());
    assert_eq!(first["examination_ended"], true);
    let after = first["next_after"]
        .as_str()
        .ok_or("no continuation")?
        .to_owned();
    let mut next = page(1)?;
    next.selection.version = Some(serde_json::from_value(
        first["selection"]["version"].clone(),
    )?);
    next.after = Some(after);
    next.relations = Some(names(&["sent_by"]));
    let second = export(&fixture.service, next)?;
    assert_eq!(edges(&second), vec![edge("sent_by", "z", "p1")]);
    assert_eq!(second["examination_ended"], false);
    assert_eq!(second["next_after"], Value::Null);
    Ok(())
}

/// GX2 red: m1, p1 and p2 named. m1's `posted_in` r1 has an end not named
/// and is left out; m2 is not named, so m2 `sent_by` p2 is not answered.
/// m1's edge and p1's edge leave two named records, two prefixes, and come
/// on one page.
#[test]
fn records_need_both_ends_and_one_page_reads_every_prefix() -> TestResult {
    let fixture = service()?;
    let mut asked = page(10)?;
    asked.records = Some(vec![
        node("messages", "m1"),
        node("people", "p1"),
        node("people", "p2"),
    ]);
    let answer = export(&fixture.service, asked)?;
    assert_eq!(
        edges(&answer),
        vec![edge("sent_by", "m1", "p1"), edge("knows", "p1", "p2")]
    );
    assert_eq!(answer["next_after"], Value::Null);
    assert_eq!(answer["examination_ended"], false);
    Ok(())
}

/// GX2 red: a continuation made under named records is taken under the
/// same records, from the next prefix, and refused under others.
#[test]
fn a_records_continuation_is_refused_under_other_records() -> TestResult {
    let fixture = service()?;
    let service = &fixture.service;
    let named = || {
        vec![
            node("messages", "m1"),
            node("people", "p1"),
            node("people", "p2"),
        ]
    };
    let mut asked = page(1)?;
    asked.records = Some(named());
    let first = export(service, asked)?;
    assert_eq!(edges(&first), vec![edge("sent_by", "m1", "p1")]);
    let after = first["next_after"]
        .as_str()
        .ok_or("no continuation")?
        .to_owned();
    let version: Version = serde_json::from_value(first["selection"]["version"].clone())?;
    let mut next = page(1)?;
    next.selection.version = Some(version);
    next.after = Some(after.clone());
    next.records = Some(named());
    let second = export(service, next)?;
    assert_eq!(edges(&second), vec![edge("knows", "p1", "p2")]);
    assert_eq!(second["next_after"], Value::Null);
    let mut changed = page(1)?;
    changed.selection.version = Some(version);
    changed.after = Some(after);
    changed.records = Some(vec![node("messages", "m1"), node("people", "p1")]);
    let refused = export(service, changed).err().ok_or("answered")?;
    assert!(refused.to_string().starts_with("graph_filter_changed"));
    Ok(())
}

/// `GX2c` red: the fixture's page bound is 100, so its examination bound is
/// 100 × 64 = 6,400; 6,401 named records are refused by name before any is
/// looked up, and a record not stored is refused.
#[test]
fn records_over_the_bound_or_unknown_are_refused() -> TestResult {
    let fixture = service()?;
    let service = &fixture.service;
    let mut asked = page(10)?;
    asked.records = Some(
        (0..6_401)
            .map(|index| node("people", &format!("q{index}")))
            .collect(),
    );
    let refused = export(service, asked).err().ok_or("answered")?;
    assert!(refused.to_string().starts_with("graph_records_bound"));
    let mut asked = page(10)?;
    asked.records = Some(vec![node("people", "p1"), node("people", "p9")]);
    let refused = export(service, asked).err().ok_or("answered")?;
    assert!(refused.to_string().starts_with("graph_record_unknown"));
    Ok(())
}

/// alpha a00..a39 and beta b00..b39, every one with one edge: a00 and b39
/// to targets t1, every other to other o1. 80 edges.
fn two_from_collections() -> Result<Fixture, Box<dyn Error>> {
    let (directory, mut storage) = create()?;
    for collection in ["alpha", "beta", "other", "targets"] {
        apply(
            &mut storage,
            collection,
            Change::Collection(CollectionWrite {
                collection: collection.to_owned(),
                schema: None,
            }),
        )?;
    }
    for (collection, letter) in [("alpha", 'a'), ("beta", 'b')] {
        let documents = (0..40)
            .map(|index| {
                Ok(ImportedDocument {
                    document_id: format!("{letter}{index:02}"),
                    document: RawValue::from_string("{}".to_owned())?,
                })
            })
            .collect::<Result<Vec<_>, serde_json::Error>>()?;
        apply(
            &mut storage,
            &format!("{collection}-documents"),
            Change::PutMany(DocumentImport {
                collection: collection.to_owned(),
                documents,
            }),
        )?;
    }
    for (collection, id) in [("other", "o1"), ("targets", "t1")] {
        apply(
            &mut storage,
            id,
            Change::Put(DocumentWrite {
                collection: collection.to_owned(),
                document_id: id.to_owned(),
                document: RawValue::from_string("{}".to_owned())?,
            }),
        )?;
    }
    for from in ["alpha", "beta"] {
        for to in ["other", "targets"] {
            apply(
                &mut storage,
                &format!("{from}_{to}"),
                Change::RelationPut(Relation {
                    relation: format!("{from}_{to}"),
                    from_collection: from.to_owned(),
                    to_collection: to.to_owned(),
                    cardinality: Cardinality::ManyToMany,
                }),
            )?;
        }
    }
    let mut edges_put = Vec::new();
    for (collection, letter, wanted) in [("alpha", 'a', 0), ("beta", 'b', 39)] {
        for index in 0..40 {
            let to = if index == wanted {
                node("targets", "t1")
            } else {
                node("other", "o1")
            };
            edges_put.push(Edge {
                relation: format!("{collection}_{}", to.collection),
                from: node(collection, &format!("{letter}{index:02}")),
                to,
                properties: json!({}),
            });
        }
    }
    apply(
        &mut storage,
        "edges",
        Change::EdgePutMany(EdgeImport { edges: edges_put }),
    )?;
    serve(directory, storage, 1)
}

/// `GX2b` red: the examination bound is one per page, shared by its
/// prefixes. At `max_page` 1 a page examines 64 keys. Page 1 takes a00's
/// edge, examines alpha's 40 keys and beta's first 24 (40 + 24 = 64), and
/// ends inside beta, the second prefix, without b39's edge. Following the
/// continuation answers b39's edge, and every wanted edge comes once.
#[test]
fn the_examination_bound_is_shared_by_a_page_s_prefixes() -> TestResult {
    let fixture = two_from_collections()?;
    let asked = || -> Result<GraphPage, Box<dyn Error>> {
        let mut asked = page(1)?;
        asked.collections = Some(names(&["alpha", "beta", "targets"]));
        Ok(asked)
    };
    let first = export(&fixture.service, asked()?)?;
    assert_eq!(edges(&first), vec![edge("alpha_targets", "a00", "t1")]);
    assert_eq!(first["examination_ended"], true);
    let mut after = first["next_after"]
        .as_str()
        .ok_or("no continuation")?
        .to_owned();
    let mut parts = after.split('.').skip(1);
    assert_eq!(parts.next(), Some("1"));
    assert_ne!(parts.next(), Some(""));
    let version: Version = serde_json::from_value(first["selection"]["version"].clone())?;
    let mut seen = edges(&first);
    for _ in 0..4 {
        let mut next = asked()?;
        next.selection.version = Some(version);
        next.after = Some(after);
        let answer = export(&fixture.service, next)?;
        seen.extend(edges(&answer));
        let Some(text) = answer["next_after"].as_str() else {
            assert_eq!(
                seen,
                vec![
                    edge("alpha_targets", "a00", "t1"),
                    edge("beta_targets", "b39", "t1"),
                ]
            );
            return Ok(());
        };
        after = text.to_owned();
    }
    Err("the export did not end within four more pages".into())
}

/// `GX2c` red: named records whose JSON is over the frame are refused by name,
/// before any is looked up, though their count is within the bound. Each of
/// the 100 is `{"collection":"people","document_id":"qNNN"}`, 44 bytes, so
/// with 99 commas and the brackets 4,501 bytes, over a 4,096-byte frame.
#[test]
fn records_over_the_frame_are_refused() -> TestResult {
    let fixture = service()?;
    let mut asked = page(10)?;
    asked.records = Some(
        (0..100)
            .map(|index| node("people", &format!("q{index:03}")))
            .collect(),
    );
    let frame = std::num::NonZeroUsize::new(4_096).ok_or("not zero")?;
    let refused = fixture
        .service
        .dispatch_bounded(Operation::GraphExport(asked), frame)
        .err()
        .ok_or("answered")?;
    let refused = serde_json::to_value(&refused)?;
    assert_eq!(refused["code"], "graph_records_bytes");
    assert_eq!(refused["path"], "/records");
    Ok(())
}

/// GR1: `notes` notes `n00`… with a title and no relation; `people` people
/// `p00`…; `knows` edges `p00` → `p01`… (`knows` of them); `messages` m1
/// `sent_by` p00. Served at `max_page`, with the version before the
/// relations beside it. `GR1b`: each collection's records are one import and
/// every edge one import, one durable operation each; only the declarations
/// go one by one, as no change declares several.
fn listed(
    notes: usize,
    people: usize,
    knows: usize,
    max_page: u32,
) -> Result<(Fixture, Version), Box<dyn Error>> {
    listed_citing(notes, people, knows, 0, max_page)
}

/// As [`listed`], and with `cites` above 0 a `cites` relation from `notes`
/// to `messages` and `cites` edges `n00` → m1, `n01` → m1, ….
fn listed_citing(
    notes: usize,
    people: usize,
    knows: usize,
    cites: usize,
    max_page: u32,
) -> Result<(Fixture, Version), Box<dyn Error>> {
    let (directory, mut storage) = create()?;
    for collection in ["messages", "notes", "people"] {
        apply(
            &mut storage,
            collection,
            Change::Collection(CollectionWrite {
                collection: collection.to_owned(),
                schema: None,
            }),
        )?;
    }
    let imported = |id: String, body: String| -> Result<ImportedDocument, serde_json::Error> {
        Ok(ImportedDocument {
            document_id: id,
            document: RawValue::from_string(body)?,
        })
    };
    let notes = (0..notes)
        .map(|note| {
            let body = format!(r#"{{"title": "note {note}", "body": "not drawn"}}"#);
            imported(format!("n{note:02}"), body)
        })
        .collect::<Result<Vec<_>, _>>()?;
    let people = (0..people)
        .map(|person| imported(format!("p{person:02}"), r#"{"name": "Ann"}"#.to_owned()))
        .collect::<Result<Vec<_>, _>>()?;
    let messages = vec![imported("m1".to_owned(), r#"{"weight": 1}"#.to_owned())?];
    for (collection, documents) in [("messages", messages), ("notes", notes), ("people", people)] {
        if documents.is_empty() {
            continue;
        }
        apply(
            &mut storage,
            &format!("{collection}-documents"),
            Change::PutMany(DocumentImport {
                collection: collection.to_owned(),
                documents,
            }),
        )?;
    }
    let older = head(&storage)?;
    let citing = (cites > 0).then_some(("cites", "notes", "messages"));
    for (relation, from, to) in [
        ("sent_by", "messages", "people"),
        ("knows", "people", "people"),
    ]
    .into_iter()
    .chain(citing)
    {
        apply(
            &mut storage,
            relation,
            Change::RelationPut(Relation {
                relation: relation.to_owned(),
                from_collection: from.to_owned(),
                to_collection: to.to_owned(),
                cardinality: Cardinality::ManyToMany,
            }),
        )?;
    }
    let mut edges = vec![("sent_by", node("messages", "m1"), node("people", "p00"))];
    for to in 1..=knows {
        edges.push((
            "knows",
            node("people", "p00"),
            node("people", &format!("p{to:02}")),
        ));
    }
    for from in 0..cites {
        edges.push((
            "cites",
            node("notes", &format!("n{from:02}")),
            node("messages", "m1"),
        ));
    }
    let edges = edges
        .into_iter()
        .map(|(relation, from, to)| Edge {
            relation: relation.to_owned(),
            from,
            to,
            properties: json!({}),
        })
        .collect();
    apply(
        &mut storage,
        "edges",
        Change::EdgePutMany(EdgeImport { edges }),
    )?;
    Ok((serve(directory, storage, max_page)?, older))
}

fn unrelated(per_collection: u64, most: u64) -> Result<Unrelated, Box<dyn Error>> {
    Ok(Unrelated {
        per_collection: std::num::NonZeroU64::new(per_collection).ok_or("not zero")?,
        most: std::num::NonZeroU64::new(most).ok_or("not zero")?,
    })
}

/// The ids of `collection` among an answer's nodes, in the order given.
fn nodes_of(answer: &Value, collection: &str) -> Vec<String> {
    answer["nodes"]
        .as_array()
        .map(|nodes| {
            nodes
                .iter()
                .filter(|node| node["collection"] == collection)
                .filter_map(|node| node["document_id"].as_str().map(str::to_owned))
                .collect()
        })
        .unwrap_or_default()
}

/// GR1 red: three collections, `notes` named by no relation: with
/// `unrelated` and no `collections`, its records come back as nodes with
/// their named members, never their bodies, beside the edge's ends.
#[test]
fn records_no_relation_names_come_back_as_nodes() -> TestResult {
    let (fixture, _older) = listed(3, 1, 0, 100)?;
    let mut asked = page(10)?;
    asked.unrelated = Some(unrelated(10, 100)?);
    asked.members = Some(std::iter::once(("notes".to_owned(), names(&["title"]))).collect());
    let answer = export(&fixture.service, asked)?;
    assert_eq!(nodes_of(&answer, "notes"), ["n00", "n01", "n02"]);
    let notes: Vec<&Value> = answer["nodes"]
        .as_array()
        .ok_or("no nodes")?
        .iter()
        .filter(|node| node["collection"] == "notes")
        .collect();
    assert_eq!(notes[0]["members"], json!({"title": "\"note 0\""}));
    assert!(!answer.to_string().contains("not drawn"), "{answer}");
    assert_eq!(edges(&answer), vec![edge("sent_by", "m1", "p00")]);
    assert_eq!(answer["next_after"], Value::Null);
    Ok(())
}

/// GR1 red: at `max_page` 1 a page asks for one edge (F4: the service
/// refuses a page larger than `max_page`) and examines 64 keys, and
/// `per_collection` and `most` may be at most those 64 (RED4). Page one
/// examines the 20 `cites` edges leaving `notes`, none answered as `messages`
/// is not asked for, leaving 44: it lists `notes` n00…n43 and ends its
/// examination. Page two lists n44…n49, reaching `per_collection` 50, then
/// `people` p00…p09, reaching `most` 60 of the 11 people, and ends the series.
#[test]
fn per_collection_and_most_hold_across_pages() -> TestResult {
    let (fixture, _older) = listed_citing(70, 11, 0, 20, 1)?;
    let asked = || -> Result<GraphPage, Box<dyn Error>> {
        let mut asked = page(1)?;
        asked.collections = Some(names(&["notes", "people"]));
        asked.unrelated = Some(unrelated(50, 60)?);
        Ok(asked)
    };
    let first = export(&fixture.service, asked()?)?;
    assert_eq!(first["examination_ended"], true);
    assert!(edges(&first).is_empty());
    assert_eq!(nodes_of(&first, "notes").len(), 44);
    assert_eq!(nodes_of(&first, "people").len(), 0);
    let version: Version = serde_json::from_value(first["selection"]["version"].clone())?;
    let mut next = asked()?;
    next.selection.version = Some(version);
    next.after = first["next_after"].as_str().map(str::to_owned);
    let second = export(&fixture.service, next)?;
    let notes = nodes_of(&second, "notes");
    assert_eq!(notes.first().map(String::as_str), Some("n44"));
    assert_eq!(notes.len(), 6);
    let people = nodes_of(&second, "people");
    assert_eq!(people.first().map(String::as_str), Some("p00"));
    assert_eq!(people.len(), 10);
    assert_eq!(second["next_after"], Value::Null);
    Ok(())
}

/// GR1 red: a listing's continuation is refused at another version (one
/// write later) and under another `unrelated`. Pages of one, as
/// `max_page` 1 allows (F4), and `unrelated` within its 64 (RED4): the
/// first page ends inside the listing as in
/// `per_collection_and_most_hold_across_pages`.
#[test]
fn a_listing_continuation_is_refused_at_another_version_or_unrelated() -> TestResult {
    let (fixture, _older) = listed_citing(70, 11, 0, 20, 1)?;
    let mut asked = page(1)?;
    asked.collections = Some(names(&["notes", "people"]));
    asked.unrelated = Some(unrelated(50, 60)?);
    let first = export(&fixture.service, asked)?;
    let after = first["next_after"]
        .as_str()
        .ok_or("no continuation")?
        .to_owned();
    let version: Version = serde_json::from_value(first["selection"]["version"].clone())?;
    let receipt = fixture.service.dispatch(Operation::Mutate(Mutation {
        request_id: "later".to_owned(),
        branch: "main".to_owned(),
        expected: version,
        change: Change::Collection(CollectionWrite {
            collection: "later".to_owned(),
            schema: None,
        }),
    }))?;
    let later: Version = serde_json::from_value(receipt["version"].clone())?;
    for (at, per_collection) in [(later, 50), (version, 49)] {
        let mut next = page(1)?;
        next.collections = Some(names(&["notes", "people"]));
        next.unrelated = Some(unrelated(per_collection, 60)?);
        next.selection.version = Some(at);
        next.after = Some(after.clone());
        let refused = export(&fixture.service, next)
            .err()
            .ok_or("a changed listing continued")?;
        assert!(
            refused.to_string().starts_with("graph_filter_changed"),
            "{refused}"
        );
    }
    Ok(())
}

/// GR1 red: a caller who may read no note gets no note, and nothing in the
/// answer names one.
#[test]
fn a_collection_the_caller_may_not_read_gives_no_record() -> TestResult {
    let (fixture, _older) = listed(3, 1, 0, 100)?;
    let access = request(&TestSource::denying(&[
        ("notes", "n00"),
        ("notes", "n01"),
        ("notes", "n02"),
    ]))?;
    let mut asked = page(10)?;
    asked.unrelated = Some(unrelated(10, 100)?);
    let answer = super::export(
        &fixture.reader,
        &fixture.at,
        &access,
        &asked,
        PageLimits {
            rows: std::num::NonZeroUsize::new(10).ok_or("not zero")?,
            bytes: std::num::NonZeroUsize::new(1 << 20).ok_or("not zero")?,
        },
        NonZeroU32::new(100).ok_or("not zero")?,
    )?;
    assert!(answer.nodes.iter().all(|node| node.collection != "notes"));
    assert_eq!(answer.next_after, None);
    Ok(())
}

/// GR1 red: without `unrelated` a filtered page is today's: its
/// continuation's fingerprint is FNV-1a over the GX1 filter's JSON alone,
/// and its nodes are its edge's ends.
#[test]
fn without_unrelated_a_filtered_page_is_unchanged() -> TestResult {
    let fixture = service()?;
    let mut asked = page(1)?;
    asked.collections = Some(names(&["messages", "people"]));
    let answer = export(&fixture.service, asked)?;
    let filter = br#"{"relations":null,"collections":["messages","people"],"members":{}}"#;
    let fingerprint = filter.iter().fold(0xcbf2_9ce4_8422_2325_u64, |hash, byte| {
        (hash ^ u64::from(*byte)).wrapping_mul(0x0100_0000_01b3)
    });
    let next = answer["next_after"].as_str().ok_or("no continuation")?;
    assert!(
        next.starts_with(&format!("{fingerprint:016x}.0.")),
        "{next}"
    );
    assert_eq!(answer["nodes"].as_array().map(Vec::len), Some(2));
    Ok(())
}

/// `GR1b` red: a listed node larger than everything a page has, on a page that
/// holds nothing, is refused by its address and its bytes against the
/// frame, never answered as an empty page whose continuation asks again.
#[test]
fn a_listed_node_no_page_could_hold_is_refused_by_name() -> TestResult {
    let (fixture, _older) = listed(1, 1, 0, 100)?;
    let mut asked = page(10)?;
    asked.collections = Some(names(&["notes"]));
    asked.unrelated = Some(unrelated(10, 10)?);
    let refused = super::export(
        &fixture.reader,
        &fixture.at,
        &request(&Standalone)?,
        &asked,
        PageLimits {
            rows: std::num::NonZeroUsize::new(10).ok_or("not zero")?,
            bytes: std::num::NonZeroUsize::new(40).ok_or("not zero")?,
        },
        NonZeroU32::new(100).ok_or("not zero")?,
    )
    .err()
    .ok_or("a page answered without the node")?;
    assert_eq!(refused.code, "graph_node_bytes", "{}", refused.message);
    assert!(refused.message.contains("notes/n00"), "{}", refused.message);
    assert!(
        refused.message.contains("bytes, over the"),
        "{}",
        refused.message
    );
    Ok(())
}

/// `GR1b` red: so is an edge, by its address, where the budget alone answered
/// `response_limit` naming nothing.
#[test]
fn an_edge_no_page_could_hold_is_refused_by_name() -> TestResult {
    let fixture = service()?;
    let mut asked = page(10)?;
    asked.relations = Some(names(&["sent_by"]));
    let refused = super::export(
        &fixture.reader,
        &fixture.at,
        &request(&Standalone)?,
        &asked,
        PageLimits {
            rows: std::num::NonZeroUsize::new(10).ok_or("not zero")?,
            bytes: std::num::NonZeroUsize::new(60).ok_or("not zero")?,
        },
        NonZeroU32::new(100).ok_or("not zero")?,
    )
    .err()
    .ok_or("a page answered without the edge")?;
    assert_eq!(refused.code, "graph_edge_bytes", "{}", refused.message);
    assert!(
        refused.message.contains("sent_by messages/m1 -> people/p1"),
        "{}",
        refused.message
    );
    Ok(())
}

const fn fixture_long_wait_interval() -> std::time::Duration {
    const LONG_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
    LONG_WAIT_INTERVAL
}