haematite-cli 0.21.7

//! GX1: `graph.export` answering what was asked.
//!
//! With none of `relations`, `collections` or `members`, the export is
//! today's ([`super::edges::visible_export`]). With any of them, only the
//! named relations and the edges whose both ends are in the named
//! collections are answered, and every end on the page is a node carrying
//! the named members' stored JSON text.
//!
//! An edge's key leads with its from collection and record, then its
//! relation (`e\0{from}\0{id}\0{relation}\0…`), so no key prefix selects one
//! relation. A filtered page scans `e\0{collection}\0` for each from
//! collection the filter allows (a named relation's declared from
//! collection, within `collections`), one prefix after another, and reads
//! every edge leaving those collections once across a walk. One page
//! examines at most `max_page × 64` keys, the product the service already
//! bounds an import's rows by, and when that is spent answers what it found
//! (perhaps nothing), a continuation, and `examination_ended: true`, so a
//! short page is never read as the end of the data.
//!
//! GX2: `records` names at most `max_page × 64` records (`GX2c`: the
//! examination bound, so one paged series reads any drawing a query can
//! name), whose JSON fits the response frame, and only edges with both
//! ends among them are answered. Its scan is one prefix per named record
//! allowed as a from end (`e\0{collection}\0{id}\0`), so a page reads only
//! the edges leaving those records. One page continues from prefix to prefix
//! within the one examination bound until it is full or the prefixes end.
//!
//! GR1: `unrelated` ({`per_collection`, most}) lists, after the edges, up to
//! `per_collection` records of each chosen collection (`collections`, or
//! every collection at the captured version) and at most `most` in the
//! whole series, as nodes with the named members. The addresses are read
//! from the revision entries (`m\0{collection}\0{id}`), which hold no body;
//! a body is read only for a record whose collection names members, as for
//! an edge's end. Each entry examined counts against the same examination
//! bound; a record the caller may not read is passed over and not counted,
//! so nothing says it exists. Its continuation carries the counts so far
//! and the captured version, so it is refused at another version or under
//! another `unrelated`.
//!
//! A known property (`GR1b`, Apollo's ruling of 17:58): examined entries the
//! caller may not read cost the bound, as an edge's end does, so page
//! boundaries can differ by entries the caller cannot see. Access holds only
//! per-record denials (access.rs), so a collection denied whole cannot be
//! told from one partly denied without examining its entries. What closes
//! it is a whole-collection denial in Access, passed over before any entry
//! is examined; it belongs with the Lys permissions source and must land
//! before any source that denies by collection. Standalone, the one source
//! that ships, denies nothing.
//!
//! `GR1b`: an edge or a listed node larger than everything a page has left,
//! on a page that has taken nothing, is refused by name with its address and
//! its bytes against the frame (`graph_edge_bytes`, `graph_node_bytes`),
//! never answered as an empty page whose continuation would ask again.

use std::collections::{BTreeMap, BTreeSet};
use std::num::{NonZeroU32, NonZeroUsize};

use haematite::db::ReadOnlyDatabase;
use serde::Serialize;
use serde_json::value::RawValue;

use super::edges::{decode_edge, decode_key, page_start};
use super::reads::{PageLimits, bounded_key_page};
use super::{Captured, Failure, collection_key, document_key, identifier, reads, relations};
use crate::serve::access::Access;
use crate::serve::protocol::{Edge, GraphPage, Node as Record, Unrelated, Version};

/// Member names one collection may name: what a drawing maps (size, edge
/// thickness, colour, label and a few more) with room; each name costs one
/// lookup in a record already read for the page.
pub const MEMBER_NAMES_MOST: usize = 16;

/// Bytes of one member's stored JSON text a node carries: a number, a date,
/// a label or a short kind fits; a message body does not, and is named in
/// `over_bound` with its length instead of cut.
pub const VALUE_BYTES_MOST: usize = 1024;

/// One record at an edge's end.
#[derive(Debug, Serialize)]
pub struct Node {
    pub collection: String,
    pub document_id: String,
    /// Each named member present in the record, as its stored JSON text.
    pub members: BTreeMap<String, String>,
    /// Each named member present whose text is over [`VALUE_BYTES_MOST`].
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub over_bound: Vec<OverBound>,
}

#[derive(Debug, Serialize)]
pub struct OverBound {
    pub member: String,
    pub bytes: usize,
}

