lodestone_wire/
document_mutation_semantic_digest.rs

1use crate::{
2    DOCUMENT_MUTATION_SEMANTIC_DIGEST_DOMAIN, DOCUMENT_MUTATION_SEMANTIC_DIGEST_FORMAT_VERSION, Digest,
3    DocumentMutationCondition,
4};
5
6const MAXIMUM_DOCUMENT_ID_BYTES: usize = 512;
7const MAXIMUM_SOURCE_BYTES: usize = 2 * 1024 * 1024;
8const MAXIMUM_ROUTING_BYTES: usize = 1_024;
9
10#[derive(Clone, Copy, Debug, Eq, PartialEq)]
11pub enum DocumentMutationSemanticDigestError {
12    EmptyDocumentId,
13    DocumentIdTooLong(usize),
14    InvalidOperation,
15    SourceRequired,
16    SourceForbidden,
17    SourceTooLarge(usize),
18    InvalidSourceJson,
19    RoutingTooLong(usize),
20    InvalidSequenceNumber,
21    InvalidPrimaryTerm,
22}
23
24impl std::fmt::Display for DocumentMutationSemanticDigestError {
25    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        write!(
27            formatter,
28            "invalid Lodestone document mutation semantic digest input: {self:?}"
29        )
30    }
31}
32
33impl std::error::Error for DocumentMutationSemanticDigestError {}
34
35/// Derives the exact semantic digest Core validates for one raw-source
36/// document mutation or document-bulk item.
37///
38/// `source` is the exact UTF-8 JSON byte sequence used in the request. Put and
39/// update require it; delete forbids it. Mapping-owned native-record
40/// projection and ingest-pipeline selection require authoritative server
41/// preparation and are intentionally outside this pure helper.
42pub fn document_mutation_semantic_digest(
43    document_id: &str,
44    operation: &str,
45    source: Option<&[u8]>,
46    routing: Option<&str>,
47    condition: Option<&DocumentMutationCondition>,
48) -> Result<Digest, DocumentMutationSemanticDigestError> {
49    let document_id = document_id.as_bytes();
50    if document_id.is_empty() {
51        return Err(DocumentMutationSemanticDigestError::EmptyDocumentId);
52    }
53    if document_id.len() > MAXIMUM_DOCUMENT_ID_BYTES {
54        return Err(DocumentMutationSemanticDigestError::DocumentIdTooLong(
55            document_id.len(),
56        ));
57    }
58    if let Some(routing) = routing
59        && routing.len() > MAXIMUM_ROUTING_BYTES
60    {
61        return Err(DocumentMutationSemanticDigestError::RoutingTooLong(routing.len()));
62    }
63
64    let (operation_tag, source) = match (operation, source) {
65        ("put", Some(source)) => (1_u8, source),
66        ("update", Some(source)) => (2_u8, source),
67        ("delete", None) => (3_u8, &[][..]),
68        ("put" | "update", None) => return Err(DocumentMutationSemanticDigestError::SourceRequired),
69        ("delete", Some(_)) => return Err(DocumentMutationSemanticDigestError::SourceForbidden),
70        _ => return Err(DocumentMutationSemanticDigestError::InvalidOperation),
71    };
72    if source.len() > MAXIMUM_SOURCE_BYTES {
73        return Err(DocumentMutationSemanticDigestError::SourceTooLarge(source.len()));
74    }
75    if operation_tag != 3 && serde_json::from_slice::<&serde_json::value::RawValue>(source).is_err() {
76        return Err(DocumentMutationSemanticDigestError::InvalidSourceJson);
77    }
78
79    let condition = canonical_condition(condition)?;
80    let mut hasher = blake3::Hasher::new();
81    hasher.update(DOCUMENT_MUTATION_SEMANTIC_DIGEST_DOMAIN);
82    hasher.update(&DOCUMENT_MUTATION_SEMANTIC_DIGEST_FORMAT_VERSION.to_be_bytes());
83    hasher.update(&[operation_tag]);
84    update_bounded(&mut hasher, document_id);
85    match routing {
86        Some(value) => {
87            hasher.update(&[1]);
88            update_bounded(&mut hasher, value.as_bytes());
89        }
90        None => {
91            hasher.update(&[0]);
92        }
93    }
94    update_bounded(&mut hasher, source);
95    update_bounded(&mut hasher, condition.as_bytes());
96    Ok(Digest {
97        value: hasher.finalize().to_hex().to_string(),
98    })
99}
100
101fn canonical_condition(
102    condition: Option<&DocumentMutationCondition>,
103) -> Result<String, DocumentMutationSemanticDigestError> {
104    match condition {
105        None | Some(DocumentMutationCondition::Any) => Ok("any".to_owned()),
106        Some(DocumentMutationCondition::MustNotExist) => Ok("must_not_exist".to_owned()),
107        Some(DocumentMutationCondition::IfSeqNo { seq_no }) => {
108            seq_no
109                .validate()
110                .map_err(|_| DocumentMutationSemanticDigestError::InvalidSequenceNumber)?;
111            Ok(format!("if_seq_no={}", seq_no.value))
112        }
113        Some(DocumentMutationCondition::IfSeqNoAndPrimaryTerm { seq_no, primary_term }) => {
114            seq_no
115                .validate()
116                .map_err(|_| DocumentMutationSemanticDigestError::InvalidSequenceNumber)?;
117            primary_term
118                .validate()
119                .map_err(|_| DocumentMutationSemanticDigestError::InvalidPrimaryTerm)?;
120            if primary_term.value == "0" {
121                return Err(DocumentMutationSemanticDigestError::InvalidPrimaryTerm);
122            }
123            Ok(format!(
124                "if_seq_no={};if_primary_term={}",
125                seq_no.value, primary_term.value
126            ))
127        }
128    }
129}
130
131fn update_bounded(hasher: &mut blake3::Hasher, value: &[u8]) {
132    hasher.update(&(value.len() as u64).to_be_bytes());
133    hasher.update(value);
134}
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139    use crate::U64;
140
141    #[test]
142    fn digest_preserves_exact_source_lexemes_and_condition() -> Result<(), Box<dyn std::error::Error>> {
143        let condition = DocumentMutationCondition::IfSeqNoAndPrimaryTerm {
144            seq_no: U64 { value: "9".to_owned() },
145            primary_term: U64 { value: "3".to_owned() },
146        };
147        let first = document_mutation_semantic_digest(
148            "document-1",
149            "update",
150            Some(br#"{"n":9007199254740993,"d":1.2300}"#),
151            Some("tenant-a"),
152            Some(&condition),
153        )?;
154        let second = document_mutation_semantic_digest(
155            "document-1",
156            "update",
157            Some(br#"{"n":9007199254740993,"d":1.23}"#),
158            Some("tenant-a"),
159            Some(&condition),
160        )?;
161        assert_ne!(first, second);
162        Ok(())
163    }
164
165    #[test]
166    fn digest_rejects_invalid_source_operation_and_primary_term() {
167        assert_eq!(
168            document_mutation_semantic_digest("document-1", "put", None, None, None),
169            Err(DocumentMutationSemanticDigestError::SourceRequired)
170        );
171        assert_eq!(
172            document_mutation_semantic_digest("document-1", "delete", Some(b"{}"), None, None),
173            Err(DocumentMutationSemanticDigestError::SourceForbidden)
174        );
175        assert_eq!(
176            document_mutation_semantic_digest("document-1", "put", Some(b"not-json"), None, None),
177            Err(DocumentMutationSemanticDigestError::InvalidSourceJson)
178        );
179        assert_eq!(
180            document_mutation_semantic_digest(
181                "document-1",
182                "update",
183                Some(b"{}"),
184                None,
185                Some(&DocumentMutationCondition::IfSeqNoAndPrimaryTerm {
186                    seq_no: U64 { value: "0".to_owned() },
187                    primary_term: U64 { value: "0".to_owned() },
188                }),
189            ),
190            Err(DocumentMutationSemanticDigestError::InvalidPrimaryTerm)
191        );
192    }
193}
lodestone_wire/document_mutation_semantic_digest.rs

lodestone_wire/
document_mutation_semantic_digest.rs
