radion_sdk/realtime/

protocol.rs

//! Wire protocol: subscriptions, filters, and inbound/outbound frames.

use serde::{Deserialize, Serialize};

use super::channels::{Channel, ClobChannel, FilterKey, SubscribableChannel, join_filter_keys};
use super::payloads::Payload;
use crate::error::{RadionError, Result};

/// Server-side filters narrowing the events delivered on a channel.
///
/// Some channels require a filter (for example `wallets` needs `wallets`,
/// `markets` needs `market_ids` or `token_ids`); see the channel docs.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct ChannelFilters {
    /// Wallet addresses to match (required by `wallets`, optional on `trading`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub wallets: Option<Vec<String>>,
    /// Condition / market ids to match (required by `markets`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub market_ids: Option<Vec<String>>,
    /// ERC-1155 token ids to match (required by `markets` and every `clob.*`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub token_ids: Option<Vec<String>>,
    /// Minimum trade notional in USD (optional on `trading`).
    ///
    /// On the confirmed feed this is the actual filled USD; on the pending feed
    /// it is the intended fill notional (`call.notional_usd`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub min_usd: Option<f64>,
}

impl ChannelFilters {
    /// Whether the given filter key carries a value.
    fn has(&self, key: FilterKey) -> bool {
        match key {
            FilterKey::Wallets => self.wallets.as_ref().is_some_and(|v| !v.is_empty()),
            FilterKey::MarketIds => self.market_ids.as_ref().is_some_and(|v| !v.is_empty()),
            FilterKey::TokenIds => self.token_ids.as_ref().is_some_and(|v| !v.is_empty()),
            FilterKey::MinUsd => self.min_usd.is_some(),
        }
    }
}

/// A single channel subscription.
///
/// `id` is a client-defined string echoed back on acknowledgements and on every
/// event frame, so multiple subscriptions to the same channel can be told
/// apart. `channel` is a topic channel or a CLOB channel.
///
/// `confirmed` picks the feed: `true` (the default) streams confirmed on-chain
/// events; `false` streams pending mempool transactions before block inclusion.
/// It has no effect on CLOB channels, which have no pending feed.
#[derive(Debug, Clone, PartialEq)]
pub struct Subscription {
    /// Client-defined id, echoed back on confirmations and event frames.
    pub id: String,
    /// Channel name: a topic channel or a `clob.`-prefixed channel.
    pub channel: SubscribableChannel,
    /// Which feed to stream: `true` = confirmed (default), `false` = pending.
    pub confirmed: bool,
    /// Optional server-side filters.
    pub filters: Option<ChannelFilters>,
}

impl Subscription {
    /// Create a subscription with no filters, on the confirmed feed.
    pub fn new(id: impl Into<String>, channel: impl Into<SubscribableChannel>) -> Self {
        Self {
            id: id.into(),
            channel: channel.into(),
            confirmed: true,
            filters: None,
        }
    }

    /// Attach server-side filters.
    #[must_use]
    pub fn with_filters(mut self, filters: ChannelFilters) -> Self {
        self.filters = Some(filters);
        self
    }

    /// Choose the feed: `true` = confirmed on-chain events, `false` = pending
    /// mempool transactions. Ignored for CLOB channels.
    #[must_use]
    pub fn confirmed(mut self, confirmed: bool) -> Self {
        self.confirmed = confirmed;
        self
    }

    /// Stream the pending (mempool) feed instead of confirmed events. Shorthand
    /// for [`confirmed(false)`](Self::confirmed).
    #[must_use]
    pub fn pending(self) -> Self {
        self.confirmed(false)
    }

    /// Validate that this subscription carries the filters its channel requires.
    ///
    /// # Errors
    ///
    /// Returns [`RadionError::Connection`] describing the first violation. The
    /// requirement is the same for the confirmed and pending feeds; every CLOB
    /// channel requires a `token_ids` filter.
    pub fn validate(&self) -> Result<()> {
        let Some(requirement) = self.channel.filter_requirement() else {
            return Ok(());
        };
        let satisfied = requirement
            .required_any_of
            .iter()
            .any(|key| self.filters.as_ref().is_some_and(|f| f.has(*key)));
        if satisfied {
            return Ok(());
        }
        Err(RadionError::connection(format!(
            "channel \"{}\" requires a {} filter",
            self.channel,
            join_filter_keys(requirement.required_any_of),
        )))
    }
}

