ruststream 0.7.5

//! Protocol bindings: what a broker adds to the document that no broker-agnostic core can know.
//!
//! The specification keeps a closed list of protocols and, per protocol, a binding object for each
//! of the four levels (server, channel, operation, message). A broker crate fills the levels it
//! has something to say about; the core carries the values without ever naming a broker's field.

use std::collections::BTreeMap;

use serde::ser::SerializeMap;
use serde::{Serialize, Serializer};
use serde_json::Value;
use thiserror::Error;

/// The protocol keys the specification allows in a bindings object, in its own order.
///
/// A key outside this list is rejected at construction: the document would carry an object no
/// tool can read, and the mistake surfaces at startup rather than in a published file.
const PROTOCOLS: &[&str] = &[
    "amqp",
    "amqp1",
    "anypointmq",
    "googlepubsub",
    "http",
    "ibmmq",
    "jms",
    "kafka",
    "mercure",
    "mqtt",
    "mqtt5",
    "nats",
    "pulsar",
    "redis",
    "ros2",
    "sns",
    "solace",
    "sqs",
    "stomp",
    "websockets",
];

/// One binding object: the protocol it belongs to, the binding version, and the body.
///
/// The body is whatever the broker's own descriptor or policy knows, serialized once at
/// construction. The core writes `bindingVersion` itself, so a broker cannot forget it, and the
/// protocol key is checked against the specification's list.
///
/// # Examples
///
/// ```
/// use ruststream::asyncapi::{Binding, BindingError, Bindings};
/// use ruststream::{DescribeServer, ServerSpec};
/// use serde::Serialize;
/// # use ruststream::{Broker, ConnectedBroker};
/// # struct ConnectedKafka;
/// # impl Broker for KafkaBroker {
/// #     type Error = std::io::Error;
/// #     type Connected = ConnectedKafka;
/// #     async fn connect(self) -> Result<ConnectedKafka, Self::Error> { Ok(ConnectedKafka) }
/// # }
/// # impl ConnectedBroker for ConnectedKafka {
/// #     type Error = std::io::Error;
/// #     type Closed = ();
/// #     async fn shutdown(self) -> Result<(), Self::Error> { Ok(()) }
/// # }
///
/// struct KafkaBroker {
///     url: String,
///     bindings: Bindings,
/// }
///
/// #[derive(Serialize)]
/// struct KafkaServer<'a> {
///     #[serde(rename = "schemaRegistryUrl")]
///     schema_registry_url: &'a str,
/// }
///
/// impl KafkaBroker {
///     /// The binding is built with the broker, so a bad one fails construction.
///     fn new(url: &str, registry: &str) -> Result<Self, BindingError> {
///         let body = KafkaServer { schema_registry_url: registry };
///         let binding = Binding::new("kafka", "0.5.0", &body)?;
///         Ok(Self { url: url.to_owned(), bindings: Bindings::new().with(binding) })
///     }
/// }
///
/// impl DescribeServer for KafkaBroker {
///     fn describe_server(&self) -> ServerSpec {
///         ServerSpec::from_url(&self.url, "kafka").bindings(self.bindings.clone())
///     }
/// }
/// # KafkaBroker::new("kafka://broker:9092", "http://registry:8081")?;
/// # Ok::<_, BindingError>(())
/// ```
///
/// # Errors
///
/// See [`BindingError`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Binding {
    key: &'static str,
    body: String,
}

impl Binding {
    /// A binding for `protocol`, at `version`, carrying `body`.
    ///
    /// `body` must serialize to a JSON object: a binding object holds named fields, and
    /// `bindingVersion` is written into it here.
    ///
    /// # Errors
    ///
    /// [`BindingError::UnknownProtocol`] when `protocol` is not one the specification lists,
    /// [`BindingError::NotAnObject`] when `body` does not serialize to an object, and
    /// [`BindingError::Serialize`] when serializing it fails.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::asyncapi::{Binding, BindingError, Bindings};
    /// use serde::Serialize;
    ///
    /// #[derive(Serialize)]
    /// struct MqttServer<'a> {
    ///     #[serde(rename = "clientId")]
    ///     client_id: &'a str,
    /// }
    ///
    /// /// The server bindings an MQTT broker crate describes its connection with.
    /// fn server_bindings(client_id: &str) -> Result<Bindings, BindingError> {
    ///     let binding = Binding::new("mqtt", "0.2.0", &MqttServer { client_id })?;
    ///     Ok(Bindings::new().with(binding))
    /// }
    /// ```
    pub fn new<T: Serialize>(
        protocol: &'static str,
        version: &'static str,
        body: &T,
    ) -> Result<Self, BindingError> {
        if !PROTOCOLS.contains(&protocol) {
            return Err(BindingError::UnknownProtocol { protocol });
        }
        let mut value = serde_json::to_value(body)?;
        let object = value
            .as_object_mut()
            .ok_or(BindingError::NotAnObject { key: protocol })?;
        object.insert(
            "bindingVersion".to_owned(),
            Value::String(version.to_owned()),
        );
        Ok(Self {
            key: protocol,
            body: serde_json::to_string(&value)?,
        })
    }

