haematite-cli 0.21.7

//! A request's attribution, independent of the number of things it changed.

use serde::{Deserialize, Serialize};

use super::{Caller, ResolvedMutation, identifier};
use crate::serve::protocol::{Change, Node, Version};
use crate::serve::storage::Failure;

#[path = "attribution_batch.rs"]
mod batch;
pub use batch::Deletion;
#[path = "attribution_key.rs"]
mod keys;
pub use keys::{decode_key, key, version_prefix};

pub const TAG: &[u8] = b"a\0";
pub const MAX_ROW_BYTES: usize = 64 * 1024;
const FORMAT: u32 = 1;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ActorKind {
    Person,
    Agent,
}

/// Only identity metadata belongs here; payloads stay in their own rows.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "kind", deny_unknown_fields)]
pub enum Changed {
    #[serde(rename = "collection.put")]
    CollectionPut { collection: String },
    #[serde(rename = "document.put")]
    DocumentPut {
        collection: String,
        document_id: String,
    },
    #[serde(rename = "document.append_prune")]
    AppendPrune {
        collection: String,
        scope: Vec<String>,
        suffix: String,
        keep: u32,
    },
    #[serde(rename = "document.delete")]
    DocumentDelete {
        collection: String,
        document_id: String,
    },
    #[serde(rename = "documents.delete_many")]
    DeleteMany { changes: Vec<Deletion> },
    #[serde(rename = "documents.put_many")]
    Import { collection: String, count: u64 },
    #[serde(rename = "edge.put")]
    EdgePut {
        relation: String,
        from: Node,
        to: Node,
    },
    #[serde(rename = "edge.delete")]
    EdgeDelete {
        relation: String,
        from: Node,
        to: Node,
    },
    #[serde(rename = "edges.put_many")]
    EdgeImport { count: u64 },
    #[serde(rename = "index.declare")]
    IndexDeclare { collection: String, field: String },
    #[serde(rename = "vector.declare")]
    VectorDeclare { collection: String, field: String },
    #[serde(rename = "vector.put")]
    VectorPut {
        collection: String,
        document_id: String,
        field: String,
    },
    #[serde(rename = "vectors.put_many")]
    VectorImport { count: u64 },
    #[serde(rename = "vector.delete")]
    VectorDelete {
        collection: String,
        document_id: String,
        field: String,
    },
    #[serde(rename = "relation.put")]
    RelationPut { relation: String },
    #[serde(rename = "relation.delete")]
    RelationDelete { relation: String },
    #[serde(rename = "branch.fork")]
    Fork {
        name: String,
        from_branch: String,
        from: Version,
    },
    #[serde(rename = "branch.discard")]
    Discard { name: String },
    #[serde(rename = "branch.merge")]
    Merge { from: String, reviewed: Version },
}

impl Changed {
    fn from_request(request: &ResolvedMutation) -> Result<Self, Failure> {
        validate_input(request.change(), request.branch())?;
        let changed = match request.change() {
            Change::Collection(write) => Self::CollectionPut {
                collection: write.collection.clone(),
            },
            Change::Put(write) => Self::DocumentPut {
                collection: write.collection.clone(),
                document_id: write.document_id.clone(),
            },
            Change::AppendPrune(write) => Self::AppendPrune {
                collection: write.collection.clone(),
                scope: write.scope.clone(),
                suffix: write.suffix.clone(),
                keep: write.keep,
            },
            Change::Delete(write) => Self::DocumentDelete {
                collection: write.collection.clone(),
                document_id: write.document_id.clone(),
            },
            Change::DeleteMany(delete) => Self::DeleteMany {
                changes: batch::entries(delete),
            },
            Change::PutMany(import) => Self::Import {
                collection: import.collection.clone(),
                count: u64::try_from(import.documents.len()).map_err(|_| {
                    Failure::new("change_record_count", "import row count exceeds u64")
                })?,
            },
            Change::EdgePut(edge) => Self::EdgePut {
                relation: edge.relation.clone(),
                from: edge.from.clone(),
                to: edge.to.clone(),
            },
            Change::EdgeDelete(edge) => Self::EdgeDelete {
                relation: edge.relation.clone(),
                from: edge.from.clone(),
                to: edge.to.clone(),
            },
            Change::EdgePutMany(import) => Self::EdgeImport {
                count: u64::try_from(import.edges.len()).map_err(|_| {
                    Failure::new("change_record_count", "edge import row count exceeds u64")
                })?,
            },
            Change::IndexDeclare(field) => Self::IndexDeclare {
                collection: field.collection.clone(),
                field: field.field.clone(),
            },
            Change::VectorDeclare(field) => Self::VectorDeclare {
                collection: field.collection.clone(),
                field: field.field.clone(),
            },
            Change::VectorPut(vector) => Self::VectorPut {
                collection: vector.collection.clone(),
                document_id: vector.document_id.clone(),
                field: vector.field.clone(),
            },
            Change::VectorPutMany(import) => Self::VectorImport {
                count: u64::try_from(import.vectors.len()).map_err(|_| {
                    Failure::new("change_record_count", "vector import row count exceeds u64")
                })?,
            },
            Change::VectorDelete(vector) => Self::VectorDelete {
                collection: vector.collection.clone(),
                document_id: vector.document_id.clone(),
                field: vector.field.clone(),
            },
            Change::RelationPut(relation) => Self::RelationPut {
                relation: relation.relation.clone(),
            },
            Change::RelationDelete(relation) => Self::RelationDelete {
                relation: relation.relation.clone(),
            },
            Change::Fork(fork) => Self::Fork {
                name: fork.name.clone(),
                from_branch: request.branch().to_owned(),
                from: fork.from,
            },
            Change::Discard(discard) => Self::Discard {
                name: discard.name.clone(),
            },
            Change::Merge(merge) => Self::Merge {
                from: merge.from.clone(),
                reviewed: merge.reviewed,
            },
        };
        Ok(changed)
    }

