haematite-cli 0.21.8

//! Relation declarations and edge constraints share the captured branch roots.

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

use haematite::db::ReadOnlyDatabase;

use super::mutations::PreparedWrite;
use super::reads::{PageLimits, bounded_key_page};
use super::{Captured, Failure, counts, edges, identifier, reads};
use crate::serve::access::Access;
use crate::serve::protocol::{Cardinality, Edge, Relation, RelationDelete, RelationTotal};

fn key(relation: &str) -> Result<Vec<u8>, Failure> {
    identifier(relation)?;
    Ok(format!("n\0{relation}").into_bytes())
}

fn validate_declaration(declaration: &Relation) -> Result<(), Failure> {
    identifier(&declaration.relation)?;
    identifier(&declaration.from_collection)?;
    identifier(&declaration.to_collection)
}

pub(super) fn decode(name: &str, bytes: &[u8]) -> Result<Relation, Failure> {
    identifier(name)?;
    let declaration: Relation = serde_json::from_slice(bytes).map_err(Failure::storage)?;
    validate_declaration(&declaration)?;
    if declaration.relation != name {
        return Err(Failure::new(
            "relation_corrupt",
            "stored relation name differs from its key",
        ));
    }
    Ok(declaration)
}

pub(super) fn get(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    name: &str,
) -> Result<Option<Relation>, Failure> {
    reads::get_bytes(reader, at, &key(name)?)?
        .map(|bytes| decode(name, &bytes))
        .transpose()
}

fn refusal(code: &str, declaration: &Relation, edge: &Edge, detail: &str) -> Failure {
    Failure::new(
        code,
        format!(
            "relation {} refuses edge {}/{} -> {}/{}: {detail}",
            declaration.relation,
            edge.from.collection,
            edge.from.document_id,
            edge.to.collection,
            edge.to.document_id,
        ),
    )
}

fn validate_collections(declaration: &Relation, edge: &Edge) -> Result<(), Failure> {
    for (side, actual, expected) in [
        ("from", &edge.from.collection, &declaration.from_collection),
        ("to", &edge.to.collection, &declaration.to_collection),
    ] {
        if actual != expected {
            return Err(Failure {
                path: Some(format!("/{side}/collection")),
                ..refusal(
                    "relation_collection",
                    declaration,
                    edge,
                    &format!("{side} collection must be {expected}, received {actual}"),
                )
            });
        }
    }
    Ok(())
}

pub(super) fn validate_edge(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    edge: &Edge,
) -> Result<(), Failure> {
    let Some(declaration) = get(reader, at, &edge.relation)? else {
        return Ok(());
    };
    validate_collections(&declaration, edge)?;
    let limit = NonZeroUsize::new(2)
        .ok_or_else(|| Failure::new("page_limit", "cardinality page limit is zero"))?;
    let candidates = match declaration.cardinality {
        Cardinality::OneToOne => {
            let mut candidates = edges::out(
                reader,
                at,
                &edge.from,
                Some(&edge.relation),
                None,
                limit,
                None,
            )?
            .entries;
            candidates.extend(
                edges::incoming(
                    reader,
                    at,
                    &edge.to,
                    Some(&edge.relation),
                    None,
                    limit,
                    None,
                )?
                .entries,
            );
            candidates
        }
        Cardinality::OneToMany => {
            edges::incoming(
                reader,
                at,
                &edge.to,
                Some(&edge.relation),
                None,
                limit,
                None,
            )?
            .entries
        }
        Cardinality::ManyToMany => return Ok(()),
    };
    for existing in candidates {
        if existing.from != edge.from || existing.to != edge.to {
            return Err(refusal(
                "relation_cardinality",
                &declaration,
                edge,
                &format!(
                    "conflicts with stored edge {}/{} -> {}/{} under {:?}",
                    existing.from.collection,
                    existing.from.document_id,
                    existing.to.collection,
                    existing.to.document_id,
                    declaration.cardinality,
                ),
            ));
        }
    }
    Ok(())
}

/// One end of an edge under one relation, as a batch keys it.
type End = (String, String, String);

/// The ends one `edges.put_many` batch has bound so far, by relation: what a
/// stored edge would show the next edge, before the batch is stored.
#[derive(Default)]
pub struct Batch {
    /// (relation, from) to the `to` it was bound to, for one to one.
    from: BTreeMap<End, (String, String)>,
    /// (relation, to) to the `from` it was bound to, for one to one and one
    /// to many.
    to: BTreeMap<End, (String, String)>,
}

