haematite_cli/serve/

protocol.rs

//! Versioned request vocabulary shared by the socket service and clients.

use std::io;

use serde::{Deserialize, Serialize};
use serde_json::{Value, value::RawValue};

use super::storage::Failure;

mod equality;
pub use equality::IndexDeclaration;

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CountReason {
    Withheld,
    CountsNotReady,
}

#[derive(Debug, Serialize)]
pub struct KeptCount {
    pub value: Option<u64>,
    pub reason: Option<CountReason>,
}

impl KeptCount {
    pub(in crate::serve) fn insert(self, answer: &mut Value, member: &str) -> Result<(), Failure> {
        let object = answer
            .as_object_mut()
            .ok_or_else(|| Failure::new("answer_shape", "count answer must be an object"))?;
        object.insert(
            member.to_owned(),
            serde_json::to_value(self.value).map_err(Failure::storage)?,
        );
        if let Some(reason) = self.reason {
            object.insert(
                format!("{member}_reason"),
                serde_json::to_value(reason).map_err(Failure::storage)?,
            );
        }
        Ok(())
    }
}

#[derive(Serialize)]
pub struct RelationTotal {
    #[serde(flatten)]
    pub declaration: Relation,
    pub relationships: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub relationships_reason: Option<CountReason>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CountsRebuild {
    pub branch: String,
    pub request_id: String,
}

/// EMB-E2 (j): the embedded field `embed.rewalk_longer` walks again.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EmbedRewalk {
    pub branch: String,
    pub collection: String,
    pub field: String,
}

#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CountBucket {
    Documents { collection: String },
    Relationships { relation: String },
    Vectors { collection: String, field: String },
}

#[derive(Debug, Serialize)]
#[serde(tag = "result", rename_all = "snake_case")]
pub enum AuditVerdict {
    Agree,
    Differ { kept: Option<u64>, counted: u64 },
}

#[derive(Debug, Serialize)]
pub struct AuditedBucket {
    #[serde(flatten)]
    pub bucket: CountBucket,
    #[serde(flatten)]
    pub verdict: AuditVerdict,
}

#[derive(Debug, Serialize)]
#[serde(tag = "result", rename_all = "snake_case")]
pub enum CountsAudit {
    Compared { buckets: Vec<AuditedBucket> },
    CannotTell { reason: CountReason },
}

#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MutationOutcome {
    NotApplied,
    Unknown,
}

#[derive(Clone, Debug, Serialize)]
pub struct MutationContext {
    pub outcome: MutationOutcome,
    pub request_id: Value,
    pub receipt_branch: Value,
}

impl MutationContext {
    pub(super) fn from_operation(operation: &Operation) -> Option<Self> {
        if let Operation::CountsRebuild(request) = operation {
            return Some(Self {
                outcome: MutationOutcome::Unknown,
                request_id: Value::String(request.request_id.clone()),
                receipt_branch: Value::String(request.branch.clone()),
            });
        }
        let Operation::Mutate(request) = operation else {
            return None;
        };
        let branch = match &request.change {
            Change::Fork(fork) => &fork.name,
            _ => &request.branch,
        };
        Some(Self {
            outcome: MutationOutcome::Unknown,
            request_id: Value::String(request.request_id.clone()),
            receipt_branch: Value::String(branch.clone()),
        })
    }
}

/// Mutation context survives errors after the effect, including response limits.
#[derive(Clone, Debug, Serialize)]
pub struct ResponseFailure {
    #[serde(flatten)]
    failure: Failure,
    #[serde(flatten)]
    mutation: Option<Box<MutationContext>>,
}

impl ResponseFailure {
    pub(super) fn new(failure: Failure, mut mutation: Option<MutationContext>) -> Self {
        if let Some(context) = &mut mutation {
            context.outcome = if proves_not_applied(&failure.code) {
                MutationOutcome::NotApplied
            } else {
                MutationOutcome::Unknown
            };
        }
        Self {
            failure,
            mutation: mutation.map(Box::new),
        }
    }

    pub(super) fn after_dispatch(failure: Failure, mutation: Option<MutationContext>) -> Self {
        Self {
            failure,
            mutation: mutation.map(Box::new),
        }
    }

