feat(hermes): add sse endpoint (#1425)

* add initial sse code

* fix typo

* add more error handling

* fix formatting

* revert import format

* add error handling for nonexistent price feeds in the middle of sub

* refactor

* format

* add comment

* Update hermes/src/api/sse.rs

Co-authored-by: Reisen <Reisen@users.noreply.github.com>

* refactor

* bump

---------

Co-authored-by: Reisen <Reisen@users.noreply.github.com>

hermes/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hermes"
-version = "0.5.3"
+version = "0.5.4"
 description = "Hermes is an agent that provides Verified Prices from the Pythnet Pyth Oracle."
 edition = "2021"
 
@@ -42,6 +42,7 @@ serde_wormhole = { git = "https://github.com/wormhole-foundation/wormhol
 sha3 = { version = "0.10.4" }
 strum = { version = "0.24.1", features = ["derive"] }
 tokio = { version = "1.26.0", features = ["full"] }
+tokio-stream = { version = "0.1.15", features = ["full"] }
 tonic = { version = "0.10.1", features = ["tls"] }
 tower-http = { version = "0.4.0", features = ["cors"] }
 tracing = { version = "0.1.37", features = ["log"] }

hermes/src/api.rs

@@ -23,6 +23,7 @@ use {
 mod doc_examples;
 mod metrics_middleware;
 mod rest;
+mod sse;
 pub mod types;
 mod ws;
 
@@ -143,6 +144,10 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> {
  .route("/api/latest_price_feeds", get(rest::latest_price_feeds))
  .route("/api/latest_vaas", get(rest::latest_vaas))
  .route("/api/price_feed_ids", get(rest::price_feed_ids))
+ .route(
+ "/v2/updates/price/stream",
+ get(sse::price_stream_sse_handler),
+ )
  .route("/v2/updates/price/latest", get(rest::latest_price_updates))
  .route(
  "/v2/updates/price/:publish_time",

hermes/src/api/rest.rs

@@ -21,6 +21,7 @@ mod price_feed_ids;
 mod ready;
 mod v2;
 
+
 pub use {
  get_price_feed::*,
  get_vaa::*,
@@ -38,6 +39,7 @@ pub use {
  },
 };
 
+#[derive(Debug)]
 pub enum RestError {
  BenchmarkPriceNotUnique,
  UpdateDataNotFound,

hermes/src/api/sse.rs

@@ -0,0 +1,173 @@
+use {
+ crate::{
+ aggregate::{
+ AggregationEvent,
+ RequestTime,
+ },
+ api::{
+ rest::{
+ verify_price_ids_exist,
+ RestError,
+ },
+ types::{
+ BinaryPriceUpdate,
+ EncodingType,
+ ParsedPriceUpdate,
+ PriceIdInput,
+ PriceUpdate,
+ },
+ ApiState,
+ },
+ },
+ anyhow::Result,
+ axum::{
+ extract::State,
+ response::sse::{
+ Event,
+ KeepAlive,
+ Sse,
+ },
+ },
+ futures::Stream,
+ pyth_sdk::PriceIdentifier,
+ serde::Deserialize,
+ serde_qs::axum::QsQuery,
+ std::convert::Infallible,
+ tokio::sync::broadcast,
+ tokio_stream::{
+ wrappers::BroadcastStream,
+ StreamExt as _,
+ },
+ utoipa::IntoParams,
+};
+
+#[derive(Debug, Deserialize, IntoParams)]
+#[into_params(parameter_in = Query)]
+pub struct StreamPriceUpdatesQueryParams {
+ /// Get the most recent price update for this set of price feed ids.
+ ///
+ /// This parameter can be provided multiple times to retrieve multiple price updates,
+ /// for example see the following query string:
+ ///
+ /// ```
+ /// ?ids[]=a12...&ids[]=b4c...
+ /// ```
+ #[param(rename = "ids[]")]
+ #[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")]
+ ids: Vec<PriceIdInput>,
+
+ /// If true, include the parsed price update in the `parsed` field of each returned feed.
+ #[serde(default)]
+ encoding: EncodingType,
+
+ /// If true, include the parsed price update in the `parsed` field of each returned feed.
+ #[serde(default = "default_true")]
+ parsed: bool,
+}
+
+fn default_true() -> bool {
+ true
+}
+
+#[utoipa::path(
+ get,
+ path = "/v2/updates/price/stream",
+ responses(
+ (status = 200, description = "Price updates retrieved successfully", body = PriceUpdate),
+ (status = 404, description = "Price ids not found", body = String)
+ ),
+ params(StreamPriceUpdatesQueryParams)
+)]
+/// SSE route handler for streaming price updates.
+pub async fn price_stream_sse_handler(
+ State(state): State<ApiState>,
+ QsQuery(params): QsQuery<StreamPriceUpdatesQueryParams>,
+) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, RestError> {
+ let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(Into::into).collect();
+
+ verify_price_ids_exist(&state, &price_ids).await?;
+
+ // Clone the update_tx receiver to listen for new price updates
+ let update_rx: broadcast::Receiver<AggregationEvent> = state.update_tx.subscribe();
+
+ // Convert the broadcast receiver into a Stream
+ let stream = BroadcastStream::new(update_rx);
+
+ let sse_stream = stream.then(move |message| {
+ let state_clone = state.clone(); // Clone again to use inside the async block
+ let price_ids_clone = price_ids.clone(); // Clone again for use inside the async block
+ async move {
+ match message {
+ Ok(event) => {
+ match handle_aggregation_event(
+ event,
+ state_clone,
+ price_ids_clone,
+ params.encoding,
+ params.parsed,
+ )
+ .await
+ {
+ Ok(price_update) => Ok(Event::default().json_data(price_update).unwrap()),
+ Err(e) => Ok(error_event(e)),
+ }
+ }
+ Err(e) => Ok(error_event(e)),
+ }
+ }
+ });
+
+ Ok(Sse::new(sse_stream).keep_alive(KeepAlive::default()))
+}
+
+async fn handle_aggregation_event(
+ event: AggregationEvent,
+ state: ApiState,
+ mut price_ids: Vec<PriceIdentifier>,
+ encoding: EncodingType,
+ parsed: bool,
+) -> Result<PriceUpdate> {
+ // We check for available price feed ids to ensure that the price feed ids provided exists since price feeds can be removed.
+ let available_price_feed_ids = crate::aggregate::get_price_feed_ids(&*state.state).await;
+
+ price_ids.retain(|price_feed_id| available_price_feed_ids.contains(price_feed_id));
+
+ let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data(
+ &*state.state,
+ &price_ids,
+ RequestTime::AtSlot(event.slot()),
+ )
+ .await?;
+ let price_update_data = price_feeds_with_update_data.update_data;
+ let encoded_data: Vec<String> = price_update_data
+ .into_iter()
+ .map(|data| encoding.encode_str(&data))
+ .collect();
+ let binary_price_update = BinaryPriceUpdate {
+ encoding,
+ data: encoded_data,
+ };
+ let parsed_price_updates: Option<Vec<ParsedPriceUpdate>> = if parsed {
+ Some(
+ price_feeds_with_update_data
+ .price_feeds
+ .into_iter()
+ .map(|price_feed| price_feed.into())
+ .collect(),
+ )
+ } else {
+ None
+ };
+
+
+ Ok(PriceUpdate {
+ binary: binary_price_update,
+ parsed: parsed_price_updates,
+ })
+}
+
+fn error_event<E: std::fmt::Debug>(e: E) -> Event {
+ Event::default()
+ .event("error")
+ .data(format!("Error receiving update: {:?}", e))
+}

hermes/src/main.rs

@@ -28,14 +28,14 @@ mod state;
 
 lazy_static! {
  /// A static exit flag to indicate to running threads that we're shutting down. This is used to
- /// gracefully shutdown the application.
+ /// gracefully shut down the application.
  ///
  /// We make this global based on the fact the:
  /// - The `Sender` side does not rely on any async runtime.
  /// - Exit logic doesn't really require carefully threading this value through the app.
  /// - The `Receiver` side of a watch channel performs the detection based on if the change
  /// happened after the subscribe, so it means all listeners should always be notified
- /// currectly.
+ /// correctly.
  pub static ref EXIT: watch::Sender<bool> = watch::channel(false).0;
 }