[hermes] Improve latest feeds rest api (#752)

* [hermes] Improve latest feeds rest api

This change adds verbose and binary option to latest_price_feeds
endpoint. Unfortunately it exposes many internal information which
required touching different components to expose batch_vaa specific
information. The code is now coupled to batch_vaa and we need to
refactor it when we add other proof types (and eventually remove it when
it get deprecated).

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

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

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

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

* Fix merge

---------

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

Author: ali-bahjati

hermes/src/network/rpc.rs

@@ -14,6 +14,7 @@ use {
 };
 
 mod rest;
+mod rpc_price_feed;
 
 #[derive(Clone)]
 pub struct State {

hermes/src/network/rpc/rest.rs

@@ -1,4 +1,8 @@
 use {
+    super::rpc_price_feed::{
+        RpcPriceFeed,
+        RpcPriceFeedMetadata,
+    },
     crate::store::RequestTime,
     crate::{
         impl_deserialize_for_hex_string_wrapper,
@@ -23,10 +27,7 @@ use {
         Deref,
         DerefMut,
     },
-    pyth_sdk::{
-        PriceFeed,
-        PriceIdentifier,
-    },
+    pyth_sdk::PriceIdentifier,
 };
 
 /// PriceIdInput is a wrapper around a 32-byte hex string.
@@ -93,8 +94,8 @@ pub async fn latest_vaas(
         .map_err(|_| RestError::UpdateDataNotFound)?;
     Ok(Json(
         price_feeds_with_update_data
-            .update_data
             .batch_vaa
+            .update_data
             .iter()
             .map(|vaa_bytes| base64_standard_engine.encode(vaa_bytes)) // TODO: Support multiple
             // encoding formats
@@ -104,22 +105,42 @@ pub async fn latest_vaas(
 
 #[derive(Debug, serde::Deserialize)]
 pub struct LatestPriceFeedsQueryParams {
-    ids: Vec<PriceIdInput>,
+    ids:     Vec<PriceIdInput>,
+    #[serde(default)]
+    verbose: bool,
+    #[serde(default)]
+    binary:  bool,
 }
 
 pub async fn latest_price_feeds(
     State(state): State<super::State>,
     Query(params): Query<LatestPriceFeedsQueryParams>,
-) -> Result<Json<Vec<PriceFeed>>, RestError> {
+) -> 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
         .store
         .get_price_feeds_with_update_data(price_ids, RequestTime::Latest)
         .map_err(|_| RestError::UpdateDataNotFound)?;
     Ok(Json(
         price_feeds_with_update_data
-            .price_feeds
+            .batch_vaa
+            .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
+            })
             .collect(),
     ))
 }
@@ -152,20 +173,18 @@ pub async fn get_vaa(
         .map_err(|_| RestError::UpdateDataNotFound)?;
 
     let vaa = price_feeds_with_update_data
-        .update_data
         .batch_vaa
+        .update_data
         .get(0)
         .map(|vaa_bytes| base64_standard_engine.encode(vaa_bytes))
         .ok_or(RestError::UpdateDataNotFound)?;
 
     let publish_time = price_feeds_with_update_data
-        .price_feeds
+        .batch_vaa
+        .price_infos
         .get(&price_id)
-        .map(|price_feed| price_feed.get_price_unchecked().publish_time)
+        .map(|price_info| price_info.publish_time)
         .ok_or(RestError::UpdateDataNotFound)?;
-    let publish_time: UnixTimestamp = publish_time
-        .try_into()
-        .map_err(|_| RestError::UpdateDataNotFound)?;
 
     Ok(Json(GetVaaResponse { vaa, publish_time }))
 }
@@ -197,13 +216,11 @@ pub async fn get_vaa_ccip(
         .map_err(|_| RestError::CcipUpdateDataNotFound)?;
 
     let vaa = price_feeds_with_update_data
-        .update_data
         .batch_vaa
+        .update_data
         .get(0) // One price feed has only a single VAA as proof.
         .ok_or(RestError::UpdateDataNotFound)?;
 
-    // FIXME: We should return 5xx when the vaa is not found and 4xx when the price id is not there
-
     Ok(Json(GetVaaCcipResponse {
         data: format!("0x{}", hex::encode(vaa)),
     }))
@@ -221,7 +238,7 @@ pub async fn index() -> impl IntoResponse {
     Json([
         "/live",
         "/api/price_feed_ids",
-        "/api/latest_price_feeds?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..",
+        "/api/latest_price_feeds?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..(&verbose=true)(&binary=true)",
         "/api/latest_vaas?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&...",
         "/api/get_vaa?id=<price_feed_id>&publish_time=<publish_time_in_unix_timestamp>",
         "/api/get_vaa_ccip?data=<0x<price_feed_id_32_bytes>+<publish_time_unix_timestamp_be_8_bytes>>",

hermes/src/network/rpc/rpc_price_feed.rs

@@ -0,0 +1,40 @@
+use {
+    crate::store::UnixTimestamp,
+    pyth_sdk::{
+        Price,
+        PriceFeed,
+        PriceIdentifier,
+    },
+};
+
+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 From<PriceFeed> for RpcPriceFeed {
+    fn from(price_feed: PriceFeed) -> Self {
+        Self {
+            id:        price_feed.id,
+            price:     price_feed.get_price_unchecked(),
+            ema_price: price_feed.get_ema_price_unchecked(),
+            metadata:  None,
+            vaa:       None,
+        }
+    }
+}

hermes/src/store.rs

@@ -1,18 +1,11 @@
 use {
-    self::storage::Storage,
-    anyhow::Result,
-    pyth_sdk::{
-        PriceFeed,
-        PriceIdentifier,
-    },
-    serde::{
-        Deserialize,
-        Serialize,
-    },
-    std::{
-        collections::HashMap,
-        sync::Arc,
+    self::{
+        proof::batch_vaa::PriceInfosWithUpdateData,
+        storage::Storage,
     },
+    anyhow::Result,
+    pyth_sdk::PriceIdentifier,
+    std::sync::Arc,
 };
 
 mod proof;
@@ -30,19 +23,8 @@ pub enum Update {
     Vaa(Vec<u8>),
 }
 
-#[derive(Clone, Default, Serialize, Deserialize)]
-pub struct UpdateData {
-    pub batch_vaa: Vec<Vec<u8>>,
-}
-
-// TODO: A price feed might not have update data in all different
-// formats. For example, Batch VAA and Merkle updates will result
-// in different price feeds. We need to figure out how to handle
-// it properly.
-#[derive(Clone, Default)]
 pub struct PriceFeedsWithUpdateData {
-    pub price_feeds: HashMap<PriceIdentifier, PriceFeed>,
-    pub update_data: UpdateData,
+    pub batch_vaa: PriceInfosWithUpdateData,
 }
 
 pub type State = Arc<Box<dyn Storage>>;
@@ -75,11 +57,13 @@ impl Store {
         price_ids: Vec<PriceIdentifier>,
         request_time: RequestTime,
     ) -> Result<PriceFeedsWithUpdateData> {
-        proof::batch_vaa::get_price_feeds_with_update_data(
-            self.state.clone(),
-            price_ids,
-            request_time,
-        )
+        Ok(PriceFeedsWithUpdateData {
+            batch_vaa: proof::batch_vaa::get_price_infos_with_update_data(
+                self.state.clone(),
+                price_ids,
+                request_time,
+            )?,
+        })
     }
 
     pub fn get_price_feed_ids(&self) -> Vec<PriceIdentifier> {

hermes/src/store/proof/batch_vaa.rs

@@ -4,11 +4,9 @@ use {
             Key,
             StorageData,
         },
-        PriceFeedsWithUpdateData,
         RequestTime,
         State,
         UnixTimestamp,
-        UpdateData,
     },
     anyhow::{
         anyhow,
@@ -24,38 +22,59 @@ use {
         PriceAttestation,
         PriceStatus,
     },
-    std::collections::{
-        HashMap,
-        HashSet,
+    std::{
+        collections::{
+            HashMap,
+            HashSet,
+        },
+        time::{
+            SystemTime,
+            UNIX_EPOCH,
+        },
     },
     wormhole::VAA,
 };
 
 // TODO: We need to add more metadata to this struct.
 #[derive(Clone, Default, PartialEq, Debug)]
 pub struct PriceInfo {
-    pub price_feed:   PriceFeed,
-    pub vaa_bytes:    Vec<u8>,
-    pub publish_time: UnixTimestamp,
+    pub price_feed:       PriceFeed,
+    pub vaa_bytes:        Vec<u8>,
+    pub publish_time:     UnixTimestamp,
+    pub emitter_chain:    u16,
+    pub attestation_time: UnixTimestamp,
+    pub receive_time:     UnixTimestamp,
+    pub sequence_number:  u64,
 }
 
+#[derive(Clone, Default)]
+pub struct PriceInfosWithUpdateData {
+    pub price_infos: HashMap<PriceIdentifier, PriceInfo>,
+    pub update_data: Vec<Vec<u8>>,
+}
 
 pub fn store_vaa_update(state: State, vaa_bytes: Vec<u8>) -> Result<()> {
     // FIXME: Vaa bytes might not be a valid Pyth BatchUpdate message nor originate from Our emitter.
     // We should check that.
+    // FIXME: We receive multiple vaas for the same update (due to different signedVAAs). We need
+    // to drop them.
     let vaa = VAA::from_bytes(&vaa_bytes)?;
     let batch_price_attestation = BatchPriceAttestation::deserialize(vaa.payload.as_slice())
         .map_err(|_| anyhow!("Failed to deserialize VAA"))?;
 
     for price_attestation in batch_price_attestation.price_attestations {
-        let price_feed = price_attestation_to_price_feed(price_attestation);
+        let price_feed = price_attestation_to_price_feed(price_attestation.clone());
 
         let publish_time = price_feed.get_price_unchecked().publish_time.try_into()?;
 
         let price_info = PriceInfo {
             price_feed,
             vaa_bytes: vaa_bytes.clone(),
             publish_time,
+            emitter_chain: vaa.emitter_chain.into(),
+            attestation_time: price_attestation.attestation_time.try_into()?,
+            receive_time: SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs(),
+            sequence_number: vaa.sequence,
         };
 
         let key = Key::BatchVaa(price_feed.id);
@@ -65,32 +84,30 @@ pub fn store_vaa_update(state: State, vaa_bytes: Vec<u8>) -> Result<()> {
 }
 
 
-pub fn get_price_feeds_with_update_data(
+pub fn get_price_infos_with_update_data(
     state: State,
     price_ids: Vec<PriceIdentifier>,
     request_time: RequestTime,
-) -> Result<PriceFeedsWithUpdateData> {
-    let mut price_feeds = HashMap::new();
+) -> Result<PriceInfosWithUpdateData> {
+    let mut price_infos = HashMap::new();
     let mut vaas: HashSet<Vec<u8>> = HashSet::new();
     for price_id in price_ids {
         let key = Key::BatchVaa(price_id);
         let maybe_data = state.get(key, request_time.clone())?;
 
         match maybe_data {
             Some(StorageData::BatchVaa(price_info)) => {
-                price_feeds.insert(price_info.price_feed.id, price_info.price_feed);
-                vaas.insert(price_info.vaa_bytes);
+                vaas.insert(price_info.vaa_bytes.clone());
+                price_infos.insert(price_id, price_info);
             }
             None => {
                 return Err(anyhow!("No price feed found for price id: {:?}", price_id));
             }
         }
     }
-    let update_data = UpdateData {
-        batch_vaa: vaas.into_iter().collect(),
-    };
-    Ok(PriceFeedsWithUpdateData {
-        price_feeds,
+    let update_data: Vec<Vec<u8>> = vaas.into_iter().collect();
+    Ok(PriceInfosWithUpdateData {
+        price_infos,
         update_data,
     })
 }

hermes/src/store/storage.rs

@@ -5,10 +5,6 @@ use {
         UnixTimestamp,
     },
     anyhow::Result,
-    derive_more::{
-        Deref,
-        DerefMut,
-    },
     pyth_sdk::PriceIdentifier,
 };