dyniak 1.7.0

//! Object envelope exchanged over the Riak HTTP gateway.
//!
//! The HTTP K/V endpoints carry a single logical object in one of the
//! three baseline serialisations the gateway negotiates
//! ([`crate::proto::http::content_type::SUPPORTED_CONTENT_TYPES`]):
//! `application/json`, `application/cbor`, and
//! `application/x-protobuf`. To make a value stored under one
//! encoding fetchable under any other, the gateway does not persist
//! the raw request bytes. Instead it decodes the request body into an
//! [`HttpObject`] and stores a canonical, encoding-independent form of
//! that struct (its protobuf serialisation). A later `GET` decodes the
//! canonical form back into an [`HttpObject`] and re-encodes it with
//! whatever codec the client negotiated, so the same logical object
//! survives a json-in / cbor-out (or protobuf-out) round-trip.
//!
//! # Wire shape
//!
//! The envelope is a fixed-schema struct so the same Rust type can be
//! registered with the JSON, CBOR, and protobuf codecs uniformly.
//! Expressed as JSON it is:
//!
//! ```json
//! {
//!   "value": [104, 101, 108, 108, 111],
//!   "content_type": "text/plain",
//!   "indexes": [{"name": "age_int", "value": "42"}]
//! }
//! ```
//!
//! * `value` -- the opaque object payload bytes. In JSON these are a
//!   numeric array; in CBOR an integer array; in protobuf a `bytes`
//!   field. The bytes themselves are preserved verbatim across codecs.
//! * `content_type` -- optional declared media type of `value`. This
//!   is object metadata; it is distinct from the HTTP `Content-Type`
//!   header, which names the codec that framed the envelope.
//! * `indexes` -- secondary-index `(name, value)` pairs. `name` ends
//!   in `_int` (integer index) or `_bin` (binary index), mirroring the
//!   PBC path's [`crate::proto::pb::messages::RpbContent::indexes`].
//! * `links` -- typed object-to-object pointers. Each link is a
//!   `(bucket, key, tag)` triple naming a target object and the tag
//!   that classifies the relationship. Over HTTP these ride in
//!   `Link:` headers; over PBC they ride in
//!   [`crate::proto::pb::messages::RpbContent::links`]. The storage
//!   form keeps them on the envelope so they survive every codec
//!   round-trip, so a put over either transport persists them, and so
//!   a MapReduce link phase can walk them.
//!
//! A dedicated envelope is used rather than reusing
//! [`crate::proto::pb::messages::RpbContent`] directly so the HTTP
//! object store has one serde- and codec-friendly type with `String`
//! index and link components; the PBC path maps between
//! [`crate::proto::pb::messages::RpbContent`] and this envelope on the
//! way in and out.

use std::sync::OnceLock;

use dyn_encoding::{CborCodec, CodecRegistry, JsonCodec, ProtobufCodec, WireTypeId, WireValue};
use prost::Message;
use serde::{Deserialize, Serialize};

/// One secondary-index entry attached to an [`HttpObject`].
///
/// `name` selects the index encoding the storage layer applies: a
/// name ending in `_int` is stored as a big-endian integer (so range
/// scans iterate numerically), anything else is stored as raw bytes.
#[derive(Clone, Eq, PartialEq, Message, Serialize, Deserialize)]
pub struct HttpIndex {
    /// Index name, for example `age_int` or `city_bin`.
    #[prost(string, tag = "1")]
    pub name: String,
    /// Index value as supplied by the client (textual form).
    #[prost(string, tag = "2")]
    pub value: String,
}

/// One typed link attached to an [`HttpObject`].
///
/// A link is a directed, tagged pointer from the carrying object to a
/// target object named by `(bucket, key)`. The `tag` classifies the
/// relationship (Riak calls it the `riaktag`); a MapReduce link phase
/// filters on `bucket` and `tag` when deciding which links to walk.
///
/// # Examples
///
/// ```
/// use dyniak::proto::http::object::HttpLink;
/// let link = HttpLink {
///     bucket: "people".to_string(),
///     key: "bob".to_string(),
///     tag: "friend".to_string(),
/// };
/// assert_eq!(link.tag, "friend");
/// ```
#[derive(Clone, Eq, PartialEq, Message, Serialize, Deserialize)]
pub struct HttpLink {
    /// Target object's bucket.
    #[prost(string, tag = "1")]
    pub bucket: String,
    /// Target object's key.
    #[prost(string, tag = "2")]
    pub key: String,
    /// Relationship tag (Riak's `riaktag`).
    #[prost(string, tag = "3")]
    pub tag: String,
}