    fn validate(&self) -> Result<(), Failure> {
        match self {
            Self::CollectionPut { collection } => identifier(collection),
            Self::DocumentPut {
                collection,
                document_id,
            }
            | Self::DocumentDelete {
                collection,
                document_id,
            } => names([collection, document_id]),
            Self::AppendPrune {
                collection,
                scope,
                suffix,
                keep,
            } => {
                identifier(collection)?;
                crate::serve::storage::append_prune_address(scope, suffix)?;
                if *keep == 0 {
                    return Err(Failure::new(
                        "change_record_retention",
                        "invalid retention metadata",
                    ));
                }
                Ok(())
            }
            Self::DeleteMany { changes } => batch::validate(changes),
            Self::Import { collection, count } => {
                identifier(collection)?;
                if *count == 0 {
                    return Err(Failure::new(
                        "change_record_count",
                        "an accepted import cannot be empty",
                    ));
                }
                Ok(())
            }
            Self::EdgePut { relation, from, to } | Self::EdgeDelete { relation, from, to } => {
                names([
                    relation,
                    &from.collection,
                    &from.document_id,
                    &to.collection,
                    &to.document_id,
                ])
            }
            Self::EdgeImport { count } => {
                if *count == 0 {
                    return Err(Failure::new(
                        "change_record_count",
                        "an accepted edge import cannot be empty",
                    ));
                }
                Ok(())
            }
            Self::VectorImport { count } => {
                if *count == 0 {
                    return Err(Failure::new(
                        "change_record_count",
                        "an accepted vector import cannot be empty",
                    ));
                }
                Ok(())
            }
            Self::IndexDeclare { collection, field } => {
                crate::serve::storage::field_index::validate_names(collection, field)
            }
            Self::VectorDeclare { collection, field } => names([collection, field]),
            Self::VectorPut {
                collection,
                document_id,
                field,
            }
            | Self::VectorDelete {
                collection,
                document_id,
                field,
            } => names([collection, document_id, field]),
            Self::RelationPut { relation } | Self::RelationDelete { relation } => {
                identifier(relation)
            }
            Self::Fork {
                name, from_branch, ..
            } => names([name, from_branch]),
            Self::Discard { name } => identifier(name),
            Self::Merge { from, .. } => identifier(from),
        }
    }
}

fn validate_input(change: &Change, source: &str) -> Result<(), Failure> {
    match change {
        Change::Collection(write) => identifier(&write.collection),
        Change::Put(write) => names([&write.collection, &write.document_id]),
        Change::Delete(write) => names([&write.collection, &write.document_id]),
        Change::DeleteMany(delete) => batch::validate(&batch::entries(delete)),
        Change::AppendPrune(write) => {
            identifier(&write.collection)?;
            crate::serve::storage::append_prune_address(&write.scope, &write.suffix)?;
            Ok(())
        }
        Change::PutMany(import) => {
            identifier(&import.collection)?;
            if import.documents.is_empty() {
                return Err(Failure::new(
                    "change_record_count",
                    "an accepted import cannot be empty",
                ));
            }
            Ok(())
        }
        Change::EdgePut(edge) => names([
            &edge.relation,
            &edge.from.collection,
            &edge.from.document_id,
            &edge.to.collection,
            &edge.to.document_id,
        ]),
        Change::EdgeDelete(edge) => names([
            &edge.relation,
            &edge.from.collection,
            &edge.from.document_id,
            &edge.to.collection,
            &edge.to.document_id,
        ]),
        Change::EdgePutMany(import) => {
            if import.edges.is_empty() {
                return Err(Failure::new(
                    "change_record_count",
                    "an accepted edge import cannot be empty",
                ));
            }
            for edge in &import.edges {
                names([
                    &edge.relation,
                    &edge.from.collection,
                    &edge.from.document_id,
                    &edge.to.collection,
                    &edge.to.document_id,
                ])?;
            }
            Ok(())
        }
        Change::IndexDeclare(field) => {
            crate::serve::storage::field_index::validate_names(&field.collection, &field.field)
        }
        Change::VectorDeclare(field) => names([&field.collection, &field.field]),
        Change::VectorPut(vector) => {
            names([&vector.collection, &vector.document_id, &vector.field])
        }
        Change::VectorPutMany(import) => {
            if import.vectors.is_empty() {
                return Err(Failure::new(
                    "change_record_count",
                    "an accepted vector import cannot be empty",
                ));
            }
            for vector in &import.vectors {
                names([&vector.collection, &vector.document_id, &vector.field])?;
            }
            Ok(())
        }
        Change::VectorDelete(vector) => {
            names([&vector.collection, &vector.document_id, &vector.field])
        }
        Change::RelationPut(relation) => identifier(&relation.relation),
        Change::RelationDelete(relation) => identifier(&relation.relation),
        Change::Fork(fork) => names([&fork.name, source]),
        Change::Discard(discard) => identifier(&discard.name),
        Change::Merge(merge) => identifier(&merge.from),
    }
}

