lodestone_wire/
bulk_batch_id.rs

1use std::collections::BTreeMap;
2
3use crate::{BULK_BATCH_ID_DOMAIN, BULK_BATCH_ID_FORMAT_VERSION, Digest, Identity, Resource};
4const MAXIMUM_ITEM_ACTION_IDS: usize = 1_024;
5
6#[derive(Clone, Debug, Eq, PartialEq)]
7pub enum BulkBatchIdError {
8    InvalidCollectionIdentity,
9    ZeroCollectionIdentity,
10    InvalidIndexKind,
11    InvalidIndexName,
12    InvalidIndexIncarnation,
13    ZeroIndexIncarnation,
14    TooManyItemActionIds,
15    InvalidItemActionId { position: usize },
16    ZeroItemActionId { position: usize },
17    DuplicateItemActionId { first: usize, duplicate: usize },
18}
19
20impl std::fmt::Display for BulkBatchIdError {
21    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
22        write!(formatter, "invalid Lodestone API bulk batch identity input: {self:?}")
23    }
24}
25
26impl std::error::Error for BulkBatchIdError {}
27
28/// Derives the sole current Lodestone API batch identity used by document bulk
29/// and retained-append bulk requests.
30///
31/// The BLAKE3-256 input is the versioned domain followed by this canonical
32/// binary payload, in order: format version as `u16` big-endian, collection as raw
33/// `u128` big-endian, index-name byte length as `u16` big-endian, exact UTF-8
34/// index-name bytes, index incarnation as raw `u128` big-endian, item count as
35/// `u32` big-endian, then each ordered action identity as 32 raw bytes.
36pub fn bulk_batch_id(
37    collection: &Identity,
38    index: &Resource,
39    item_action_ids: &[Digest],
40) -> Result<Digest, BulkBatchIdError> {
41    let payload = canonical_bulk_batch_payload_v1(collection, index, item_action_ids)?;
42    let mut hasher = blake3::Hasher::new();
43    hasher.update(BULK_BATCH_ID_DOMAIN);
44    hasher.update(&payload);
45    Ok(Digest {
46        value: hasher.finalize().to_hex().to_string(),
47    })
48}
49
50fn canonical_bulk_batch_payload_v1(
51    collection: &Identity,
52    index: &Resource,
53    item_action_ids: &[Digest],
54) -> Result<Vec<u8>, BulkBatchIdError> {
55    let collection = decode_identity(&collection.value).ok_or(BulkBatchIdError::InvalidCollectionIdentity)?;
56    if collection == 0 {
57        return Err(BulkBatchIdError::ZeroCollectionIdentity);
58    }
59    if index.kind != "index" {
60        return Err(BulkBatchIdError::InvalidIndexKind);
61    }
62    let index_name = index.name.as_bytes();
63    if !(1..=255).contains(&index_name.len()) || index_name.iter().any(|byte| *byte <= 0x1f || *byte == 0x7f) {
64        return Err(BulkBatchIdError::InvalidIndexName);
65    }
66    let incarnation = decode_identity(&index.incarnation.value).ok_or(BulkBatchIdError::InvalidIndexIncarnation)?;
67    if incarnation == 0 {
68        return Err(BulkBatchIdError::ZeroIndexIncarnation);
69    }
70    if item_action_ids.len() > MAXIMUM_ITEM_ACTION_IDS {
71        return Err(BulkBatchIdError::TooManyItemActionIds);
72    }
73
74    let name_length = u16::try_from(index_name.len()).map_err(|_| BulkBatchIdError::InvalidIndexName)?;
75    let item_count = u32::try_from(item_action_ids.len()).map_err(|_| BulkBatchIdError::TooManyItemActionIds)?;
76    let mut payload = Vec::with_capacity(40 + index_name.len() + item_action_ids.len() * 32);
77    payload.extend_from_slice(&BULK_BATCH_ID_FORMAT_VERSION.to_be_bytes());
78    payload.extend_from_slice(&collection.to_be_bytes());
79    payload.extend_from_slice(&name_length.to_be_bytes());
80    payload.extend_from_slice(index_name);
81    payload.extend_from_slice(&incarnation.to_be_bytes());
82    payload.extend_from_slice(&item_count.to_be_bytes());
83
84    let mut positions = BTreeMap::new();
85    for (position, action) in item_action_ids.iter().enumerate() {
86        let bytes = decode_digest(&action.value).ok_or(BulkBatchIdError::InvalidItemActionId { position })?;
87        if bytes == [0; 32] {
88            return Err(BulkBatchIdError::ZeroItemActionId { position });
89        }
90        if let Some(first) = positions.insert(bytes, position) {
91            return Err(BulkBatchIdError::DuplicateItemActionId {
92                first,
93                duplicate: position,
94            });
95        }
96        payload.extend_from_slice(&bytes);
97    }
98    Ok(payload)
99}
100
101fn decode_identity(value: &str) -> Option<u128> {
102    if value.len() != 32 || !value.bytes().all(is_lower_hex) {
103        return None;
104    }
105    u128::from_str_radix(value, 16).ok()
106}
107
108fn decode_digest(value: &str) -> Option<[u8; 32]> {
109    if value.len() != 64 || !value.bytes().all(is_lower_hex) {
110        return None;
111    }
112    let mut decoded = [0_u8; 32];
113    let (pairs, []) = value.as_bytes().as_chunks::<2>() else {
114        return None;
115    };
116    for (output, pair) in decoded.iter_mut().zip(pairs) {
117        *output = (lower_hex_value(pair[0])? << 4) | lower_hex_value(pair[1])?;
118    }
119    Some(decoded)
120}
121
122const fn is_lower_hex(value: u8) -> bool {
123    matches!(value, b'0'..=b'9' | b'a'..=b'f')
124}
125
126const fn lower_hex_value(value: u8) -> Option<u8> {
127    match value {
128        b'0'..=b'9' => Some(value - b'0'),
129        b'a'..=b'f' => Some(value - b'a' + 10),
130        _ => None,
131    }
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137
138    const VECTORS: &str = include_str!("../../../clients/native/fixtures/lodestone-bulk-batch-id.json");
139
140    fn identity(value: u128) -> Identity {
141        Identity {
142            value: format!("{value:032x}"),
143        }
144    }
145
146    fn digest(byte: u8) -> Digest {
147        Digest {
148            value: format!("{byte:02x}").repeat(32),
149        }
150    }
151
152    fn index(name: &str, incarnation: u128) -> Resource {
153        Resource {
154            kind: "index".to_owned(),
155            name: name.to_owned(),
156            incarnation: identity(incarnation),
157        }
158    }
159
160    #[test]
161    fn bulk_batch_identity_is_deterministic_ordered_and_scope_bound() -> Result<(), Box<dyn std::error::Error>> {
162        let collection = identity(1);
163        let resource = index("audit-events", 2);
164        let actions = [digest(0x11), digest(0x22)];
165        let first = bulk_batch_id(&collection, &resource, &actions)?;
166        assert_eq!(
167            first.value,
168            "25b1694f4d2883cc7c475d51ef6e9fc6ce1d42ef09a9283db30a4acd7ebd0f06"
169        );
170        let escaped = bulk_batch_id(
171            &Identity {
172                value: "0123456789abcdef0123456789abcdef".to_owned(),
173            },
174            &Resource {
175                kind: "index".to_owned(),
176                name: "quote\"\\雪".to_owned(),
177                incarnation: Identity {
178                    value: "fedcba9876543210fedcba9876543210".to_owned(),
179                },
180            },
181            &[
182                Digest {
183                    value: format!("{}01", "00".repeat(31)),
184                },
185                digest(0xff),
186            ],
187        )?;
188        assert_eq!(
189            escaped.value,
190            "619b49a49aa5a91437679522dbc272e3344118eea59f6ccd1fda7655b78d5f29"
191        );
192        assert_eq!(bulk_batch_id(&collection, &resource, &actions)?, first);
193        assert_ne!(
194            bulk_batch_id(&collection, &resource, &[actions[1].clone(), actions[0].clone()])?,
195            first
196        );
197        assert_ne!(bulk_batch_id(&identity(3), &resource, &actions)?, first);
198        assert_ne!(
199            bulk_batch_id(&collection, &index("audit-events-2", 2), &actions)?,
200            first
201        );
202        assert_ne!(bulk_batch_id(&collection, &index("audit-events", 3), &actions)?, first);
203        Ok(())
204    }
205
206    #[test]
207    fn bulk_batch_identity_rejects_duplicate_malformed_and_zero_actions() {
208        let collection = identity(1);
209        let resource = index("audit-events", 2);
210        assert_eq!(
211            bulk_batch_id(&collection, &resource, &[digest(0x11), digest(0x11)]),
212            Err(BulkBatchIdError::DuplicateItemActionId { first: 0, duplicate: 1 })
213        );
214        assert_eq!(
215            bulk_batch_id(&collection, &resource, &[Digest { value: "GG".repeat(32) }]),
216            Err(BulkBatchIdError::InvalidItemActionId { position: 0 })
217        );
218        assert_eq!(
219            bulk_batch_id(&collection, &resource, &[digest(0)]),
220            Err(BulkBatchIdError::ZeroItemActionId { position: 0 })
221        );
222    }
223
224    #[test]
225    fn bulk_batch_identity_rejects_invalid_scope_and_over_bound_inputs() {
226        let resource = index("audit-events", 2);
227        assert_eq!(
228            bulk_batch_id(&Identity { value: "01".to_owned() }, &resource, &[digest(0x11)],),
229            Err(BulkBatchIdError::InvalidCollectionIdentity)
230        );
231        assert_eq!(
232            bulk_batch_id(&identity(0), &resource, &[digest(0x11)]),
233            Err(BulkBatchIdError::ZeroCollectionIdentity)
234        );
235        assert_eq!(
236            bulk_batch_id(
237                &identity(1),
238                &Resource {
239                    kind: "alias".to_owned(),
240                    name: "audit-events".to_owned(),
241                    incarnation: identity(2),
242                },
243                &[digest(0x11)],
244            ),
245            Err(BulkBatchIdError::InvalidIndexKind)
246        );
247        assert_eq!(
248            bulk_batch_id(&identity(1), &index("", 2), &[digest(0x11)]),
249            Err(BulkBatchIdError::InvalidIndexName)
250        );
251        assert_eq!(
252            bulk_batch_id(&identity(1), &index("audit-events", 0), &[digest(0x11)]),
253            Err(BulkBatchIdError::ZeroIndexIncarnation)
254        );
255        assert_eq!(
256            bulk_batch_id(&identity(1), &resource, &vec![digest(0x11); 1_025]),
257            Err(BulkBatchIdError::TooManyItemActionIds)
258        );
259    }
260
261    #[test]
262    fn fixed_language_neutral_vectors_match_payload_and_digest() -> Result<(), Box<dyn std::error::Error>> {
263        let document: serde_json::Value = serde_json::from_str(VECTORS)?;
264        assert_eq!(document["format_version"].as_u64(), Some(1));
265        assert_eq!(
266            document["batch_id_format_version"].as_u64(),
267            Some(u64::from(BULK_BATCH_ID_FORMAT_VERSION))
268        );
269        assert_eq!(document["algorithm"].as_str(), Some("blake3-256"));
270        assert_eq!(
271            document["domain_separator_utf8"].as_str().map(str::as_bytes),
272            Some(BULK_BATCH_ID_DOMAIN)
273        );
274        let vectors = document["vectors"].as_array().ok_or("bulk batch vectors are absent")?;
275        for vector in vectors {
276            let collection = Identity {
277                value: vector["collection_identity"]
278                    .as_str()
279                    .ok_or("bulk batch vector collection is absent")?
280                    .to_owned(),
281            };
282            let index: Resource = serde_json::from_value(vector["index"].clone())?;
283            let action_values = vector["item_action_ids"]
284                .as_array()
285                .ok_or("bulk batch vector actions are absent")?;
286            let actions = action_values
287                .iter()
288                .map(|value| {
289                    value
290                        .as_str()
291                        .map(|value| Digest {
292                            value: value.to_owned(),
293                        })
294                        .ok_or("bulk batch vector action is invalid")
295                })
296                .collect::<Result<Vec<_>, _>>()?;
297            let expected_payload = decode_test_hex(
298                vector["canonical_payload_hex"]
299                    .as_str()
300                    .ok_or("bulk batch vector payload is absent")?,
301            )?;
302            assert_eq!(
303                canonical_bulk_batch_payload_v1(&collection, &index, &actions)?,
304                expected_payload
305            );
306            assert_eq!(
307                bulk_batch_id(&collection, &index, &actions)?.value,
308                vector["batch_id"]
309                    .as_str()
310                    .ok_or("bulk batch vector result is absent")?
311            );
312        }
313        Ok(())
314    }
315
316    fn decode_test_hex(value: &str) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
317        if !value.len().is_multiple_of(2) {
318            return Err("bulk batch vector payload has odd hex width".into());
319        }
320        value
321            .as_bytes()
322            .as_chunks::<2>()
323            .0
324            .iter()
325            .map(|pair| Ok(u8::from_str_radix(std::str::from_utf8(pair)?, 16)?))
326            .collect()
327    }
328}
lodestone_wire/bulk_batch_id.rs

lodestone_wire/
bulk_batch_id.rs