/// A frame sent from the client to the Radion server.
#[cfg(feature = "realtime")]
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "action", rename_all = "snake_case")]
pub(crate) enum OutboundFrame {
    Subscribe {
        id: String,
        channel: SubscribableChannel,
        #[serde(skip_serializing_if = "Option::is_none")]
        confirmed: Option<bool>,
        #[serde(skip_serializing_if = "Option::is_none")]
        filters: Option<ChannelFilters>,
    },
    Unsubscribe {
        id: String,
    },
    Ping,
}

#[cfg(feature = "realtime")]
impl OutboundFrame {
    pub(crate) fn subscribe(subscription: &Subscription) -> Self {
        let confirmed = subscription.channel.topic().map(|_| subscription.confirmed);
        Self::Subscribe {
            id: subscription.id.clone(),
            channel: subscription.channel,
            confirmed,
            filters: subscription.filters.clone(),
        }
    }
}

/// A data event delivered on a subscribed channel.
///
/// `id` identifies the subscription it belongs to; `channel` is the bare channel
/// name (same for both feeds). `confirmed` says which feed it came from: `true` =
/// confirmed on-chain event, `false` = pending mempool transaction. Route by
/// `id`, not by the channel name. `seq` and `sent_at_ms` describe the envelope:
/// gap detection and server send time. `data` is the typed payload — match on it
/// to handle a specific event shape.
#[derive(Debug, Clone)]
pub struct ChannelEvent {
    /// Subscription id this event belongs to.
    pub id: String,
    /// Bare channel name.
    pub channel: String,
    /// Which feed this event came from: `true` = confirmed, `false` = pending.
    pub confirmed: bool,
    /// Per-connection monotonic counter, starting at 0 and incremented by one
    /// for every event frame across all subscriptions. A jump means frames
    /// were dropped.
    pub seq: u64,
    /// Unix milliseconds when the server sent the frame. Server→client latency
    /// is your receive time (ms) minus `sent_at_ms`.
    pub sent_at_ms: u64,
    /// Typed payload.
    pub data: Payload,
}

/// A frame received from the server.
#[derive(Debug, Clone, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub(crate) enum InboundFrame {
    Event {
        id: String,
        channel: String,
        #[serde(default = "confirmed_default")]
        confirmed: bool,
        seq: u64,
        sent_at_ms: u64,
        data: serde_json::Value,
    },
    Subscribed {
        #[allow(dead_code)]
        id: String,
        #[allow(dead_code)]
        channel: Option<String>,
        #[allow(dead_code)]
        confirmed: Option<bool>,
    },
    Unsubscribed {
        #[allow(dead_code)]
        id: String,
        #[allow(dead_code)]
        channel: Option<String>,
    },
    #[cfg_attr(not(feature = "realtime"), allow(dead_code))]
    Warning {
        code: String,
        id: Option<String>,
        message: String,
    },
    Pong,
    #[cfg_attr(not(feature = "realtime"), allow(dead_code))]
    Error {
        message: String,
        code: Option<String>,
        id: Option<String>,
        channel: Option<String>,
        #[allow(dead_code)]
        skipped: Option<u64>,
    },
}

/// Default for a missing `confirmed` envelope field: confirmed feed.
fn confirmed_default() -> bool {
    true
}

/// Parse and validate a raw text frame into a typed [`InboundFrame`].
///
/// Returns `None` when the payload is not valid JSON or does not match a known
/// frame envelope, so callers can drop malformed frames without erroring.
pub(crate) fn parse_inbound_frame(raw: &str) -> Option<InboundFrame> {
    serde_json::from_str(raw).ok()
}

