reddb_engine/storage/
metadata_codec.rs

1//! Canonical native metadata bytes and entity-record adapter, shared with WASM.
2#[path = "metadata_codec/budget.rs"]
3mod budget;
4use super::entity::{EntityId, UnifiedEntity};
5use super::metadata::{Metadata, MetadataValue};
6
7fn read_entity(data: &[u8], version: u32) -> Result<UnifiedEntity, String> {
8    super::entity_codec::read_entity_binary(data, &mut 0, version).map_err(invalid)
9}
10
11fn invalid(error: impl std::fmt::Display) -> String {
12    error.to_string()
13}
14
15fn read_count(data: &[u8], position: &mut usize, bytes_min: usize) -> Result<usize, String> {
16    let count = usize::try_from(read_u32(data, position)?).map_err(invalid)?;
17    if count > data.len().saturating_sub(*position) / bytes_min {
18        return Err(invalid("metadata count exceeds the remaining payload"));
19    }
20    Ok(count)
21}
22
23/// Preserve native entity/value and optional typed metadata framing.
24pub fn serialize_entity_record(
25    entity: &UnifiedEntity,
26    metadata: Option<&Metadata>,
27    version: u32,
28) -> Vec<u8> {
29    let mut bytes = Vec::with_capacity(256);
30    super::entity_codec::write_entity_binary(&mut bytes, entity, version);
31    let entity = bytes;
32    if metadata.is_some_and(|metadata| !metadata.fields.is_empty()) {
33        reddb_file::encode_native_entity_record_frame(&entity, Some(&serialize_metadata(metadata)))
34    } else {
35        reddb_file::encode_native_entity_record_frame(&entity, None)
36    }
37}
38
39/// Decode the existing framed record, retaining the pre-V7 entity-only fallback.
40pub fn deserialize_entity_record(
41    data: &[u8],
42    version: u32,
43) -> Result<(UnifiedEntity, Option<Metadata>), String> {
44    let Some(frame) = reddb_file::decode_native_entity_record_frame(data).map_err(invalid)? else {
45        return read_entity(data, version).map(|entity| (entity, None));
46    };
47    let entity = read_entity(frame.entity, version)?;
48    let metadata = if frame.metadata.is_empty() {
49        None
50    } else {
51        let metadata = deserialize_metadata(frame.metadata)?;
52        (!metadata.is_empty()).then_some(metadata)
53    };
54    Ok((entity, metadata))
55}
56
57pub fn serialize_metadata(metadata: Option<&Metadata>) -> Vec<u8> {
58    let Some(metadata) = metadata else {
59        return Vec::new();
60    };
61    if metadata.is_empty() {
62        return Vec::new();
63    }
64
65    let mut entries: Vec<_> = metadata.iter().collect();
66    entries.sort_by_key(|(a, _)| *a);
67
68    let mut buf = Vec::new();
69    buf.extend_from_slice(&(entries.len() as u32).to_le_bytes());
70    for (key, value) in entries {
71        write_string(&mut buf, key);
72        write_metadata_value(&mut buf, value);
73    }
74    buf
75}
76
77pub fn deserialize_metadata(data: &[u8]) -> Result<Metadata, String> {
78    let mut pos = 0usize;
79    let count = read_count(data, &mut pos, 5)?;
80    let mut metadata = Metadata::new();
81    for _ in 0..count {
82        let key = read_string(data, &mut pos)?;
83        let value = read_metadata_value(data, &mut pos, 0)?;
84        metadata.set(key, value);
85    }
86    Ok(metadata)
87}
88
89fn write_string(buf: &mut Vec<u8>, value: &str) {
90    reddb_file::encode_native_len_prefixed_str(buf, value);
91}
92
93fn write_collection_name(buf: &mut Vec<u8>, value: &str) {
94    write_string(
95        buf,
96        super::system_collection_names::canonical_or_self(value),
97    );
98}
99
100fn write_bytes(buf: &mut Vec<u8>, value: &[u8]) {
101    reddb_file::encode_native_len_prefixed_bytes(buf, value);
102}
103
104fn write_ref_target(buf: &mut Vec<u8>, target: &super::metadata::RefTarget) {
105    use super::metadata::RefTarget;
106
107    match target {
108        RefTarget::TableRow { table, row_id } => {
109            buf.push(0);
110            write_collection_name(buf, table);
111            buf.extend_from_slice(&row_id.to_le_bytes());
112        }
113        RefTarget::Node {
114            collection,
115            node_id,
116        } => {
117            buf.push(1);
118            write_collection_name(buf, collection);
119            buf.extend_from_slice(&node_id.raw().to_le_bytes());
120        }
121        RefTarget::Edge {
122            collection,
123            edge_id,
124        } => {
125            buf.push(2);
126            write_collection_name(buf, collection);
127            buf.extend_from_slice(&edge_id.raw().to_le_bytes());
128        }
129        RefTarget::Vector {
130            collection,
131            vector_id,
132        } => {
133            buf.push(3);
134            write_collection_name(buf, collection);
135            buf.extend_from_slice(&vector_id.raw().to_le_bytes());
136        }
137        RefTarget::Entity {
138            collection,
139            entity_id,
140        } => {
141            buf.push(4);
142            write_collection_name(buf, collection);
143            buf.extend_from_slice(&entity_id.raw().to_le_bytes());
144        }
145    }
146}
147
148fn write_metadata_value(buf: &mut Vec<u8>, value: &MetadataValue) {
149    match value {
150        MetadataValue::Null => buf.push(0),
151        MetadataValue::Bool(v) => {
152            buf.push(1);
153            buf.push(u8::from(*v));
154        }
155        MetadataValue::Int(v) => {
156            buf.push(2);
157            buf.extend_from_slice(&v.to_le_bytes());
158        }
159        MetadataValue::Float(v) => {
160            buf.push(3);
161            buf.extend_from_slice(&v.to_le_bytes());
162        }
163        MetadataValue::String(v) => {
164            buf.push(4);
165            write_string(buf, v);
166        }
167        MetadataValue::Bytes(v) => {
168            buf.push(5);
169            write_bytes(buf, v);
170        }
171        MetadataValue::Array(values) => {
172            buf.push(6);
173            buf.extend_from_slice(&(values.len() as u32).to_le_bytes());
174            for value in values {
175                write_metadata_value(buf, value);
176            }
177        }
178        MetadataValue::Object(values) => {
179            buf.push(7);
180            let mut entries: Vec<_> = values.iter().collect();
181            entries.sort_by_key(|(a, _)| *a);
182            buf.extend_from_slice(&(entries.len() as u32).to_le_bytes());
183            for (key, value) in entries {
184                write_string(buf, key);
185                write_metadata_value(buf, value);
186            }
187        }
188        MetadataValue::Timestamp(v) => {
189            buf.push(8);
190            buf.extend_from_slice(&v.to_le_bytes());
191        }
192        MetadataValue::Geo { lat, lon } => {
193            buf.push(9);
194            buf.extend_from_slice(&lat.to_le_bytes());
195            buf.extend_from_slice(&lon.to_le_bytes());
196        }
197        MetadataValue::Reference(target) => {
198            buf.push(10);
199            write_ref_target(buf, target);
200        }
201        MetadataValue::References(targets) => {
202            buf.push(11);
203            buf.extend_from_slice(&(targets.len() as u32).to_le_bytes());
204            for target in targets {
205                write_ref_target(buf, target);
206            }
207        }
208    }
209}
210
211fn read_exact_slice<'a>(data: &'a [u8], pos: &mut usize, len: usize) -> Result<&'a [u8], String> {
212    if *pos > data.len() || len > data.len() - *pos {
213        return Err(invalid("truncated metadata payload".to_string()));
214    }
215    let slice = &data[*pos..*pos + len];
216    *pos += len;
217    Ok(slice)
218}
219
220fn read_u32(data: &[u8], pos: &mut usize) -> Result<u32, String> {
221    let bytes = read_exact_slice(data, pos, 4)?;
222    let mut raw = [0u8; 4];
223    raw.copy_from_slice(bytes);
224    Ok(u32::from_le_bytes(raw))
225}
226
227fn read_u64(data: &[u8], pos: &mut usize) -> Result<u64, String> {
228    let bytes = read_exact_slice(data, pos, 8)?;
229    let mut raw = [0u8; 8];
230    raw.copy_from_slice(bytes);
231    Ok(u64::from_le_bytes(raw))
232}
233
234fn read_i64(data: &[u8], pos: &mut usize) -> Result<i64, String> {
235    let bytes = read_exact_slice(data, pos, 8)?;
236    let mut raw = [0u8; 8];
237    raw.copy_from_slice(bytes);
238    Ok(i64::from_le_bytes(raw))
239}
240
241fn read_f64(data: &[u8], pos: &mut usize) -> Result<f64, String> {
242    let bytes = read_exact_slice(data, pos, 8)?;
243    let mut raw = [0u8; 8];
244    raw.copy_from_slice(bytes);
245    Ok(f64::from_le_bytes(raw))
246}
247
248fn read_u8(data: &[u8], pos: &mut usize) -> Result<u8, String> {
249    let bytes = read_exact_slice(data, pos, 1)?;
250    Ok(bytes[0])
251}
252
253fn read_string(data: &[u8], pos: &mut usize) -> Result<String, String> {
254    reddb_file::decode_native_len_prefixed_string(data, pos).map_err(|err| invalid(err.to_string()))
255}
256
257fn read_bytes(data: &[u8], pos: &mut usize) -> Result<Vec<u8>, String> {
258    reddb_file::decode_native_len_prefixed_bytes(data, pos)
259        .map(|bytes| bytes.to_vec())
260        .map_err(|err| invalid(err.to_string()))
261}
262
263fn read_ref_target(data: &[u8], pos: &mut usize) -> Result<super::metadata::RefTarget, String> {
264    use super::metadata::RefTarget;
265
266    match read_u8(data, pos)? {
267        0 => Ok(RefTarget::TableRow {
268            table: super::system_collection_names::canonical_owned(read_string(data, pos)?),
269            row_id: read_u64(data, pos)?,
270        }),
271        1 => Ok(RefTarget::Node {
272            collection: super::system_collection_names::canonical_owned(read_string(data, pos)?),
273            node_id: EntityId::new(read_u64(data, pos)?),
274        }),
275        2 => Ok(RefTarget::Edge {
276            collection: super::system_collection_names::canonical_owned(read_string(data, pos)?),
277            edge_id: EntityId::new(read_u64(data, pos)?),
278        }),
279        3 => Ok(RefTarget::Vector {
280            collection: super::system_collection_names::canonical_owned(read_string(data, pos)?),
281            vector_id: EntityId::new(read_u64(data, pos)?),
282        }),
283        4 => Ok(RefTarget::Entity {
284            collection: super::system_collection_names::canonical_owned(read_string(data, pos)?),
285            entity_id: EntityId::new(read_u64(data, pos)?),
286        }),
287        tag => Err(invalid(format!("unknown metadata ref target tag {tag}"))),
288    }
289}
290
291fn read_metadata_value(data: &[u8], pos: &mut usize, depth: u32) -> Result<MetadataValue, String> {
292    if depth > reddb_rql::limits::JSON_LITERAL_MAX_DEPTH {
293        return Err(invalid("metadata nesting exceeds the maximum depth"));
294    }
295    match read_u8(data, pos)? {
296        0 => Ok(MetadataValue::Null),
297        1 => Ok(MetadataValue::Bool(read_u8(data, pos)? != 0)),
298        2 => Ok(MetadataValue::Int(read_i64(data, pos)?)),
299        3 => Ok(MetadataValue::Float(read_f64(data, pos)?)),
300        4 => Ok(MetadataValue::String(read_string(data, pos)?)),
301        5 => Ok(MetadataValue::Bytes(read_bytes(data, pos)?)),
302        6 => {
303            let count = read_count(data, pos, 1)?;
304            let mut values = Vec::with_capacity(count);
305            for _ in 0..count {
306                values.push(read_metadata_value(data, pos, depth + 1)?);
307            }
308            Ok(MetadataValue::Array(values))
309        }
310        7 => {
311            let count = read_count(data, pos, 5)?;
312            let mut values = std::collections::HashMap::with_capacity(count);
313            for _ in 0..count {
314                let key = read_string(data, pos)?;
315                let value = read_metadata_value(data, pos, depth + 1)?;
316                values.insert(key, value);
317            }
318            Ok(MetadataValue::Object(values))
319        }
320        8 => Ok(MetadataValue::Timestamp(read_u64(data, pos)?)),
321        9 => Ok(MetadataValue::Geo {
322            lat: read_f64(data, pos)?,
323            lon: read_f64(data, pos)?,
324        }),
325        10 => Ok(MetadataValue::Reference(read_ref_target(data, pos)?)),
326        11 => {
327            let count = read_count(data, pos, 13)?;
328            let mut targets = Vec::with_capacity(count);
329            for _ in 0..count {
330                targets.push(read_ref_target(data, pos)?);
331            }
332            Ok(MetadataValue::References(targets))
333        }
334        tag => Err(invalid(format!("unknown metadata value tag {tag}"))),
335    }
336}
337
338/// Admit native value expansion and metadata containers before allocation.
339pub fn deserialize_entity_record_with_budget(
340    data: &[u8],
341    version: u32,
342    budget: &mut reddb_file::native_value::DecodeBudget,
343) -> Result<(UnifiedEntity, Option<Metadata>), super::error::StoreError> {
344    use super::error::StoreError;
345    let frame = reddb_file::decode_native_entity_record_frame(data).map_err(decode_error)?;
346    let (entity_bytes, metadata_bytes) =
347        frame.map_or((data, &[][..]), |frame| (frame.entity, frame.metadata));
348    let entity =
349        super::entity_codec::read_entity_binary_with_budget(entity_bytes, &mut 0, version, budget)
350            .map_err(decode_error)?;
351    let metadata = if metadata_bytes.is_empty() {
352        None
353    } else {
354        self::budget::admit(metadata_bytes, budget).map_err(decode_error)?;
355        Some(deserialize_metadata(metadata_bytes).map_err(StoreError::Serialization)?)
356    };
357    Ok((entity, metadata))
358}
359
360pub(super) fn decode_error(error: reddb_file::RdbFileError) -> super::error::StoreError {
361    match error {
362        reddb_file::RdbFileError::ReadLimitExceeded {
363            required_bytes,
364            limit_bytes,
365        } => super::error::StoreError::ReadLimitExceeded {
366            required_bytes,
367            limit_bytes,
368        },
369        other => super::error::StoreError::Serialization(other.to_string()),
370    }
371}
reddb_engine/storage/metadata_codec.rs

reddb_engine/storage/
metadata_codec.rs
