prolly/prolly/
value.rs

1//! Typed value encoding helpers.
2//!
3//! The core tree stores raw byte values. This module provides small helpers for
4//! serializing typed application values into those bytes, plus a deterministic
5//! versioned envelope for values that need schema and migration metadata.
6
7use serde::{de::DeserializeOwned, Deserialize, Serialize};
8
9use super::encoding::Encoding;
10use super::error::Error;
11
12const VERSIONED_VALUE_MAGIC: &[u8; 4] = b"PLVV";
13const VERSIONED_VALUE_WIRE_VERSION: u8 = 1;
14const ENCODING_RAW: u8 = 0;
15const ENCODING_CBOR: u8 = 1;
16const ENCODING_JSON: u8 = 2;
17const ENCODING_CUSTOM: u8 = 3;
18const HEADER_LEN: usize = 4 + 1 + 1 + 8 + 4 + 4 + 8;
19
20/// Codec for converting typed application values to and from stored bytes.
21///
22/// The core tree stores `Vec<u8>` leaf values. A `ValueCodec` is a small
23/// reusable adapter that keeps application encode/decode policy next to the
24/// schema that owns it.
25pub trait ValueCodec {
26    /// Encode a typed value into bytes suitable for storing in a tree.
27    fn encode<T: Serialize>(&self, value: &T) -> Result<Vec<u8>, Error>;
28
29    /// Decode a typed value from bytes read from a tree.
30    fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, Error>;
31}
32
33/// JSON codec for serde-backed application values.
34#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
35pub struct JsonCodec;
36
37/// CBOR codec for serde-backed application values.
38#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
39pub struct CborCodec;
40
41/// Versioned JSON codec with schema/version validation on decode.
42#[derive(Clone, Debug, PartialEq, Eq)]
43pub struct VersionedJsonCodec {
44    schema: String,
45    version: u64,
46}
47
48/// Versioned CBOR codec with schema/version validation on decode.
49#[derive(Clone, Debug, PartialEq, Eq)]
50pub struct VersionedCborCodec {
51    schema: String,
52    version: u64,
53}
54
55/// Serialize a typed value as compact JSON bytes.
56pub fn encode_json<T: Serialize>(value: &T) -> Result<Vec<u8>, Error> {
57    serde_json::to_vec(value).map_err(|err| Error::Serialize(err.to_string()))
58}
59
60/// Deserialize typed JSON value bytes.
61pub fn decode_json<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, Error> {
62    serde_json::from_slice(bytes).map_err(|err| Error::Deserialize(err.to_string()))
63}
64
65/// Serialize a typed value as compact CBOR bytes.
66pub fn encode_cbor<T: Serialize>(value: &T) -> Result<Vec<u8>, Error> {
67    serde_cbor::ser::to_vec_packed(value).map_err(|err| Error::Serialize(err.to_string()))
68}
69
70/// Deserialize typed CBOR value bytes.
71pub fn decode_cbor<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, Error> {
72    serde_cbor::from_slice(bytes).map_err(|err| Error::Deserialize(err.to_string()))
73}
74
75impl ValueCodec for JsonCodec {
76    fn encode<T: Serialize>(&self, value: &T) -> Result<Vec<u8>, Error> {
77        encode_json(value)
78    }
79
80    fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, Error> {
81        decode_json(bytes)
82    }
83}
84
85impl ValueCodec for CborCodec {
86    fn encode<T: Serialize>(&self, value: &T) -> Result<Vec<u8>, Error> {
87        encode_cbor(value)
88    }
89
90    fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, Error> {
91        decode_cbor(bytes)
92    }
93}
94
95impl VersionedJsonCodec {
96    /// Create a JSON codec that wraps values in a [`VersionedValue`] envelope.
97    pub fn new(schema: impl Into<String>, version: u64) -> Self {
98        Self {
99            schema: schema.into(),
100            version,
101        }
102    }
103
104    /// Schema expected by this codec.
105    pub fn schema(&self) -> &str {
106        &self.schema
107    }
108
109    /// Schema version expected by this codec.
110    pub fn version(&self) -> u64 {
111        self.version
112    }
113}
114
115impl ValueCodec for VersionedJsonCodec {
116    fn encode<T: Serialize>(&self, value: &T) -> Result<Vec<u8>, Error> {
117        VersionedValue::json(&self.schema, self.version, value)?.to_bytes()
118    }
119
120    fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, Error> {
121        let envelope = VersionedValue::from_bytes(bytes)?;
122        envelope.require_schema(&self.schema, self.version)?;
123        envelope.decode_json()
124    }
125}
126
127impl VersionedCborCodec {
128    /// Create a CBOR codec that wraps values in a [`VersionedValue`] envelope.
129    pub fn new(schema: impl Into<String>, version: u64) -> Self {
130        Self {
131            schema: schema.into(),
132            version,
133        }
134    }
135
136    /// Schema expected by this codec.
137    pub fn schema(&self) -> &str {
138        &self.schema
139    }
140
141    /// Schema version expected by this codec.
142    pub fn version(&self) -> u64 {
143        self.version
144    }
145}
146
147impl ValueCodec for VersionedCborCodec {
148    fn encode<T: Serialize>(&self, value: &T) -> Result<Vec<u8>, Error> {
149        VersionedValue::cbor(&self.schema, self.version, value)?.to_bytes()
150    }
151
152    fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, Error> {
153        let envelope = VersionedValue::from_bytes(bytes)?;
154        envelope.require_schema(&self.schema, self.version)?;
155        envelope.decode_cbor()
156    }
157}
158
159/// Deterministic schema/version envelope for an application value.
160///
161/// The envelope is useful when a tree contains values that may evolve over
162/// time. It keeps the core map byte-oriented while giving applications a
163/// stable place to record a schema name, schema version, value encoding, and
164/// payload bytes.
165#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
166pub struct VersionedValue {
167    /// Application schema or type name.
168    pub schema: String,
169    /// Application-controlled schema version.
170    pub version: u64,
171    /// Encoding used for `payload`.
172    pub encoding: Encoding,
173    /// Encoded application payload.
174    pub payload: Vec<u8>,
175}
176
177impl VersionedValue {
178    /// Create a versioned raw-byte value.
179    pub fn raw(schema: impl Into<String>, version: u64, payload: impl Into<Vec<u8>>) -> Self {
180        Self {
181            schema: schema.into(),
182            version,
183            encoding: Encoding::Raw,
184            payload: payload.into(),
185        }
186    }
187
188    /// Create a versioned JSON value.
189    pub fn json<T: Serialize>(
190        schema: impl Into<String>,
191        version: u64,
192        value: &T,
193    ) -> Result<Self, Error> {
194        Ok(Self {
195            schema: schema.into(),
196            version,
197            encoding: Encoding::Json,
198            payload: encode_json(value)?,
199        })
200    }
201
202    /// Create a versioned CBOR value.
203    pub fn cbor<T: Serialize>(
204        schema: impl Into<String>,
205        version: u64,
206        value: &T,
207    ) -> Result<Self, Error> {
208        Ok(Self {
209            schema: schema.into(),
210            version,
211            encoding: Encoding::Cbor,
212            payload: encode_cbor(value)?,
213        })
214    }
215
216    /// Create a versioned value with an explicit encoding marker.
217    pub fn with_encoding(
218        schema: impl Into<String>,
219        version: u64,
220        encoding: Encoding,
221        payload: impl Into<Vec<u8>>,
222    ) -> Self {
223        Self {
224            schema: schema.into(),
225            version,
226            encoding,
227            payload: payload.into(),
228        }
229    }
230
231    /// Encode the envelope as deterministic bytes suitable for a leaf value.
232    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
233        let schema = self.schema.as_bytes();
234        let custom = custom_encoding_name(&self.encoding);
235        let schema_len = checked_u32_len(schema.len(), "schema")?;
236        let custom_len = checked_u32_len(custom.len(), "custom encoding")?;
237
238        let mut out =
239            Vec::with_capacity(HEADER_LEN + schema.len() + custom.len() + self.payload.len());
240        out.extend_from_slice(VERSIONED_VALUE_MAGIC);
241        out.push(VERSIONED_VALUE_WIRE_VERSION);
242        out.push(encoding_tag(&self.encoding));
243        out.extend_from_slice(&self.version.to_be_bytes());
244        out.extend_from_slice(&schema_len.to_be_bytes());
245        out.extend_from_slice(&custom_len.to_be_bytes());
246        out.extend_from_slice(&(self.payload.len() as u64).to_be_bytes());
247        out.extend_from_slice(schema);
248        out.extend_from_slice(custom);
249        out.extend_from_slice(&self.payload);
250        Ok(out)
251    }
252
253    /// Decode a deterministic versioned value envelope.
254    pub fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
255        decode_versioned_value(bytes)
256    }
257
258    /// Decode the payload as JSON, failing if the envelope is not JSON encoded.
259    pub fn decode_json<T: DeserializeOwned>(&self) -> Result<T, Error> {
260        if self.encoding != Encoding::Json {
261            return Err(invalid_value(format!(
262                "expected JSON payload, got {:?}",
263                self.encoding
264            )));
265        }
266        decode_json(&self.payload)
267    }
268
269    /// Decode the payload as CBOR, failing if the envelope is not CBOR encoded.
270    pub fn decode_cbor<T: DeserializeOwned>(&self) -> Result<T, Error> {
271        if self.encoding != Encoding::Cbor {
272            return Err(invalid_value(format!(
273                "expected CBOR payload, got {:?}",
274                self.encoding
275            )));
276        }
277        decode_cbor(&self.payload)
278    }
279
280    /// Verify the envelope schema and version before decoding a payload.
281    pub fn require_schema(&self, schema: &str, version: u64) -> Result<(), Error> {
282        if self.schema == schema && self.version == version {
283            return Ok(());
284        }
285
286        Err(invalid_value(format!(
287            "schema/version mismatch: expected {schema}@{version}, got {}@{}",
288            self.schema, self.version
289        )))
290    }
291
292    /// Return true when this value has the requested schema and version.
293    pub fn matches_schema(&self, schema: &str, version: u64) -> bool {
294        self.schema == schema && self.version == version
295    }
296}
297
298fn decode_versioned_value(bytes: &[u8]) -> Result<VersionedValue, Error> {
299    if bytes.len() < HEADER_LEN {
300        return Err(invalid_value("versioned value envelope is too short"));
301    }
302    if !bytes.starts_with(VERSIONED_VALUE_MAGIC) {
303        return Err(invalid_value("versioned value missing PLVV magic"));
304    }
305
306    let wire_version = bytes[4];
307    if wire_version != VERSIONED_VALUE_WIRE_VERSION {
308        return Err(invalid_value(format!(
309            "unsupported versioned value wire version {wire_version}"
310        )));
311    }
312
313    let encoding = bytes[5];
314    let version = read_u64(bytes, 6)?;
315    let schema_len = read_u32(bytes, 14)? as usize;
316    let custom_len = read_u32(bytes, 18)? as usize;
317    let payload_len = usize::try_from(read_u64(bytes, 22)?)
318        .map_err(|_| invalid_value("payload length does not fit in usize"))?;
319
320    let schema_start = HEADER_LEN;
321    let custom_start = checked_add(schema_start, schema_len, "schema")?;
322    let payload_start = checked_add(custom_start, custom_len, "custom encoding")?;
323    let expected_len = checked_add(payload_start, payload_len, "payload")?;
324    if expected_len != bytes.len() {
325        return Err(invalid_value(format!(
326            "versioned value length mismatch: header expects {expected_len} bytes, got {}",
327            bytes.len()
328        )));
329    }
330
331    let schema = decode_utf8(&bytes[schema_start..custom_start], "schema")?;
332    let custom = decode_utf8(&bytes[custom_start..payload_start], "custom encoding")?;
333    let encoding = decode_encoding(encoding, custom)?;
334    let payload = bytes[payload_start..expected_len].to_vec();
335
336    Ok(VersionedValue {
337        schema,
338        version,
339        encoding,
340        payload,
341    })
342}
343
344fn checked_u32_len(len: usize, field: &str) -> Result<u32, Error> {
345    u32::try_from(len).map_err(|_| invalid_value(format!("{field} is too large")))
346}
347
348fn checked_add(base: usize, len: usize, field: &str) -> Result<usize, Error> {
349    base.checked_add(len)
350        .ok_or_else(|| invalid_value(format!("{field} length overflows usize")))
351}
352
353fn read_u32(bytes: &[u8], offset: usize) -> Result<u32, Error> {
354    let value = bytes
355        .get(offset..offset + 4)
356        .ok_or_else(|| invalid_value("versioned value header is truncated"))?;
357    Ok(u32::from_be_bytes(
358        value.try_into().expect("fixed slice length"),
359    ))
360}
361
362fn read_u64(bytes: &[u8], offset: usize) -> Result<u64, Error> {
363    let value = bytes
364        .get(offset..offset + 8)
365        .ok_or_else(|| invalid_value("versioned value header is truncated"))?;
366    Ok(u64::from_be_bytes(
367        value.try_into().expect("fixed slice length"),
368    ))
369}
370
371fn decode_utf8(bytes: &[u8], field: &str) -> Result<String, Error> {
372    std::str::from_utf8(bytes)
373        .map(str::to_owned)
374        .map_err(|err| invalid_value(format!("{field} is not valid UTF-8: {err}")))
375}
376
377fn custom_encoding_name(encoding: &Encoding) -> &[u8] {
378    match encoding {
379        Encoding::Custom(name) => name.as_bytes(),
380        _ => &[],
381    }
382}
383
384fn encoding_tag(encoding: &Encoding) -> u8 {
385    match encoding {
386        Encoding::Raw => ENCODING_RAW,
387        Encoding::Cbor => ENCODING_CBOR,
388        Encoding::Json => ENCODING_JSON,
389        Encoding::Custom(_) => ENCODING_CUSTOM,
390    }
391}
392
393fn decode_encoding(tag: u8, custom: String) -> Result<Encoding, Error> {
394    match tag {
395        ENCODING_RAW => require_no_custom(custom, Encoding::Raw),
396        ENCODING_CBOR => require_no_custom(custom, Encoding::Cbor),
397        ENCODING_JSON => require_no_custom(custom, Encoding::Json),
398        ENCODING_CUSTOM => Ok(Encoding::Custom(custom)),
399        tag => Err(invalid_value(format!("unknown value encoding tag {tag}"))),
400    }
401}
402
403fn require_no_custom(custom: String, encoding: Encoding) -> Result<Encoding, Error> {
404    if custom.is_empty() {
405        Ok(encoding)
406    } else {
407        Err(invalid_value(format!(
408            "non-custom encoding {:?} included custom encoding name",
409            encoding
410        )))
411    }
412}
413
414fn invalid_value(message: impl Into<String>) -> Error {
415    Error::Deserialize(format!("invalid versioned value: {}", message.into()))
416}
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421
422    #[derive(Debug, PartialEq, Serialize, Deserialize)]
423    struct Example {
424        name: String,
425        score: u64,
426    }
427
428    #[test]
429    fn json_and_cbor_helpers_round_trip_typed_values() {
430        let value = Example {
431            name: "memory".to_string(),
432            score: 42,
433        };
434
435        let json = encode_json(&value).unwrap();
436        assert_eq!(decode_json::<Example>(&json).unwrap(), value);
437
438        let cbor = encode_cbor(&value).unwrap();
439        assert_eq!(decode_cbor::<Example>(&cbor).unwrap(), value);
440    }
441
442    #[test]
443    fn versioned_value_round_trips_schema_encoding_and_payload() {
444        let value = Example {
445            name: "chunk".to_string(),
446            score: 9,
447        };
448        let envelope = VersionedValue::json("memory.chunk", 2, &value).unwrap();
449
450        let bytes = envelope.to_bytes().unwrap();
451        let decoded = VersionedValue::from_bytes(&bytes).unwrap();
452
453        assert_eq!(decoded.schema, "memory.chunk");
454        assert_eq!(decoded.version, 2);
455        assert_eq!(decoded.encoding, Encoding::Json);
456        decoded.require_schema("memory.chunk", 2).unwrap();
457        assert_eq!(decoded.decode_json::<Example>().unwrap(), value);
458    }
459
460    #[test]
461    fn versioned_value_rejects_bad_magic_and_schema_mismatch() {
462        assert!(matches!(
463            VersionedValue::from_bytes(b"raw"),
464            Err(Error::Deserialize(_))
465        ));
466
467        let value = VersionedValue::raw("memory", 1, b"bytes");
468        let err = value.require_schema("memory", 2).unwrap_err();
469        assert!(matches!(err, Error::Deserialize(_)));
470    }
471}
prolly/prolly/value.rs

prolly/prolly/
value.rs
