yeti_types/

transcode.rs

//! Zero-`Value` `MessagePack` → JSON transcoding for the read path.
//!
//! The storage layer persists records as `MessagePack` (see
//! [`crate::encoding::encode`]) but the HTTP wire format is JSON. The
//! historical read path paid for this twice: it deserialized the stored
//! bytes into a [`serde_json::Value`] tree and then re-serialized that
//! tree to JSON bytes. Both the tree allocation and the re-walk are pure
//! overhead when the record needs no in-memory transformation.
//!
//! This module streams the `MessagePack` deserializer directly into a
//! [`serde_json::Serializer`] using the established serde "transcode"
//! pattern: a [`serde::Serialize`] adapter that, when asked to serialize,
//! pulls one value out of a [`serde::Deserializer`] and forwards it into
//! the serializer event-by-event. No `serde_json::Value` is ever
//! materialized.
//!
//! ## Byte-identity with the round-trip path
//!
//! The write path serializes a [`serde_json::Value`] to `MessagePack`. A
//! `serde_json::Value`'s object is a `BTreeMap` (the workspace does not
//! enable serde_json's `preserve_order` feature), so stored map keys are
//! already in sorted order. `serde_json::Serializer` emits map entries in
//! the order it receives them, so transcoding preserves that sorted order
//! — byte-identical to `serde_json::to_vec(&decode::<Value>(bytes))`.
//! Numbers and strings are formatted by the same `serde_json` serializer
//! in both paths, so their byte representation matches as well.
//!
//! ## Top-level key filtering
//!
//! [`transcode_msgpack_to_json_filtered`] reproduces the read path's
//! schema-strip (`retain(|k| k == "id" || allowed.contains(k))`) during
//! the transcode itself, so the fast path stays correct for records that
//! carry fields no longer in the current schema. Filtering happens only
//! at the top-level object; nested objects are forwarded verbatim, exactly
//! as the `Value`-based strip behaved (it only ever called `as_object_mut`
//! on the root record).

use crate::error::{EncodingError, Result, YetiError};
use std::cell::RefCell;
use std::collections::HashSet;

use serde::de::{DeserializeSeed, MapAccess, SeqAccess, Visitor};
use serde::ser::{SerializeMap, SerializeSeq, Serializer};
use serde::{Deserializer, Serialize};

/// Transcode `MessagePack` bytes directly to JSON bytes without
/// materializing a [`serde_json::Value`].
///
/// Produces output byte-identical to
/// `serde_json::to_vec(&decode::<serde_json::Value>(bytes)?)` for any
/// record written through [`crate::encoding::encode`]. See the module
/// docs for why the byte-identity holds.
///
/// # Errors
///
/// Returns [`YetiError::Encoding`] with [`EncodingError::MessagePack`] if
/// `bytes` is not valid `MessagePack` or the nested JSON serialization
/// fails (the latter cannot happen for well-formed `MessagePack` that
/// originated from a `serde_json::Value`).
pub fn transcode_msgpack_to_json(bytes: &[u8]) -> Result<Vec<u8>> {
    transcode(bytes, None)
}

/// Transcode `MessagePack` bytes to JSON bytes, retaining only top-level
/// object keys that are `"id"` or present in `allowed`.
///
/// This mirrors the read path's schema-field enforcement
/// (`obj.retain(|k, _| k == "id" || valid_fields.contains(k))`) without
/// building a [`serde_json::Value`]. When the stored record's keys are a
/// subset of `allowed ∪ {"id"}` (the common case) the output is identical
/// to the unfiltered transcode.
///
/// # Errors
///
/// Same conditions as [`transcode_msgpack_to_json`].
// The read path's `cached_schema_fields` is a std `HashSet<String>`
// (default hasher), so a generic `BuildHasher` parameter would only add
// noise here.
#[allow(clippy::implicit_hasher)]
pub fn transcode_msgpack_to_json_filtered(
    bytes: &[u8],
    allowed: &HashSet<String>,
) -> Result<Vec<u8>> {
    transcode(bytes, Some(allowed))
}

fn transcode(bytes: &[u8], allowed: Option<&HashSet<String>>) -> Result<Vec<u8>> {
    let mut de = rmp_serde::Deserializer::new(bytes);
    let mut out = Vec::with_capacity(bytes.len() * 2);
    let mut ser = serde_json::Serializer::new(&mut out);

    // `&mut rmp_serde::Deserializer` is itself a `serde::Deserializer`, so
    // the root transcoder takes the deserializer by mutable reference.
    let transcoder = RootTranscoder {
        de: RefCell::new(Some(&mut de)),
        allowed,
    };

    transcoder.serialize(&mut ser).map_err(|e| {
        YetiError::Encoding(EncodingError::MessagePack(format!(
            "Failed to transcode MessagePack to JSON: {e}"
        )))
    })?;

    Ok(out)
}