impl Batch {
    /// Refuses `edge` when an earlier edge of the batch already bound one of
    /// its ends to another record under its relation's cardinality.
    ///
    /// # Errors
    /// `relation_cardinality`; whatever reading the declaration answers.
    pub fn admit(
        &mut self,
        reader: &ReadOnlyDatabase,
        at: &Captured,
        edge: &Edge,
    ) -> Result<(), Failure> {
        let Some(declaration) = get(reader, at, &edge.relation)? else {
            return Ok(());
        };
        let from = (edge.from.collection.clone(), edge.from.document_id.clone());
        let to = (edge.to.collection.clone(), edge.to.document_id.clone());
        let from_end = (edge.relation.clone(), from.0.clone(), from.1.clone());
        let to_end = (edge.relation.clone(), to.0.clone(), to.1.clone());
        if declaration.cardinality == Cardinality::OneToOne
            && self.from.get(&from_end).is_some_and(|bound| *bound != to)
        {
            return Err(refusal(
                "relation_cardinality",
                &declaration,
                edge,
                "the from document has another edge earlier in this batch",
            ));
        }
        if declaration.cardinality != Cardinality::ManyToMany
            && self.to.get(&to_end).is_some_and(|bound| *bound != from)
        {
            return Err(refusal(
                "relation_cardinality",
                &declaration,
                edge,
                "the to document has another edge earlier in this batch",
            ));
        }
        if declaration.cardinality == Cardinality::OneToOne {
            self.from.insert(from_end, to);
        }
        if declaration.cardinality != Cardinality::ManyToMany {
            self.to.insert(to_end, from);
        }
        Ok(())
    }
}

fn validate_existing(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    declaration: &Relation,
) -> Result<(), Failure> {
    let limit = NonZeroUsize::new(128)
        .ok_or_else(|| Failure::new("page_limit", "relation validation page limit is zero"))?;
    let mut after = None;
    let mut from = BTreeSet::new();
    let mut to = BTreeSet::new();
    loop {
        let page = edges::export(reader, at, after.as_deref(), limit, None)?;
        for edge in page.entries {
            if edge.relation != declaration.relation {
                continue;
            }
            validate_collections(declaration, &edge)?;
            if declaration.cardinality == Cardinality::OneToOne
                && !from.insert(edge.from.document_id.clone())
            {
                return Err(refusal(
                    "relation_cardinality",
                    declaration,
                    &edge,
                    "the from document already has another edge under this relation",
                ));
            }
            if declaration.cardinality != Cardinality::ManyToMany
                && !to.insert(edge.to.document_id.clone())
            {
                return Err(refusal(
                    "relation_cardinality",
                    declaration,
                    &edge,
                    "the to document already has another edge under this relation",
                ));
            }
        }
        let Some(next) = page.next_after else {
            return Ok(());
        };
        after = Some(next);
    }
}

pub(super) fn put(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    declaration: &Relation,
) -> Result<Vec<PreparedWrite>, Failure> {
    validate_declaration(declaration)?;
    for (field, collection) in [
        ("from_collection", &declaration.from_collection),
        ("to_collection", &declaration.to_collection),
    ] {
        if reads::collection(reader, at, collection)?.is_none() {
            return Err(Failure {
                path: Some(format!("/{field}")),
                ..Failure::new(
                    "collection_missing",
                    format!(
                        "relation {} {field} {collection} does not exist at the captured version",
                        declaration.relation
                    ),
                )
            });
        }
    }
    validate_existing(reader, at, declaration)?;
    Ok(vec![(
        key(&declaration.relation)?,
        Some(serde_json::to_vec(declaration).map_err(Failure::storage)?),
    )])
}

pub(super) fn delete(declaration: &RelationDelete) -> Result<Vec<PreparedWrite>, Failure> {
    Ok(vec![(key(&declaration.relation)?, None)])
}

pub(in crate::serve) fn list(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    byte_limit: NonZeroUsize,
) -> Result<Vec<RelationTotal>, Failure> {
    let counts = counts::View::new(reader, at)?;
    let limit = NonZeroUsize::new(128)
        .ok_or_else(|| Failure::new("page_limit", "relation list page limit is zero"))?;
    let prefix = b"n\0";
    let mut start = prefix.to_vec();
    let mut result = Vec::new();
    let mut bytes_used = 2_usize;
    loop {
        let (entries, next) = bounded_key_page(
            reader,
            at,
            prefix,
            &start,
            PageLimits {
                rows: limit,
                bytes: NonZeroUsize::MAX,
            },
            |name, bytes| {
                let declaration = decode(&name, &bytes)?;
                let total = counts.relationships(access, &name)?;
                Ok(RelationTotal {
                    declaration,
                    relationships: total.value,
                    relationships_reason: total.reason,
                })
            },
            |entry| {
                serde_json::to_vec(entry)
                    .map(|bytes| bytes.len())
                    .map_err(Failure::storage)
            },
        )?;
        for entry in entries {
            let entry_len = serde_json::to_vec(&entry).map_err(Failure::storage)?.len();
            bytes_used = bytes_used
                .checked_add(entry_len)
                .and_then(|bytes| bytes.checked_add(usize::from(!result.is_empty())))
                .ok_or_else(|| Failure::new("response_limit", "relation list size overflow"))?;
            if bytes_used > byte_limit.get() {
                return Err(Failure::new(
                    "response_limit",
                    "relation list exceeds the response byte budget",
                ));
            }
            result.push(entry);
        }
        let Some(next) = next else { return Ok(result) };
        identifier(&next)?;
        start = prefix.to_vec();
        start.extend_from_slice(next.as_bytes());
        start.push(0);
    }
}