    /// An `x-` extension at the level a binding would sit at, for a transport the specification
    /// has no binding for.
    ///
    /// `ZeroMQ`, Kinesis and a file transport have no binding object of their own, and the
    /// protocol keys are a closed list, so an extension is the only lawful place for what such a
    /// transport knows. There is no `bindingVersion` here: the field belongs to a binding, and
    /// an extension is not one.
    ///
    /// # Errors
    ///
    /// [`BindingError::NotAnExtension`] when `name` does not start with `x-` or carries a
    /// character the specification's extension pattern forbids, [`BindingError::NotAnObject`]
    /// when `body` does not serialize to an object, and [`BindingError::Serialize`] when
    /// serializing it fails.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::asyncapi::{Binding, BindingError, Bindings};
    /// use ruststream::{DescribeServer, ServerSpec};
    /// use serde::Serialize;
    /// # use ruststream::{Broker, ConnectedBroker};
    /// # struct ConnectedZmq;
    /// # impl Broker for ZmqBroker {
    /// #     type Error = std::io::Error;
    /// #     type Connected = ConnectedZmq;
    /// #     async fn connect(self) -> Result<ConnectedZmq, Self::Error> { Ok(ConnectedZmq) }
    /// # }
    /// # impl ConnectedBroker for ConnectedZmq {
    /// #     type Error = std::io::Error;
    /// #     type Closed = ();
    /// #     async fn shutdown(self) -> Result<(), Self::Error> { Ok(()) }
    /// # }
    ///
    /// struct ZmqBroker {
    ///     endpoint: String,
    ///     bindings: Bindings,
    /// }
    ///
    /// #[derive(Serialize)]
    /// struct ZmqSocket {
    ///     socket: &'static str,
    /// }
    ///
    /// impl ZmqBroker {
    ///     fn new(endpoint: &str) -> Result<Self, BindingError> {
    ///         // ZeroMQ has no binding object, so its socket type rides an extension.
    ///         let extension = Binding::extension("x-zeromq", &ZmqSocket { socket: "PULL" })?;
    ///         Ok(Self { endpoint: endpoint.to_owned(), bindings: Bindings::new().with(extension) })
    ///     }
    /// }
    ///
    /// impl DescribeServer for ZmqBroker {
    ///     fn describe_server(&self) -> ServerSpec {
    ///         ServerSpec::new(ServerSpec::host_from_url(&self.endpoint), "zmq")
    ///             .bindings(self.bindings.clone())
    ///     }
    /// }
    /// # ZmqBroker::new("tcp://feed:5555")?;
    /// # Ok::<_, BindingError>(())
    /// ```
    pub fn extension<T: Serialize>(name: &'static str, body: &T) -> Result<Self, BindingError> {
        let legal = name.starts_with("x-")
            && name.len() > 2
            && name[2..]
                .chars()
                .all(|c| c.is_alphanumeric() || matches!(c, '_' | '-' | '.'));
        if !legal {
            return Err(BindingError::NotAnExtension { name });
        }
        let value = serde_json::to_value(body)?;
        if !value.is_object() {
            return Err(BindingError::NotAnObject { key: name });
        }
        Ok(Self {
            key: name,
            body: serde_json::to_string(&value)?,
        })
    }

    /// The protocol key this binding sits under, or the extension name.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::asyncapi::Binding;
    /// use std::collections::BTreeMap;
    ///
    /// let binding = Binding::new("redis", "0.1.0", &BTreeMap::<String, String>::new())?;
    /// assert_eq!(binding.protocol(), "redis");
    /// # Ok::<_, ruststream::asyncapi::BindingError>(())
    /// ```
    #[must_use]
    pub const fn protocol(&self) -> &'static str {
        self.key
    }
}