/// Top-level transcoder. Drives the `MessagePack` deserializer once and
/// forwards the single contained value into the JSON serializer, applying
/// the optional top-level key filter when the value is a map.
///
/// The deserializer is held as `RefCell<Option<&mut D>>` and `take`n on
/// the single `serialize` call; serde drives the whole transcode through
/// exactly one such call, so the `Option` is always `Some` when reached.
struct RootTranscoder<'a, D> {
    de: RefCell<Option<D>>,
    allowed: Option<&'a HashSet<String>>,
}

impl<'de, D: Deserializer<'de>> Serialize for RootTranscoder<'_, D> {
    fn serialize<S: Serializer>(&self, ser: S) -> std::result::Result<S::Ok, S::Error> {
        // serde drives the whole transcode through exactly one `serialize`
        // call, so the deserializer is always present here. The `ok_or_else`
        // keeps the function total without a panic in the unreachable case.
        let de = self.de.borrow_mut().take().ok_or_else(|| {
            serde::ser::Error::custom("RootTranscoder deserializer already consumed")
        })?;
        // `deserialize_any` yields `Result<Result<S::Ok, S::Error>, D::Error>`:
        // the outer error is a deserialize failure (bad MessagePack), the
        // inner is a serialize failure (cannot happen for JSON output). The
        // `?` lifts the deserialize error into `S::Error`; the inner result
        // is the transcode outcome.
        de.deserialize_any(RootVisitor {
            ser,
            allowed: self.allowed,
        })
        .map_err(serde::ser::Error::custom)?
    }
}

/// Visitor that maps each deserialized event to a JSON serializer event.
struct RootVisitor<'a, S: Serializer> {
    ser: S,
    allowed: Option<&'a HashSet<String>>,
}

impl<'de, S: Serializer> Visitor<'de> for RootVisitor<'_, S> {
    type Value = std::result::Result<S::Ok, S::Error>;

    fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        f.write_str("any MessagePack value")
    }

    fn visit_map<A: MapAccess<'de>>(
        self,
        mut map: A,
    ) -> std::result::Result<Self::Value, A::Error> {
        let mut sm = match self.ser.serialize_map(None) {
            Ok(sm) => sm,
            Err(e) => return Ok(Err(e)),
        };
        while let Some(key) = map.next_key::<String>()? {
            let keep = self
                .allowed
                .is_none_or(|set| key == "id" || set.contains(&key));
            if keep {
                if let Err(e) = sm.serialize_key(&key) {
                    return Ok(Err(e));
                }
                let value = map.next_value_seed(JsonSeed { sm: &mut sm })?;
                if let Err(e) = value {
                    return Ok(Err(e));
                }
            } else {
                // Drop the value without forwarding it.
                map.next_value::<serde::de::IgnoredAny>()?;
            }
        }
        Ok(sm.end())
    }

    fn visit_seq<A: SeqAccess<'de>>(
        self,
        mut seq: A,
    ) -> std::result::Result<Self::Value, A::Error> {
        let mut ss = match self.ser.serialize_seq(None) {
            Ok(ss) => ss,
            Err(e) => return Ok(Err(e)),
        };
        loop {
            let stepped = seq.next_element_seed(SeqSeed { ss: &mut ss })?;
            match stepped {
                Some(Ok(())) => {},
                Some(Err(e)) => return Ok(Err(e)),
                None => break,
            }
        }
        Ok(ss.end())
    }

    fn visit_bool<E>(self, v: bool) -> std::result::Result<Self::Value, E> {
        Ok(self.ser.serialize_bool(v))
    }
    fn visit_i64<E>(self, v: i64) -> std::result::Result<Self::Value, E> {
        Ok(self.ser.serialize_i64(v))
    }
    fn visit_u64<E>(self, v: u64) -> std::result::Result<Self::Value, E> {
        Ok(self.ser.serialize_u64(v))
    }
    fn visit_f64<E>(self, v: f64) -> std::result::Result<Self::Value, E> {
        Ok(self.ser.serialize_f64(v))
    }
    fn visit_str<E>(self, v: &str) -> std::result::Result<Self::Value, E> {
        Ok(self.ser.serialize_str(v))
    }
    fn visit_string<E>(self, v: String) -> std::result::Result<Self::Value, E> {
        Ok(self.ser.serialize_str(&v))
    }
    fn visit_bytes<E>(self, v: &[u8]) -> std::result::Result<Self::Value, E> {
        // serde_json serializes a byte slice as an array of numbers, which
        // matches how a `serde_json::Value` produced from these bytes would
        // render. Records written via `encode(&Value)` never contain raw
        // msgpack bin, but this keeps the transcoder total.
        use serde::ser::SerializeSeq;
        let mut seq = match self.ser.serialize_seq(Some(v.len())) {
            Ok(s) => s,
            Err(e) => return Ok(Err(e)),
        };
        for b in v {
            if let Err(e) = seq.serialize_element(b) {
                return Ok(Err(e));
            }
        }
        Ok(seq.end())
    }
    fn visit_none<E>(self) -> std::result::Result<Self::Value, E> {
        Ok(self.ser.serialize_none())
    }
    fn visit_unit<E>(self) -> std::result::Result<Self::Value, E> {
        Ok(self.ser.serialize_unit())
    }
    fn visit_some<D2: Deserializer<'de>>(
        self,
        de: D2,
    ) -> std::result::Result<Self::Value, D2::Error> {
        de.deserialize_any(self)
    }
}