impl InboundFrame {
    /// Convert an `event` frame into a typed [`ChannelEvent`], decoding `data`
    /// against the channel and the envelope's `confirmed` flag. Returns `None`
    /// for non-event frames.
    pub(crate) fn into_channel_event(self) -> Option<ChannelEvent> {
        let Self::Event {
            id,
            channel,
            confirmed,
            seq,
            sent_at_ms,
            data,
        } = self
        else {
            return None;
        };
        let payload = if let Ok(clob) = channel.parse::<ClobChannel>() {
            Payload::from_clob_channel(clob, data)
        } else if confirmed {
            match channel.parse::<Channel>() {
                Ok(topic) => Payload::from_channel(topic, data),
                Err(_) => Payload::Other(data),
            }
        } else {
            Payload::from_pending(data)
        };
        Some(ChannelEvent {
            id,
            channel,
            confirmed,
            seq,
            sent_at_ms,
            data: payload,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::realtime::channels::ClobChannel;
    use crate::realtime::payloads::{OrderSide, Payload, TradingEventType};

    #[test]
    fn validates_required_filters() {
        // `wallets` requires a wallets filter.
        assert!(Subscription::new("w", Channel::Wallets).validate().is_err());
        let ok = Subscription::new("w", Channel::Wallets).with_filters(ChannelFilters {
            wallets: Some(vec!["0x1".into()]),
            ..Default::default()
        });
        assert!(ok.validate().is_ok());

        // `markets` accepts either market_ids or token_ids.
        let markets = Subscription::new("m", Channel::Markets).with_filters(ChannelFilters {
            token_ids: Some(vec!["1".into()]),
            ..Default::default()
        });
        assert!(markets.validate().is_ok());

        // `trading` requires nothing.
        assert!(Subscription::new("t", Channel::Trading).validate().is_ok());

        // Every clob channel requires token_ids.
        assert!(
            Subscription::new("b", ClobChannel::Book)
                .validate()
                .is_err()
        );
        let clob = Subscription::new("b", ClobChannel::Book).with_filters(ChannelFilters {
            token_ids: Some(vec!["1".into()]),
            ..Default::default()
        });
        assert!(clob.validate().is_ok());
    }

    #[test]
    fn parses_and_types_clob_event_frames() {
        let raw = r#"{"type":"event","id":"mid","channel":"clob.midpoint","seq":42,"sent_at_ms":1721818200123,"data":{"asset_id":"7","market":"0xm","midpoint":0.5,"timestamp":1}}"#;
        let event = parse_inbound_frame(raw)
            .expect("valid frame")
            .into_channel_event()
            .expect("event");
        assert_eq!(event.channel, "clob.midpoint");
        assert_eq!(event.seq, 42);
        assert_eq!(event.sent_at_ms, 1_721_818_200_123);
        match event.data {
            Payload::ClobMidpoint(mid) => {
                assert_eq!(mid.asset_id, "7");
                assert_eq!(mid.midpoint, 0.5);
            }
            other => panic!("expected clob midpoint payload, got {other:?}"),
        }
    }

    #[test]
    fn serializes_outbound_frames() {
        let ping = serde_json::to_string(&OutboundFrame::Ping).unwrap();
        assert_eq!(ping, r#"{"action":"ping"}"#);

        let unsub = serde_json::to_string(&OutboundFrame::Unsubscribe { id: "x".into() }).unwrap();
        assert_eq!(unsub, r#"{"action":"unsubscribe","id":"x"}"#);

        let sub = OutboundFrame::subscribe(&Subscription::new("trading", Channel::Trading));
        let json: serde_json::Value =
            serde_json::from_str(&serde_json::to_string(&sub).unwrap()).unwrap();
        assert_eq!(json["action"], "subscribe");
        assert_eq!(json["channel"], "trading");
        assert_eq!(json["confirmed"], true);
        // No filters key when none are set.
        assert!(json.get("filters").is_none());

        let pending = OutboundFrame::subscribe(&Subscription::new("t", Channel::Trading).pending());
        let json: serde_json::Value =
            serde_json::from_str(&serde_json::to_string(&pending).unwrap()).unwrap();
        assert_eq!(json["channel"], "trading");
        assert_eq!(json["confirmed"], false);

        let clob = OutboundFrame::subscribe(
            &Subscription::new("b", ClobChannel::Book).with_filters(ChannelFilters {
                token_ids: Some(vec!["1".into()]),
                ..Default::default()
            }),
        );
        let json: serde_json::Value =
            serde_json::from_str(&serde_json::to_string(&clob).unwrap()).unwrap();
        assert!(json.get("confirmed").is_none());
    }

    #[test]
    fn parses_and_types_confirmed_event_frames() {
        let raw = r#"{"type":"event","id":"t","channel":"trading","confirmed":true,"seq":42,"sent_at_ms":1721818200123,"data":{"type":"order_filled_v2","side":1,"tokenId":"0xabc"}}"#;
        let frame = parse_inbound_frame(raw).expect("valid frame");
        let event = frame.into_channel_event().expect("event");
        assert_eq!(event.id, "t");
        assert!(event.confirmed);
        assert_eq!(event.seq, 42);
        assert_eq!(event.sent_at_ms, 1_721_818_200_123);
        match event.data {
            Payload::Trading(trade) => {
                assert_eq!(trade.kind, TradingEventType::OrderFilledV2);
                assert_eq!(trade.side, Some(1));
                assert_eq!(trade.token_id.as_deref(), Some("0xabc"));
            }
            other => panic!("expected trading payload, got {other:?}"),
        }
    }

    #[test]
    fn parses_and_types_pending_event_frames() {
        let raw = r#"{"type":"event","id":"t","channel":"trading","confirmed":false,"seq":42,"sent_at_ms":1721818200123,"data":{"seen_at_ms":1782027489000,"transaction_hash":"0xhash","from":"0xfrom","to":"0xto","contract_kinds":["exchange"],"method_selector":"0xabcdef12","input":"0xdead","value":"0","call":{"method":"fillOrder","market_ids":["0xm"],"token_ids":["7"],"wallets":["0xw"],"notional_usd":192.5,"orders":[{"maker":"0xa","taker":null,"token_id":"7","side":"buy","maker_amount":"100","taker_amount":"50"}]}}}"#;
        let event = parse_inbound_frame(raw)
            .expect("valid frame")
            .into_channel_event()
            .expect("event");
        assert_eq!(event.channel, "trading");
        assert!(!event.confirmed);
        match event.data {
            Payload::Mempool(tx) => {
                assert_eq!(tx.transaction_hash, "0xhash");
                assert_eq!(tx.from, "0xfrom");
                let call = tx.call.expect("call present");
                assert_eq!(call.method, "fillOrder");
                assert_eq!(call.notional_usd, Some(192.5));
                assert_eq!(call.orders.len(), 1);
                assert_eq!(call.orders[0].side, OrderSide::Buy);
                assert!(call.orders[0].taker.is_none());
            }
            other => panic!("expected mempool payload, got {other:?}"),
        }
    }

    #[test]
    fn missing_confirmed_defaults_to_confirmed_feed() {
        let raw = r#"{"type":"event","id":"t","channel":"trading","seq":42,"sent_at_ms":1721818200123,"data":{"type":"order_cancelled"}}"#;
        let event = parse_inbound_frame(raw)
            .unwrap()
            .into_channel_event()
            .unwrap();
        assert!(event.confirmed);
    }

    #[test]
    fn unknown_channel_falls_back_to_other() {
        let raw = r#"{"type":"event","id":"m","channel":"unknownz","confirmed":true,"seq":42,"sent_at_ms":1721818200123,"data":{"foo":1}}"#;
        let event = parse_inbound_frame(raw)
            .unwrap()
            .into_channel_event()
            .unwrap();
        assert!(matches!(event.data, Payload::Other(_)));
    }

    #[test]
    fn parses_warning_frame() {
        let raw = r#"{"type":"warning","code":"mempool_unavailable","id":"t","message":"no pending stream"}"#;
        assert!(matches!(
            parse_inbound_frame(raw),
            Some(InboundFrame::Warning { .. })
        ));
    }

    #[test]
    fn subscribed_ack_echoes_confirmed() {
        let raw = r#"{"type":"subscribed","id":"t","channel":"trading","confirmed":false}"#;
        assert!(matches!(
            parse_inbound_frame(raw),
            Some(InboundFrame::Subscribed {
                confirmed: Some(false),
                ..
            })
        ));
    }

    #[test]
    fn drops_malformed_frames() {
        assert!(parse_inbound_frame("not json").is_none());
        assert!(parse_inbound_frame(r#"{"type":"mystery"}"#).is_none());
    }

    #[test]
    fn drops_event_frames_missing_seq_or_sent_at_ms() {
        let no_envelope =
            r#"{"type":"event","id":"t","channel":"trading","data":{"type":"order_cancelled"}}"#;
        assert!(parse_inbound_frame(no_envelope).is_none());

        let no_sent_at_ms = r#"{"type":"event","id":"t","channel":"trading","seq":42,"data":{"type":"order_cancelled"}}"#;
        assert!(parse_inbound_frame(no_sent_at_ms).is_none());
    }

    #[test]
    fn parses_error_frame() {
        let raw = r#"{"type":"error","message":"boom","code":"bad","id":"x"}"#;
        assert!(matches!(
            parse_inbound_frame(raw),
            Some(InboundFrame::Error { .. })
        ));
    }
}