feat(pyth-lazer) Create schema for lazer agent JRPC endpoint

bplatak · bplatak · commit dac9d93cfa8c · 2025-07-07T15:56:32.000+01:00
diff --git a/lazer/publisher_sdk/rust/src/lib.rs b/lazer/publisher_sdk/rust/src/lib.rs
@@ -1,9 +1,12 @@
 use std::{collections::BTreeMap, time::Duration};
 
+use crate::publisher_update::feed_update::Update;
+use crate::publisher_update::{FeedUpdate, FundingRateUpdate, PriceUpdate};
 use ::protobuf::MessageField;
 use anyhow::{bail, ensure, Context};
 use humantime::format_duration;
 use protobuf::dynamic_value::{dynamic_value, DynamicValue};
+use pyth_lazer_protocol::jrpc::{FeedUpdateParams, UpdateParams};
 use pyth_lazer_protocol::router::TimestampUs;
 
 pub mod transaction_envelope {
@@ -164,3 +167,38 @@ impl TryFrom<DynamicValue> for serde_value::Value {
         }
     }
 }
+
+impl From<FeedUpdateParams> for FeedUpdate {
+    fn from(value: FeedUpdateParams) -> Self {
+        FeedUpdate {
+            feed_id: Some(value.feed_id.0),
+            source_timestamp: value.source_timestamp.into(),
+            update: Some(value.update.into()),
+            special_fields: Default::default(),
+        }
+    }
+}
+
+impl From<UpdateParams> for Update {
+    fn from(value: UpdateParams) -> Self {
+        match value {
+            UpdateParams::PriceUpdate {
+                price,
+                best_bid_price,
+                best_ask_price,
+            } => Update::PriceUpdate(PriceUpdate {
+                price: Some(price.0.into()),
+                best_bid_price: Some(best_bid_price.0.into()),
+                best_ask_price: Some(best_ask_price.0.into()),
+                special_fields: Default::default(),
+            }),
+            UpdateParams::FundingRateUpdate { price, rate } => {
+                Update::FundingRateUpdate(FundingRateUpdate {
+                    price: Some(price.0.into()),
+                    rate: Some(rate.0.into()),
+                    special_fields: Default::default(),
+                })
+            }
+        }
+    }
+}
diff --git a/lazer/sdk/rust/protocol/src/jrpc.rs b/lazer/sdk/rust/protocol/src/jrpc.rs
@@ -0,0 +1,238 @@
+use std::time::Duration;
+use crate::router::{Price, PriceFeedId, Rate, TimestampUs};
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub struct PythLazerAgentJrpcV1 {
+    pub jsonrpc: JsonRpcVersion,
+    #[serde(flatten)]
+    pub params: JrpcParams,
+    pub id: i64,
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+#[serde(tag = "method", content = "params")]
+pub enum JrpcParams {
+    #[serde(rename = "send_updates")]
+    SendUpdates(FeedUpdateParams),
+    #[serde(rename = "get_symbols")]
+    GetMetadata(GetMetadataParams),
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub struct FeedUpdateParams {
+    pub feed_id: PriceFeedId,
+    pub source_timestamp: TimestampUs,
+    pub update: UpdateParams,
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+#[serde(tag = "type")]
+pub enum UpdateParams {
+    #[serde(rename = "price")]
+    PriceUpdate {
+        price: Price,
+        best_bid_price: Price,
+        best_ask_price: Price,
+    },
+    #[serde(rename = "funding_rate")]
+    FundingRateUpdate { price: Price, rate: Rate },
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub struct Filter {
+    name: Option<String>,
+    asset_type: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub struct GetMetadataParams {
+    filters: Option<Vec<Filter>>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub enum JsonRpcVersion {
+    #[serde(rename = "2.0")]
+    V2,
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub struct JrpcResponse<T> {
+    pub jsonrpc: JsonRpcVersion,
+    pub result: T,
+    pub id: i64,
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
+pub struct ErrorResponse {
+    pub message: String
+}
+
+#[derive(Serialize, Deserialize)]
+struct SymbolMetadata {
+    pub asset_type: String,
+    pub cmc_id: i64,
+    pub description: String,
+    pub exponent: i64,
+    pub hermes_id: String,
+    #[serde(default, with = "humantime_serde", alias = "interval")]
+    pub interval: Option<Duration>,
+    pub min_channel: String,
+    pub min_publishers: i64,
+    pub name: String,
+    pub pyth_lazer_id: i64,
+    pub schedule: String,
+    pub state: String,
+    pub symbol: String,
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::jrpc::JrpcParams::{GetMetadata, SendUpdates};
+    use crate::jrpc::{
+        FeedUpdateParams, Filter, GetMetadataParams, JsonRpcVersion, PythLazerAgentJrpcV1,
+        UpdateParams,
+    };
+    use crate::router::{Price, PriceFeedId, Rate, TimestampUs};
+
+    #[test]
+    fn test_send_updates_price() {
+        let json = r#"
+        {
+          "jsonrpc": "2.0",
+          "method": "send_updates",
+          "params": {
+            "feed_id": 1,
+            "source_timestamp": 124214124124,
+
+            "update": {
+              "type": "price",
+              "price": 1234567890,
+              "best_bid_price": 1234567891,
+              "best_ask_price": 1234567892
+            }
+          },
+          "id": 1
+        }
+        "#;
+
+        let expected = PythLazerAgentJrpcV1 {
+            jsonrpc: JsonRpcVersion::V2,
+            params: SendUpdates(FeedUpdateParams {
+                feed_id: PriceFeedId(1),
+                source_timestamp: TimestampUs(124214124124),
+                update: UpdateParams::PriceUpdate {
+                    price: Price::from_integer(1234567890, 0).unwrap(),
+                    best_bid_price: Price::from_integer(1234567891, 0).unwrap(),
+                    best_ask_price: Price::from_integer(1234567892, 0).unwrap(),
+                },
+            }),
+            id: 1,
+        };
+
+        assert_eq!(
+            serde_json::from_str::<PythLazerAgentJrpcV1>(json).unwrap(),
+            expected
+        );
+    }
+
+    #[test]
+    fn test_send_updates_funding_rate() {
+        let json = r#"
+        {
+          "jsonrpc": "2.0",
+          "method": "send_updates",
+          "params": {
+            "feed_id": 1,
+            "source_timestamp": 124214124124,
+
+            "update": {
+              "type": "funding_rate",
+              "price": 1234567890,
+              "rate": 1234567891
+            }
+          },
+          "id": 1
+        }
+        "#;
+
+        let expected = PythLazerAgentJrpcV1 {
+            jsonrpc: JsonRpcVersion::V2,
+            params: SendUpdates(FeedUpdateParams {
+                feed_id: PriceFeedId(1),
+                source_timestamp: TimestampUs(124214124124),
+                update: UpdateParams::FundingRateUpdate {
+                    price: Price::from_integer(1234567890, 0).unwrap(),
+                    rate: Rate::from_integer(1234567891, 0).unwrap(),
+                },
+            }),
+            id: 1,
+        };
+
+        assert_eq!(
+            serde_json::from_str::<PythLazerAgentJrpcV1>(json).unwrap(),
+            expected
+        );
+    }
+    #[test]
+    fn test_send_get_symbols() {
+        let json = r#"
+        {
+          "jsonrpc": "2.0",
+          "method": "get_symbols",
+          "params": {
+            "filters": [
+              {"name":  "BTC/USD"},
+              {"asset_type": "crypto"}
+            ]
+          },
+          "id": 1
+        }
+        "#;
+
+        let expected = PythLazerAgentJrpcV1 {
+            jsonrpc: JsonRpcVersion::V2,
+            params: GetMetadata(GetMetadataParams {
+                filters: Some(vec![
+                    Filter {
+                        name: Some("BTC/USD".to_string()),
+                        asset_type: None,
+                    },
+                    Filter {
+                        name: None,
+                        asset_type: Some("crypto".to_string()),
+                    },
+                ]),
+            }),
+            id: 1,
+        };
+
+        assert_eq!(
+            serde_json::from_str::<PythLazerAgentJrpcV1>(json).unwrap(),
+            expected
+        );
+    }
+
+    #[test]
+    fn test_get_symbols_without_filters() {
+        let json = r#"
+        {
+          "jsonrpc": "2.0",
+          "method": "get_symbols",
+          "params": {},
+          "id": 1
+        }
+        "#;
+
+        let expected = PythLazerAgentJrpcV1 {
+            jsonrpc: JsonRpcVersion::V2,
+            params: GetMetadata(GetMetadataParams { filters: None }),
+            id: 1,
+        };
+
+        assert_eq!(
+            serde_json::from_str::<PythLazerAgentJrpcV1>(json).unwrap(),
+            expected
+        );
+    }
+}
diff --git a/lazer/sdk/rust/protocol/src/lib.rs b/lazer/sdk/rust/protocol/src/lib.rs
@@ -2,6 +2,7 @@
 
 pub mod api;
 pub mod binary_update;
+pub mod jrpc;
 pub mod message;
 pub mod payload;
 pub mod publisher;
@@ -23,7 +24,7 @@ fn magics_in_big_endian() {
     };
 
     // The values listed in this test can be used when reading the magic headers in BE format
-    // (e.g. on EVM).
+    // (e.g., on EVM).
 
     assert_eq!(u32::swap_bytes(BINARY_UPDATE_FORMAT_MAGIC), 1937213467);
     assert_eq!(u32::swap_bytes(PAYLOAD_FORMAT_MAGIC), 1976813459);
@@ -44,6 +45,6 @@ fn magics_in_big_endian() {
         LE_UNSIGNED_FORMAT_MAGIC,
     ] {
         // Required to distinguish between byte orders.
-        assert!(u32::swap_bytes(magic) != magic);
+        assert_ne!(u32::swap_bytes(magic), magic);
     }
 }
diff --git a/lazer/sdk/rust/protocol/src/router.rs b/lazer/sdk/rust/protocol/src/router.rs
@@ -1,5 +1,6 @@
-//! WebSocket JSON protocol types for API the router provides to consumers and publishers.
+//! WebSocket JSON protocol types for the API the router provides to consumers and publishers.
 
+use protobuf::MessageField;
 use {
     crate::payload::AggregatedPriceFeedData,
     anyhow::{bail, Context},
@@ -37,6 +38,26 @@ impl TryFrom<&Timestamp> for TimestampUs {
     }
 }
 
+impl Into<Timestamp> for TimestampUs {
+    fn into(self) -> Timestamp {
+        Timestamp {
+            #[allow(
+                clippy::cast_possible_wrap,
+                reason = "u64 to i64 after this division can never overflow because the value cannot be too big"
+            )]
+            seconds: (self.0 / 1_000_000) as i64,
+            nanos: (self.0 % 1_000_000) as i32 * 1000,
+            special_fields: Default::default(),
+        }
+    }
+}
+
+impl Into<MessageField<Timestamp>> for TimestampUs {
+    fn into(self) -> MessageField<Timestamp> {
+        MessageField::some(self.into())
+    }
+}
+
 impl TimestampUs {
     pub fn now() -> Self {
         let value = SystemTime::now()
@@ -304,7 +325,7 @@ impl<'de> Deserialize<'de> for Channel {
         D: serde::Deserializer<'de>,
     {
         let value = <String>::deserialize(deserializer)?;
-        parse_channel(&value).ok_or_else(|| D::Error::custom("unknown channel"))
+        parse_channel(&value).ok_or_else(|| Error::custom("unknown channel"))
     }
 }
 
@@ -341,12 +362,14 @@ fn fixed_rate_values() {
         "values must be unique and sorted"
     );
     for value in FixedRate::ALL {
-        assert!(
-            1000 % value.ms == 0,
+        assert_eq!(
+            1000 % value.ms,
+            0,
             "1 s must contain whole number of intervals"
         );
-        assert!(
-            value.value_us() % FixedRate::MIN.value_us() == 0,
+        assert_eq!(
+            value.value_us() % FixedRate::MIN.value_us(),
+            0,
             "the interval's borders must be a subset of the minimal interval's borders"
         );
     }
@@ -383,7 +406,7 @@ impl<'de> Deserialize<'de> for SubscriptionParams {
         D: serde::Deserializer<'de>,
     {
         let value = SubscriptionParamsRepr::deserialize(deserializer)?;
-        Self::new(value).map_err(D::Error::custom)
+        Self::new(value).map_err(Error::custom)
     }
 }
 