/// `DeserializeSeed` that forwards one map value into the open JSON map.
struct JsonSeed<'a, M: SerializeMap> {
    sm: &'a mut M,
}

impl<'de, M: SerializeMap> DeserializeSeed<'de> for JsonSeed<'_, M> {
    type Value = std::result::Result<(), M::Error>;

    fn deserialize<D2: Deserializer<'de>>(
        self,
        de: D2,
    ) -> std::result::Result<Self::Value, D2::Error> {
        let v = ValueTranscoder {
            de: RefCell::new(Some(de)),
        };
        Ok(self.sm.serialize_value(&v))
    }
}

/// `DeserializeSeed` that forwards one sequence element into the open JSON
/// sequence.
struct SeqSeed<'a, Q: SerializeSeq> {
    ss: &'a mut Q,
}

impl<'de, Q: SerializeSeq> DeserializeSeed<'de> for SeqSeed<'_, Q> {
    type Value = std::result::Result<(), Q::Error>;

    fn deserialize<D2: Deserializer<'de>>(
        self,
        de: D2,
    ) -> std::result::Result<Self::Value, D2::Error> {
        let v = ValueTranscoder {
            de: RefCell::new(Some(de)),
        };
        Ok(self.ss.serialize_element(&v))
    }
}

/// A `Serialize` adapter holding one not-yet-consumed `Deserializer`. When
/// serialized, it pulls the single value out of the deserializer and
/// forwards every nested event into the serializer. Nested maps are never
/// filtered — only the root record is.
struct ValueTranscoder<D> {
    de: RefCell<Option<D>>,
}