    pub(super) fn mutation_context(&self) -> Option<MutationContext> {
        self.mutation.as_deref().cloned()
    }

    pub(super) const fn failure(&self) -> &Failure {
        &self.failure
    }
}

impl std::ops::Deref for ResponseFailure {
    type Target = Failure;
    fn deref(&self) -> &Self::Target {
        self.failure()
    }
}

impl std::fmt::Display for ResponseFailure {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Display::fmt(&self.failure, formatter)
    }
}

impl std::error::Error for ResponseFailure {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(&self.failure)
    }
}

/// Refusals which establish that the requested mutation was not applied.
///
/// Both response context and write-client exit classification use this table.
pub const NOT_APPLIED_REFUSALS: &[&str] = &[
    "invalid_identifier",
    "version_conflict",
    "main_protected",
    "discard_target_mismatch",
    "request_conflict",
    "branch_exists",
    "collection_missing",
    "document_missing",
    "relation_missing",
    "relation_collection",
    "relation_cardinality",
    "edge_properties",
    "vector_dimension_missing",
    "vector_dimension",
    "vector_empty",
    "vector_non_finite",
    "schema_refused",
    "json_text_invalid",
    "import_empty",
    "import_duplicate",
    "invalid_document",
    "import_limit",
    "import_limit_required",
    "sequence_exhausted",
    "invalid_operation",
    "busy",
    "writer_unavailable",
    "embed_admission_wait_exceeded",
    "invalid_params",
    "unknown_change_kind",
    "hello_required",
    "access_unavailable",
    "write_refused",
    "counts_corrupt",
    "count_overflow",
    "count_underflow",
    "counts_merge_pending",
    "counts_discard_pending",
    "counts_scan_refused",
    "index_field_not_top_level",
    "index_already_declared",
    "index_vector_field",
    "index_declaration_corrupt",
    "vector_field_embedded",
    "vector_field_supplied",
    "embed_model_change_not_available",
    "embed_manifest_invalid",
    "counts_access_refused",
    "discard_access_refused",
    "counts_branch_missing",
    "append_prune_scope_components_invalid",
    "append_prune_scope_component_invalid",
    "append_prune_suffix_invalid",
    "append_prune_document_id_invalid",
    "append_prune_keep_zero",
    "append_prune_keep_limit",
    "append_prune_key_exists",
    "append_prune_vector_field",
    "append_prune_limit_overflow",
    "append_prune_key_would_be_removed",
    "append_prune_outgoing_edge",
    "append_prune_incoming_edge",
    "append_prune_revision_missing",
    "append_prune_preparation_required",
    "route_unclassified",
    "permission_refused",
    "held_for_approval",
    "pass_missing",
    "lys_could_not_be_asked",
    "wrong_audience",
    "wrong_issuer",
    "pass_expired",
    "pass_not_yet_valid",
    "unpublished_key",
    "signature_refused",
    "pass_encoding_refused",
    "contract_refused",
    "rights_outside_audience",
    "rights_truncated",
    "unknown_method",
    "discard_pending",
    "branch_missing",
    "edge_key",
];

fn proves_not_applied(code: &str) -> bool {
    NOT_APPLIED_REFUSALS.contains(&code)
}