/// A filtered page.
#[derive(Debug)]
pub struct Filtered {
    pub edges: Vec<Edge>,
    pub nodes: Vec<Node>,
    pub next_after: Option<String>,
    /// The page ended at the examination bound, not at the end of the data.
    pub examination_ended: bool,
}

/// Whether `page` asks for a filtered export.
#[must_use]
pub const fn is_filtered(page: &GraphPage) -> bool {
    page.relations.is_some()
        || page.collections.is_some()
        || page.members.is_some()
        || page.records.is_some()
        || page.unrelated.is_some()
}

/// The filter, checked against the store.
#[derive(Serialize)]
struct Asked {
    relations: Option<BTreeSet<String>>,
    collections: Option<BTreeSet<String>>,
    members: BTreeMap<String, Vec<String>>,
    /// Absent from the JSON when not asked, so a GX1 continuation keeps its
    /// fingerprint.
    #[serde(skip_serializing_if = "Option::is_none")]
    records: Option<BTreeSet<(String, String)>>,
    /// GR1: absent when not asked, so every earlier continuation keeps its
    /// fingerprint; with it, the captured version too.
    #[serde(skip_serializing_if = "Option::is_none")]
    unrelated: Option<(Unrelated, Version)>,
    /// The key prefixes scanned, in order.
    #[serde(skip)]
    prefixes: Vec<Vec<u8>>,
    /// GR1: the collections whose records are listed after the edges.
    #[serde(skip)]
    listed: Vec<String>,
}

fn unknown(code: &str, what: &str, name: &str, path: String) -> Failure {
    Failure {
        path: Some(path),
        ..Failure::new(
            code,
            format!("{what} {name} does not exist at the captured version"),
        )
    }
}

fn collection_exists(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    name: &str,
) -> Result<bool, Failure> {
    identifier(name)?;
    Ok(reads::get_bytes(reader, at, &collection_key(name))?.is_some())
}

/// What a page may name and examine: the examination bound (`max_page × 64`
/// keys, and as many records) and the response frame's bytes.
#[derive(Clone, Copy)]
struct Bounds {
    examined: usize,
    bytes: usize,
}

impl Bounds {
    fn of(max_page: NonZeroU32, bytes: NonZeroUsize) -> Result<Self, Failure> {
        let examined = usize::try_from(max_page.get())
            .ok()
            .and_then(|pages| pages.checked_mul(64))
            .ok_or_else(|| {
                Failure::new("limit_overflow", "graph examination bound is too large")
            })?;
        Ok(Self {
            examined,
            bytes: bytes.get(),
        })
    }
}

/// The named records, at most the examination bound of them and their JSON
/// within the frame, both checked before any is looked up; each stored and
/// readable by the caller: one it may not read is refused as unknown, so
/// naming it does not tell whether it exists.
fn named_records(
    reading: (&ReadOnlyDatabase, &Captured, &Access),
    named: &[Record],
    bounds: Bounds,
) -> Result<BTreeSet<(String, String)>, Failure> {
    let (reader, at, access) = reading;
    let most = bounds.examined;
    if named.len() > most {
        return Err(Failure {
            path: Some(format!("/records/{most}")),
            ..Failure::new(
                "graph_records_bound",
                format!("at most {most} records may be named, the store's examination bound"),
            )
        });
    }
    let length = serde_json::to_vec(named).map_err(Failure::storage)?.len();
    if length > bounds.bytes {
        return Err(Failure {
            path: Some("/records".to_owned()),
            ..Failure::new(
                "graph_records_bytes",
                format!(
                    "the named records are {length} bytes, over the frame's {}",
                    bounds.bytes
                ),
            )
        });
    }
    let mut records = BTreeSet::new();
    for (index, record) in named.iter().enumerate() {
        let (collection, id) = (&record.collection, &record.document_id);
        if !access.can_read_document(collection, id)
            || !reads::document_exists_for_constraint(reader, at, collection, id)?
        {
            return Err(Failure {
                path: Some(format!("/records/{index}")),
                ..Failure::new(
                    "graph_record_unknown",
                    "a named record does not exist at the captured version",
                )
            });
        }
        records.insert((collection.clone(), id.clone()));
    }
    Ok(records)
}