impl<'de, D: Deserializer<'de>> Serialize for ValueTranscoder<D> {
    fn serialize<S: Serializer>(&self, ser: S) -> std::result::Result<S::Ok, S::Error> {
        // Each `ValueTranscoder` is serialized exactly once (one map value /
        // one sequence element), so the deserializer is always present.
        let de = self.de.borrow_mut().take().ok_or_else(|| {
            serde::ser::Error::custom("ValueTranscoder deserializer already consumed")
        })?;
        de.deserialize_any(RootVisitor { ser, allowed: None })
            .map_err(serde::ser::Error::custom)?
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

    use super::*;
    use crate::encoding::{decode, encode};
    use serde_json::{Value, json};

    /// The byte-for-byte oracle: the exact round-trip the read path used
    /// before this transcoder existed.
    fn round_trip_oracle(msgpack: &[u8]) -> Vec<u8> {
        let value: Value = decode(msgpack).unwrap();
        serde_json::to_vec(&value).unwrap()
    }

    fn assert_identical(value: &Value) {
        let msgpack = encode(value).unwrap();
        let expected = round_trip_oracle(&msgpack);
        let actual = transcode_msgpack_to_json(&msgpack).unwrap();
        assert_eq!(
            String::from_utf8_lossy(&actual),
            String::from_utf8_lossy(&expected),
            "transcode diverged from the round-trip oracle"
        );
        assert_eq!(
            actual, expected,
            "transcode produced non-byte-identical JSON"
        );
    }

    #[test]
    fn scalars_round_trip_identically() {
        assert_identical(&json!(null));
        assert_identical(&json!(true));
        assert_identical(&json!(false));
        assert_identical(&json!(0));
        assert_identical(&json!(-42));
        assert_identical(&json!(9_007_199_254_740_993u64));
        assert_identical(&json!(3.5));
        assert_identical(&json!("hello"));
        assert_identical(&json!("with \"quotes\" and \n newlines \t and unicode ✓"));
    }

    #[test]
    fn arrays_round_trip_identically() {
        assert_identical(&json!([]));
        assert_identical(&json!([1, 2, 3]));
        assert_identical(&json!(["a", "b", "c"]));
        assert_identical(&json!([{"x": 1}, {"y": 2}]));
        assert_identical(&json!([[1, [2, [3]]], null, "x"]));
    }

    #[test]
    fn objects_round_trip_identically() {
        assert_identical(&json!({}));
        assert_identical(&json!({"id": "abc", "name": "n"}));
        // Keys deliberately out of order — serde_json's BTreeMap sorts
        // them on both paths, so the output must still match.
        assert_identical(&json!({"zeta": 1, "alpha": 2, "mid": 3}));
        assert_identical(&json!({
            "id": "rec_00000001",
            "name": "User",
            "nested": {"b": 2, "a": 1, "deep": {"y": 2, "x": 1}},
            "list": [1, 2, {"k": "v"}],
            "flag": true,
            "missing": null
        }));
    }

    #[test]
    fn realistic_record_round_trips_identically() {
        let value = json!({
            "id": "rec_00000042",
            "name": "User Name 42",
            "email": "user42@example.com",
            "category": "category_2",
            "description": "A detailed description with punctuation, commas, and \"quotes\".",
            "metadata": "{\"created\":\"42\",\"tags\":[\"a\",\"b\"]}",
            "payload": "X".repeat(550),
        });
        assert_identical(&value);
    }

    #[test]
    fn filter_drops_unknown_top_level_keys() {
        let value = json!({
            "id": "rec_1",
            "name": "keep",
            "stale_field": "drop me",
            "another_stale": 99,
        });
        let msgpack = encode(&value).unwrap();

        let mut allowed = HashSet::new();
        allowed.insert("name".to_owned());
        // `id` is always kept implicitly.

        let actual = transcode_msgpack_to_json_filtered(&msgpack, &allowed).unwrap();

        // Oracle: strip via the same retain rule, then serialize.
        let mut oracle_value = value;
        if let Some(obj) = oracle_value.as_object_mut() {
            obj.retain(|k, _| k == "id" || allowed.contains(k));
        }
        let expected = serde_json::to_vec(&oracle_value).unwrap();

        assert_eq!(actual, expected);
        // Sanity: the dropped keys really are gone.
        let parsed: Value = serde_json::from_slice(&actual).unwrap();
        assert!(parsed.get("stale_field").is_none());
        assert!(parsed.get("another_stale").is_none());
        assert_eq!(parsed["id"], json!("rec_1"));
        assert_eq!(parsed["name"], json!("keep"));
    }

    #[test]
    fn filter_keeps_all_when_every_key_allowed() {
        let value = json!({"id": "x", "a": 1, "b": 2, "c": 3});
        let msgpack = encode(&value).unwrap();
        let allowed: HashSet<String> = ["a", "b", "c"].iter().map(|s| (*s).to_owned()).collect();

        let filtered = transcode_msgpack_to_json_filtered(&msgpack, &allowed).unwrap();
        let unfiltered = transcode_msgpack_to_json(&msgpack).unwrap();
        assert_eq!(filtered, unfiltered);
        assert_eq!(filtered, round_trip_oracle(&msgpack));
    }

    #[test]
    fn filter_does_not_touch_nested_objects() {
        // A nested key matching a dropped top-level name must survive.
        let value = json!({
            "id": "x",
            "keep": {"stale_field": "nested stays", "n": 1},
        });
        let msgpack = encode(&value).unwrap();
        let allowed: HashSet<String> = std::iter::once("keep".to_owned()).collect();

        let actual = transcode_msgpack_to_json_filtered(&msgpack, &allowed).unwrap();
        let parsed: Value = serde_json::from_slice(&actual).unwrap();
        assert_eq!(parsed["keep"]["stale_field"], json!("nested stays"));
        assert_eq!(parsed["keep"]["n"], json!(1));
    }

    #[test]
    fn invalid_msgpack_is_an_error() {
        // 0xc1 is the never-used MessagePack byte.
        let err = transcode_msgpack_to_json(&[0xc1]);
        assert!(err.is_err());
    }
}