dusk-rusk 1.6.0

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
//
// Copyright (c) DUSK NETWORK. All rights reserved.

use futures_util::stream::Iter as StreamIter;
use futures_util::{Stream, stream};
use http_body_util::{
    BodyExt, Either, Full, LengthLimitError, Limited, StreamBody,
};
use hyper::body::{Body, Bytes, Frame, Incoming};
use hyper::header::{InvalidHeaderName, InvalidHeaderValue};
use hyper::{Request, Response};
use pin_project::pin_project;
use rand::Rng;
use rand::distributions::{Distribution, Standard};
use semver::{Prerelease, Version, VersionReq};
use serde::{Deserialize, Serialize};
use serde_with::As;
use serde_with::hex::Hex;
use std::collections::HashMap;
use std::fmt::{Display, Formatter};
use std::pin::Pin;
use std::str::FromStr;
use std::sync::mpsc;
use std::task::{Context, Poll};
use tungstenite::http::HeaderValue;

use super::{RUSK_VERSION_HEADER, RUSK_VERSION_STRICT_HEADER};

#[derive(Debug, Deserialize, Serialize)]
pub struct MessageResponse {
    pub headers: serde_json::Map<String, serde_json::Value>,

    /// The data returned by the contract call.
    pub data: DataType,

    /// A possible error happening during the contract call.
    pub error: Option<(String, u16)>,

    pub force_binary: bool,
}

impl MessageResponse {
    // ExecutionError is intentionally large; boxing it would add complexity
    // without meaningful benefit here.
    #[allow(clippy::result_large_err)]
    pub fn into_http(
        self,
        is_binary: bool,
    ) -> Result<Response<FullOrStreamBody>, ExecutionError> {
        if let Some((error, code)) = self.error {
            let code = hyper::StatusCode::from_u16(code)
                .unwrap_or(hyper::StatusCode::INTERNAL_SERVER_ERROR);
            return Ok(hyper::Response::builder()
                .status(code)
                .body(Full::new(error.into()).into())?);
        }

        let mut headers = HashMap::new();

        let body = {
            match self.data {
                DataType::Binary(wrapper) => {
                    let data = match is_binary {
                        true => wrapper.inner,
                        false => hex::encode(wrapper.inner).as_bytes().to_vec(),
                    };
                    Full::from(Bytes::from(data)).into()
                }
                DataType::Text(text) => Full::from(Bytes::from(text)).into(),
                DataType::Json(value) => {
                    headers.insert(
                        CONTENT_TYPE,
                        HeaderValue::from_static(CONTENT_TYPE_JSON),
                    );
                    Full::from(Bytes::from(value.to_string())).into()
                }
                DataType::Channel(receiver) => FullOrStreamBody {
                    either: Either::Right(StreamBody::new(
                        BinaryOrTextStream {
                            hex: !is_binary,
                            stream: stream::iter(receiver),
                        },
                    )),
                },
                DataType::JsonChannel(receiver) => {
                    headers.insert(
                        CONTENT_TYPE,
                        HeaderValue::from_static(CONTENT_TYPE_JSON),
                    );
                    FullOrStreamBody {
                        either: Either::Right(StreamBody::new(
                            BinaryOrTextStream {
                                hex: false,
                                stream: stream::iter(receiver),
                            },
                        )),
                    }
                }
                DataType::None => Full::new(Bytes::new()).into(),
            }
        };
        let mut response = Response::new(body);
        for (k, v) in headers {
            response.headers_mut().insert(k, v);
        }

        Ok(response)
    }

    pub fn set_header(&mut self, key: &str, value: serde_json::Value) {
        // search for the key in a case-insensitive way
        let v = self
            .headers
            .iter_mut()
            .find_map(|(k, v)| k.eq_ignore_ascii_case(key).then_some(v));

        if let Some(v) = v {
            *v = value;
        } else {
            self.headers.insert(key.into(), value);
        }
    }
}

#[pin_project]
pub struct FullOrStreamBody {
    #[pin]
    either: Either<Full<Bytes>, StreamBody<BinaryOrTextStream>>,
}