// Classification and typed decoding must share one vocabulary.
macro_rules! tagged_vocabulary {
    (
        $(#[$enum_doc:meta])*
        $name:ident ($tag:literal, $content:literal, $known:ident) {
            $( $(#[$doc:meta])* $variant:ident $(($parameter:ty))? => $wire:literal, )*
        }
    ) => {
        $(#[$enum_doc])*
        #[derive(Debug, Serialize, Deserialize)]
        #[serde(tag = $tag, content = $content, deny_unknown_fields)]
        pub enum $name {
            $( $(#[$doc])* #[serde(rename = $wire)] $variant $(($parameter))?, )*
        }

        impl $name {
            /// Every wire name of this vocabulary, in declaration order; the
            /// route matrix is checked against it.
            pub const WIRE_NAMES: &'static [&'static str] = &[$($wire),*];

            fn $known(name: &str) -> bool {
                matches!(name, $($wire)|*)
            }
        }
    };
}

/// Protocol identifier negotiated before issuing requests.
pub const PROTOCOL: &str = "haem/1";
pub const RECEIPT_CAPABILITY: &str = "typed_receipt_fragments";

/// A correlation identifier belongs to one connection, not durable history.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Request {
    pub id: String,
    pub operation: Operation,
}

tagged_vocabulary! {
    /// Reads select a captured version; mutations have a separate stable identity.
    Operation ("method", "params", known_method) {
        Hello(HelloAgreement) => "hello",
        /// Presents the connection's Lys pass; every later request on the
        /// connection is decided from it (ACCESS-001 R2). Answers
        /// `{holder, expires}` once the pass verifies.
        Present(PassPresent) => "session.pass",
        /// Runs one page of the drafts Lys approved for this database, with
        /// the connection's pass as the connector's: each is applied exactly
        /// once and acknowledged (ACCESS-001 R2). Answers `{acknowledged,
        /// next}`; nothing polls or retries on its own.
        DraftsRun(DraftsRun) => "drafts.run",
        Branches(BranchPage) => "branches.list",
        Collections(CollectionPage) => "collections.list",
        Get(DocumentRead) => "documents.get",
        List(DocumentPage) => "documents.list",
        History(HistoryPage) => "history.list",
        /// Answers a stream of frames, all with the request's id: the
        /// branch's head `{branch, version}` at once, then one each time its
        /// published version changes, ended by a refusal: `pass_expired` when
        /// the connection's pass expires, `branch_unknown` when the branch is
        /// gone. Pushed by the publish itself; nothing polls (serve/follow.rs).
        Follow(HeadFollow) => "head.follow",
        Review(ReviewPage) => "branch.review",
        /// Answers partial owner-key counts; sum pages until `done` is true.
        ReviewCounts(ReviewPage) => "branch.review_counts",
        /// Answers pinned three-way owner lines without admitting a mutation.
        MergePrepare(ReviewPage) => "branch.merge_prepare",
        /// Answers merge endpoint findings without choosing a conflict winner.
        MergeDangling(ReviewPage) => "branch.merge_dangling",
        Mutate(Mutation) => "mutation.apply",
        Receipt(ReceiptQuery) => "mutation.receipt",
        /// Recounts visible rows atomically; a repeated request replays its receipt.
        CountsRebuild(CountsRebuild) => "counts.rebuild",
        /// Compares kept counts with visible rows without changing the selected version.
        CountsAudit(Selection) => "counts.audit",
        /// Answers `{selection, edges: [Edge], next_after}`: edges leaving `node`.
        EdgesOut(EdgePage) => "edge.out",
        /// Answers `{selection, edges: [Edge], next_after}`: edges arriving at `node`.
        EdgesIn(EdgePage) => "edge.in",
        /// Answers `{selection, edges: [Edge], next_after}`: every edge of the version.
        GraphExport(GraphPage) => "graph.export",
        /// Answers `{selection, vector}`; `vector` is null when none is stored.
        VectorGet(VectorRead) => "vectors.get",
        /// Answers `{selection, vectors: [{document_id, vector}], next_after}`.
        VectorList(VectorPage) => "vectors.list",
        /// Answers `{selection, hits: [{document_id, score}], scanned}`.
        VectorSearch(VectorSearch) => "vectors.search",
        /// Answers `{selection, hits: [{document_id, score}], compared, stale,
        /// coverage}`: `text` embedded as a question, then the field searched.
        VectorSearchText(VectorSearchText) => "vectors.search_text",
        StoreStats(StoreRead) => "store.stats",
        VectorFields(StoreRead) => "vectors.fields",
        /// Answers `{branch, collection, field, through}`: opens one walk of
        /// the field toward `through` for the texts it passed as longer than
        /// the model reads (EMB-E2 (j)).
        EmbedRewalk(EmbedRewalk) => "embed.rewalk_longer",
        /// Answers `{selection, relations: [Relation]}` at the captured version.
        Relations(RelationsRead) => "relations.list",
        /// Answers `{selection, rows: [{value, origin}], next, engine, reading,
        /// collection, records_read}`: one page of a query's rows at one captured
        /// moment; `collection` is the one a pipeline read, else null.
        QueryRun(QueryRequest) => "query.run",
        QueryCheck(crate::serve::query::check::CheckRequest) => "query.check",
        QueryWords(crate::serve::query::check::WordsRequest) => "query.words",
        /// Answers `{from, to, rows: [{kind, origin, before, after}], next,
        /// changes_read}`: the documents changed between two versions.
        ChangesList(crate::serve::changes::ChangesRequest) => "changes.list",
    }
}

/// A Lys pass for audience haematite, as Lys issued it.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PassPresent {
    pub pass: String,
}

/// The approved page to run: the first, or the one after `after`.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DraftsRun {
    #[serde(default)]
    pub after: Option<String>,
}

/// A query's text and the page its answer may fill.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct QueryRequest {
    pub selection: Selection,
    pub query: String,
    /// Bound as jq variables beside the text, never pasted into it.
    #[serde(default)]
    pub parameters: std::collections::BTreeMap<String, Box<RawValue>>,
    pub page: QueryPage,
    #[serde(default)]
    pub continuation: Option<String>,
    /// `"origins"` leaves each row's value out; absent is `"values"`.
    #[serde(default)]
    pub rows: crate::serve::query::Rows,
}

/// At most `rows` rows, and at most `bytes` bytes for the whole answer.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct QueryPage {
    pub rows: std::num::NonZeroU64,
    pub bytes: std::num::NonZeroU64,
}

