Add initial version

Author: ali-behjati

hermes/Cargo.lock

@@ -5,7 +5,6 @@ edition                        = "2021"
 
 [dependencies]
 axum                           = { version = "0.6.9", features = ["json", "ws", "macros"] }
-axum-extra                     = { version = "0.7.2", features = ["query"] }
 axum-macros                    = { version = "0.3.4" }
 anyhow                         = { version = "1.0.69" }
 base64                         = { version = "0.21.0" }
@@ -15,7 +14,7 @@ dashmap                        = { version = "5.4.0" }
 der                            = { version = "0.7.0" }
 derive_more                    = { version = "0.99.17" }
 env_logger                     = { version = "0.10.0" }
-futures                        = { version = "0.3.26" }
+futures                        = { version = "0.3.28" }
 hex                            = { version = "0.4.3" }
 rand                           = { version = "0.8.5" }
 reqwest                        = { version = "0.11.14", features = ["blocking", "json"] }
@@ -39,24 +38,30 @@ log                            = { version = "0.4.17" }
 wormhole-core                  = { git = "https://github.com/guibescos/wormhole", branch = "reisen/sdk-solana"}
 
 # Parse Wormhole attester price attestations.
-pyth-wormhole-attester-sdk     = { path = "../wormhole_attester/sdk/rust/", version = "0.1.2" }
+pythnet-sdk                    = { path = "../pythnet/pythnet_sdk/", version = "=1.13.6" }
 
 # Setup LibP2P. Unfortunately the dependencies required by libp2p are shared
 # with the dependencies required by solana's geyser plugin. This means that we
 # would have to use the same version of libp2p as solana. Luckily we don't need
 # to do this yet but it's something to keep in mind.
