[hermes] Add WS (#773)

* [hermes] Add WS

* Address Jayant comments

* Update hermes/src/network/rpc/ws.rs

Co-authored-by: Reisen <[email protected]>

* Address David comments

---------

Co-authored-by: Reisen <[email protected]>

Author: ali-behjati

hermes/src/network/rpc.rs

@@ -1,4 +1,5 @@
 use {
+    self::ws::dispatch_updates,
     crate::{
         network::p2p::OBSERVATIONS,
         store::{
@@ -11,19 +12,25 @@ use {
         routing::get,
         Router,
     },
+    std::sync::Arc,
 };
 
 mod rest;
-mod rpc_price_feed;
+mod types;
+mod ws;
 
 #[derive(Clone)]
 pub struct State {
     pub store: Store,
+    pub ws:    Arc<ws::WsState>,
 }
 
 impl State {
     pub fn new(store: Store) -> Self {
-        Self { store }
+        Self {
+            store,
+            ws: Arc::new(ws::WsState::new()),
+        }
     }
 }
 
@@ -40,6 +47,7 @@ pub async fn spawn(rpc_addr: String, store: Store) -> Result<()> {
     let app = app
         .route("/", get(rest::index))
         .route("/live", get(rest::live))
+        .route("/ws", get(ws::ws_route_handler))
         .route("/api/latest_price_feeds", get(rest::latest_price_feeds))
         .route("/api/latest_vaas", get(rest::latest_vaas))
         .route("/api/get_vaa", get(rest::get_vaa))
@@ -51,8 +59,11 @@ pub async fn spawn(rpc_addr: String, store: Store) -> Result<()> {
     tokio::spawn(async move {
         loop {
             if let Ok(observation) = OBSERVATIONS.1.lock().unwrap().recv() {
-                if let Err(e) = state.store.store_update(Update::Vaa(observation)) {
-                    log::error!("Failed to process VAA: {:?}", e);
+                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),
                 }
             }
         }

hermes/src/network/rpc/rest.rs

@@ -1,8 +1,6 @@
+use super::types::PriceIdInput;
 use {
-    super::rpc_price_feed::{
-        RpcPriceFeed,
-        RpcPriceFeedMetadata,
-    },
+    super::types::RpcPriceFeed,
     crate::store::RequestTime,
     crate::{
         impl_deserialize_for_hex_string_wrapper,
@@ -30,21 +28,6 @@ use {
     pyth_sdk::PriceIdentifier,
 };
 
-/// PriceIdInput is a wrapper around a 32-byte hex string.
-/// that supports a flexible deserialization from a hex string.
-/// It supports both 0x-prefixed and non-prefixed hex strings,
-/// and also supports both lower and upper case characters.
-#[derive(Debug, Clone, Deref, DerefMut)]
-pub struct PriceIdInput([u8; 32]);
-// TODO: Use const generics instead of macro.
-impl_deserialize_for_hex_string_wrapper!(PriceIdInput, 32);
-
-impl From<PriceIdInput> for PriceIdentifier {
-    fn from(id: PriceIdInput) -> Self {
-        Self::new(*id)
-    }
-}
-
 pub enum RestError {
     UpdateDataNotFound,
     CcipUpdateDataNotFound,
@@ -127,19 +110,7 @@ pub async fn latest_price_feeds(
             .price_infos
             .into_values()
             .map(|price_info| {
-                let mut rpc_price_feed: RpcPriceFeed = price_info.price_feed.into();
-                rpc_price_feed.metadata = params.verbose.then_some(RpcPriceFeedMetadata {
-                    emitter_chain:              price_info.emitter_chain,
-                    sequence_number:            price_info.sequence_number,
-                    attestation_time:           price_info.attestation_time,
-                    price_service_receive_time: price_info.receive_time,
-                });
-
-                rpc_price_feed.vaa = params
-                    .binary
-                    .then_some(base64_standard_engine.encode(&price_info.vaa_bytes));
-
-                rpc_price_feed
+                RpcPriceFeed::from_price_info(price_info, params.verbose, params.binary)
             })
             .collect(),
     ))

hermes/src/network/rpc/rpc_price_feed.rs

@@ -0,0 +1,76 @@
+use {
+    crate::{
+        impl_deserialize_for_hex_string_wrapper,
+        store::{
+            proof::batch_vaa::PriceInfo,
+            UnixTimestamp,
+        },
+    },
+    base64::{
+        engine::general_purpose::STANDARD as base64_standard_engine,
+        Engine as _,
+    },
+    derive_more::{
+        Deref,
+        DerefMut,
+    },
+    pyth_sdk::{
+        Price,
+        PriceIdentifier,
+    },
+};
+
+
+/// PriceIdInput is a wrapper around a 32-byte hex string.
+/// that supports a flexible deserialization from a hex string.
+/// It supports both 0x-prefixed and non-prefixed hex strings,
+/// and also supports both lower and upper case characters.
+#[derive(Debug, Clone, Deref, DerefMut)]
+pub struct PriceIdInput([u8; 32]);
+// TODO: Use const generics instead of macro.
+impl_deserialize_for_hex_string_wrapper!(PriceIdInput, 32);
+
+impl From<PriceIdInput> for PriceIdentifier {
+    fn from(id: PriceIdInput) -> Self {
+        Self::new(*id)
+    }
+}
+
+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,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct RpcPriceFeed {
+    pub id:        PriceIdentifier,
+    pub price:     Price,
+    pub ema_price: Price,
+    pub metadata:  Option<RpcPriceFeedMetadata>,
+    /// Vaa binary represented in base64.
+    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 {
+        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)),
+        }
+    }
+}