haematite-cli 0.21.7

//! Strict stored pointers; document bodies remain in their separate data rows.

use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::Value;

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

const FORMAT: u32 = 1;

/// The written version is independent of the branch from which a row is read.
pub trait Attributed {
    fn written_at(&self) -> Version;
    fn request_id(&self) -> &str;
}

/// Locate an attributed envelope's author row.
///
/// # Errors
/// Returns `invalid_identifier` when the envelope's request id cannot form an attribution key.
pub fn attribution_key(envelope: &impl Attributed) -> Result<Vec<u8>, Failure> {
    attribution::key(envelope.written_at(), envelope.request_id())
}

#[derive(Deserialize)]
struct Header {
    format: Option<u32>,
    written_at: Option<Version>,
    changed_at: Option<Version>,
    request_id: Option<String>,
}

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

fn decode<T: DeserializeOwned>(bytes: &[u8], collection: bool) -> Result<T, Failure> {
    // Inspect only metadata before allocating a payload; no old-shape fallback.
    let header: Header = serde_json::from_slice(bytes).map_err(corrupt)?;
    if header.format != Some(FORMAT) {
        return Err(Failure::new(
            "stored_envelope_format_unsupported",
            "stored envelope requires format 1; older shapes are not readable",
        ));
    }
    let version = if collection {
        header.changed_at
    } else {
        header.written_at
    };
    let (Some(version), Some(request_id)) = (version, header.request_id) else {
        return Err(Failure::new(
            "stored_envelope_pointer_missing",
            "stored envelope requires its written version and request id",
        ));
    };
    validate_pointer(version, &request_id)?;
    serde_json::from_slice(bytes).map_err(corrupt)
}

fn encode(value: &impl Serialize) -> Result<Vec<u8>, Failure> {
    serde_json::to_vec(value).map_err(corrupt)
}

fn validate_pointer(version: Version, request_id: &str) -> Result<(), Failure> {
    attribution::key(version, request_id)
        .map(|_| ())
        .map_err(|error| Failure::new("stored_envelope_pointer_invalid", error.message))
}

fn validate_name(name: &str) -> Result<(), Failure> {
    identifier(name).map_err(|error| Failure::new("stored_envelope_payload_invalid", error.message))
}

fn wrong_kind() -> Failure {
    Failure::new(
        "stored_envelope_request_kind",
        "request does not write this kind of stored envelope",
    )
}

#[derive(Debug, Serialize)]
#[serde(transparent)]
pub struct Document {
    wire: DocumentWire,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct DocumentWire {
    format: u32,
    collection_branch: String,
    collection_version: Version,
    written_in: String,
    written_at: Version,
    request_id: String,
}

impl Document {
    /// Build document provenance for a document-writing request.
    ///
    /// # Errors
    /// Refuses another change kind, an invalid attribution pointer or invalid branch names.
    pub fn for_request(
        request: &ResolvedMutation,
        written_at: Version,
        collection_branch: String,
        collection_version: Version,
    ) -> Result<Self, Failure> {
        if !matches!(request.change(), Change::Put(_) | Change::PutMany(_)) {
            return Err(wrong_kind());
        }
        Self::from_wire(DocumentWire {
            format: FORMAT,
            collection_branch,
            collection_version,
            written_in: request.branch().to_owned(),
            written_at,
            request_id: request.request_id().to_owned(),
        })
    }

    fn from_wire(wire: DocumentWire) -> Result<Self, Failure> {
        validate_pointer(wire.written_at, &wire.request_id)?;
        validate_name(&wire.collection_branch)?;
        validate_name(&wire.written_in)?;
        Ok(Self { wire })
    }

    /// Decode and validate document provenance.
    ///
    /// # Errors
    /// Refuses malformed JSON, unsupported formats, missing or invalid attribution pointers and invalid branch names.
    pub fn decode(bytes: &[u8]) -> Result<Self, Failure> {
        Self::from_wire(decode(bytes, false)?)
    }

    /// Encode document provenance.
    ///
    /// # Errors
    /// Returns `stored_envelope_corrupt` if serialization fails.
    pub fn encode(&self) -> Result<Vec<u8>, Failure> {
        encode(self)
    }

    #[must_use]
    pub fn collection_branch(&self) -> &str {
        &self.wire.collection_branch
    }

    #[must_use]
    pub const fn collection_version(&self) -> Version {
        self.wire.collection_version
    }

    #[must_use]
    pub fn written_in(&self) -> &str {
        &self.wire.written_in
    }
}

impl Attributed for Document {
    fn written_at(&self) -> Version {
        self.wire.written_at
    }