impl From<Full<Bytes>> for FullOrStreamBody {
    fn from(body: Full<Bytes>) -> Self {
        Self {
            either: Either::Left(body),
        }
    }
}

impl Body for FullOrStreamBody {
    type Data =
        <Either<Full<Bytes>, StreamBody<BinaryOrTextStream>> as Body>::Data;
    type Error =
        <Either<Full<Bytes>, StreamBody<BinaryOrTextStream>> as Body>::Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        let this = self.project();
        this.either.poll_frame(cx)
    }
}

#[pin_project]
pub struct BinaryOrTextStream {
    hex: bool,
    #[pin]
    stream: StreamIter<<mpsc::Receiver<Vec<u8>> as IntoIterator>::IntoIter>,
}

impl Stream for BinaryOrTextStream {
    type Item = Result<Frame<Bytes>, std::convert::Infallible>;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        let this = self.project();
        this.stream.poll_next(cx).map(|next| {
            next.map(|x| match this.hex {
                true => Ok(Frame::data(Bytes::from(
                    hex::encode(x).as_bytes().to_vec(),
                ))),
                false => Ok(Frame::data(Bytes::from(x))),
            })
        })
    }
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)]
pub enum RequestData {
    Binary(BinaryWrapper),
    Text(String),
}

impl RequestData {
    pub fn as_bytes(&self) -> &[u8] {
        match self {
            Self::Binary(w) => &w.inner,
            Self::Text(s) => s.as_bytes(),
        }
    }

    pub fn as_string(&self) -> String {
        match self {
            Self::Binary(w) => {
                String::from_utf8(w.inner.clone()).unwrap_or_default()
            }
            Self::Text(s) => s.clone(),
        }
    }
}

impl From<String> for RequestData {
    fn from(value: String) -> Self {
        Self::Text(value)
    }
}
impl From<Vec<u8>> for RequestData {
    fn from(value: Vec<u8>) -> Self {
        Self::Binary(BinaryWrapper { inner: value })
    }
}

#[derive(Debug, Clone, Eq, PartialEq)]
pub struct ResponseData {
    data: DataType,
    header: serde_json::Map<String, serde_json::Value>,
    force_binary: bool,
}

impl ResponseData {
    pub fn new<D: Into<DataType>>(data: D) -> Self {
        Self {
            data: data.into(),
            header: serde_json::Map::new(),
            force_binary: false,
        }
    }

    pub fn add_header<K: Into<String>, V: Into<serde_json::Value>>(
        &mut self,
        key: K,
        value: V,
    ) {
        self.header.insert(key.into(), value.into());
    }

    pub fn with_header<K: Into<String>, V: Into<serde_json::Value>>(
        mut self,
        key: K,
        value: V,
    ) -> Self {
        self.add_header(key, value);
        self
    }

    pub fn into_inner(
        self,
    ) -> (DataType, serde_json::Map<String, serde_json::Value>, bool) {
        (self.data, self.header, self.force_binary)
    }

    pub fn data(&self) -> &DataType {
        &self.data
    }

    pub fn with_force_binary(mut self, force: bool) -> Self {
        self.force_binary = force;
        self
    }
}

/// Data in a response.
#[derive(Debug, Serialize, Deserialize, Default)]
#[serde(untagged)]
pub enum DataType {
    Binary(BinaryWrapper),
    Text(String),
    Json(serde_json::Value),
    #[serde(skip)]
    Channel(mpsc::Receiver<Vec<u8>>),
    #[serde(skip)]
    JsonChannel(mpsc::Receiver<Vec<u8>>),
    #[default]
    None,
}

impl Eq for DataType {}

impl PartialEq for DataType {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Self::Channel(_), Self::Channel(_)) => true,
            (Self::Text(a), Self::Text(b)) => a == b,
            (Self::Json(a), Self::Json(b)) => a == b,
            (Self::Binary(a), Self::Binary(b)) => a == b,
            (Self::None, Self::None) => true,
            _ => false,
        }
    }
}

