aion-core 0.31.0

//! The JSON shape of payload bytes on every aion wire and in every stored row.
//!
//! A `Vec<u8>` under `serde_json` is an array of integers: 3.5 to 4 times the
//! bytes it carries, multiplied again at every nesting (a [`Payload`] inside an
//! `Event` inside a wire envelope is inflated twice). This module is the one
//! codec for those bytes, applied with `#[serde(with = "aion_core::payload_bytes")]`
//! to every byte field that crosses a JSON boundary: [`Payload`] here, the proto
//! payload in `aion-proto`, and the liminal dispatch request's `input` on both
//! sides of that wire.
//!
//! It DECODES two shapes and refuses everything else by name:
//!
//! - a **base64 string** (RFC 4648 standard alphabet; padding accepted either
//!   way, as proto-JSON readers accept it), which is what gRPC's JSON mapping
//!   already renders for `bytes` and what the wire moves to;
//! - an **array of integers 0..=255**, which is the shape every history,
//!   outbox row and schedule row written before this module carries on disk,
//!   and every frame a pre-module writer still emits.
//!
//! It ENCODES the integer array. That is deliberate and temporary: readers
//! move first, so that no writer emits a base64 string before every reader on
//! every wire (Rust workers in the fleet, the Python and TypeScript SDKs, the
//! console) decodes one. The encoder flips to base64 in the writer landing,
//! and the decoder keeps both shapes for as long as an array exists on disk.
//!
//! [`Payload`]: crate::payload::Payload

use std::fmt;

use base64::Engine as _;
use base64::alphabet::STANDARD;
use base64::engine::general_purpose::GeneralPurposeConfig;
use base64::engine::{DecodePaddingMode, GeneralPurpose};
use serde::de::{self, Deserializer, SeqAccess, Visitor};
use serde::ser::Serializer;

/// The base64 engine for payload bytes: the standard alphabet, padding
/// accepted with or without on decode.
const ENGINE: GeneralPurpose = GeneralPurpose::new(
    &STANDARD,
    GeneralPurposeConfig::new().with_decode_padding_mode(DecodePaddingMode::Indifferent),
);

/// What the decoder accepts, in the words every refusal repeats.
const EXPECTED: &str = "payload bytes as a base64 string or an array of integers 0..=255";

/// Serializes payload bytes in the shape the current wire carries.
///
/// # Errors
///
/// Returns the serializer's own error when it refuses the sequence.
pub fn serialize<S>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error>
where
    S: Serializer,
{
    serializer.collect_seq(bytes)
}

/// Deserializes payload bytes from either accepted shape.
///
/// # Errors
///
/// Returns the deserializer's error, carrying a refusal that names the shape
/// that was found and the two that are accepted.
pub fn deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
where
    D: Deserializer<'de>,
{
    if deserializer.is_human_readable() {
        deserializer.deserialize_any(BytesVisitor)
    } else {
        deserializer.deserialize_seq(BytesVisitor)
    }
}

/// Decodes the `bytes` field of a payload object that was read as a JSON
/// value by hand, through the same codec the typed path uses.
///
/// # Errors
///
/// Returns the refusal naming the shape found, exactly as the typed path would.
pub fn from_json_value(value: &serde_json::Value) -> Result<Vec<u8>, serde_json::Error> {
    deserialize(value)
}

/// The number of bytes a `bytes` value of either shape carries, without
/// materializing them: an array's length, or a base64 string's decoded length
/// from its character count. For the projections that elide a payload larger
/// than a caller's limit, where decoding a payload only to measure it would
/// cost the very bytes the limit exists to keep off the wire.
///
/// `None` for any other shape (an elision marker, a number, `null`): not a
/// payload's bytes, so not something to measure. A malformed base64 string
/// still yields its arithmetic length here; the typed decode refuses it.
#[must_use]
pub fn decoded_len(value: &serde_json::Value) -> Option<u64> {
    match value {
        serde_json::Value::Array(bytes) => u64::try_from(bytes.len()).ok(),
        serde_json::Value::String(text) => {
            let unpadded = text.trim_end_matches('=').len();
            u64::try_from(unpadded * 6 / 8).ok()
        }
        _ => None,
    }
}

struct BytesVisitor;

impl<'de> Visitor<'de> for BytesVisitor {
    type Value = Vec<u8>;

    fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(EXPECTED)
    }

    fn visit_str<E>(self, text: &str) -> Result<Self::Value, E>
    where
        E: de::Error,
    {
        ENGINE.decode(text).map_err(|error| {
            E::custom(format!(
                "payload bytes are a string but not base64 in the standard alphabet ({error}); \
                 expected {EXPECTED}"
            ))
        })
    }

    fn visit_bytes<E>(self, bytes: &[u8]) -> Result<Self::Value, E>
    where
        E: de::Error,
    {
        Ok(bytes.to_vec())
    }

    fn visit_byte_buf<E>(self, bytes: Vec<u8>) -> Result<Self::Value, E>
    where
        E: de::Error,
    {
        Ok(bytes)
    }

    fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
    where
        A: SeqAccess<'de>,
    {
        let mut bytes = Vec::with_capacity(seq.size_hint().unwrap_or(0));
        while let Some(element) = seq.next_element::<serde_json::Value>()? {
            let index = bytes.len();
            let byte = match element.as_u64().map(u8::try_from) {
                Some(Ok(byte)) => byte,
                Some(Err(_)) => {
                    return Err(de::Error::custom(format!(
                        "payload byte {index} is {element}, outside 0..=255; expected {EXPECTED}"
                    )));
                }
                None => {
                    return Err(de::Error::custom(format!(
                        "payload byte {index} is {}, not an integer; expected {EXPECTED}",
                        json_kind(&element)
                    )));
                }
            };
            bytes.push(byte);
        }
        Ok(bytes)
    }

    fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
    where
        A: de::MapAccess<'de>,
    {
        let mut keys = Vec::new();
        while let Some((key, _)) = map.next_entry::<String, serde::de::IgnoredAny>()? {
            keys.push(key);
        }
        let what = if keys.iter().any(|key| key == "__elided") {
            "an elided projection (`__elided`), which carries no bytes"
        } else {
            "an object"
        };
        Err(de::Error::custom(format!(
            "payload bytes are {what}; expected {EXPECTED}"
        )))
    }
}