impl Asked {
    fn read(
        reader: &ReadOnlyDatabase,
        at: &Captured,
        access: &Access,
        page: &GraphPage,
        bounds: Bounds,
    ) -> Result<Self, Failure> {
        let mut from = None;
        let relations = match &page.relations {
            None => None,
            Some(names) => {
                let mut froms = BTreeSet::new();
                for (index, name) in names.iter().enumerate() {
                    identifier(name)?;
                    let declaration = relations::get(reader, at, name)?.ok_or_else(|| {
                        unknown(
                            "graph_relation_unknown",
                            "relation",
                            name,
                            format!("/relations/{index}"),
                        )
                    })?;
                    froms.insert(declaration.from_collection);
                }
                from = Some(froms);
                Some(names.iter().cloned().collect::<BTreeSet<_>>())
            }
        };
        let collections = match &page.collections {
            None => None,
            Some(names) => {
                for (index, name) in names.iter().enumerate() {
                    if !collection_exists(reader, at, name)? {
                        return Err(unknown(
                            "graph_collection_unknown",
                            "collection",
                            name,
                            format!("/collections/{index}"),
                        ));
                    }
                }
                Some(names.iter().cloned().collect::<BTreeSet<_>>())
            }
        };
        let members = page.members.clone().unwrap_or_default();
        members_named(reader, at, &members)?;
        let from = match (from, &collections) {
            (Some(from), Some(collections)) => Some(
                from.intersection(collections)
                    .cloned()
                    .collect::<BTreeSet<_>>(),
            ),
            (Some(from), None) => Some(from),
            (None, Some(collections)) => Some(collections.clone()),
            (None, None) => None,
        };
        let records = page
            .records
            .as_deref()
            .map(|named| named_records((reader, at, access), named, bounds))
            .transpose()?;
        let prefixes = match (&records, from) {
            (Some(records), from) => records
                .iter()
                .filter(|(collection, _)| {
                    from.as_ref().is_none_or(|from| from.contains(collection))
                })
                .map(|(collection, id)| format!("e\0{collection}\0{id}\0").into_bytes())
                .collect(),
            (None, Some(from)) => from
                .iter()
                .map(|collection| format!("e\0{collection}\0").into_bytes())
                .collect(),
            (None, None) => vec![b"e\0".to_vec()],
        };
        let (unrelated, listed) = match page.unrelated {
            None => (None, Vec::new()),
            Some(unrelated) => {
                unrelated_within(unrelated, bounds)?;
                let listed = match &collections {
                    Some(collections) => collections.iter().cloned().collect(),
                    None => every_collection(reader, at)?,
                };
                (Some((unrelated, at.version)), listed)
            }
        };
        Ok(Self {
            relations,
            collections,
            members,
            records,
            unrelated,
            prefixes,
            listed,
        })
    }

    fn wants(&self, edge: &Edge) -> bool {
        self.relations
            .as_ref()
            .is_none_or(|relations| relations.contains(&edge.relation))
            && self.collections.as_ref().is_none_or(|collections| {
                collections.contains(&edge.from.collection)
                    && collections.contains(&edge.to.collection)
            })
            && self.records.as_ref().is_none_or(|records| {
                among(records, &edge.from.collection, &edge.from.document_id)
                    && among(records, &edge.to.collection, &edge.to.document_id)
            })
    }

    /// FNV-1a over the filter as JSON: a continuation names the filter it
    /// was made under, and is refused under another.
    fn fingerprint(&self) -> Result<u64, Failure> {
        let bytes = serde_json::to_vec(self).map_err(Failure::storage)?;
        Ok(bytes.iter().fold(0xcbf2_9ce4_8422_2325_u64, |hash, byte| {
            (hash ^ u64::from(*byte)).wrapping_mul(0x0100_0000_01b3)
        }))
    }
}

/// Each collection a page names members for exists, and names at most
/// [`MEMBER_NAMES_MOST`] members, none of them empty.
fn members_named(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    members: &BTreeMap<String, Vec<String>>,
) -> Result<(), Failure> {
    for (collection, names) in members {
        if !collection_exists(reader, at, collection)? {
            return Err(unknown(
                "graph_collection_unknown",
                "collection",
                collection,
                format!("/members/{collection}"),
            ));
        }
        if names.len() > MEMBER_NAMES_MOST {
            return Err(Failure {
                path: Some(format!("/members/{collection}/{MEMBER_NAMES_MOST}")),
                ..Failure::new(
                    "graph_members_bound",
                    format!("at most {MEMBER_NAMES_MOST} members may be named per collection"),
                )
            });
        }
        if let Some(index) = names.iter().position(String::is_empty) {
            return Err(Failure {
                path: Some(format!("/members/{collection}/{index}")),
                ..Failure::new("graph_member_name", "a member name cannot be empty")
            });
        }
    }
    Ok(())
}

