Use pyth-oracle types

hermes/Cargo.toml

@@ -52,9 +52,12 @@ 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"
@@ -62,6 +65,9 @@ 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/src/api.rs

@@ -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/rest.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/types.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/network/pythnet.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/store.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/storage.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/local_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()
     }
 }

hermes/src/store/types.rs

@@ -8,6 +8,12 @@ use {
         Result,
     },
     borsh::BorshDeserialize,
+    pyth_oracle::{
+        Message,
+        PriceFeedMessage,
+    },
+    pyth_sdk::PriceIdentifier,
+    strum::EnumIter,
 };
 
 #[derive(Clone, Debug, PartialEq)]
@@ -47,22 +53,49 @@ impl WormholePayload {
     }
 }
 
-pub type RawMessage = Vec<u8>;
-pub type MessageIdentifier = [u8; 32];
 
-#[derive(Clone, PartialEq, Eq, Debug, Hash)]
+// TODO: We can use strum on Message enum to derive this.
+#[derive(Clone, Debug, Eq, PartialEq, Hash, EnumIter)]
 pub enum MessageType {
-    PriceFeed,
-    TwapPrice,
+    PriceFeedMessage,
+    TwapMessage,
+}
+
+// TODO: Move this methods to Message enum
+pub trait MessageExt {
+    fn type_(&self) -> MessageType;
+    fn id(&self) -> MessageIdentifier;
+    fn publish_time(&self) -> UnixTimestamp;
 }
 