    fn request_id(&self) -> &str {
        &self.wire.request_id
    }
}

#[derive(Debug, Serialize)]
#[serde(transparent)]
pub struct Collection {
    wire: CollectionWire,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct CollectionWire {
    format: u32,
    name: String,
    schema: Option<Value>,
    changed_in: String,
    changed_at: Version,
    request_id: String,
}

impl Collection {
    /// Build a collection envelope from its declaring request.
    ///
    /// # Errors
    /// Refuses another change kind, an invalid attribution pointer or invalid collection and branch names.
    pub fn for_request(request: &ResolvedMutation, changed_at: Version) -> Result<Self, Failure> {
        let Change::Collection(write) = request.change() else {
            return Err(wrong_kind());
        };
        Self::from_wire(CollectionWire {
            format: FORMAT,
            name: write.collection.clone(),
            schema: write.schema.clone(),
            changed_in: request.branch().to_owned(),
            changed_at,
            request_id: request.request_id().to_owned(),
        })
    }

    fn from_wire(wire: CollectionWire) -> Result<Self, Failure> {
        validate_pointer(wire.changed_at, &wire.request_id)?;
        validate_name(&wire.name)?;
        validate_name(&wire.changed_in)?;
        Ok(Self { wire })
    }

    /// Decode and validate a collection envelope.
    ///
    /// # Errors
    /// Refuses malformed JSON, unsupported formats, missing or invalid attribution pointers and invalid names.
    pub fn decode(bytes: &[u8]) -> Result<Self, Failure> {
        Self::from_wire(decode(bytes, true)?)
    }

    /// Encode this collection envelope.
    ///
    /// # Errors
    /// Returns `stored_envelope_corrupt` if serialization fails.
    pub fn encode(&self) -> Result<Vec<u8>, Failure> {
        encode(self)
    }

    #[must_use]
    pub fn name(&self) -> &str {
        &self.wire.name
    }

    #[must_use]
    pub const fn schema(&self) -> Option<&Value> {
        self.wire.schema.as_ref()
    }

    #[must_use]
    pub fn changed_in(&self) -> &str {
        &self.wire.changed_in
    }

    #[must_use]
    pub const fn changed_at(&self) -> Version {
        self.wire.changed_at
    }
}

impl Attributed for Collection {
    fn written_at(&self) -> Version {
        self.wire.changed_at
    }

    fn request_id(&self) -> &str {
        &self.wire.request_id
    }
}

#[derive(Debug, Serialize)]
#[serde(transparent)]
pub struct Edge {
    wire: EdgeWire,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct EdgeWire {
    format: u32,
    properties: Value,
    written_in: String,
    written_at: Version,
    request_id: String,
}

impl Edge {
    /// Build an edge envelope from its writing request.
    ///
    /// # Errors
    /// Refuses another change kind, an invalid attribution pointer, an invalid branch name or properties that are not an object.
    pub fn for_request(request: &ResolvedMutation, written_at: Version) -> Result<Self, Failure> {
        let Change::EdgePut(edge) = request.change() else {
            return Err(wrong_kind());
        };
        Self::from_wire(EdgeWire {
            format: FORMAT,
            properties: edge.properties.clone(),
            written_in: request.branch().to_owned(),
            written_at,
            request_id: request.request_id().to_owned(),
        })
    }

    fn from_wire(wire: EdgeWire) -> Result<Self, Failure> {
        validate_pointer(wire.written_at, &wire.request_id)?;
        validate_name(&wire.written_in)?;
        if !wire.properties.is_object() {
            return Err(Failure::new(
                "stored_envelope_payload_invalid",
                "edge properties must be a JSON object",
            ));
        }
        Ok(Self { wire })
    }

    /// Decode and validate an edge envelope.
    ///
    /// # Errors
    /// Refuses malformed JSON, unsupported formats, invalid provenance or edge properties that are not an object.
    pub fn decode(bytes: &[u8]) -> Result<Self, Failure> {
        Self::from_wire(decode(bytes, false)?)
    }

    /// Encode this edge envelope.
    ///
    /// # Errors
    /// Returns `stored_envelope_corrupt` if serialization fails.
    pub fn encode(&self) -> Result<Vec<u8>, Failure> {
        encode(self)
    }

    #[must_use]
    pub const fn properties(&self) -> &Value {
        &self.wire.properties
    }

    #[must_use]
    pub fn written_in(&self) -> &str {
        &self.wire.written_in
    }
}

impl Attributed for Edge {
    fn written_at(&self) -> Version {
        self.wire.written_at
    }

