lodestone-wire 0.0.31

use crate::{
    DOCUMENT_MUTATION_SEMANTIC_DIGEST_DOMAIN, DOCUMENT_MUTATION_SEMANTIC_DIGEST_FORMAT_VERSION, Digest,
    DocumentMutationCondition,
};

const MAXIMUM_DOCUMENT_ID_BYTES: usize = 512;
const MAXIMUM_SOURCE_BYTES: usize = 2 * 1024 * 1024;
const MAXIMUM_ROUTING_BYTES: usize = 1_024;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DocumentMutationSemanticDigestError {
    EmptyDocumentId,
    DocumentIdTooLong(usize),
    InvalidOperation,
    SourceRequired,
    SourceForbidden,
    SourceTooLarge(usize),
    InvalidSourceJson,
    RoutingTooLong(usize),
    InvalidSequenceNumber,
    InvalidPrimaryTerm,
}

impl std::fmt::Display for DocumentMutationSemanticDigestError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            formatter,
            "invalid Lodestone document mutation semantic digest input: {self:?}"
        )
    }
}

impl std::error::Error for DocumentMutationSemanticDigestError {}

/// Derives the exact semantic digest Core validates for one raw-source
/// document mutation or document-bulk item.
///
/// `source` is the exact UTF-8 JSON byte sequence used in the request. Put and
/// update require it; delete forbids it. Mapping-owned native-record
/// projection and ingest-pipeline selection require authoritative server
/// preparation and are intentionally outside this pure helper.
pub fn document_mutation_semantic_digest(
    document_id: &str,
    operation: &str,
    source: Option<&[u8]>,
    routing: Option<&str>,
    condition: Option<&DocumentMutationCondition>,
) -> Result<Digest, DocumentMutationSemanticDigestError> {
    let document_id = document_id.as_bytes();
    if document_id.is_empty() {
        return Err(DocumentMutationSemanticDigestError::EmptyDocumentId);
    }
    if document_id.len() > MAXIMUM_DOCUMENT_ID_BYTES {
        return Err(DocumentMutationSemanticDigestError::DocumentIdTooLong(
            document_id.len(),
        ));
    }
    if let Some(routing) = routing
        && routing.len() > MAXIMUM_ROUTING_BYTES
    {
        return Err(DocumentMutationSemanticDigestError::RoutingTooLong(routing.len()));
    }

    let (operation_tag, source) = match (operation, source) {
        ("put", Some(source)) => (1_u8, source),
        ("update", Some(source)) => (2_u8, source),
        ("delete", None) => (3_u8, &[][..]),
        ("put" | "update", None) => return Err(DocumentMutationSemanticDigestError::SourceRequired),
        ("delete", Some(_)) => return Err(DocumentMutationSemanticDigestError::SourceForbidden),
        _ => return Err(DocumentMutationSemanticDigestError::InvalidOperation),
    };
    if source.len() > MAXIMUM_SOURCE_BYTES {
        return Err(DocumentMutationSemanticDigestError::SourceTooLarge(source.len()));
    }
    if operation_tag != 3 && serde_json::from_slice::<&serde_json::value::RawValue>(source).is_err() {
        return Err(DocumentMutationSemanticDigestError::InvalidSourceJson);
    }

    let condition = canonical_condition(condition)?;
    let mut hasher = blake3::Hasher::new();
    hasher.update(DOCUMENT_MUTATION_SEMANTIC_DIGEST_DOMAIN);
    hasher.update(&DOCUMENT_MUTATION_SEMANTIC_DIGEST_FORMAT_VERSION.to_be_bytes());
    hasher.update(&[operation_tag]);
    update_bounded(&mut hasher, document_id);
    match routing {
        Some(value) => {
            hasher.update(&[1]);
            update_bounded(&mut hasher, value.as_bytes());
        }
        None => {
            hasher.update(&[0]);
        }
    }
    update_bounded(&mut hasher, source);
    update_bounded(&mut hasher, condition.as_bytes());
    Ok(Digest {
        value: hasher.finalize().to_hex().to_string(),
    })
}

fn canonical_condition(
    condition: Option<&DocumentMutationCondition>,
) -> Result<String, DocumentMutationSemanticDigestError> {
    match condition {
        None | Some(DocumentMutationCondition::Any) => Ok("any".to_owned()),
        Some(DocumentMutationCondition::MustNotExist) => Ok("must_not_exist".to_owned()),
        Some(DocumentMutationCondition::IfSeqNo { seq_no }) => {
            seq_no
                .validate()
                .map_err(|_| DocumentMutationSemanticDigestError::InvalidSequenceNumber)?;
            Ok(format!("if_seq_no={}", seq_no.value))
        }
        Some(DocumentMutationCondition::IfSeqNoAndPrimaryTerm { seq_no, primary_term }) => {
            seq_no
                .validate()
                .map_err(|_| DocumentMutationSemanticDigestError::InvalidSequenceNumber)?;
            primary_term
                .validate()
                .map_err(|_| DocumentMutationSemanticDigestError::InvalidPrimaryTerm)?;
            if primary_term.value == "0" {
                return Err(DocumentMutationSemanticDigestError::InvalidPrimaryTerm);
            }
            Ok(format!(
                "if_seq_no={};if_primary_term={}",
                seq_no.value, primary_term.value
            ))
        }
    }
}

fn update_bounded(hasher: &mut blake3::Hasher, value: &[u8]) {
    hasher.update(&(value.len() as u64).to_be_bytes());
    hasher.update(value);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::U64;

    #[test]
    fn digest_preserves_exact_source_lexemes_and_condition() -> Result<(), Box<dyn std::error::Error>> {
        let condition = DocumentMutationCondition::IfSeqNoAndPrimaryTerm {
            seq_no: U64 { value: "9".to_owned() },
            primary_term: U64 { value: "3".to_owned() },
        };
        let first = document_mutation_semantic_digest(
            "document-1",
            "update",
            Some(br#"{"n":9007199254740993,"d":1.2300}"#),
            Some("tenant-a"),
            Some(&condition),
        )?;
        let second = document_mutation_semantic_digest(
            "document-1",
            "update",
            Some(br#"{"n":9007199254740993,"d":1.23}"#),
            Some("tenant-a"),
            Some(&condition),
        )?;
        assert_ne!(first, second);
        Ok(())
    }

    #[test]
    fn digest_rejects_invalid_source_operation_and_primary_term() {
        assert_eq!(
            document_mutation_semantic_digest("document-1", "put", None, None, None),
            Err(DocumentMutationSemanticDigestError::SourceRequired)
        );
        assert_eq!(
            document_mutation_semantic_digest("document-1", "delete", Some(b"{}"), None, None),
            Err(DocumentMutationSemanticDigestError::SourceForbidden)
        );
        assert_eq!(
            document_mutation_semantic_digest("document-1", "put", Some(b"not-json"), None, None),
            Err(DocumentMutationSemanticDigestError::InvalidSourceJson)
        );
        assert_eq!(
            document_mutation_semantic_digest(
                "document-1",
                "update",
                Some(b"{}"),
                None,
                Some(&DocumentMutationCondition::IfSeqNoAndPrimaryTerm {
                    seq_no: U64 { value: "0".to_owned() },
                    primary_term: U64 { value: "0".to_owned() },
                }),
            ),
            Err(DocumentMutationSemanticDigestError::InvalidPrimaryTerm)
        );
    }
}