/// The bindings of one level: one body per protocol.
///
/// Empty by default, and an empty set never reaches the document. A broker that says nothing
/// changes nothing.
///
/// # Where a broker crate fills them
///
/// - The server level is a field: [`ServerSpec::bindings`](crate::ServerSpec::bindings) in the
///   broker's [`DescribeServer`](crate::DescribeServer) impl, because a server is described once
///   per broker.
/// - A subscription descriptor fills its channel, its `receive` operation and its messages
///   ([`SubscriptionSource::channel_bindings`](crate::SubscriptionSource::channel_bindings) and
///   its two siblings). A subscription opened by a bare name carries none, so a broker that wants
///   bindings ships a descriptor.
/// - A publish policy fills the same three for the channel it publishes to
///   ([`PublishPolicy::channel_bindings`](crate::PublishPolicy::channel_bindings) and its
///   siblings), handed the destination the mount site resolved: a reply's name, a slot entry's
///   own name, a `dead_letter(..)` destination. A reply has no `send` operation of its own, so a
///   reply policy's operation bindings reach no document.
///   [`reply_address_location`](crate::PublishPolicy::reply_address_location) tells a client
///   where an answer routed through a reply-to header is addressed.
///
/// Each hook computes its value from the descriptor or policy alone and never carries a
/// credential: configure the broker, a descriptor and a policy with a known password, and
/// `conformance::harness::describes_without_credentials`,
/// `conformance::message_shape::publishes_without_credentials` and
/// `describes_addresses_without_credentials` scan the description for it. A protocol the specification does not
/// list fills nothing on the publish side either: the in-memory broker's `MemoryPublish` stays
/// silent rather than invent a key.
///
/// The hooks are gated on the core's `asyncapi` feature. A broker crate forwards it
/// (`asyncapi = ["ruststream/asyncapi"]`) and puts `#[cfg(feature = "asyncapi")]` on each method
/// it fills in.
///
/// # Examples
///
/// ```
/// use ruststream::asyncapi::{Binding, BindingError, Bindings};
/// use ruststream::{DescribeServer, ServerSpec};
/// use serde::Serialize;
/// # use ruststream::{Broker, ConnectedBroker};
/// # struct ConnectedKafka;
/// # impl Broker for KafkaBroker {
/// #     type Error = std::io::Error;
/// #     type Connected = ConnectedKafka;
/// #     async fn connect(self) -> Result<ConnectedKafka, Self::Error> { Ok(ConnectedKafka) }
/// # }
/// # impl ConnectedBroker for ConnectedKafka {
/// #     type Error = std::io::Error;
/// #     type Closed = ();
/// #     async fn shutdown(self) -> Result<(), Self::Error> { Ok(()) }
/// # }
///
/// struct KafkaBroker {
///     url: String,
///     bindings: Bindings,
/// }
///
/// #[derive(Serialize)]
/// struct KafkaServer<'a> {
///     #[serde(rename = "schemaRegistryUrl")]
///     schema_registry_url: &'a str,
/// }
///
/// impl KafkaBroker {
///     fn new(url: &str, registry: Option<&str>) -> Result<Self, BindingError> {
///         // Without a schema registry there is nothing to say, and the empty set leaves the
///         // server's bindings out of the document.
///         let mut bindings = Bindings::new();
///         if let Some(registry) = registry {
///             let body = KafkaServer { schema_registry_url: registry };
///             bindings = bindings.with(Binding::new("kafka", "0.5.0", &body)?);
///         }
///         Ok(Self { url: url.to_owned(), bindings })
///     }
/// }
///
/// impl DescribeServer for KafkaBroker {
///     fn describe_server(&self) -> ServerSpec {
///         ServerSpec::from_url(&self.url, "kafka").bindings(self.bindings.clone())
///     }
/// }
/// # KafkaBroker::new("kafka://broker:9092", None)?;
/// # Ok::<_, BindingError>(())
/// ```
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Bindings(BTreeMap<&'static str, String>);

impl Bindings {
    /// An empty set.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::asyncapi::{Binding, BindingError, Bindings};
    /// use serde::Serialize;
    ///
    /// #[derive(Serialize)]
    /// struct SqsQueue<'a> {
    ///     name: &'a str,
    /// }
    ///
    /// /// What an SQS queue descriptor says about its channel: the queue name, or nothing at all.
    /// fn channel_bindings(queue: Option<&str>) -> Result<Bindings, BindingError> {
    ///     let Some(name) = queue else {
    ///         return Ok(Bindings::new());
    ///     };
    ///     Ok(Bindings::new().with(Binding::new("sqs", "0.3.0", &SqsQueue { name })?))
    /// }
    /// ```
    #[must_use]
    pub const fn new() -> Self {
        Self(BTreeMap::new())
    }

    /// Adds one binding, replacing whatever was under the same protocol key.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::asyncapi::{Binding, BindingError, Bindings};
    /// use serde::Serialize;
    ///
    /// #[derive(Serialize)]
    /// struct KafkaMessage {
    ///     #[serde(rename = "schemaIdLocation")]
    ///     schema_id_location: &'static str,
    /// }
    ///
    /// #[derive(Serialize)]
    /// struct XRetention {
    ///     days: u32,
    /// }
    ///
    /// /// A topic descriptor's message bindings: the Kafka object, plus what no binding covers.
    /// fn message_bindings() -> Result<Bindings, BindingError> {
    ///     Ok(Bindings::new()
    ///         .with(Binding::new(
    ///             "kafka",
    ///             "0.5.0",
    ///             &KafkaMessage {
    ///                 schema_id_location: "payload",
    ///             },
    ///         )?)
    ///         .with(Binding::extension("x-retention", &XRetention { days: 7 })?))
    /// }
    /// ```
    #[must_use]
    pub fn with(mut self, binding: Binding) -> Self {
        self.0.insert(binding.key, binding.body);
        self
    }

    /// Adds every binding of `other` this set has no entry for, keeping its own on a clash.
    ///
    /// One message component is shared by every registration that carries the type, so two
    /// descriptors may each contribute to it. First one wins per protocol, the way a conflicting
    /// headers schema does.
    pub(crate) fn fill_from(&mut self, other: &Self) {
        for (key, body) in &other.0 {
            self.0.entry(key).or_insert_with(|| body.clone());
        }
    }

    /// True when nothing was added, so the document leaves the object out.
    ///
    /// # Examples
    ///
    /// ```
    /// use ruststream::asyncapi::{Binding, BindingError, Bindings};
    /// use ruststream::{DescribeServer, ServerSpec};
    /// use serde::Serialize;
    /// # use ruststream::{Broker, ConnectedBroker};
    /// # struct ConnectedKafka;
    /// # impl Broker for KafkaBroker {
    /// #     type Error = std::io::Error;
    /// #     type Connected = ConnectedKafka;
    /// #     async fn connect(self) -> Result<ConnectedKafka, Self::Error> { Ok(ConnectedKafka) }
    /// # }
    /// # impl ConnectedBroker for ConnectedKafka {
    /// #     type Error = std::io::Error;
    /// #     type Closed = ();
    /// #     async fn shutdown(self) -> Result<(), Self::Error> { Ok(()) }
    /// # }
    /// # #[derive(Serialize)]
    /// # struct KafkaServer<'a> {
    /// #     #[serde(rename = "schemaRegistryUrl")]
    /// #     schema_registry_url: &'a str,
    /// # }
    /// # struct KafkaBroker { url: String, bindings: Bindings }
    /// # impl KafkaBroker {
    /// #     fn new(url: &str) -> Result<Self, BindingError> {
    /// #         let body = KafkaServer { schema_registry_url: "http://registry:8081" };
    /// #         let binding = Binding::new("kafka", "0.5.0", &body)?;
    /// #         Ok(Self { url: url.to_owned(), bindings: Bindings::new().with(binding) })
    /// #     }
    /// # }
    /// # impl DescribeServer for KafkaBroker {
    /// #     fn describe_server(&self) -> ServerSpec {
    /// #         ServerSpec::from_url(&self.url, "kafka").bindings(self.bindings.clone())
    /// #     }
    /// # }
    ///
    /// // A broker crate's check that its server description reaches the document.
    /// let broker = KafkaBroker::new("kafka://broker:9092")?;
    /// assert!(!broker.describe_server().bindings.is_empty());
    /// # Ok::<_, BindingError>(())
    /// ```
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
}

impl Serialize for Bindings {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let mut map = serializer.serialize_map(Some(self.0.len()))?;
        for (key, body) in &self.0 {
            // The body round-trips through the string the constructor serialized, so parsing it
            // back cannot fail; Null is the unreachable fallback, not an error path.
            let value = serde_json::from_str::<Value>(body).unwrap_or(Value::Null);
            map.serialize_entry(key, &value)?;
        }
        map.end()
    }
}