/// Decimal strings avoid loss of nanosecond identities in browser JSON numbers.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Version {
    #[serde(with = "decimal")]
    pub generation: u64,
    #[serde(with = "decimal")]
    pub sequence: u64,
}

/// An omitted version requests a fresh capture of the current branch head.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Selection {
    pub branch: String,
    pub version: Option<Version>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BranchPage {
    pub after: Option<String>,
    pub limit: std::num::NonZeroU32,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CollectionPage {
    pub selection: Selection,
    pub after: Option<String>,
    pub limit: std::num::NonZeroU32,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentRead {
    pub selection: Selection,
    pub collection: String,
    pub document_id: String,
}

/// Continuations carry the response's captured version, never a fresh head.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentPage {
    pub selection: Selection,
    pub collection: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prefix: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub content: Option<String>,
    pub after: Option<String>,
    pub limit: std::num::NonZeroU32,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HistoryPage {
    pub branch: String,
    pub after: Option<Version>,
    pub limit: std::num::NonZeroU32,
}

/// The branch whose head a connection follows.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HeadFollow {
    pub branch: String,
}

/// The continuation retains the review's captured versions on every page.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ReviewPage {
    pub branch: String,
    pub after: Option<String>,
    pub limit: std::num::NonZeroU32,
}

/// Query the current named branch; the response identifies its generation.
///
/// A fork's receipt is queried using the destination name, so a client need
/// not have received the new generation identifier before reconciling it.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ReceiptQuery {
    pub branch: String,
    pub request_id: String,
}

/// The service checks an existing receipt before refusing a stale expectation.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Mutation {
    pub request_id: String,
    pub branch: String,
    pub expected: Version,
    pub change: Change,
    /// STOR-050 R1: the stage this change is the result of, written as one
    /// row in the change's own commit (serve/storage/lineage.rs). Absent for
    /// a change that is no stage; refused beside branch control.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stage: Option<Box<Stage>>,
}

/// STOR-050 R1: a stage of a line's run, by name, and the versions it read.
/// Every member is a name or a version, never free JSON.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Stage {
    pub line: String,
    pub run: String,
    pub stage: String,
    pub subject: String,
    /// Every earlier result the stage read, in the order it read them; one
    /// is the common case, a join names each.
    pub inputs: Vec<StageInput>,
}