/// GR1: `per_collection` and `most` within the examination bound, the one
/// ceiling the store sets; refused by name past it, never cut short.
fn unrelated_within(unrelated: Unrelated, bounds: Bounds) -> Result<(), Failure> {
    let most = u64::try_from(bounds.examined).unwrap_or(u64::MAX);
    for (member, asked) in [
        ("per_collection", unrelated.per_collection),
        ("most", unrelated.most),
    ] {
        if asked.get() > most {
            return Err(Failure {
                path: Some(format!("/unrelated/{member}")),
                ..Failure::new(
                    "graph_unrelated_bound",
                    format!("at most {most} records may be asked, the store's examination bound"),
                )
            });
        }
    }
    Ok(())
}

/// Every collection at the captured version, in name order.
fn every_collection(reader: &ReadOnlyDatabase, at: &Captured) -> Result<Vec<String>, Failure> {
    let mut names = BTreeSet::new();
    reads::scan_rows(reader, at, b"c\0", 1, |parts, _| {
        names.insert(parts[0].to_owned());
        Ok(())
    })?;
    Ok(names.into_iter().collect())
}

fn among(records: &BTreeSet<(String, String)>, collection: &str, id: &str) -> bool {
    records.contains(&(collection.to_owned(), id.to_owned()))
}

/// `{fingerprint}.{prefix ordinal}.{key after the prefix, or empty}`.
fn continuation(text: &str, fingerprint: u64) -> Result<(usize, Option<String>), Failure> {
    let invalid = || Failure::new("invalid_continuation", "graph continuation is malformed");
    let mut parts = text.splitn(3, '.');
    let (Some(made), Some(ordinal), Some(after)) = (parts.next(), parts.next(), parts.next())
    else {
        return Err(invalid());
    };
    let made = u64::from_str_radix(made, 16).map_err(|_| invalid())?;
    if made != fingerprint {
        return Err(Failure::new(
            "graph_filter_changed",
            "a continuation must carry the filter of the page that made it",
        ));
    }
    let ordinal = ordinal.parse().map_err(|_| invalid())?;
    Ok((ordinal, (!after.is_empty()).then(|| after.to_owned())))
}

fn readable(access: &Access, edge_ends: (&str, &str, &str, &str)) -> bool {
    let (from_collection, from_id, to_collection, to_id) = edge_ends;
    access.can_read_document(from_collection, from_id)
        && access.can_read_document(to_collection, to_id)
}

/// One prefix's scan within a page: the wanted edges taken, the key of the
/// last one taken, whether a wanted edge was left because the page filled,
/// the key the examination stopped at, and the bytes the taken edges cost.
struct Scanned {
    resume: Option<String>,
    overflow: bool,
    stopped: Option<String>,
    examined: usize,
    bytes: usize,
}

/// What every prefix of one page is read from and for.
struct Reading<'a> {
    reader: &'a ReadOnlyDatabase,
    at: &'a Captured,
    access: &'a Access,
    asked: &'a Asked,
}