/// A logical Riak object as carried by the HTTP K/V endpoints.
///
/// The struct is the unit of cross-encoding round-tripping: a body
/// decoded from one codec re-encodes losslessly under any other.
///
/// # Examples
///
/// ```
/// use dyniak::proto::http::object::HttpObject;
/// let obj = HttpObject {
///     value: b"hello".to_vec(),
///     content_type: Some("text/plain".to_string()),
///     indexes: Vec::new(),
///     links: Vec::new(),
///     context: Vec::new(),
/// };
/// assert_eq!(obj.value, b"hello");
/// ```
#[derive(Clone, Eq, PartialEq, Message, Serialize, Deserialize)]
pub struct HttpObject {
    /// Opaque object payload bytes.
    #[prost(bytes = "vec", tag = "1")]
    #[serde(default)]
    pub value: Vec<u8>,
    /// Optional declared media type of [`Self::value`]. Object
    /// metadata, not the codec content-type of the envelope.
    #[prost(string, optional, tag = "2")]
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub content_type: Option<String>,
    /// Secondary-index entries associated with the object.
    #[prost(message, repeated, tag = "3")]
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub indexes: Vec<HttpIndex>,
    /// Typed links from this object to other objects. Tag 4 was
    /// previously unused, so objects stored before links existed
    /// decode here with an empty list (backward compatible).
    #[prost(message, repeated, tag = "4")]
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub links: Vec<HttpLink>,
    /// Encoded per-object causal context (an interval tree clock).
    /// Tag 5 was previously unused, so objects stored before causal
    /// context existed decode here with an empty context (backward
    /// compatible); an empty context is treated as the seed clock.
    #[prost(bytes = "vec", tag = "5")]
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub context: Vec<u8>,
}

impl WireValue for HttpObject {
    fn wire_type_id() -> WireTypeId {
        WireTypeId::new("riak.http.Object")
    }
}

/// Storage-form marker byte for a sibling set: distinguishes a
/// multi-value blob from a legacy single-object blob. A legacy
/// `HttpObject` protobuf always starts with a field tag byte (the
/// smallest being 0x0A for tag 1), never 0xFF, so this marker is
/// unambiguous.
const SIBLING_SET_MARKER: u8 = 0xFF;

/// A set of concurrent object values (siblings) stored under one key.
///
/// Riak keeps every concurrent write as a sibling when `allow_mult` is
/// set, rather than dropping one. The storage form is a marker byte
/// followed by the protobuf encoding of this message; a blob without
/// the marker is a legacy single [`HttpObject`] and decodes as a
/// one-element set (backward compatible).
#[derive(Clone, Eq, PartialEq, Message, Serialize, Deserialize)]
pub struct SiblingSet {
    /// The concurrent object values. A causally-resolved key holds
    /// exactly one; a key with unresolved concurrent writes holds
    /// more than one.
    #[prost(message, repeated, tag = "1")]
    #[serde(default)]
    pub siblings: Vec<HttpObject>,
}

impl SiblingSet {
    /// A set holding a single object.
    #[must_use]
    pub fn single(obj: HttpObject) -> Self {
        Self {
            siblings: vec![obj],
        }
    }

    /// The representative object for a read that expects a single
    /// value (a MapReduce input, a transaction read): the causally
    /// dominant sibling, or the first when they are concurrent. Returns
    /// `None` for an empty set.
    #[must_use]
    pub fn primary(&self) -> Option<&HttpObject> {
        self.siblings.first()
    }

    /// Serialise to the canonical storage form (marker byte + protobuf).
    #[must_use]
    pub fn to_storage_bytes(&self) -> Vec<u8> {
        let mut out = Vec::with_capacity(1 + self.encoded_len());
        out.push(SIBLING_SET_MARKER);
        out.extend_from_slice(&self.encode_to_vec());
        out
    }