    fn request_id(&self) -> &str {
        &self.wire.request_id
    }
}

#[derive(Debug)]
pub struct Vector {
    header: VectorHeader,
    vector: Vec<f32>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct VectorHeader {
    format: u32,
    written_in: String,
    written_at: Version,
    request_id: String,
    dimensions: usize,
}

fn vector_header_error(error: Failure) -> Failure {
    if error.code == "stored_envelope_corrupt" {
        Failure::new("vector_envelope_header_invalid", error.message)
    } else {
        error
    }
}

fn vector_tail_length(header: &VectorHeader) -> Result<usize, Failure> {
    validate_pointer(header.written_at, &header.request_id)?;
    validate_name(&header.written_in)?;
    if header.dimensions == 0 {
        return Err(Failure::new(
            "vector_envelope_dimensions_invalid",
            "vector dimensions must be at least one",
        ));
    }
    header.dimensions.checked_mul(4).ok_or_else(|| {
        Failure::new(
            "vector_envelope_dimensions_invalid",
            "vector byte length exceeds the addressable size",
        )
    })
}

fn finite(member: f32) -> Result<f32, Failure> {
    if !member.is_finite() {
        return Err(Failure::new(
            "vector_envelope_non_finite",
            "vector members must be finite",
        ));
    }
    Ok(member)
}

impl Vector {
    /// Build a vector envelope from its writing request.
    ///
    /// # Errors
    /// Refuses another change kind, invalid provenance, zero or unrepresentable dimensions and non-finite vector members.
    pub fn for_request(request: &ResolvedMutation, written_at: Version) -> Result<Self, Failure> {
        let Change::VectorPut(vector) = request.change() else {
            return Err(wrong_kind());
        };
        let header = VectorHeader {
            format: FORMAT,
            written_in: request.branch().to_owned(),
            written_at,
            request_id: request.request_id().to_owned(),
            dimensions: vector.vector.len(),
        };
        vector_tail_length(&header)?;
        for member in &vector.vector {
            finite(*member)?;
        }
        Ok(Self {
            header,
            vector: vector.vector.clone(),
        })
    }

    /// Decode the vector's JSON header and binary tail.
    ///
    /// # Errors
    /// Refuses a missing or malformed header, unsupported format, invalid provenance or dimensions, a mismatched tail length or non-finite members.
    pub fn decode(bytes: &[u8]) -> Result<Self, Failure> {
        let newline = bytes
            .iter()
            .position(|byte| *byte == b'\n')
            .ok_or_else(|| {
                Failure::new(
                    "vector_envelope_header_missing",
                    "vector requires a JSON header followed by one LF",
                )
            })?;
        let (line, remainder) = bytes.split_at(newline);
        if line.contains(&b'\r')
            || line.iter().find(|byte| !byte.is_ascii_whitespace()) != Some(&b'{')
        {
            return Err(Failure::new(
                "vector_envelope_header_invalid",
                "vector header must be a single-line JSON object",
            ));
        }
        let header: VectorHeader = decode(line, false).map_err(vector_header_error)?;
        let tail = remainder.get(1..).ok_or_else(|| {
            Failure::new(
                "vector_envelope_header_missing",
                "vector header has no LF separator",
            )
        })?;
        if tail.len() != vector_tail_length(&header)? {
            return Err(Failure::new(
                "vector_envelope_tail_invalid",
                "vector tail must contain exactly four bytes per dimension",
            ));
        }
        let vector = tail
            .as_chunks::<4>()
            .0
            .iter()
            .map(|chunk| finite(f32::from_le_bytes(*chunk)))
            .collect::<Result<Vec<_>, _>>()?;
        Ok(Self { header, vector })
    }

    /// Encode the vector header, separator and binary tail.
    ///
    /// # Errors
    /// Refuses header serialization failure or invalid provenance and dimensions.
    pub fn encode(&self) -> Result<Vec<u8>, Failure> {
        let mut bytes = encode(&self.header)?;
        bytes.push(b'\n');
        bytes.reserve(vector_tail_length(&self.header)?);
        for member in &self.vector {
            bytes.extend_from_slice(&member.to_le_bytes());
        }
        Ok(bytes)
    }

    #[must_use]
    pub fn vector(&self) -> &[f32] {
        &self.vector
    }

    #[must_use]
    pub const fn dimension(&self) -> usize {
        self.header.dimensions
    }

    #[must_use]
    pub fn written_in(&self) -> &str {
        &self.header.written_in
    }
}

impl Attributed for Vector {
    fn written_at(&self) -> Version {
        self.header.written_at
    }

    fn request_id(&self) -> &str {
        &self.header.request_id
    }
}

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