/// What one subscription contributes to the document, one set per level.
///
/// Assembled by the runtime at mount time from a
/// [`SubscriptionSource`](crate::SubscriptionSource), and carried in the registration's
/// metadata until the document is built.
///
/// # Examples
///
/// ```
/// use ruststream::asyncapi::SubscriptionBindings;
///
/// let bindings = SubscriptionBindings::default();
///
/// assert!(bindings.channel.is_empty());
/// assert!(bindings.operation.is_empty());
/// assert!(bindings.message.is_empty());
/// ```
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct SubscriptionBindings {
    /// What the subscription says about its channel.
    pub channel: Bindings,
    /// What it says about the `receive` operation.
    pub operation: Bindings,
    /// What it says about the messages that arrive on it.
    pub message: Bindings,
}

/// What one publish position contributes to the document, one set per level.
///
/// Assembled by the runtime at mount time from the [`PublishPolicy`](crate::PublishPolicy) bound
/// on the mount chain - a reply, an [`Out`](crate::runtime::Out) slot, the publisher a dead-lettered
/// delivery leaves through - and carried in the registration's metadata until the document is
/// built.
///
/// A reply has no `send` operation of its own (it is the `reply` of the operation it answers, and
/// the specification gives that object no bindings), so on a reply the operation set is dropped
/// and the channel and message sets apply.
///
/// # Examples
///
/// ```
/// use ruststream::asyncapi::PublishBindings;
///
/// let bindings = PublishBindings::default();
///
/// assert!(bindings.channel.is_empty());
/// assert!(bindings.operation.is_empty());
/// assert!(bindings.message.is_empty());
/// ```
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct PublishBindings {
    /// What the policy says about the channel it publishes to.
    pub channel: Bindings,
    /// What it says about the `send` operation.
    pub operation: Bindings,
    /// What it says about the messages that leave through it.
    pub message: Bindings,
}