/// Scans `prefix` from `after` within `limits` (keys examined, bytes taken),
/// taking wanted edges into `edges` until it holds `limit`.
fn scan_prefix(
    reading: &Reading<'_>,
    prefix: &[u8],
    after: Option<&str>,
    limits: PageLimits,
    edges: &mut Vec<Edge>,
    limit: usize,
) -> Result<Scanned, Failure> {
    let Reading {
        reader,
        at,
        access,
        asked,
    } = *reading;
    // GR1b: a wanted edge larger than all the page has left is refused by
    // name on a page that has taken nothing; on a page that holds edges the
    // next page starts at it. The budget alone would refuse the first row
    // of a scan as `response_limit`, naming nothing, even on a page that
    // holds edges from an earlier prefix.
    let page_empty = edges.is_empty();
    let mut taken = 0_usize;
    let scanned = bounded_key_page(
        reader,
        at,
        prefix,
        &page_start(prefix, after)?,
        limits,
        |suffix, bytes| {
            let edge = decode_edge(prefix, &suffix, &bytes)?;
            let ends = (
                edge.from.collection.as_str(),
                edge.from.document_id.as_str(),
                edge.to.collection.as_str(),
                edge.to.document_id.as_str(),
            );
            let keep = asked.wants(&edge) && readable(access, ends);
            Ok((suffix, keep.then_some(edge)))
        },
        |(_, edge)| {
            let Some(edge) = edge else {
                return Ok(0);
            };
            let length = serialized_len(edge)?;
            if length > limits.bytes.get() && taken == 0 {
                if page_empty {
                    return Err(Failure::new(
                        "graph_edge_bytes",
                        format!(
                            "edge {} is {length} bytes, over the {} bytes a page has",
                            address(edge),
                            limits.bytes
                        ),
                    ));
                }
                return Err(Failure::new(WIDE, "the next page starts at this edge"));
            }
            taken = taken.saturating_add(1);
            Ok(length)
        },
    );
    let (rows, stopped) = match scanned {
        Err(failure) if failure.code == WIDE => {
            return Ok(Scanned {
                resume: None,
                overflow: true,
                stopped: None,
                examined: 0,
                bytes: 0,
            });
        }
        scanned => scanned?,
    };
    let mut scanned = Scanned {
        resume: None,
        overflow: false,
        stopped,
        examined: rows.len(),
        bytes: 0,
    };
    for (suffix, edge) in rows {
        let Some(edge) = edge else {
            continue;
        };
        if edges.len() == limit {
            scanned.overflow = true;
            break;
        }
        scanned.bytes = scanned.bytes.saturating_add(serialized_len(&edge)?);
        edges.push(edge);
        scanned.resume = Some(suffix);
    }
    Ok(scanned)
}

/// Within this module only: a scan met a wanted edge too large for what the
/// page has left, on a page that already holds edges.
const WIDE: &str = "graph_edge_next_page";

/// An edge's address, as a refusal names it.
fn address(edge: &Edge) -> String {
    format!(
        "{} {}/{} -> {}/{}",
        edge.relation,
        edge.from.collection,
        edge.from.document_id,
        edge.to.collection,
        edge.to.document_id
    )
}

fn serialized_len(edge: &Edge) -> Result<usize, Failure> {
    serde_json::to_vec(edge)
        .map(|bytes| bytes.len())
        .map_err(Failure::storage)
}

/// Refuses a continuation at `key` under `prefix` when the caller may not
/// read both ends of the edge it names.
fn readable_at(access: &Access, prefix: &[u8], key: &str) -> Result<(), Failure> {
    let mut whole = prefix.to_vec();
    whole.extend_from_slice(key.as_bytes());
    let ends = decode_key(&whole)?;
    let ends = (
        ends.from.collection.as_str(),
        ends.from.document_id.as_str(),
        ends.to.collection.as_str(),
        ends.to.document_id.as_str(),
    );
    if readable(access, ends) {
        Ok(())
    } else {
        Err(Failure::new(
            "access_scan_limit",
            "graph examination ended on an edge the caller may not read",
        ))
    }
}