impl Clone for DataType {
    fn clone(&self) -> Self {
        match self {
            Self::Binary(b) => b.inner.clone().into(),
            Self::Text(s) => s.clone().into(),
            Self::Json(v) => v.clone().into(),
            _ => Self::None,
        }
    }
}

impl DataType {
    pub fn to_bytes(&self) -> Vec<u8> {
        match self {
            Self::Binary(b) => b.inner.clone(),
            Self::Text(s) => s.as_bytes().to_vec(),
            Self::Json(s) => s.to_string().as_bytes().to_vec(),
            _ => vec![],
        }
    }
}

impl From<serde_json::Value> for DataType {
    fn from(value: serde_json::Value) -> Self {
        Self::Json(value)
    }
}

impl From<String> for DataType {
    fn from(text: String) -> Self {
        Self::Text(text)
    }
}

impl From<Vec<u8>> for DataType {
    fn from(bytes: Vec<u8>) -> Self {
        Self::Binary(BinaryWrapper { inner: bytes })
    }
}

impl From<mpsc::Receiver<Vec<u8>>> for DataType {
    fn from(receiver: mpsc::Receiver<Vec<u8>>) -> Self {
        Self::Channel(receiver)
    }
}

#[derive(Debug, Deserialize, Serialize, Eq, PartialEq)]
#[serde(transparent)]
pub struct BinaryWrapper {
    #[serde(with = "As::<Hex>")]
    pub inner: Vec<u8>,
}

const CONTENT_TYPE: &str = "content-type";
const ACCEPT: &str = "accept";
const CONTENT_TYPE_BINARY: &str = "application/octet-stream";
const CONTENT_TYPE_JSON: &str = "application/json";