/// One version a stage read.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StageInput {
    pub branch: String,
    pub version: Version,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentAppendPrune {
    pub collection: String,
    pub scope: Vec<String>,
    pub suffix: String,
    pub document: Box<RawValue>,
    pub keep: u32,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RetentionResult {
    pub document_id: String,
    pub removed: u32,
    pub more: bool,
}

tagged_vocabulary! {
    /// Collection shapes, documents, edges and vectors share the branch commit
    /// boundary.
    ///
    /// Deleting a document also removes, in the same version, every
    /// edge that leaves or arrives at it and every vector stored for it.
    Change ("kind", "value", known_kind) {
        Collection(CollectionWrite) => "collection.put",
        Put(DocumentWrite) => "document.put",
        AppendPrune(DocumentAppendPrune) => "document.append_prune",
        Delete(DocumentDelete) => "document.delete",
        DeleteMany(DocumentDeleteMany) => "documents.delete_many",
        Fork(Fork) => "branch.fork",
        Discard(BranchDiscard) => "branch.discard",
        /// The branch `from`, at the version its review read, merged into
        /// the mutation's branch (serve/storage/merges.rs).
        Merge(BranchMerge) => "branch.merge",
        EdgePut(Edge) => "edge.put",
        EdgeDelete(EdgeDelete) => "edge.delete",
        IndexDeclare(IndexDeclaration) => "index.declare",
        VectorDeclare(VectorDeclaration) => "vector.declare",
        VectorPut(VectorWrite) => "vector.put",
        VectorPutMany(VectorImport) => "vectors.put_many",
        VectorDelete(VectorDelete) => "vector.delete",
        PutMany(DocumentImport) => "documents.put_many",
        /// Every edge checked as `edge.put` checks it, and each relation's
        /// cardinality across the batch too; all or none, in one commit.
        EdgePutMany(EdgeImport) => "edges.put_many",
        RelationPut(Relation) => "relation.put",
        RelationDelete(RelationDelete) => "relation.delete",
    }
}

pub(super) enum DecodedRequest {
    Request(Request),
    Refused {
        id: String,
        failure: ResponseFailure,
    },
}

#[derive(Deserialize)]
struct Correlation {
    id: String,
}

#[derive(Deserialize)]
struct Envelope<'a> {
    #[serde(borrow)]
    operation: Option<&'a RawValue>,
}

#[derive(Deserialize)]
struct OperationProbe<'a> {
    method: Option<String>,
    #[serde(borrow)]
    params: Option<&'a RawValue>,
}

#[derive(Deserialize)]
struct MutationProbe<'a> {
    request_id: Option<Value>,
    branch: Option<Value>,
    #[serde(borrow)]
    change: Option<&'a RawValue>,
}

#[derive(Deserialize)]
struct ChangeProbe {
    kind: Option<String>,
}

#[derive(Deserialize)]
struct ReceiptBranchProbe<'a> {
    kind: Option<String>,
    #[serde(borrow)]
    value: Option<&'a RawValue>,
}

#[derive(Deserialize)]
struct ForkNameProbe {
    name: Option<Value>,
}

pub(super) fn decode_request(bytes: &[u8]) -> io::Result<DecodedRequest> {
    if let Ok(request) = serde_json::from_slice::<Request>(bytes) {
        validate_id(&request.id)?;
        return Ok(DecodedRequest::Request(request));
    }
    let correlation: Correlation = serde_json::from_slice(bytes)
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
    validate_id(&correlation.id)?;
    let failure = serde_json::from_slice::<Envelope<'_>>(bytes).map_or_else(
        |_| invalid_params(),
        |envelope| classify_refusal(envelope.operation),
    );
    Ok(DecodedRequest::Refused {
        id: correlation.id,
        failure: ResponseFailure::new(failure, malformed_mutation_context(bytes)),
    })
}

fn malformed_mutation_context(bytes: &[u8]) -> Option<MutationContext> {
    let envelope = serde_json::from_slice::<Envelope<'_>>(bytes).ok()?;
    let operation = serde_json::from_str::<OperationProbe<'_>>(envelope.operation?.get()).ok()?;
    let rebuild = operation.method.as_deref() == Some("counts.rebuild");
    if !rebuild && operation.method.as_deref() != Some("mutation.apply") {
        return None;
    }
    let mutation = operation
        .params
        .and_then(|params| serde_json::from_str::<MutationProbe<'_>>(params.get()).ok());
    let (request_id, mut receipt_branch, change) = match mutation {
        Some(mutation) => (
            mutation.request_id.unwrap_or(Value::Null),
            mutation.branch.unwrap_or(Value::Null),
            mutation.change,
        ),
        None => (Value::Null, Value::Null, None),
    };
    if !rebuild
        && let Some(change) = change
        && let Ok(change) = serde_json::from_str::<ReceiptBranchProbe<'_>>(change.get())
        && change.kind.as_deref() == Some("branch.fork")
    {
        receipt_branch = change
            .value
            .and_then(|value| serde_json::from_str::<ForkNameProbe>(value.get()).ok())
            .and_then(|fork| fork.name)
            .unwrap_or(Value::Null);
    }
    Some(MutationContext {
        outcome: MutationOutcome::NotApplied,
        request_id,
        receipt_branch,
    })
}

fn validate_id(id: &str) -> io::Result<()> {
    if id.is_empty() || id.len() > 512 {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "invalid correlation ID",
        ));
    }
    Ok(())
}