    /// Reconstruct from the canonical storage form. A blob starting
    /// with the sibling marker decodes as a set; any other blob is a
    /// legacy single [`HttpObject`] and becomes a one-element set.
    ///
    /// # Errors
    ///
    /// Returns [`prost::DecodeError`] when the payload after the marker
    /// is not a valid `SiblingSet`, or a legacy blob is not a valid
    /// `HttpObject`.
    pub fn from_storage_bytes(bytes: &[u8]) -> Result<Self, prost::DecodeError> {
        match bytes.split_first() {
            Some((&SIBLING_SET_MARKER, rest)) => Self::decode(rest),
            _ => Ok(Self::single(HttpObject::from_storage_bytes(bytes)?)),
        }
    }
}

impl WireValue for SiblingSet {
    fn wire_type_id() -> WireTypeId {
        WireTypeId::new("riak.http.SiblingSet")
    }
}

impl HttpObject {
    /// Serialise the object into its canonical, encoding-independent
    /// storage form (its protobuf bytes).
    ///
    /// This is the form persisted under the primary K/V key so a
    /// subsequent fetch can re-encode it under any negotiated codec.
    #[must_use]
    pub fn to_storage_bytes(&self) -> Vec<u8> {
        self.encode_to_vec()
    }

    /// Reconstruct an object from its canonical storage form.
    ///
    /// # Errors
    ///
    /// Returns [`prost::DecodeError`] when `bytes` is not a valid
    /// protobuf encoding of the envelope (indicating corruption or a
    /// schema mismatch).
    pub fn from_storage_bytes(bytes: &[u8]) -> Result<Self, prost::DecodeError> {
        Self::decode(bytes)
    }

    /// Convert the envelope's index list into the `(name, value)`
    /// byte-pair form the storage layer's 2i API expects.
    #[must_use]
    pub fn index_pairs(&self) -> Vec<(Vec<u8>, Vec<u8>)> {
        self.indexes
            .iter()
            .map(|i| (i.name.clone().into_bytes(), i.value.clone().into_bytes()))
            .collect()
    }
}