/// Why a value is not a [`Binding`].
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum BindingError {
    /// The protocol key is not one the specification lists.
    #[error(
        "`{protocol}` is not an AsyncAPI protocol key; a transport the specification has no \
         binding for goes in an `x-` extension instead (Binding::extension)"
    )]
    UnknownProtocol {
        /// The key that was offered.
        protocol: &'static str,
    },
    /// The extension name does not match the specification's `x-` pattern.
    #[error(
        "`{name}` is not an extension name; it must start with `x-` and continue with letters, \
         digits, `_`, `-` or `.`"
    )]
    NotAnExtension {
        /// The name that was offered.
        name: &'static str,
    },
    /// The body did not serialize to a JSON object.
    #[error("the body of `{key}` must serialize to a JSON object: a binding holds named fields")]
    NotAnObject {
        /// The protocol key or extension name the body was offered under.
        key: &'static str,
    },
    /// Serializing the body failed.
    #[error("serializing a binding body failed: {0}")]
    Serialize(#[from] serde_json::Error),
}

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

    #[derive(Serialize)]
    struct Group {
        #[serde(rename = "groupId")]
        group_id: &'static str,
    }

    #[test]
    fn an_unlisted_protocol_is_refused_at_construction() {
        let error = Binding::new("kinesis", "0.1.0", &Group { group_id: "x" }).unwrap_err();
        assert!(matches!(
            error,
            BindingError::UnknownProtocol {
                protocol: "kinesis"
            }
        ));
    }

    #[test]
    fn an_extension_needs_the_x_prefix_and_carries_no_binding_version() {
        assert!(matches!(
            Binding::extension("kinesis", &Group { group_id: "x" }).unwrap_err(),
            BindingError::NotAnExtension { name: "kinesis" },
        ));
        assert!(matches!(
            Binding::extension("x-", &Group { group_id: "x" }).unwrap_err(),
            BindingError::NotAnExtension { name: "x-" },
        ));

        let bindings = Bindings::new()
            .with(Binding::extension("x-kinesis", &Group { group_id: "x" }).unwrap());
        let json = serde_json::to_value(&bindings).unwrap();
        assert_eq!(json["x-kinesis"]["groupId"], "x");
        assert!(json["x-kinesis"].get("bindingVersion").is_none());
    }

    #[test]
    fn a_body_that_is_not_an_object_has_no_binding_to_be() {
        assert!(matches!(
            Binding::new("kafka", "0.5.0", &"a string").unwrap_err(),
            BindingError::NotAnObject { key: "kafka" },
        ));
    }

    #[test]
    fn one_protocol_carries_one_body() {
        let bindings = Bindings::new()
            .with(Binding::new("kafka", "0.5.0", &Group { group_id: "first" }).unwrap())
            .with(Binding::new("kafka", "0.5.0", &Group { group_id: "second" }).unwrap());
        let json = serde_json::to_value(&bindings).unwrap();

        assert_eq!(json["kafka"]["groupId"], "second");
    }
}