fn names<const N: usize>(names: [&str; N]) -> Result<(), Failure> {
    names.into_iter().try_for_each(identifier)
}

/// A stored author is mandatory and never resolved from a current session.
#[derive(Debug, Serialize)]
pub struct Row {
    format: u32,
    version: Version,
    request_id: String,
    caller: Caller,
    origin: Version,
    change: Changed,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Wire {
    format: u32,
    version: Version,
    request_id: String,
    caller: Caller,
    origin: Version,
    change: Changed,
}

#[derive(Deserialize)]
struct Header {
    format: Option<u64>,
}

impl Row {
    /// Describe the change and its resolved caller at the written version.
    ///
    /// # Errors
    /// Propagates invalid change identifiers, malformed change input or an import count that exceeds the stored range.
    pub fn for_request(request: &ResolvedMutation, written_at: Version) -> Result<Self, Failure> {
        Ok(Self {
            format: FORMAT,
            version: written_at,
            request_id: request.request_id().to_owned(),
            caller: request.caller().clone(),
            origin: request.origin(),
            change: Changed::from_request(request)?,
        })
    }

    /// Encode this row's version and request identity.
    ///
    /// # Errors
    /// Returns `invalid_identifier` for an invalid stored request id.
    pub fn key(&self) -> Result<Vec<u8>, Failure> {
        key(self.version, &self.request_id)
    }

    /// Serialize one attribution row within the stored row bound.
    ///
    /// # Errors
    /// Returns `storage_unavailable` on serialization failure or the named row-size refusal when the row is too large.
    pub fn encode(&self) -> Result<Vec<u8>, Failure> {
        let bytes = serde_json::to_vec(self).map_err(Failure::storage)?;
        row_bound(&bytes)?;
        Ok(bytes)
    }

    /// Decode a bounded row and verify that its identity matches its key.
    ///
    /// # Errors
    /// Refuses oversized or malformed rows, unsupported formats, invalid keys, mismatched identities and invalid change payloads.
    pub fn decode(key: &[u8], value: &[u8]) -> Result<Self, Failure> {
        row_bound(value)?;
        let (version, request_id) = decode_key(key)?;
        let header: Header = serde_json::from_slice(value).map_err(corrupt)?;
        if header.format != Some(u64::from(FORMAT)) {
            return Err(Failure::new(
                "change_record_format_unsupported",
                "attribution row is not format 1",
            ));
        }
        let wire: Wire = serde_json::from_slice(value).map_err(corrupt)?;
        if wire.version != version || wire.request_id != request_id {
            return Err(Failure::new(
                "change_record_key_mismatch",
                "attribution version or request id differs from its key",
            ));
        }
        wire.change.validate().map_err(corrupt)?;
        Ok(Self {
            format: wire.format,
            version: wire.version,
            request_id: wire.request_id,
            caller: wire.caller,
            origin: wire.origin,
            change: wire.change,
        })
    }

    #[must_use]
    pub const fn version(&self) -> Version {
        self.version
    }
    #[must_use]
    pub fn request_id(&self) -> &str {
        &self.request_id
    }
    #[must_use]
    pub const fn caller(&self) -> &Caller {
        &self.caller
    }
    #[must_use]
    pub const fn origin(&self) -> Version {
        self.origin
    }
    #[must_use]
    pub const fn changed(&self) -> &Changed {
        &self.change
    }
    #[must_use]
    pub const fn actor_kind(&self) -> ActorKind {
        if self.caller.responsible.is_some() {
            ActorKind::Agent
        } else {
            ActorKind::Person
        }
    }
}

fn corrupt(error: impl std::fmt::Display) -> Failure {
    Failure::new("change_record_corrupt", error.to_string())
}

fn row_bound(bytes: &[u8]) -> Result<(), Failure> {
    if bytes.len() > MAX_ROW_BYTES {
        return Err(Failure::new(
            "change_record_too_large",
            "attribution row exceeds 65536 bytes",
        ));
    }
    Ok(())
}

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