/// Codec registry shared by the HTTP object endpoints.
///
/// Holds a JSON, CBOR, and protobuf codec, each with [`HttpObject`]
/// registered, keyed by the canonical content-type strings from
/// [`crate::proto::http::content_type::SUPPORTED_CONTENT_TYPES`]. The
/// registry is built once and reused for the life of the process.
///
/// # Examples
///
/// ```
/// use dyniak::proto::http::object::object_codecs;
/// let registry = object_codecs();
/// assert!(registry.for_content_type("application/json").is_some());
/// assert!(registry.for_content_type("application/cbor").is_some());
/// assert!(registry.for_content_type("application/x-protobuf").is_some());
/// ```
#[must_use]
pub fn object_codecs() -> &'static CodecRegistry {
    static REGISTRY: OnceLock<CodecRegistry> = OnceLock::new();
    REGISTRY.get_or_init(|| {
        let mut json = JsonCodec::new();
        json.register::<HttpObject>();
        let mut cbor = CborCodec::new();
        cbor.register::<HttpObject>();
        let mut protobuf = ProtobufCodec::new();
        protobuf.register::<HttpObject>();

        let mut registry = CodecRegistry::new();
        registry.register(json);
        registry.register(cbor);
        registry.register(protobuf);
        registry
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use dyn_encoding::WireValue;

    fn fixture() -> HttpObject {
        HttpObject {
            value: b"hello world".to_vec(),
            content_type: Some("text/plain".to_string()),
            indexes: vec![
                HttpIndex {
                    name: "age_int".to_string(),
                    value: "42".to_string(),
                },
                HttpIndex {
                    name: "city_bin".to_string(),
                    value: "seattle".to_string(),
                },
            ],
            links: vec![HttpLink {
                bucket: "people".to_string(),
                key: "bob".to_string(),
                tag: "friend".to_string(),
            }],
            context: Vec::new(),
        }
    }

    #[test]
    fn storage_form_round_trips() {
        let obj = fixture();
        let bytes = obj.to_storage_bytes();
        let back = HttpObject::from_storage_bytes(&bytes).expect("decode");
        assert_eq!(back, obj);
    }

    #[test]
    fn links_round_trip_through_storage_form() {
        let obj = fixture();
        let bytes = obj.to_storage_bytes();
        let back = HttpObject::from_storage_bytes(&bytes).expect("decode");
        assert_eq!(back.links, obj.links);
        assert_eq!(back.links.len(), 1);
        assert_eq!(back.links[0].tag, "friend");
    }

    #[test]
    fn objects_stored_before_links_decode_with_empty_links() {
        // Encode an envelope that carries only tags 1/2/3 (the
        // pre-link schema) and confirm it decodes with no links and
        // no error. This is the on-disk backward-compatibility
        // guarantee: tag 4 was unused, so old bytes are still valid.
        #[derive(Clone, PartialEq, ::prost::Message)]
        struct LegacyObject {
            #[prost(bytes = "vec", tag = "1")]
            value: Vec<u8>,
            #[prost(string, optional, tag = "2")]
            content_type: Option<String>,
            #[prost(message, repeated, tag = "3")]
            indexes: Vec<HttpIndex>,
        }
        let legacy = LegacyObject {
            value: b"old".to_vec(),
            content_type: Some("text/plain".to_string()),
            indexes: vec![HttpIndex {
                name: "age_int".to_string(),
                value: "7".to_string(),
            }],
        };
        let bytes = legacy.encode_to_vec();
        let obj = HttpObject::from_storage_bytes(&bytes).expect("decode legacy");
        assert_eq!(obj.value, b"old");
        assert_eq!(obj.indexes.len(), 1);
        assert!(obj.links.is_empty());
    }

    #[test]
    fn corrupt_storage_form_is_an_error() {
        // A length-delimited field (tag 1, wire type 2) with a length
        // that overruns the buffer is a hard protobuf decode error.
        let err = HttpObject::from_storage_bytes(&[0x0a, 0xff]);
        assert!(err.is_err());
    }

    #[test]
    fn cross_encoding_preserves_logical_object() {
        // Encode through JSON, decode the JSON, re-encode through CBOR
        // and protobuf, and confirm every hop reconstructs the same
        // logical object.
        let obj = fixture();
        let registry = object_codecs();

        let json = registry.for_content_type("application/json").expect("json");
        let cbor = registry.for_content_type("application/cbor").expect("cbor");
        let pb = registry
            .for_content_type("application/x-protobuf")
            .expect("protobuf");

        let json_bytes = json.encode(&obj).expect("json encode");
        let from_json = json
            .decode(HttpObject::wire_type_id(), &json_bytes)
            .expect("json decode");
        let from_json = from_json
            .as_any()
            .downcast_ref::<HttpObject>()
            .expect("downcast json");
        assert_eq!(from_json, &obj);

        let cbor_bytes = cbor.encode(from_json).expect("cbor encode");
        let from_cbor = cbor
            .decode(HttpObject::wire_type_id(), &cbor_bytes)
            .expect("cbor decode");
        let from_cbor = from_cbor
            .as_any()
            .downcast_ref::<HttpObject>()
            .expect("downcast cbor");
        assert_eq!(from_cbor, &obj);

        let pb_bytes = pb.encode(from_cbor).expect("pb encode");
        let from_pb = pb
            .decode(HttpObject::wire_type_id(), &pb_bytes)
            .expect("pb decode");
        let from_pb = from_pb
            .as_any()
            .downcast_ref::<HttpObject>()
            .expect("downcast pb");
        assert_eq!(from_pb, &obj);
    }

    #[test]
    fn index_pairs_render_name_value_bytes() {
        let obj = fixture();
        let pairs = obj.index_pairs();
        assert_eq!(
            pairs,
            vec![
                (b"age_int".to_vec(), b"42".to_vec()),
                (b"city_bin".to_vec(), b"seattle".to_vec()),
            ]
        );
    }

    #[test]
    fn object_codecs_is_stable_across_calls() {
        let a = object_codecs();
        let b = object_codecs();
        assert!(std::ptr::eq(a, b));
    }
}