-libp2p                         = { version = "0.51.1", features = [
-    "dns",
+libp2p                         = { version = "0.42.2", features = [
     "gossipsub",
     "identify",
-    "macros",
     "mplex",
     "noise",
-    "quic",
     "secp256k1",
-    "tcp",
-    "tls",
-    "tokio",
     "websocket",
     "yamux",
 ]}
+
+async-trait = "0.1.68"
+solana-client = "=1.15.2"
+solana-sdk = "=1.15.2"
+solana-account-decoder = "=1.15.2"
+moka = { version = "0.11.0", features = ["future"] }
+derive_builder = "0.12.0"
+byteorder = "1.4.3"
+serde_qs = { version = "0.12.0", features = ["axum"] }
+
+[patch.crates-io]
+serde_wormhole = { git = "https://github.com/wormhole-foundation/wormhole" }

hermes/Cargo.toml

@@ -71,8 +71,6 @@ fn main() {
     // Tell Rust to link our Go library at compile time.
     println!("cargo:rustc-link-search=native={out_var}");
     println!("cargo:rustc-link-lib=static=pythnet");
-
-    #[cfg(target_arch = "aarch64")]
     println!("cargo:rustc-link-lib=resolv");
 
     let status = cmd.status().unwrap();

hermes/build.rs

@@ -1,12 +1,6 @@
 use {
     self::ws::dispatch_updates,
-    crate::{
-        network::p2p::OBSERVATIONS,
-        store::{
-            Store,
-            Update,
-        },
-    },
+    crate::store::Store,
     anyhow::Result,
     axum::{
         routing::get,
@@ -36,7 +30,7 @@ impl State {
 
 /// This method provides a background service that responds to REST requests
 ///
-/// Currently this is based on Axum due to the simplicity and strong ecosyjtem support for the
+/// Currently this is based on Axum due to the simplicity and strong ecosystem support for the
 /// packages they are based on (tokio & hyper).
 pub async fn spawn(rpc_addr: String, store: Store) -> Result<()> {
     let state = State::new(store);
@@ -55,25 +49,25 @@ pub async fn spawn(rpc_addr: String, store: Store) -> Result<()> {
         .route("/api/price_feed_ids", get(rest::price_feed_ids))
         .with_state(state.clone());
 
-    // Listen in the background for new VAA's from the Wormhole RPC.
+
+    // Binds the axum's server to the configured address and port. This is a blocking call and will
+    // not return until the server is shutdown.
+    tokio::spawn(async move {
+        // FIXME handle errors properly
+        axum::Server::bind(&rpc_addr.parse().unwrap())
+            .serve(app.into_make_service())
+            .await
+            .unwrap();
+    });
+
+    // Call dispatch updates to websocket every 1 seconds
+    // FIXME use a channel to get updates from the store
     tokio::spawn(async move {
         loop {
-            if let Ok(observation) = OBSERVATIONS.1.lock().unwrap().recv() {
-                match state.store.store_update(Update::Vaa(observation)) {
-                    Ok(updated_feed_ids) => {
-                        tokio::spawn(dispatch_updates(updated_feed_ids, state.clone()));
-                    }
-                    Err(e) => log::error!("Failed to process VAA: {:?}", e),
-                }
-            }
+            dispatch_updates(state.store.get_price_feed_ids(), state.clone()).await;
+            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
         }
     });
 
-    // Binds the axum's server to the configured address and port. This is a blocking call and will
-    // not return until the server is shutdown.
-    axum::Server::bind(&rpc_addr.parse()?)
-        .serve(app.into_make_service())
-        .await?;
-
     Ok(())
 }

hermes/src/network/rpc.rs renamed to hermes/src/api.rs

@@ -1,10 +1,14 @@
-use super::types::PriceIdInput;
 use {
-    super::types::RpcPriceFeed,
-    crate::store::RequestTime,
+    super::types::{
+        PriceIdInput,
+        RpcPriceFeed,
+    },
     crate::{
         impl_deserialize_for_hex_string_wrapper,
-        store::UnixTimestamp,
+        store::types::{
+            RequestTime,
+            UnixTimestamp,
+        },
     },
     anyhow::Result,
     axum::{
@@ -16,7 +20,6 @@ use {
         },
         Json,
     },
-    axum_extra::extract::Query, // Axum extra Query allows us to parse multi-value query parameters.
     base64::{
         engine::general_purpose::STANDARD as base64_standard_engine,
         Engine as _,
@@ -26,6 +29,7 @@ use {
         DerefMut,
     },
     pyth_sdk::PriceIdentifier,
+    serde_qs::axum::QsQuery,
 };
 
 pub enum RestError {
@@ -68,7 +72,7 @@ pub struct LatestVaasQueryParams {
 
 pub async fn latest_vaas(
     State(state): State<super::State>,
-    Query(params): Query<LatestVaasQueryParams>,
+    QsQuery(params): QsQuery<LatestVaasQueryParams>,
 ) -> Result<Json<Vec<String>>, RestError> {
     let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(|id| id.into()).collect();
     let price_feeds_with_update_data = state
@@ -77,10 +81,9 @@ pub async fn latest_vaas(
         .map_err(|_| RestError::UpdateDataNotFound)?;
     Ok(Json(
         price_feeds_with_update_data
-            .batch_vaa
-            .update_data
+            .wormhole_merkle_update_data
             .iter()
-            .map(|vaa_bytes| base64_standard_engine.encode(vaa_bytes)) // TODO: Support multiple
+            .map(|bytes| base64_standard_engine.encode(bytes)) // TODO: Support multiple
             // encoding formats
             .collect(),
     ))
@@ -97,7 +100,7 @@ pub struct LatestPriceFeedsQueryParams {
 
 pub async fn latest_price_feeds(
     State(state): State<super::State>,
-    Query(params): Query<LatestPriceFeedsQueryParams>,
+    QsQuery(params): QsQuery<LatestPriceFeedsQueryParams>,
 ) -> Result<Json<Vec<RpcPriceFeed>>, RestError> {
     let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(|id| id.into()).collect();
     let price_feeds_with_update_data = state
@@ -106,11 +109,10 @@ pub async fn latest_price_feeds(
         .map_err(|_| RestError::UpdateDataNotFound)?;
     Ok(Json(
         price_feeds_with_update_data
-            .batch_vaa
-            .price_infos
-            .into_values()
-            .map(|price_info| {
-                RpcPriceFeed::from_price_info(price_info, params.verbose, params.binary)
+            .price_feeds
+            .into_iter()
+            .map(|price_feed| {
+                RpcPriceFeed::from_price_feed_message(price_feed, params.verbose, params.binary)
             })
             .collect(),
     ))
@@ -131,7 +133,7 @@ pub struct GetVaaResponse {
 
 pub async fn get_vaa(
     State(state): State<super::State>,
-    Query(params): Query<GetVaaQueryParams>,
+    QsQuery(params): QsQuery<GetVaaQueryParams>,
 ) -> Result<Json<GetVaaResponse>, RestError> {
     let price_id: PriceIdentifier = params.id.into();
 
@@ -144,18 +146,16 @@ pub async fn get_vaa(
         .map_err(|_| RestError::UpdateDataNotFound)?;
 
     let vaa = price_feeds_with_update_data
-        .batch_vaa
-        .update_data
+        .wormhole_merkle_update_data
         .get(0)
-        .map(|vaa_bytes| base64_standard_engine.encode(vaa_bytes))
+        .map(|bytes| base64_standard_engine.encode(bytes))
         .ok_or(RestError::UpdateDataNotFound)?;
 
     let publish_time = price_feeds_with_update_data
-        .batch_vaa
-        .price_infos
-        .get(&price_id)
-        .map(|price_info| price_info.publish_time)
-        .ok_or(RestError::UpdateDataNotFound)?;
+        .price_feeds
+        .get(0)
+        .ok_or(RestError::UpdateDataNotFound)?
+        .publish_time; // TODO: This should never happen.
 
     Ok(Json(GetVaaResponse { vaa, publish_time }))
 }
@@ -176,7 +176,7 @@ pub struct GetVaaCcipResponse {
 
 pub async fn get_vaa_ccip(
     State(state): State<super::State>,
-    Query(params): Query<GetVaaCcipQueryParams>,
+    QsQuery(params): QsQuery<GetVaaCcipQueryParams>,
 ) -> Result<Json<GetVaaCcipResponse>, RestError> {
     let price_id: PriceIdentifier = PriceIdentifier::new(params.data[0..32].try_into().unwrap());
     let publish_time = UnixTimestamp::from_be_bytes(params.data[32..40].try_into().unwrap());
@@ -186,14 +186,13 @@ pub async fn get_vaa_ccip(
         .get_price_feeds_with_update_data(vec![price_id], RequestTime::FirstAfter(publish_time))
         .map_err(|_| RestError::CcipUpdateDataNotFound)?;
 
-    let vaa = price_feeds_with_update_data
-        .batch_vaa
-        .update_data
+    let bytes = price_feeds_with_update_data
+        .wormhole_merkle_update_data
         .get(0) // One price feed has only a single VAA as proof.
         .ok_or(RestError::UpdateDataNotFound)?;
 
     Ok(Json(GetVaaCcipResponse {
-        data: format!("0x{}", hex::encode(vaa)),
+        data: format!("0x{}", hex::encode(bytes)),
     }))
 }
 

hermes/src/network/rpc/rest.rs renamed to hermes/src/api/rest.rs

@@ -1,15 +1,11 @@
 use {
     crate::{
         impl_deserialize_for_hex_string_wrapper,
-        store::{
-            proof::batch_vaa::PriceInfo,
+        store::types::{
+            PriceFeedMessage,
             UnixTimestamp,
         },
     },
-    base64::{
-        engine::general_purpose::STANDARD as base64_standard_engine,
-        Engine as _,
-    },
     derive_more::{
         Deref,
         DerefMut,
@@ -41,7 +37,6 @@ type Base64String = String;
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct RpcPriceFeedMetadata {
     pub emitter_chain:              u16,
-    pub attestation_time:           UnixTimestamp,
     pub sequence_number:            u64,
     pub price_service_receive_time: UnixTimestamp,
 }
@@ -51,26 +46,45 @@ pub struct RpcPriceFeed {
     pub id:        PriceIdentifier,
     pub price:     Price,
     pub ema_price: Price,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub metadata:  Option<RpcPriceFeedMetadata>,
     /// Vaa binary represented in base64.
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub vaa:       Option<Base64String>,
 }
 
 impl RpcPriceFeed {
     // TODO: Use a Encoding type to have None, Base64, and Hex variants instead of binary flag.
     // TODO: Use a Verbosity type to define None, or Full instead of verbose flag.
-    pub fn from_price_info(price_info: PriceInfo, verbose: bool, binary: bool) -> Self {
+    pub fn from_price_feed_message(
+        price_feed_message: PriceFeedMessage,
+        _verbose: bool,
+        _binary: bool,
+    ) -> Self {
         Self {
-            id:        price_info.price_feed.id,
-            price:     price_info.price_feed.get_price_unchecked(),
-            ema_price: price_info.price_feed.get_ema_price_unchecked(),
-            metadata:  verbose.then_some(RpcPriceFeedMetadata {
-                emitter_chain:              price_info.emitter_chain,
-                attestation_time:           price_info.attestation_time,
-                sequence_number:            price_info.sequence_number,
-                price_service_receive_time: price_info.receive_time,
-            }),
-            vaa:       binary.then_some(base64_standard_engine.encode(price_info.vaa_bytes)),
+            id:        PriceIdentifier::new(price_feed_message.id),
+            price:     Price {
+                price:        price_feed_message.price,
+                conf:         price_feed_message.conf,
+                expo:         price_feed_message.exponent,
+                publish_time: price_feed_message.publish_time,
+            },
+            ema_price: Price {
+                price:        price_feed_message.ema_price,
+                conf:         price_feed_message.ema_conf,
+                expo:         price_feed_message.exponent,
+                publish_time: price_feed_message.publish_time,
+            },
+            // FIXME: Handle verbose flag properly.
+            // metadata:  verbose.then_some(RpcPriceFeedMetadata {
+            //     emitter_chain:              price_feed_message.emitter_chain,
+            //     sequence_number:            price_feed_message.sequence_number,
+            //     price_service_receive_time: price_feed_message.receive_time,
+            // }),
+            metadata:  None,
+            // FIXME: The vaa is wrong, fix it
+            // vaa:       binary.then_some(base64_standard_engine.encode(message_state.proof_set.wormhole_merkle_proof.vaa)),
+            vaa:       None,
         }
     }
 }