feat: update hermes to support publisher stake caps

apps/hermes/server/src/api.rs

@@ -127,7 +127,7 @@ where
                 types::RpcPriceIdentifier,
                 types::EncodingType,
                 types::PriceUpdate,
-                types::BinaryPriceUpdate,
+                types::BinaryUpdate,
                 types::ParsedPriceUpdate,
                 types::RpcPriceFeedMetadataV2,
                 types::PriceFeedMetadata,
@@ -150,6 +150,10 @@ where
         .route("/api/get_price_feed", get(rest::get_price_feed))
         .route("/api/get_vaa", get(rest::get_vaa))
         .route("/api/get_vaa_ccip", get(rest::get_vaa_ccip))
+        .route(
+            "/api/get_publisher_stake_caps_update_data",
+            get(rest::get_publisher_stake_caps_update_data),
+        )
         .route("/api/latest_price_feeds", get(rest::latest_price_feeds))
         .route("/api/latest_vaas", get(rest::latest_vaas))
         .route("/api/price_feed_ids", get(rest::price_feed_ids))

apps/hermes/server/src/api/rest.rs

@@ -12,6 +12,7 @@ use {
 };
 
 mod get_price_feed;
+mod get_publisher_stake_caps_update_data;
 mod get_vaa;
 mod get_vaa_ccip;
 mod index;
@@ -25,6 +26,7 @@ mod v2;
 
 pub use {
     get_price_feed::*,
+    get_publisher_stake_caps_update_data::*,
     get_vaa::*,
     get_vaa_ccip::*,
     index::*,

apps/hermes/server/src/api/rest/get_publisher_stake_caps_update_data.rs

@@ -0,0 +1,101 @@
+use {
+    crate::{
+        api::{
+            rest::RestError,
+            types::{
+                BinaryUpdate,
+                EncodingType,
+                GetPublisherStakeCapsUpdateDataResponse,
+                ParsedPublisherStakeCapsUpdate,
+            },
+            ApiState,
+        },
+        state::Aggregates,
+    },
+    anyhow::Result,
+    axum::{
+        extract::State,
+        Json,
+    },
+    base64::{
+        engine::general_purpose::STANDARD as base64_standard_engine,
+        Engine as _,
+    },
+    serde::Deserialize,
+    serde_qs::axum::QsQuery,
+    utoipa::IntoParams,
+};
+
+
+#[derive(Debug, Deserialize, IntoParams)]
+#[into_params(parameter_in=Query)]
+pub struct GetPublisherStakeCapsUpdateData {
+    /// Optional encoding type. If true, return the message in the encoding specified by the encoding parameter. Default is `hex`.
+    #[serde(default)]
+    encoding: EncodingType,
+
+    /// If true, include the parsed update in the `parsed` field of each returned feed. Default is `true`.
+    #[serde(default = "default_true")]
+    parsed: bool,
+}
+
+fn default_true() -> bool {
+    true
+}
+
+
+/// Get the publisher stake caps update data
+#[utoipa::path(
+    get,
+    path = "/api/get_publisher_stake_caps_update_data",
+    responses(
+        (status = 200, description = "Publisher stake caps update data retrieved succesfully", body = Vec<PriceFeedMetadata>)
+    ),
+    params(
+        GetPublisherStakeCapsUpdateData
+    )
+)]
+pub async fn get_publisher_stake_caps_update_data<S>(
+    State(state): State<ApiState<S>>,
+    QsQuery(params): QsQuery<GetPublisherStakeCapsUpdateData>,
+) -> Result<Json<GetPublisherStakeCapsUpdateDataResponse>, RestError>
+where
+    S: Aggregates,
+{
+    let state = &*state.state;
+    let publisher_stake_caps_with_update_data =
+        Aggregates::get_publisher_stake_caps_with_update_data(state)
+            .await
+            .map_err(|e| {
+                tracing::warn!(
+                    "Error getting publisher stake caps with update data: {:?}",
+                    e
+                );
+                RestError::UpdateDataNotFound
+            })?;
+
+    let encoded_data: Vec<String> = publisher_stake_caps_with_update_data
+        .update_data
+        .into_iter()
+        .map(|data| match params.encoding {
+            EncodingType::Base64 => base64_standard_engine.encode(data),
+            EncodingType::Hex => hex::encode(data),
+        })
+        .collect();
+
+    let binary = BinaryUpdate {
+        encoding: params.encoding,
+        data:     encoded_data,
+    };
+
+    let parsed: Option<Vec<ParsedPublisherStakeCapsUpdate>> = if params.parsed {
+        Some(publisher_stake_caps_with_update_data.publisher_stake_caps)
+    } else {
+        None
+    };
+
+    Ok(Json(GetPublisherStakeCapsUpdateDataResponse {
+        binary,
+        parsed,
+    }))
+}