fn classify_refusal(operation: Option<&RawValue>) -> Failure {
    let Some(operation) = operation else {
        return invalid_params();
    };
    let Ok(operation) = serde_json::from_str::<OperationProbe<'_>>(operation.get()) else {
        return invalid_params();
    };
    let Some(method) = operation.method else {
        return invalid_params();
    };
    if !Operation::known_method(&method) {
        return Failure::new("unknown_method", "method is not in the served vocabulary");
    }
    if method == "hello" {
        return Failure::new(
            "typed_receipt_capability_missing",
            format!("hello must agree {RECEIPT_CAPABILITY} and {PROTOCOL}"),
        );
    }
    if method == "mutation.apply"
        && let Some(params) = operation.params
        && let Ok(mutation) = serde_json::from_str::<MutationProbe<'_>>(params.get())
        && let Some(change) = mutation.change
        && let Ok(change) = serde_json::from_str::<ChangeProbe>(change.get())
        && let Some(kind) = change.kind
        && !Change::known_kind(&kind)
    {
        return Failure::new(
            "unknown_change_kind",
            "change kind is not in the served vocabulary",
        );
    }
    invalid_params()
}

fn invalid_params() -> Failure {
    Failure::new(
        "invalid_params",
        "request does not match the operation shape",
    )
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BranchDiscard {
    pub name: String,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BranchMerge {
    /// The branch merged in, which is left as it is.
    pub from: String,
    /// The version of `from` its review read; a merge of any other is refused.
    pub reviewed: Version,
}

/// A declaration constrains every edge under its name at this branch version.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Relation {
    pub relation: String,
    pub from_collection: String,
    pub to_collection: String,
    pub cardinality: Cardinality,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Cardinality {
    OneToOne,
    OneToMany,
    ManyToMany,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RelationDelete {
    pub relation: String,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RelationsRead {
    pub selection: Selection,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CollectionWrite {
    pub collection: String,
    /// None explicitly removes a prior schema. JSON false remains a schema.
    #[serde(deserialize_with = "explicit_schema")]
    pub schema: Option<Value>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentWrite {
    pub collection: String,
    pub document_id: String,
    /// Keep the submitted JSON bytes; parse a separate view for validation.
    pub document: Box<RawValue>,
}

/// Every row is validated before the import advances one branch version.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentImport {
    pub collection: String,
    pub documents: Vec<ImportedDocument>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ImportedDocument {
    pub document_id: String,
    pub document: Box<RawValue>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StoreRead {
    pub selection: Selection,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentDelete {
    pub collection: String,
    pub document_id: String,
}

/// Every identifier must exist in the collection before any cascade is staged.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentDeleteMany {
    pub collection: String,
    pub document_ids: Vec<String>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Fork {
    pub name: String,
    /// Must be a retained version of the mutation's source branch.
    pub from: Version,
    /// STOR-050 R2: the new branch is a line, which keeps every version it
    /// commits until a `branch.discard` names it. Absent or false is an
    /// ordinary branch.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub line: bool,
}

/// One end of an edge: a document named by its collection and identifier.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Node {
    pub collection: String,
    pub document_id: String,
}

/// An edge is identified by its relation and both ends; putting the same
/// three again replaces its properties. Both ends must exist at the write.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Edge {
    pub relation: String,
    pub from: Node,
    pub to: Node,
    /// A JSON object; any other JSON value is refused.
    pub properties: Value,
}

/// Edges written together (REL0), counted against the same row bound as
/// `documents.put_many`.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EdgeImport {
    pub edges: Vec<Edge>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EdgeDelete {
    pub relation: String,
    pub from: Node,
    pub to: Node,
}

/// Edges at one node, in key order. An omitted relation selects every
/// relation. `after` is the previous answer's `next_after`, unchanged, and
/// requires the captured version.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EdgePage {
    pub selection: Selection,
    pub node: Node,
    pub relation: Option<String>,
    pub after: Option<String>,
    pub limit: std::num::NonZeroU32,
}

/// Every edge of the captured version, each exactly once, in key order.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct GraphPage {
    pub selection: Selection,
    pub after: Option<String>,
    pub limit: std::num::NonZeroU32,
    /// GX1: only these relations; an empty list answers no edges. Each must
    /// be declared.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub relations: Option<Vec<String>>,
    /// GX1: only edges with both ends in these collections.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub collections: Option<Vec<String>>,
    /// GX1: for each collection, the members whose stored JSON text each
    /// record at an edge's end carries in `nodes`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub members: Option<std::collections::BTreeMap<String, Vec<String>>>,
    /// GX2: only edges with both ends among these records, at most the
    /// examination bound (`max_page × 64`) of them and their JSON within the
    /// frame (`GX2c`). Each must exist.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub records: Option<Vec<Node>>,
    /// GR1: beside the edges, records of each chosen collection (those of
    /// `collections`, or every collection) as nodes, whether or not any edge
    /// names them, with the members named in `members`, never a body.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub unrelated: Option<Unrelated>,
}

/// GR1: how many records with no edge asked for to draw.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Unrelated {
    /// At most this many records of each chosen collection.
    pub per_collection: std::num::NonZeroU64,
    /// At most this many records in the whole paged series.
    pub most: std::num::NonZeroU64,
}

/// A field declaration shares the collection mutation and receipt boundary.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct VectorDeclaration {
    pub collection: String,
    pub field: String,
    pub declaration: super::storage::embedded::Declaration,
}

/// A vector is stored for a named field of an existing document.
///
/// The first
/// write for a collection and field fixes its dimension; a later vector of
/// another dimension, an empty vector and a non-finite member are refused.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct VectorWrite {
    pub collection: String,
    pub document_id: String,
    pub field: String,
    pub vector: Vec<f32>,
}

/// Supplied fields must already have a recorded dimension. Every row is
/// validated before the batch advances one branch version.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct VectorImport {
    pub vectors: Vec<VectorWrite>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct VectorDelete {
    pub collection: String,
    pub document_id: String,
    pub field: String,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct VectorRead {
    pub selection: Selection,
    pub collection: String,
    pub document_id: String,
    pub field: String,
}

/// Every vector of one field, by document identifier. `after` is the last
/// document identifier of the previous page and requires the captured version.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct VectorPage {
    pub selection: Selection,
    pub collection: String,
    pub field: String,
    pub after: Option<String>,
    pub limit: std::num::NonZeroU32,
}

/// Exact search: every vector of the field is compared with `query` and the
/// answer's `scanned` says how many were. `k` is bounded by the page limit.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct VectorSearch {
    pub selection: Selection,
    pub collection: String,
    pub field: String,
    pub query: Vec<f32>,
    pub k: std::num::NonZeroU32,
    pub metric: Metric,
}