#[derive(Debug, thiserror::Error)]
pub enum ExecutionError {
    #[error("{0}")]
    Http(#[from] hyper::http::Error),
    #[error("{0}")]
    Hyper(#[from] hyper::Error),
    #[error("{0}")]
    Json(#[from] serde_json::Error),
    #[error("{0}")]
    Protocol(#[from] tungstenite::error::ProtocolError),
    #[error("{0}")]
    Tungstenite(#[from] tungstenite::Error),
    #[error("Invalid header: {0}")]
    InvalidHeader(String),
    #[error("{0}")]
    Other(String),
}

impl From<InvalidHeaderName> for ExecutionError {
    fn from(value: InvalidHeaderName) -> Self {
        Self::InvalidHeader(value.to_string())
    }
}

impl From<InvalidHeaderValue> for ExecutionError {
    fn from(value: InvalidHeaderValue) -> Self {
        Self::InvalidHeader(value.to_string())
    }
}

impl From<super::HttpError> for ExecutionError {
    fn from(e: super::HttpError) -> Self {
        Self::Other(e.to_string())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct SessionId(u128);

impl Display for SessionId {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        let bytes = self.0.to_le_bytes();
        let hex = hex::encode(bytes);
        write!(f, "{hex}")
    }
}

impl Distribution<SessionId> for Standard {
    fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> SessionId {
        SessionId(rng.r#gen())
    }
}

impl SessionId {
    /// Parses a session ID from a request. The session ID is expected to be
    /// stored in the `Rusk-Session-Id` header.
    pub fn parse_from_req<B>(req: &Request<B>) -> Option<Self> {
        let headers = req.headers();

        let header_value = headers.get("Rusk-Session-Id")?;
        let text = header_value.to_str().ok()?;

        Self::parse(text)
    }

    pub fn parse(text: &str) -> Option<Self> {
        let bytes = hex::decode(text).ok()?;

        let mut session_id_bytes = [0u8; 16];
        if bytes.len() != 16 {
            return None;
        }

        session_id_bytes.copy_from_slice(&bytes);
        Some(SessionId(u128::from_le_bytes(session_id_bytes)))
    }
}

/// A subscription to an event.
///
/// Subscriptions have data related to the component the subscriber wishes to
/// subscribe to, the component targeted by the event (`contracts`,
/// `transactions`, etc...) and an optional entity within the component that
/// the event targets.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize)]
pub struct RuesEventUri {
    pub component: String,
    pub entity: Option<String>,
    pub topic: String,
}

pub const RUES_LOCATION_PREFIX: &str = "/on";

impl Display for RuesEventUri {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        let component = &self.component;
        let entity = self
            .entity
            .as_ref()
            .map(|e| format!(":{e}"))
            .unwrap_or_default();
        let topic = &self.topic;

        write!(f, "{RUES_LOCATION_PREFIX}/{component}{entity}/{topic}")
    }
}

impl RuesEventUri {
    pub fn inner(&self) -> (&str, Option<&String>, &str) {
        (
            self.component.as_ref(),
            self.entity.as_ref(),
            self.topic.as_ref(),
        )
    }

    /// Parses a RUES URI from an URL path.
    ///
    /// # Format
    ///
    /// `/on/component[:entity]/topic`
    ///
    /// # Rules
    ///
    /// - Path must start with `/on`
    /// - Component and topic are normalized to lowercase
    /// - Topic is mandatory and cannot be empty
    /// - Entity is optional (for blocks & transactions), separated from
    ///   component by `:`
    /// - Entity can contain colons (only first `:` is used as delimiter)
    /// - `contracts` component requires an entity (contract ID)
    ///
    /// # Returns
    ///
    /// `None` if parsing fails or validation rules are violated.
    pub fn parse_from_path(path: &str) -> Option<Self> {
        // Strip the required prefix, returning None if not present
        // Note: we already know its present
        let path = path.strip_prefix(RUES_LOCATION_PREFIX)?;

        let mut segments = path.split('/');

        // Skip the empty segment before first '/' (path starts with "/")
        segments.next()?;

        // Parse component and optional entity from first segment
        // Format: "component" or "component:entity"
        let first_segment = segments.next()?;
        let (component, entity) = match first_segment.split_once(':') {
            Some((comp, ent)) => (comp, Some(ent.to_string())),
            None => (first_segment, None),
        };
        let component = component.to_lowercase();

        // Parse topic (mandatory, must be non-empty)
        let topic = segments.next().filter(|t| !t.is_empty())?;
        let topic = topic.to_lowercase();

        // Contracts require an entity (contract ID)
        if component == "contracts" && entity.is_none() {
            return None;
        }

        Some(Self {
            component,
            entity,
            topic,
        })
    }

    /// Returns `true` if this subscription URI matches the given event.
    ///
    /// Matching rules:
    /// - Component must match exactly
    /// - Entity: `None` acts as wildcard (matches any), `Some` must match
    ///   exactly
    /// - Topic must match exactly
    pub fn matches(&self, event: &RuesEvent) -> bool {
        let event = &event.uri;

        // Component must match exactly
        let component_matches = self.component == event.component;

        // None entity acts as wildcard, Some must match exactly
        let entity_matches =
            self.entity.is_none() || self.entity == event.entity;

        // Topic must match exactly
        let topic_matches = self.topic == event.topic;

        component_matches && entity_matches && topic_matches
    }
}

/// A RUES event
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct RuesEvent {
    pub uri: RuesEventUri,
    pub headers: serde_json::Map<String, serde_json::Value>,
    pub data: DataType,
}

/// A RUES Dispatch request event
#[derive(Debug)]
pub struct RuesDispatchEvent {
    pub uri: RuesEventUri,
    pub headers: serde_json::Map<String, serde_json::Value>,
    pub data: RequestData,
}

impl RuesDispatchEvent {
    pub fn x_headers(&self) -> serde_json::Map<String, serde_json::Value> {
        let mut h = self.headers.clone();
        h.retain(|k, _| k.to_lowercase().starts_with("x-"));
        h
    }

    pub fn header(&self, name: &str) -> Option<&serde_json::Value> {
        self.headers
            .iter()
            .find_map(|(k, v)| k.eq_ignore_ascii_case(name).then_some(v))
    }

    pub fn check_rusk_version(&self) -> Result<(), super::HttpError> {
        check_rusk_version(
            self.header(RUSK_VERSION_HEADER),
            self.header(RUSK_VERSION_STRICT_HEADER).is_some(),
        )
    }

    pub fn is_binary(&self) -> bool {
        self.headers
            .get(CONTENT_TYPE)
            .and_then(|h| h.as_str())
            .map(|v| v.eq_ignore_ascii_case(CONTENT_TYPE_BINARY))
            .unwrap_or_default()
    }
    pub fn is_json(&self) -> bool {
        self.headers
            .get(CONTENT_TYPE)
            .and_then(|h| h.as_str())
            .map(|v| v.eq_ignore_ascii_case(CONTENT_TYPE_JSON))
            .unwrap_or_default()
    }
    pub async fn from_request(
        req: Request<Incoming>,
    ) -> Result<(Self, bool), super::HttpError> {
        let (parts, body) = req.into_parts();

        let uri = RuesEventUri::parse_from_path(parts.uri.path())
            .ok_or(super::HttpError::invalid_input("Invalid URL path"))?;

        let headers = parts
            .headers
            .iter()
            .map(|(k, v)| {
                let v = if v.is_empty() {
                    serde_json::Value::Null
                } else {
                    serde_json::from_slice::<serde_json::Value>(v.as_bytes())
                        .unwrap_or(serde_json::Value::String(
                            v.to_str().unwrap().to_string(),
                        ))
                };
                (k.to_string().to_lowercase(), v)
            })
            .collect();

        // HTTP REQUEST
        let content_type = parts
            .headers
            .get(CONTENT_TYPE)
            .and_then(|h| h.to_str().ok())
            .unwrap_or_default();

        let binary_request = content_type == CONTENT_TYPE_BINARY;

        let binary_response = binary_request
            || parts
                .headers
                .get(ACCEPT)
                .and_then(|h| h.to_str().ok())
                .map(|v| v.eq_ignore_ascii_case(CONTENT_TYPE_BINARY))
                .unwrap_or_default();

        let max_body_bytes = super::max_rues_request_body_bytes(&uri);
        let bytes = Limited::new(body, max_body_bytes)
            .collect()
            .await
            .map_err(|e| {
                if e.downcast_ref::<LengthLimitError>().is_some() {
                    super::HttpError::payload_too_large(format!(
                        "Request body exceeds {max_body_bytes} bytes"
                    ))
                } else {
                    super::HttpError::internal(e.to_string())
                }
            })?
            .to_bytes()
            .to_vec();
        let data = match binary_request {
            true => bytes.into(),
            _ => {
                let text = String::from_utf8(bytes).map_err(|_| {
                    super::HttpError::InvalidEncoding(
                        "Invalid utf8".to_string(),
                    )
                })?;
                if let Some(hex) = text.strip_prefix("0x") {
                    if let Ok(bytes) = hex::decode(hex) {
                        bytes.into()
                    } else {
                        text.into()
                    }
                } else {
                    text.into()
                }
            }
        };

        let ret = RuesDispatchEvent { headers, data, uri };

        Ok((ret, binary_response))
    }
}

impl RuesEvent {
    pub fn add_header<K: Into<String>, V: Into<serde_json::Value>>(
        &mut self,
        key: K,
        value: V,
    ) {
        self.headers.insert(key.into(), value.into());
    }

    /// Serialize the event into a vector of bytes.
    pub fn to_bytes(&self) -> Vec<u8> {
        let headers_bytes = serde_json::to_vec(&self.headers)
            .expect("Serializing JSON should succeed");

        let headers_len = headers_bytes.len() as u32;
        let headers_len_bytes = headers_len.to_le_bytes();

        let data_bytes = self.data.to_bytes();

        let len =
            headers_len_bytes.len() + headers_bytes.len() + data_bytes.len();
        let mut bytes = Vec::with_capacity(len);

        bytes.extend(headers_len_bytes);
        bytes.extend(headers_bytes);
        bytes.extend(data_bytes);

        bytes
    }
}

#[cfg(feature = "chain")]
impl From<node_data::events::contract::ContractTxEvent> for RuesEvent {
    fn from(tx_event: node_data::events::contract::ContractTxEvent) -> Self {
        let mut headers = serde_json::Map::new();

        headers
            .insert("Rusk-Origin".into(), hex::encode(tx_event.origin).into());

        let event = tx_event.event;
        Self {
            uri: RuesEventUri {
                component: "contracts".into(),
                entity: Some(hex::encode(event.target.to_bytes())),
                topic: event.topic,
            },
            data: event.data.into(),
            headers,
        }
    }
}

#[cfg(feature = "chain")]
impl From<node_data::events::Event> for RuesEvent {
    fn from(value: node_data::events::Event) -> Self {
        let data = value.data.map_or(DataType::None, DataType::Json);

        Self {
            uri: RuesEventUri {
                component: value.component.into(),
                entity: Some(value.entity),
                topic: value.topic.into(),
            },
            data,
            headers: Default::default(),
        }
    }
}

/// Checks the Rusk version of the incoming request against the current version
/// of the binary.
///
/// If `strict` is `true`, the version check is strict and requires the presence
/// of the version header, otherwise the check for the version header can be
/// optional.
pub fn check_rusk_version(
    version: Option<&serde_json::Value>,
    strict: bool,
) -> Result<(), super::HttpError> {
    if strict && version.is_none() {
        return Err(super::HttpError::VersionMismatch(
            "Missing Rusk-Version header while Rusk-Version-Strict is set"
                .to_string(),
        ));
    }

    if let Some(v) = version {
        let req = match v.as_str() {
            Some(v) => VersionReq::from_str(v),
            None => VersionReq::from_str(&v.to_string()),
        }?;

        let mut current = Version::from_str(&crate::VERSION)?;

        // if client is not requesting a strict check we should ignore the
        // prerelease version of the current binary
        //
        // If instead the client request a strict version we should respect
        // that and check the version as is
        //
        // This solves the issue when connecting to a node that is in`-dev`
        // mode
        if !strict {
            current.pre = Prerelease::EMPTY;
        }

        if !req.matches(&current) {
            return Err(super::HttpError::VersionMismatch(format!(
                "Mismatched rusk version: requested {req} - current {current}",
            )));
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{DataType, RuesEvent, RuesEventUri};

    const DUMMY_ENTITY: &str = "abc123";

    // valid parsing

    #[test]
    fn parse_contracts_with_entity_and_topic() {
        let uri = RuesEventUri::parse_from_path(&format!(
            "/on/contracts:{DUMMY_ENTITY}/withdraw"
        ))
        .expect("Should parse successfully");

        assert_eq!(uri.component, "contracts");
        assert_eq!(uri.entity, Some(DUMMY_ENTITY.to_string()));
        assert_eq!(uri.topic, "withdraw");
    }

    #[test]
    fn parse_blocks_without_entity() {
        // Blocks can omit entity (wildcard)
        let uri = RuesEventUri::parse_from_path("/on/blocks/accepted")
            .expect("Should parse successfully");

        assert_eq!(uri.component, "blocks");
        assert_eq!(uri.entity, None);
        assert_eq!(uri.topic, "accepted");
    }

    #[test]
    fn parse_transactions_without_entity() {
        // Transactions can omit entity (wildcard)
        let uri = RuesEventUri::parse_from_path("/on/transactions/included")
            .expect("Should parse successfully");

        assert_eq!(uri.component, "transactions");
        assert_eq!(uri.entity, None);
        assert_eq!(uri.topic, "included");
    }

    #[test]
    fn parse_normalizes_to_lowercase() {
        let uri = RuesEventUri::parse_from_path("/on/BLOCKS/ACCEPTED")
            .expect("Should parse successfully");

        assert_eq!(uri.component, "blocks");
        assert_eq!(uri.topic, "accepted");
    }

    #[test]
    fn parse_entity_with_colon() {
        // Entity can contain colons (split_once only splits on first colon)
        let uri = RuesEventUri::parse_from_path(
            "/on/contracts:entity:with:colons/topic",
        )
        .expect("Should parse successfully");

        assert_eq!(uri.component, "contracts");
        assert_eq!(uri.entity, Some("entity:with:colons".to_string()));
        assert_eq!(uri.topic, "topic");
    }

    // invalid parsing

    #[test]
    fn parse_contracts_without_entity_fails() {
        // Contracts component requires an entity (contract ID)
        let uri = RuesEventUri::parse_from_path("/on/contracts/withdraw");
        assert!(
            uri.is_none(),
            "Contracts without entity should fail parsing"
        );
    }

    #[test]
    fn parse_empty_topic_with_trailing_slash_fails() {
        let uri = RuesEventUri::parse_from_path(&format!(
            "/on/contracts:{DUMMY_ENTITY}/"
        ));
        assert!(uri.is_none(), "Empty topic should fail parsing");
    }

    #[test]
    fn parse_missing_topic_segment_fails() {
        let uri = RuesEventUri::parse_from_path(&format!(
            "/on/contracts:{DUMMY_ENTITY}"
        ));
        assert!(uri.is_none(), "Missing topic segment should fail parsing");
    }

    #[test]
    fn parse_component_only_fails() {
        let uri = RuesEventUri::parse_from_path("/on/blocks");
        assert!(uri.is_none(), "Missing topic should fail parsing");
    }

    #[test]
    fn parse_invalid_prefix_fails() {
        let uri = RuesEventUri::parse_from_path(&format!(
            "/invalid/contracts:{DUMMY_ENTITY}/topic"
        ));
        assert!(uri.is_none(), "Invalid prefix should fail parsing");
    }

    // matches test

    #[test]
    fn matches_exact_uri() {
        let subscription = RuesEventUri {
            component: "contracts".to_string(),
            entity: Some(DUMMY_ENTITY.to_string()),
            topic: "withdraw".to_string(),
        };

        let event = RuesEvent {
            uri: RuesEventUri {
                component: "contracts".to_string(),
                entity: Some(DUMMY_ENTITY.to_string()),
                topic: "withdraw".to_string(),
            },
            headers: Default::default(),
            data: DataType::None,
        };

        assert!(subscription.matches(&event));
    }

    #[test]
    fn matches_different_component_fails() {
        let subscription = RuesEventUri {
            component: "contracts".to_string(),
            entity: Some(DUMMY_ENTITY.to_string()),
            topic: "withdraw".to_string(),
        };

        let event = RuesEvent {
            uri: RuesEventUri {
                component: "blocks".to_string(),
                entity: Some(DUMMY_ENTITY.to_string()),
                topic: "withdraw".to_string(),
            },
            headers: Default::default(),
            data: DataType::None,
        };

        assert!(!subscription.matches(&event));
    }

    #[test]
    fn matches_different_entity_fails() {
        let subscription = RuesEventUri {
            component: "contracts".to_string(),
            entity: Some(DUMMY_ENTITY.to_string()),
            topic: "withdraw".to_string(),
        };

        let event = RuesEvent {
            uri: RuesEventUri {
                component: "contracts".to_string(),
                entity: Some("def456".to_string()),
                topic: "withdraw".to_string(),
            },
            headers: Default::default(),
            data: DataType::None,
        };

        assert!(!subscription.matches(&event));
    }

    #[test]
    fn matches_different_topic_fails() {
        let subscription = RuesEventUri {
            component: "contracts".to_string(),
            entity: Some(DUMMY_ENTITY.to_string()),
            topic: "withdraw".to_string(),
        };

        let event = RuesEvent {
            uri: RuesEventUri {
                component: "contracts".to_string(),
                entity: Some(DUMMY_ENTITY.to_string()),
                topic: "stake".to_string(),
            },
            headers: Default::default(),
            data: DataType::None,
        };

        assert!(!subscription.matches(&event));
    }

    #[test]
    fn matches_none_entity_acts_as_wildcard() {
        // Entity wildcard: subscription with None entity matches any event
        // entity
        let subscription = RuesEventUri {
            component: "blocks".to_string(),
            entity: None,
            topic: "accepted".to_string(),
        };

        let event = RuesEvent {
            uri: RuesEventUri {
                component: "blocks".to_string(),
                entity: Some("12345".to_string()),
                topic: "accepted".to_string(),
            },
            headers: Default::default(),
            data: DataType::None,
        };

        assert!(subscription.matches(&event));
    }
}