apps/hermes/server/src/api/rest/v2/latest_price_updates.rs

@@ -6,7 +6,7 @@ use {
                 RestError,
             },
             types::{
-                BinaryPriceUpdate,
+                BinaryUpdate,
                 EncodingType,
                 ParsedPriceUpdate,
                 PriceIdInput,
@@ -108,7 +108,7 @@ where
             EncodingType::Hex => hex::encode(data),
         })
         .collect();
-    let binary_price_update = BinaryPriceUpdate {
+    let binary_price_update = BinaryUpdate {
         encoding: params.encoding,
         data:     encoded_data,
     };

apps/hermes/server/src/api/rest/v2/sse.rs

@@ -6,7 +6,7 @@ use {
                 RestError,
             },
             types::{
-                BinaryPriceUpdate,
+                BinaryUpdate,
                 EncodingType,
                 ParsedPriceUpdate,
                 PriceIdInput,
@@ -211,7 +211,7 @@ where
         .into_iter()
         .map(|data| encoding.encode_str(&data))
         .collect();
-    let binary_price_update = BinaryPriceUpdate {
+    let binary_price_update = BinaryUpdate {
         encoding,
         data: encoded_data,
     };

apps/hermes/server/src/api/rest/v2/timestamp_price_updates.rs

@@ -7,7 +7,7 @@ use {
                 RestError,
             },
             types::{
-                BinaryPriceUpdate,
+                BinaryUpdate,
                 EncodingType,
                 ParsedPriceUpdate,
                 PriceIdInput,
@@ -123,7 +123,7 @@ where
         .into_iter()
         .map(|data| query_params.encoding.encode_str(&data))
         .collect();
-    let binary_price_update = BinaryPriceUpdate {
+    let binary_price_update = BinaryUpdate {
         encoding: query_params.encoding,
         data:     encoded_data,
     };

apps/hermes/server/src/api/types.rs

@@ -28,6 +28,7 @@ use {
         Deserialize,
         Serialize,
     },
+    solana_sdk::pubkey::Pubkey,
     std::{
         collections::BTreeMap,
         fmt::{
@@ -40,6 +41,7 @@ use {
     wormhole_sdk::Chain,
 };
 
+
 /// A price id is a 32-byte hex string, optionally prefixed with "0x".
 /// Price ids are case insensitive.
 ///
@@ -231,7 +233,7 @@ impl EncodingType {
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
-pub struct BinaryPriceUpdate {
+pub struct BinaryUpdate {
     pub encoding: EncodingType,
     pub data:     Vec<String>,
 }
@@ -271,9 +273,28 @@ impl From<PriceFeedUpdate> for ParsedPriceUpdate {
     }
 }
 
+#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize, Clone)]
+pub struct ParsedPublisherStakeCapsUpdate {
+    pub publisher_stake_caps: Vec<ParsedPublisherStakeCap>,
+}
+
+#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize, Clone)]
+pub struct ParsedPublisherStakeCap {
+    #[serde(with = "pyth_sdk::utils::as_string")]
+    pub publisher: Pubkey,
+    pub cap:       u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct GetPublisherStakeCapsUpdateDataResponse {
+    pub binary: BinaryUpdate,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub parsed: Option<Vec<ParsedPublisherStakeCapsUpdate>>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct PriceUpdate {
-    pub binary: BinaryPriceUpdate,
+    pub binary: BinaryUpdate,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub parsed: Option<Vec<ParsedPriceUpdate>>,
 }

apps/hermes/server/src/state/aggregate.rs

@@ -17,6 +17,10 @@ use {
         WormholeMerkleState,
     },
     crate::{
+        api::types::{
+            ParsedPublisherStakeCap,
+            ParsedPublisherStakeCapsUpdate,
+        },
         network::wormhole::VaaBytes,
         state::{
             benchmarks::Benchmarks,
@@ -45,6 +49,7 @@ use {
         messages::{
             Message,
             MessageType,
+            PUBLISHER_STAKE_CAPS_MESSAGE_FEED_ID,
         },
         wire::{
             from_slice,
@@ -55,6 +60,7 @@ use {
         },
     },
     serde::Serialize,
+    solana_sdk::pubkey::Pubkey,
     std::{
         collections::HashSet,
         time::Duration,
@@ -68,7 +74,6 @@ use {
     },
     wormhole_sdk::Vaa,
 };
-
 pub mod metrics;
 pub mod wormhole_merkle;
 
@@ -215,6 +220,12 @@ pub struct PriceFeedsWithUpdateData {
     pub update_data: Vec<Vec<u8>>,
 }
 
+#[derive(Debug, PartialEq)]
+pub struct PublisherStakeCapsWithUpdateData {
+    pub publisher_stake_caps: Vec<ParsedPublisherStakeCapsUpdate>,
+    pub update_data:          Vec<Vec<u8>>,
+}
+
 #[derive(Debug, Serialize)]
 pub struct ReadinessMetadata {
     pub has_completed_recently:          bool,
@@ -242,6 +253,9 @@ where
         price_ids: &[PriceIdentifier],
         request_time: RequestTime,
     ) -> Result<PriceFeedsWithUpdateData>;
+    async fn get_publisher_stake_caps_with_update_data(
+        &self,
+    ) -> Result<PublisherStakeCapsWithUpdateData>;
 }
 
 /// Allow downcasting State into CacheState for functions that depend on the `Cache` service.
@@ -403,6 +417,41 @@ where
         }
     }
 
+    async fn get_publisher_stake_caps_with_update_data(
+        &self,
+    ) -> Result<PublisherStakeCapsWithUpdateData> {
+        let messages = self
+            .fetch_message_states(
+                vec![PUBLISHER_STAKE_CAPS_MESSAGE_FEED_ID],
+                RequestTime::Latest,
+                MessageStateFilter::Only(MessageType::PublisherStakeCapsMessage),
+            )
+            .await?;
+
+        let publisher_stake_caps = messages
+            .iter()
+            .map(|message_state| match message_state.message.clone() {
+                Message::PublisherStakeCapsMessage(message) => Ok(ParsedPublisherStakeCapsUpdate {
+                    publisher_stake_caps: message
+                        .caps
+                        .iter()
+                        .map(|cap| ParsedPublisherStakeCap {
+                            publisher: Pubkey::from(cap.publisher),
+                            cap:       cap.cap,
+                        })
+                        .collect(),
+                }),
+                _ => Err(anyhow!("Invalid message state type")),
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        let update_data = construct_update_data(messages.into_iter().map(|m| m.into()).collect())?;
+        Ok(PublisherStakeCapsWithUpdateData {
+            publisher_stake_caps,
+            update_data,
+        })
+    }
+
     async fn get_price_feed_ids(&self) -> HashSet<PriceIdentifier> {
         Cache::message_state_keys(self)
             .await