feat: publisher updates

Author: alexcambose

Cargo.lock

@@ -34,6 +34,7 @@ serde_json = "1.0.128"
 solana-account-decoder = "2.0.13"
 solana-client = "2.0.13"
 solana-sdk = "2.0.13"
+solana-metrics = "2.0.13"
 tokio = { version = "1.40.0", features = ["full"] }
 tokio-stream = "0.1.16"
 tokio-tungstenite = "0.24.0"

Cargo.toml

@@ -9,6 +9,7 @@ use serde::{Deserialize, Deserializer, Serialize};
 use solana_account_decoder::UiAccountEncoding;
 use solana_client::nonblocking::pubsub_client::PubsubClient;
 use solana_client::rpc_config::{RpcAccountInfoConfig, RpcProgramAccountsConfig};
+use solana_metrics::datapoint_info;
 use solana_sdk::account::Account;
 use solana_sdk::commitment_config::CommitmentConfig;
 use solana_sdk::pubkey::Pubkey;
@@ -30,6 +31,15 @@ struct PriceUpdate {
     price_feed: PriceFeed,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+struct PublisherPriceUpdate {
+    #[serde(rename = "type")]
+    update_type: String,
+    publisher: String,
+    feed_id: String,
+    price_info: PriceInfo,
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 struct PriceFeed {
     id: String,
@@ -130,6 +140,10 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
                 continue;
             }
         };
+        // info!(
+        //     "Price Account: {:#?}, account: {:#?} \n\n",
+        //     price_account, account
+        // );
 
         // We want to send price updates whenever the aggregate changes but sometimes the accounts can change without the aggregate changing
         if price_account.agg.status == PriceStatus::Trading
@@ -162,40 +176,88 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
                 },
             };
 
-            let message = serde_json::to_string(&price_update)?;
+            let price_update_message = serde_json::to_string(&price_update)?;
 
             // Create a unique message ID
-            let message_id = format!(
+            let price_update_message_id = format!(
                 "{}:{}",
                 price_update.price_feed.id, price_update.price_feed.price.slot
             );
 
             // Create headers with the Nats-Msg-Id
-            let mut headers = HeaderMap::new();
-            headers.insert("Nats-Msg-Id", message_id.as_str());
+            let mut price_update_headers = HeaderMap::new();
+            price_update_headers.insert("Nats-Msg-Id", price_update_message_id.as_str());
 
             let jetstream_clone = jetstream.clone();
             task::spawn(async move {
                 match jetstream_clone
-                    .publish_with_headers("pyth.price.updates", headers, message.into())
+                    .publish_with_headers(
+                        "pyth.price.updates",
+                        price_update_headers,
+                        price_update_message.into(),
+                    )
                     .await
                 {
                     Ok(_) => debug!(
                         "Published price update to JetStream with ID: {}",
-                        message_id
+                        price_update_message_id
                     ),
                     Err(e) => warn!("Failed to publish price update to JetStream: {}", e),
                 }
             });
 
+            for component in price_account.comp {
+                let jetstream_clone = jetstream.clone();
+                let publisher = component.publisher.to_string();
+                let publisher_price_update_message_id = format!(
+                    "{}:{}:{}",
+                    price_update.price_feed.id, price_update.price_feed.price.slot, publisher
+                );
+
+                let mut publisher_price_updates = HeaderMap::new();
+                publisher_price_updates
+                    .insert("Nats-Msg-Id", publisher_price_update_message_id.as_str());
+
+                let publisher_price_update = PublisherPriceUpdate {
+                    update_type: "publisher_price_update".to_string(),
+                    feed_id: price_account.prod.to_string(),
+                    publisher: publisher,
+                    price_info: PriceInfo {
+                        price: price_account.agg.price.to_string(),
+                        conf: price_account.agg.conf.to_string(),
+                        expo: price_account.expo,
+                        publish_time: price_account.timestamp,
+                        slot: update.context.slot, // Add this field
+                    },
+                };
+                info!("Publisher price update: {:?}", publisher_price_update);
+                let publisher_price_update_message =
+                    serde_json::to_string(&publisher_price_update)?;
+
+                task::spawn(async move {
+                    match jetstream_clone
+                        .publish_with_headers(
+                            "pyth.publisher.updates",
+                            publisher_price_updates,
+                            publisher_price_update_message.into(),
+                        )
+                        .await
+                    {
+                        Ok(_) => debug!(
+                            "Published publisher price update to JetStream with ID: {}",
+                            publisher_price_update_message_id
+                        ),
+                        Err(e) => warn!("Failed to publish price update to JetStream: {}", e),
+                    }
+                });
+            }
             update_count += 1;
             unique_price_feeds.insert(price_account.prod);
         } else {
             debug!("Skipping price update due to invalid status or slot difference");
         }
 
-        // Report aggregate information every minute
-        // TODO: add this as metrics
+        // Report aggregate information every minute and emit metrics
         if last_report_time.elapsed() >= Duration::from_secs(60) {
             info!(
                 "Processed {} updates from {} unique price feeds in the last minute",