/// Text search (EW1): `text` is embedded by the service's worker, outside any
/// lock, and the field's embedded vectors searched with it under the field's
/// declared metric. `k` is bounded by the page limit.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct VectorSearchText {
    pub selection: Selection,
    pub collection: String,
    pub field: String,
    pub text: String,
    pub k: std::num::NonZeroU32,
}

/// Hits are ordered best first: highest score for cosine and dot, lowest
/// distance for l2; equal scores order by document identifier.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Metric {
    Cosine,
    L2,
    Dot,
}

mod decimal {
    use serde::{Deserialize, Deserializer, Serializer, de::Error};

    pub fn serialize<S: Serializer>(
        value: impl std::fmt::Display,
        serializer: S,
    ) -> Result<S::Ok, S::Error> {
        serializer.serialize_str(&value.to_string())
    }

    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<u64, D::Error> {
        let text = String::deserialize(deserializer)?;
        let value = text.parse::<u64>().map_err(D::Error::custom)?;
        if text != value.to_string() {
            return Err(D::Error::custom(
                "expected a canonical unsigned decimal string",
            ));
        }
        Ok(value)
    }
}

// A custom deserializer makes absence an error while retaining explicit null.
fn explicit_schema<'de, D: serde::Deserializer<'de>>(
    deserializer: D,
) -> Result<Option<Value>, D::Error> {
    Option::<Value>::deserialize(deserializer)
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum WireOwnerKind {
    Database,
    Api,
    LeafLog,
    Observer,
    Branch,
    BranchCommit,
    BranchRef,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireOwnedCause {
    pub kind: WireOwnerKind,
    pub message: String,
    pub database: Option<super::receipt_wire::WireCause>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WireRefusal {
    pub code: String,
    pub message: String,
    pub path: Option<String>,
    pub cause: Option<WireOwnedCause>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum WirePacket {
    Receipt {
        receipt: Box<super::receipt_wire::WireReceipt>,
        refusal: Option<WireRefusal>,
    },
    Refusal {
        refusal: WireRefusal,
    },
}
impl WirePacket {
    #[must_use]
    pub fn receipt(&self) -> Option<&super::receipt_wire::WireReceipt> {
        match self {
            Self::Receipt { receipt, .. } => Some(receipt.as_ref()),
            Self::Refusal { .. } => None,
        }
    }
    #[must_use]
    pub const fn refusal(&self) -> Option<&WireRefusal> {
        match self {
            Self::Receipt { refusal, .. } => refusal.as_ref(),
            Self::Refusal { refusal } => Some(refusal),
        }
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HelloAgreement {
    pub protocol: String,
    pub capabilities: Vec<String>,
    pub max_frame: std::num::NonZeroU32,
}
impl HelloAgreement {
    pub(crate) fn client() -> Self {
        Self {
            protocol: PROTOCOL.to_owned(),
            capabilities: vec![RECEIPT_CAPABILITY.to_owned()],
            max_frame: std::num::NonZeroU32::MAX,
        }
    }
    pub(crate) fn validate(
        &self,
        server_cap: std::num::NonZeroU32,
    ) -> Result<std::num::NonZeroU32, Failure> {
        if self.protocol != PROTOCOL {
            return Err(Failure::new(
                "receipt_protocol_mismatch",
                format!("expected {PROTOCOL}, received {}", self.protocol),
            ));
        }
        if !self
            .capabilities
            .iter()
            .any(|capability| capability == RECEIPT_CAPABILITY)
        {
            return Err(Failure::new(
                "typed_receipt_capability_missing",
                format!("hello must agree {RECEIPT_CAPABILITY}"),
            ));
        }
        let cap = self.max_frame.min(server_cap);
        let minimum = super::receipt_fragments::minimum_cap().map_err(Failure::storage)?;
        if u64::from(cap.get()) < u64::try_from(minimum).map_err(Failure::storage)? {
            return Err(Failure::new(
                "receipt_frame_too_small",
                format!("offered {}, minimum {minimum}", cap.get()),
            ));
        }
        Ok(cap)
    }
}

#[cfg(test)]
mod receipt_agreement_tests {
    use super::*;

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

    #[test]
    fn missing_capability_and_derived_floor_are_named_before_dispatch() -> TestResult {
        let mut agreement = HelloAgreement::client();
        agreement.capabilities.clear();
        let cap = std::num::NonZeroU32::MAX;
        let Err(missing) = agreement.validate(cap) else {
            return Err("a hello without the receipt capability was agreed".into());
        };
        assert_eq!(missing.code, "typed_receipt_capability_missing");
        agreement.capabilities.push(RECEIPT_CAPABILITY.to_owned());
        let minimum = super::super::receipt_fragments::minimum_cap()?;
        let below = u32::try_from(minimum - 1)?;
        agreement.max_frame = std::num::NonZeroU32::new(below).ok_or("floor is one")?;
        let Err(error) = agreement.validate(cap) else {
            return Err("a frame below the receipt floor was agreed".into());
        };
        assert_eq!(error.code, "receipt_frame_too_small");
        assert!(error.message.contains(&minimum.to_string()));
        assert!(error.message.contains(&agreement.max_frame.to_string()));
        Ok(())
    }
}