/// One filtered page of the export: prefix after prefix from the
/// continuation, within one examination bound and one byte budget, until the
/// page holds `limit` edges, the bound is spent, or the prefixes end.
///
/// # Errors
/// `graph_relation_unknown`, `graph_collection_unknown`,
/// `graph_record_unknown`, `graph_records_bound`, `graph_records_bytes`,
/// `graph_members_bound`,
/// `graph_member_name`, `graph_unrelated_bound` for the filter;
/// `graph_edge_bytes` and `graph_node_bytes` for an edge or a node no page
/// could hold; `graph_filter_changed` and
/// `invalid_continuation` for a continuation; `access_scan_limit` when the
/// page's examination ends on an edge the caller may not read; whatever a
/// read answers.
pub fn export(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    access: &Access,
    page: &GraphPage,
    limits: PageLimits,
    max_page: NonZeroU32,
) -> Result<Filtered, Failure> {
    let bounds = Bounds::of(max_page, limits.bytes)?;
    let asked = Asked::read(reader, at, access, page, bounds)?;
    let fingerprint = asked.fingerprint()?;
    let (mut ordinal, mut after) = match page.after.as_deref() {
        None => (0, None),
        Some(text) => continuation(text, fingerprint)?,
    };
    let mut examined = bounds.examined;
    let mut bytes = limits.bytes.get();
    let reading = Reading {
        reader,
        at,
        access,
        asked: &asked,
    };
    let mut edges = Vec::new();
    let mut next_after = None;
    let mut examination_ended = false;
    while let Some(prefix) = asked.prefixes.get(ordinal) {
        let (Some(rows), Some(budget)) = (NonZeroUsize::new(examined), NonZeroUsize::new(bytes))
        else {
            examination_ended = true;
            next_after = Some(format!(
                "{fingerprint:016x}.{ordinal}.{}",
                after.take().unwrap_or_default()
            ));
            break;
        };
        let scanned = scan_prefix(
            &reading,
            prefix,
            after.as_deref(),
            PageLimits {
                rows,
                bytes: budget,
            },
            &mut edges,
            limits.rows.get(),
        )?;
        examined = examined.saturating_sub(scanned.examined);
        bytes = bytes.saturating_sub(scanned.bytes);
        if scanned.overflow {
            let key = scanned.resume.or_else(|| after.take()).unwrap_or_default();
            next_after = Some(format!("{fingerprint:016x}.{ordinal}.{key}"));
            break;
        }
        if let Some(key) = scanned.stopped {
            readable_at(access, prefix, &key)?;
            examination_ended = true;
            next_after = Some(format!("{fingerprint:016x}.{ordinal}.{key}"));
            break;
        }
        ordinal = ordinal.saturating_add(1);
        after = None;
    }
    let mut nodes = nodes(reader, at, &asked, &edges)?;
    if next_after.is_none()
        && let Some((unrelated, _version)) = asked.unrelated
    {
        let listing = Listing {
            reading: &reading,
            fingerprint,
            first: asked.prefixes.len(),
            unrelated,
        };
        let budget = Budget {
            examined,
            bytes: bytes.saturating_sub(nodes_len(&nodes)?),
        };
        let listed = list_unrelated(&listing, ordinal, after.as_deref(), budget, &mut nodes)?;
        next_after = listed.next_after;
        examination_ended = listed.examination_ended;
    }
    Ok(Filtered {
        edges,
        nodes,
        next_after,
        examination_ended,
    })
}

fn nodes_len(nodes: &[Node]) -> Result<usize, Failure> {
    serde_json::to_vec(nodes)
        .map(|bytes| bytes.len())
        .map_err(Failure::storage)
}

/// GR1: what the listing after the edges reads with.
struct Listing<'a> {
    reading: &'a Reading<'a>,
    fingerprint: u64,
    /// The ordinal of the first listed collection: after every edge prefix.
    first: usize,
    unrelated: Unrelated,
}

/// What is left of one page's examination bound and byte budget.
#[derive(Clone, Copy)]
struct Budget {
    examined: usize,
    bytes: usize,
}

/// Where a listing page ended.
struct Listed {
    next_after: Option<String>,
    examination_ended: bool,
}

/// GR1: the counts a listing continuation carries: records given in the
/// series, records given of the collection at `ordinal`, and the id after
/// which it resumes (empty for its start).
fn listing_place(after: Option<&str>) -> Result<(u64, u64, Option<String>), Failure> {
    let invalid = || Failure::new("invalid_continuation", "graph continuation is malformed");
    let Some(after) = after else {
        return Ok((0, 0, None));
    };
    let mut parts = after.splitn(3, ':');
    let (Some(total), Some(here), Some(id)) = (parts.next(), parts.next(), parts.next()) else {
        return Err(invalid());
    };
    let total = total.parse().map_err(|_| invalid())?;
    let here = here.parse().map_err(|_| invalid())?;
    Ok((total, here, (!id.is_empty()).then(|| id.to_owned())))
}

