Use pyth-oracle types

Author: ali-behjati

hermes/Cargo.lock

@@ -52,16 +52,22 @@ libp2p                         = { version = "0.42.2", features = [
 ]}
 
 async-trait = "0.1.68"
-solana-client = "=1.15.2"
-solana-sdk = "=1.15.2"
-solana-account-decoder = "=1.15.2"
+
+# We around bound to this version because of pyth-oracle
+solana-client = "=1.13.3"
+solana-sdk = "=1.13.3"
+solana-account-decoder = "=1.13.3"
+
 moka = { version = "0.11.0", features = ["future"] }
 derive_builder = "0.12.0"
 byteorder = "1.4.3"
 serde_qs = { version = "0.12.0", features = ["axum"] }
 
 serde_wormhole = { git = "https://github.com/wormhole-foundation/wormhole", tag = "v2.17.1"}
 wormhole-sdk = { git = "https://github.com/wormhole-foundation/wormhole", tag = "v2.17.1" }
+pyth-oracle = { git = "https://github.com/pyth-network/pyth-client", rev = "7d593d87e07a1e2486e7ca21597d664ee72be1ec", features = ["library"] }
+
+strum = { version = "0.24", features = ["derive"] }
 
 [patch.crates-io]
 serde_wormhole = { git = "https://github.com/wormhole-foundation/wormhole", tag = "v2.17.1" }

hermes/Cargo.toml