-impl MessageType {
-    pub fn all() -> Vec<Self> {
-        // FIXME: This is a bit brittle, guard it in the future
-        vec![Self::PriceFeed, Self::TwapPrice]
+impl MessageExt for Message {
+    fn type_(&self) -> MessageType {
+        match self {
+            Message::PriceFeedMessage(_) => MessageType::PriceFeedMessage,
+            Message::TwapMessage(_) => MessageType::TwapMessage,
+        }
+    }
+
+    fn id(&self) -> MessageIdentifier {
+        MessageIdentifier {
+            price_id: match self {
+                Message::PriceFeedMessage(message) => PriceIdentifier::new(message.id),
+                Message::TwapMessage(message) => PriceIdentifier::new(message.id),
+            },
+            type_:    self.type_(),
+        }
+    }
+
+    fn publish_time(&self) -> UnixTimestamp {
+        match self {
+            Message::PriceFeedMessage(message) => message.publish_time,
+            Message::TwapMessage(message) => message.publish_time,
+        }
     }
 }
 
+pub type RawMessage = Vec<u8>;
+
 #[derive(Clone, PartialEq, Debug)]
 pub struct WormholeMerkleState {
     pub digest_proof: Vec<u8>,
@@ -70,10 +103,10 @@ pub struct WormholeMerkleState {
 }
 
 #[derive(Clone, PartialEq, Eq, Debug, Hash)]
-pub struct MessageKey {
+pub struct MessageIdentifier {
     // -> this is the real message id
-    pub id:    MessageIdentifier, // -> this is price feed id
-    pub type_: MessageType,
+    pub price_id: PriceIdentifier,
+    pub type_:    MessageType,
 }
 
 #[derive(Clone, PartialEq, Eq, Debug, PartialOrd, Ord)]
@@ -87,13 +120,11 @@ pub struct ProofSet {
     pub wormhole_merkle_proof: WormholeMerkleMessageProof,
 }
 
-
 #[derive(Clone, PartialEq, Debug)]
 pub struct MessageState {
     pub publish_time: UnixTimestamp,
     pub slot:         Slot,
     pub id:           MessageIdentifier,
-    pub type_:        MessageType,
     pub message:      Message,
     pub raw_message:  RawMessage,
     pub proof_set:    ProofSet,
@@ -107,19 +138,15 @@ impl MessageState {
         }
     }
 
-    pub fn key(&self) -> MessageKey {
-        MessageKey {
-            id:    self.id,
-            type_: self.type_.clone(),
-        }
+    pub fn key(&self) -> MessageIdentifier {
+        self.id.clone()
     }
 
     pub fn new(message: Message, raw_message: RawMessage, proof_set: ProofSet, slot: Slot) -> Self {
         Self {
             publish_time: message.publish_time(),
             slot,
-            id: *message.id(),
-            type_: message.message_type(),
+            id: message.id(),
             message,
             raw_message,
             proof_set,
@@ -127,20 +154,6 @@ impl MessageState {
     }
 }
 
-pub enum RequestType {
-    All,
-    Some(Vec<MessageType>),
-}
-
-impl From<RequestType> for Vec<MessageType> {
-    fn from(request_type: RequestType) -> Self {
-        match request_type {
-            RequestType::All => MessageType::all(),
-            RequestType::Some(types) => types,
-        }
-    }
-}
-
 pub type Slot = u64;
 pub type UnixTimestamp = i64;
 
@@ -169,143 +182,6 @@ pub enum Update {
     AccumulatorMessages(AccumulatorMessages),
 }
 
-#[repr(C)]
-#[derive(Debug, Copy, Clone, PartialEq)]
-pub struct PriceFeedMessage {
-    pub id:                [u8; 32],
-    pub price:             i64,
-    pub conf:              u64,
-    pub exponent:          i32,
-    pub publish_time:      i64,
-    pub prev_publish_time: i64,
-    pub ema_price:         i64,
-    pub ema_conf:          u64,
-}
-
-impl PriceFeedMessage {
-    // The size of the serialized message. Note that this is not the same as the size of the struct
-    // (because of the discriminator & struct padding/alignment).
-    pub const MESSAGE_SIZE: usize = 1 + 32 + 8 + 8 + 4 + 8 + 8 + 8 + 8;
-    pub const DISCRIMINATOR: u8 = 0;
-
-    pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
-        if bytes.len() != Self::MESSAGE_SIZE {
-            return Err(anyhow!("Invalid message length"));
-        }
-
-        let mut id = [0u8; 32];
-        id.copy_from_slice(&bytes[1..33]);
-
-        let price = i64::from_be_bytes(bytes[33..41].try_into()?);
-        let conf = u64::from_be_bytes(bytes[41..49].try_into()?);
-        let exponent = i32::from_be_bytes(bytes[49..53].try_into()?);
-        let publish_time = i64::from_be_bytes(bytes[53..61].try_into()?);
-        let prev_publish_time = i64::from_be_bytes(bytes[61..69].try_into()?);
-        let ema_price = i64::from_be_bytes(bytes[69..77].try_into()?);
-        let ema_conf = u64::from_be_bytes(bytes[77..85].try_into()?);
-
-        Ok(Self {
-            id,
-            price,
-            conf,
-            exponent,
-            publish_time,
-            prev_publish_time,
-            ema_price,
-            ema_conf,
-        })
-    }
-}
-
-#[repr(C)]
-#[derive(Debug, Copy, Clone, PartialEq)]
-pub struct TwapMessage {
-    pub id:                [u8; 32],
-    pub cumulative_price:  i128,
-    pub cumulative_conf:   u128,
-    pub num_down_slots:    u64,
-    pub exponent:          i32,
-    pub publish_time:      i64,
-    pub prev_publish_time: i64,
-    pub publish_slot:      u64,
-}
-
-#[allow(dead_code)]
-impl TwapMessage {
-    // The size of the serialized message. Note that this is not the same as the size of the struct
-    // (because of the discriminator & struct padding/alignment).
-    pub const MESSAGE_SIZE: usize = 1 + 32 + 16 + 16 + 8 + 4 + 8 + 8 + 8;
-    pub const DISCRIMINATOR: u8 = 1;
-
-    // FIXME: Use nom or a TLV ser/de library
-    pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
-        if bytes.len() != Self::MESSAGE_SIZE {
-            return Err(anyhow!("Invalid message length"));
-        }
-
-        let mut id = [0u8; 32];
-        id.copy_from_slice(&bytes[1..33]);
-
-        let cumulative_price = i128::from_be_bytes(bytes[33..49].try_into()?);
-        let cumulative_conf = u128::from_be_bytes(bytes[49..65].try_into()?);
-        let num_down_slots = u64::from_be_bytes(bytes[65..73].try_into()?);
-        let exponent = i32::from_be_bytes(bytes[73..77].try_into()?);
-        let publish_time = i64::from_be_bytes(bytes[77..85].try_into()?);
-        let prev_publish_time = i64::from_be_bytes(bytes[85..93].try_into()?);
-        let publish_slot = u64::from_be_bytes(bytes[93..101].try_into()?);
-
-        Ok(Self {
-            id,
-            cumulative_price,
-            cumulative_conf,
-            num_down_slots,
-            exponent,
-            publish_time,
-            prev_publish_time,
-            publish_slot,
-        })
-    }
-}
-
-#[derive(Clone, PartialEq, Debug)]
-pub enum Message {
-    PriceFeed(PriceFeedMessage),
-    TwapPrice(TwapMessage),
-}
-
-impl Message {
-    pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
-        match bytes[0] {
-            PriceFeedMessage::DISCRIMINATOR => {
-                Ok(Self::PriceFeed(PriceFeedMessage::from_bytes(bytes)?))
-            }
-            TwapMessage::DISCRIMINATOR => Ok(Self::TwapPrice(TwapMessage::from_bytes(bytes)?)),
-            _ => Err(anyhow!("Invalid message discriminator")),
-        }
-    }
-
-    pub fn message_type(&self) -> MessageType {
-        match self {
-            Self::PriceFeed(_) => MessageType::PriceFeed,
-            Self::TwapPrice(_) => MessageType::TwapPrice,
-        }
-    }
-
-    pub fn id(&self) -> &[u8; 32] {
-        match self {
-            Self::PriceFeed(msg) => &msg.id,
-            Self::TwapPrice(msg) => &msg.id,
-        }
-    }
-
-    pub fn publish_time(&self) -> i64 {
-        match self {
-            Self::PriceFeed(msg) => msg.publish_time,
-            Self::TwapPrice(msg) => msg.publish_time,
-        }
-    }
-}
-
 pub struct PriceFeedsWithUpdateData {
     pub price_feeds:                 Vec<PriceFeedMessage>,
     pub wormhole_merkle_update_data: Vec<Vec<u8>>,