/// GR1: records of each listed collection from `ordinal`, as nodes, within
/// what is left of the page's examination bound and bytes, until each
/// collection gives `per_collection`, the series `most`, or the collections
/// end. Each revision entry examined costs one of the bound.
fn list_unrelated(
    listing: &Listing<'_>,
    mut ordinal: usize,
    after: Option<&str>,
    mut budget: Budget,
    nodes: &mut Vec<Node>,
) -> Result<Listed, Failure> {
    let Reading {
        reader,
        at,
        access,
        asked,
    } = *listing.reading;
    let (mut total, mut here, mut resume) = if ordinal < listing.first {
        (0, 0, None)
    } else {
        listing_place(after)?
    };
    ordinal = ordinal.max(listing.first);
    let fingerprint = listing.fingerprint;
    let place = |ordinal: usize, total: u64, here: u64, id: Option<&str>| {
        format!(
            "{fingerprint:016x}.{ordinal}.{total}:{here}:{}",
            id.unwrap_or_default()
        )
    };
    let (per_collection, most) = (
        listing.unrelated.per_collection.get(),
        listing.unrelated.most.get(),
    );
    while let Some(collection) = asked.listed.get(ordinal - listing.first) {
        let need = per_collection
            .saturating_sub(here)
            .min(most.saturating_sub(total));
        if most <= total {
            break;
        }
        if need == 0 {
            ordinal = ordinal.saturating_add(1);
            (here, resume) = (0, None);
            continue;
        }
        let rows = usize::try_from(need)
            .unwrap_or(usize::MAX)
            .min(budget.examined);
        let Some(rows) = NonZeroUsize::new(rows) else {
            return Ok(Listed {
                next_after: Some(place(ordinal, total, here, resume.as_deref())),
                examination_ended: true,
            });
        };
        let prefix = format!("m\0{collection}\0").into_bytes();
        let mut from = prefix.clone();
        if let Some(id) = &resume {
            from.extend_from_slice(id.as_bytes());
            from.push(0);
        }
        let limits = PageLimits {
            rows,
            bytes: NonZeroUsize::MAX,
        };
        let (ids, more) = bounded_key_page(
            reader,
            at,
            &prefix,
            &from,
            limits,
            |id, _revision| Ok(id),
            |_| Ok(0),
        )?;
        budget.examined = budget.examined.saturating_sub(ids.len());
        for id in ids {
            if access.can_read_document(collection, &id) {
                let node = node(reader, at, asked, collection, &id)?;
                let length = nodes_len(std::slice::from_ref(&node))?;
                // GR1b: on a page that holds nothing, no later page could
                // hold it either.
                if length > budget.bytes && nodes.is_empty() {
                    return Err(Failure::new(
                        "graph_node_bytes",
                        format!(
                            "record {collection}/{id} as a node is {length} bytes, over the {} \
                             bytes a page has",
                            budget.bytes
                        ),
                    ));
                }
                if length > budget.bytes {
                    return Ok(Listed {
                        next_after: Some(place(ordinal, total, here, resume.as_deref())),
                        examination_ended: false,
                    });
                }
                budget.bytes = budget.bytes.saturating_sub(length);
                nodes.push(node);
                here = here.saturating_add(1);
                total = total.saturating_add(1);
            }
            resume = Some(id);
        }
        if more.is_none() {
            ordinal = ordinal.saturating_add(1);
            (here, resume) = (0, None);
        }
    }
    Ok(Listed {
        next_after: None,
        examination_ended: false,
    })
}

/// Every distinct end of `edges`, with the members named for its collection.
fn nodes(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    asked: &Asked,
    edges: &[Edge],
) -> Result<Vec<Node>, Failure> {
    let ends: BTreeSet<(&str, &str)> = edges
        .iter()
        .flat_map(|edge| {
            [
                (
                    edge.from.collection.as_str(),
                    edge.from.document_id.as_str(),
                ),
                (edge.to.collection.as_str(), edge.to.document_id.as_str()),
            ]
        })
        .collect();
    let mut nodes = Vec::with_capacity(ends.len());
    for (collection, document_id) in ends {
        nodes.push(node(reader, at, asked, collection, document_id)?);
    }
    Ok(nodes)
}

/// One record as a node, with the members named for its collection.
fn node(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    asked: &Asked,
    collection: &str,
    document_id: &str,
) -> Result<Node, Failure> {
    let mut node = Node {
        collection: collection.to_owned(),
        document_id: document_id.to_owned(),
        members: BTreeMap::new(),
        over_bound: Vec::new(),
    };
    if let Some(names) = asked.members.get(collection)
        && let Some(bytes) = reads::get_bytes(reader, at, &document_key(collection, document_id))?
    {
        let stored: BTreeMap<String, Box<RawValue>> =
            serde_json::from_slice(&bytes).map_err(Failure::storage)?;
        for name in names {
            let Some(value) = stored.get(name) else {
                continue;
            };
            let text = value.get();
            if text.len() > VALUE_BYTES_MOST {
                node.over_bound.push(OverBound {
                    member: name.clone(),
                    bytes: text.len(),
                });
            } else {
                node.members.insert(name.clone(), text.to_owned());
            }
        }
    }
    Ok(node)
}

#[cfg(test)]
#[path = "graph_tests.rs"]
mod tests;