@@ -64,7 +64,11 @@ pub async fn spawn(rpc_addr: String, store: Store) -> Result<()> {
     // FIXME use a channel to get updates from the store
     tokio::spawn(async move {
         loop {
-            dispatch_updates(state.store.get_price_feed_ids(), state.clone()).await;
+            dispatch_updates(
+                state.store.get_price_feed_ids().into_iter().collect(),
+                state.clone(),
+            )
+            .await;
             tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
         }
     });

hermes/src/api.rs

@@ -30,6 +30,7 @@ use {
     },
     pyth_sdk::PriceIdentifier,
     serde_qs::axum::QsQuery,
+    std::collections::HashSet,
 };
 
 pub enum RestError {
@@ -59,7 +60,7 @@ impl IntoResponse for RestError {
 
 pub async fn price_feed_ids(
     State(state): State<super::State>,
-) -> Result<Json<Vec<PriceIdentifier>>, RestError> {
+) -> Result<Json<HashSet<PriceIdentifier>>, RestError> {
     let price_feeds = state.store.get_price_feed_ids();
     Ok(Json(price_feeds))
 }

hermes/src/api/rest.rs

@@ -1,15 +1,13 @@
 use {
     crate::{
         impl_deserialize_for_hex_string_wrapper,
-        store::types::{
-            PriceFeedMessage,
-            UnixTimestamp,
-        },
+        store::types::UnixTimestamp,
     },
     derive_more::{
         Deref,
         DerefMut,
     },
+    pyth_oracle::PriceFeedMessage,
     pyth_sdk::{
         Price,
         PriceIdentifier,

hermes/src/api/types.rs

@@ -22,6 +22,7 @@ use {
         },
         rpc_filter::{
             Memcmp,
+            MemcmpEncodedBytes,
             RpcFilterType,
         },
     },
@@ -42,10 +43,11 @@ pub async fn spawn(pythnet_ws_endpoint: String, store: Store) -> Result<()> {
             encoding: Some(UiAccountEncoding::Base64Zstd),
             ..Default::default()
         },
-        filters: Some(vec![RpcFilterType::Memcmp(Memcmp::new_raw_bytes(
-            0,
-            b"PAS1".to_vec(),
-        ))]),
+        filters: Some(vec![RpcFilterType::Memcmp(Memcmp {
+            offset:   0,
+            bytes:    MemcmpEncodedBytes::Bytes(b"PAS1".to_vec()),
+            encoding: None,
+        })]),
         with_context: Some(true),
         ..Default::default()
     };

hermes/src/network/pythnet.rs

@@ -20,7 +20,6 @@ use {
             store_wormhole_merkle_verified_message,
         },
         types::{
-            Message,
             MessageState,
             ProofSet,
             WormholePayload,
@@ -32,8 +31,12 @@ use {
     },
     derive_builder::Builder,
     moka::future::Cache,
+    pyth_oracle::Message,
     pyth_sdk::PriceIdentifier,
-    std::time::Duration,
+    std::{
+        collections::HashSet,
+        time::Duration,
+    },
     wormhole_sdk::Vaa,
 };
 
@@ -137,7 +140,7 @@ impl Store {
             .iter()
             .enumerate()
             .map(|(idx, raw_message)| {
-                let message = Message::from_bytes(raw_message)?;
+                let message = Message::try_from_bytes(raw_message)?;
 
                 Ok(MessageState::new(
                     message,
@@ -168,18 +171,15 @@ impl Store {
         request_time: RequestTime,
     ) -> Result<PriceFeedsWithUpdateData> {
         let messages = self.storage.retrieve_message_states(
-            price_ids
-                .iter()
-                .map(|price_id| price_id.to_bytes())
-                .collect(),
-            types::RequestType::Some(vec![MessageType::PriceFeed]),
+            price_ids,
             request_time,
+            Some(&|message_type| *message_type == MessageType::PriceFeedMessage),
         )?;
 
         let price_feeds = messages
             .iter()
             .map(|message_state| match message_state.message {
-                Message::PriceFeed(price_feed) => Ok(price_feed),
+                Message::PriceFeedMessage(price_feed) => Ok(price_feed),
                 _ => Err(anyhow!("Invalid message state type")),
             })
             .collect::<Result<Vec<_>>>()?;
@@ -191,11 +191,7 @@ impl Store {
         })
     }
 
-    pub fn get_price_feed_ids(&self) -> Vec<PriceIdentifier> {
-        self.storage
-            .keys()
-            .iter()
-            .map(|key| PriceIdentifier::new(key.id))
-            .collect()
+    pub fn get_price_feed_ids(&self) -> HashSet<PriceIdentifier> {
+        self.storage.keys().iter().map(|key| key.price_id).collect()
     }
 }

hermes/src/store.rs

@@ -1,12 +1,12 @@
 use {
     super::types::{
         MessageIdentifier,
-        MessageKey,
         MessageState,
+        MessageType,
         RequestTime,
-        RequestType,
     },
     anyhow::Result,
+    pyth_sdk::PriceIdentifier,
     std::sync::Arc,
 };
 
@@ -23,11 +23,11 @@ pub trait Storage: Send + Sync {
     fn store_message_states(&self, message_states: Vec<MessageState>) -> Result<()>;
     fn retrieve_message_states(
         &self,
-        ids: Vec<MessageIdentifier>,
-        request_type: RequestType,
+        ids: Vec<PriceIdentifier>,
         request_time: RequestTime,
+        filter: Option<&dyn Fn(&MessageType) -> bool>,
     ) -> Result<Vec<MessageState>>;
-    fn keys(&self) -> Vec<MessageKey>;
+    fn keys(&self) -> Vec<MessageIdentifier>;
 }
 
 pub type StorageInstance = Arc<Box<dyn Storage>>;

hermes/src/store/storage.rs

@@ -1,10 +1,8 @@
 use {
     super::{
         MessageIdentifier,
-        MessageKey,
         MessageState,
         RequestTime,
-        RequestType,
         Storage,
         StorageInstance,
     },
@@ -14,15 +12,17 @@ use {
         Result,
     },
     dashmap::DashMap,
+    pyth_sdk::PriceIdentifier,
     std::{
         collections::VecDeque,
         sync::Arc,
     },
+    strum::IntoEnumIterator,
 };
 
 #[derive(Clone)]
 pub struct LocalStorage {
-    cache:            Arc<DashMap<MessageKey, VecDeque<MessageState>>>,
+    cache:            Arc<DashMap<MessageIdentifier, VecDeque<MessageState>>>,
     max_size_per_key: usize,
 }
 
@@ -36,7 +36,7 @@ impl LocalStorage {
 
     fn retrieve_message_state(
         &self,
-        key: MessageKey,
+        key: MessageIdentifier,
         request_time: RequestTime,
     ) -> Option<MessageState> {
         match self.cache.get(&key) {
@@ -109,19 +109,22 @@ impl Storage for LocalStorage {
 
     fn retrieve_message_states(
         &self,
-        ids: Vec<MessageIdentifier>,
-        request_type: RequestType,
+        ids: Vec<PriceIdentifier>,
         request_time: RequestTime,
+        filter: Option<&dyn Fn(&MessageType) -> bool>,
     ) -> Result<Vec<MessageState>> {
         // TODO: Should we return an error if any of the ids are not found?
-        let types: Vec<MessageType> = request_type.into();
         ids.into_iter()
             .flat_map(|id| {
                 let request_time = request_time.clone();
-                types.iter().map(move |message_type| {
-                    let key = MessageKey {
-                        id,
-                        type_: message_type.clone(),
+                let message_types: Vec<MessageType> = match filter {
+                    Some(filter) => MessageType::iter().filter(filter).collect(),
+                    None => MessageType::iter().collect(),
+                };
+                message_types.into_iter().map(move |message_type| {
+                    let key = MessageIdentifier {
+                        price_id: id,
+                        type_:    message_type,
                     };
                     self.retrieve_message_state(key, request_time.clone())
                         .ok_or(anyhow!("Message not found"))
@@ -130,7 +133,7 @@ impl Storage for LocalStorage {
             .collect()
     }
 
-    fn keys(&self) -> Vec<MessageKey> {
+    fn keys(&self) -> Vec<MessageIdentifier> {
         self.cache.iter().map(|entry| entry.key().clone()).collect()
     }
 }