/// The JSON kind of a value, in the words a refusal uses.
fn json_kind(value: &serde_json::Value) -> &'static str {
    match value {
        serde_json::Value::Null => "null",
        serde_json::Value::Bool(_) => "a boolean",
        serde_json::Value::Number(number) if number.is_f64() => "a fraction",
        serde_json::Value::Number(_) => "a negative integer",
        serde_json::Value::String(_) => "a string",
        serde_json::Value::Array(_) => "an array",
        serde_json::Value::Object(_) => "an object",
    }
}

#[cfg(test)]
mod tests {
    use serde::{Deserialize, Serialize};
    use serde_json::json;

    #[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
    struct Carrier {
        #[serde(with = "super")]
        bytes: Vec<u8>,
    }

    #[test]
    fn a_base64_string_decodes_with_or_without_padding() -> Result<(), serde_json::Error> {
        let padded: Carrier = serde_json::from_value(json!({ "bytes": "eyJvayI6dHJ1ZX0=" }))?;
        let bare: Carrier = serde_json::from_value(json!({ "bytes": "eyJvayI6dHJ1ZX0" }))?;
        assert_eq!(padded.bytes, br#"{"ok":true}"#);
        assert_eq!(bare, padded);
        Ok(())
    }

    #[test]
    fn an_integer_array_decodes() -> Result<(), serde_json::Error> {
        let carrier: Carrier = serde_json::from_value(json!({ "bytes": [0, 15, 16, 255] }))?;
        assert_eq!(carrier.bytes, vec![0, 15, 16, 255]);
        let empty: Carrier = serde_json::from_value(json!({ "bytes": [] }))?;
        assert!(empty.bytes.is_empty());
        Ok(())
    }

    #[test]
    fn the_writer_still_emits_the_array_until_every_reader_has_moved()
    -> Result<(), serde_json::Error> {
        let value = serde_json::to_value(Carrier {
            bytes: vec![123, 125],
        })?;
        assert_eq!(value, json!({ "bytes": [123, 125] }));
        Ok(())
    }

    #[test]
    fn every_refusal_names_the_shape_it_found() {
        let cases: [(serde_json::Value, &[&str]); 6] = [
            (
                json!({ "bytes": [256] }),
                &["payload byte 0 is 256", "outside 0..=255"],
            ),
            (
                json!({ "bytes": [1, "a"] }),
                &["payload byte 1 is a string", "not an integer"],
            ),
            (
                json!({ "bytes": [1, -1] }),
                &["payload byte 1 is a negative integer"],
            ),
            (
                json!({ "bytes": "not-base64!" }),
                &["a string but not base64"],
            ),
            (
                json!({ "bytes": { "__elided": true, "size_bytes": 9 } }),
                &["an elided projection", "carries no bytes"],
            ),
            (
                json!({ "bytes": 7 }),
                &["invalid type: integer `7`", "expected payload bytes"],
            ),
        ];
        for (value, words) in cases {
            let refused = serde_json::from_value::<Carrier>(value.clone());
            assert!(refused.is_err(), "{value} must be refused");
            let message = refused
                .err()
                .map(|error| error.to_string())
                .unwrap_or_default();
            for word in words {
                assert!(
                    message.contains(word),
                    "{value}: refusal {message:?} lacks {word:?}"
                );
            }
        }
    }

    #[test]
    fn decoded_len_measures_both_shapes_without_decoding() {
        assert_eq!(super::decoded_len(&json!([1, 2, 3, 4, 5])), Some(5));
        assert_eq!(super::decoded_len(&json!([])), Some(0));
        // "hello" = aGVsbG8= (one pad), "hell" = aGVsbA== (two), "hel" = aGVs (none).
        assert_eq!(super::decoded_len(&json!("aGVsbG8=")), Some(5));
        assert_eq!(super::decoded_len(&json!("aGVsbA==")), Some(4));
        assert_eq!(super::decoded_len(&json!("aGVsbA")), Some(4));
        assert_eq!(super::decoded_len(&json!("aGVs")), Some(3));
        assert_eq!(super::decoded_len(&json!("")), Some(0));
        assert_eq!(
            super::decoded_len(&json!({ "__elided": true, "size_bytes": 9 })),
            None
        );
        assert_eq!(super::decoded_len(&json!(null)), None);
    }

    #[test]
    fn the_hand_reader_and_the_typed_path_agree() -> Result<(), serde_json::Error> {
        let string = super::from_json_value(&json!("eyJvayI6dHJ1ZX0="))?;
        let array =
            super::from_json_value(&json!([123, 34, 111, 107, 34, 58, 116, 114, 117, 101, 125]))?;
        assert_eq!(string, array);
        assert!(super::from_json_value(&json!(null)).is_err());
        Ok(())
    }
}