Add lazer price for analytics

Cargo.toml

@@ -43,6 +43,7 @@ spl-associated-token-account = "6.0.0"
 spl-token = "7.0.0"
 spl-memo-client = "0.1.0"
 spl-token-2022 = "7.0.0"
+tokio-tungstenite = "0.24.0"
 
 # This patch disables debugging features in litesvm runtime_environments
 # which allows more programs to be loaded in the runtime

auction-server/.env.example

@@ -4,3 +4,6 @@ CLICKHOUSE_URL="http://localhost:18123"
 CLICKHOUSE_NAME="default"
 CLICKHOUSE_USER="default"
 CLICKHOUSE_PASSWORD="clickhouse"
+
+LAZER_API_KEY="lazer_api_key"
+LAZER_URL="wss://pyth-lazer-staging.dourolabs.app/v1/stream"

auction-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "auction-server"
-version = "0.29.0"
+version = "0.30.0"
 edition = "2021"
 license-file = "license.txt"
 
@@ -62,6 +62,7 @@ humantime-serde = "1.1.1"
 tokio-metrics = { version = "0.4.2", features = ["rt"] }
 clickhouse = { version = "0.13.2", features = ["time", "uuid", "native-tls", "inserter"] }
 sha2 = "0.10.9"
+tokio-tungstenite = { workspace = true, features = ["native-tls"] }
 
 [dev-dependencies]
 mockall = "0.13.1"

auction-server/clickhouse_migrations/0005_alter_bid_swap_searcher_usd_price.sql

@@ -0,0 +1 @@
+ALTER TABLE bid_swap RENAME COLUMN searcher_token_usd_price TO searcher_token_notional_usd_value;

auction-server/clickhouse_migrations/0006_alter_bid_swap_user_usd_price.sql

@@ -0,0 +1 @@
+ALTER TABLE bid_swap RENAME COLUMN user_token_usd_price TO user_token_notional_usd_value;

auction-server/clickhouse_migrations/0007_alter_opportunity_swap_sell_mint.sql

@@ -0,0 +1 @@
+ALTER TABLE opportunity_swap RENAME COLUMN sell_token_mint TO searcher_token_mint;

auction-server/clickhouse_migrations/0008_alter_opportunity_swap_sell_amount.sql

@@ -0,0 +1 @@
+ALTER TABLE opportunity_swap RENAME COLUMN sell_token_amount TO searcher_token_amount;

auction-server/clickhouse_migrations/0009_alter_opportunity_swap_sell_price.sql

@@ -0,0 +1 @@
+ALTER TABLE opportunity_swap RENAME COLUMN sell_token_usd_price TO searcher_token_notional_usd_value;

auction-server/clickhouse_migrations/0010_alter_opportunity_swap_buy_mint.sql

@@ -0,0 +1 @@
+ALTER TABLE opportunity_swap RENAME COLUMN buy_token_mint TO user_token_mint;

auction-server/clickhouse_migrations/0011_alter_opportunity_swap_buy_amount.sql

@@ -0,0 +1 @@
+ALTER TABLE opportunity_swap RENAME COLUMN buy_token_amount TO user_token_amount;

auction-server/clickhouse_migrations/0012_alter_opportunity_swap_buy_price.sql

@@ -0,0 +1 @@
+ALTER TABLE opportunity_swap RENAME COLUMN buy_token_usd_price TO user_token_notional_usd_value;

auction-server/clickhouse_migrations/0013_alter_opportunity_limo_buy_price.sql

@@ -0,0 +1 @@
+ALTER TABLE opportunity_limo RENAME COLUMN buy_token_usd_price TO buy_token_notional_usd_value;

auction-server/clickhouse_migrations/0014_alter_opportunity_limo_sell_price.sql

@@ -0,0 +1 @@
+ALTER TABLE opportunity_limo RENAME COLUMN sell_token_usd_price TO sell_token_notional_usd_value;

auction-server/config.sample.yaml

@@ -29,3 +29,12 @@ chains:
         - Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB
         - So11111111111111111111111111111111111111112
         - EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v
+
+lazer:
+  price_feeds:
+    - id: 7
+      mint: EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v
+      exponent: -8
+    - id: 397
+      mint: cbbtcf3aa214zXHbiAZQwf4122FBYbraNdFqgw4iMij
+      exponent: -8

auction-server/src/auction/repository/add_bid_analytics.rs

@@ -9,7 +9,11 @@ use {
             self,
             BidStatus,
         },
-        kernel::entities::Svm,
+        kernel::{
+            entities::Svm,
+            pyth_lazer::calculate_notional_value,
+        },
+        state::Price,
     },
     base64::{
         engine::general_purpose::STANDARD,
@@ -19,6 +23,8 @@ use {
         SubmitBidArgs,
         SwapV2Args,
     },
+    solana_sdk::pubkey::Pubkey,
+    std::collections::HashMap,
 };
 
 impl Repository {
@@ -27,6 +33,8 @@ impl Repository {
         &self,
         bid: entities::Bid,
         data: entities::BidTransactionData,
+        prices: HashMap<Pubkey, Price>,
+        decimals: HashMap<Pubkey, u8>,
     ) -> anyhow::Result<()> {
         let transaction = STANDARD.encode(bincode::serialize(&bid.chain_data.transaction.clone())?);
         let bid_analytics = match data {
@@ -59,6 +67,19 @@ impl Repository {
             }
             entities::BidTransactionData::Swap(transaction_data) => {
                 let status_reason = Svm::get_bid_status_reason(&bid.status);
+                let mint_user = transaction_data.accounts.mint_user;
+                let user_token_notional_usd_value = calculate_notional_value(
+                    prices.get(&mint_user).cloned(),
+                    transaction_data.data.amount_user,
+                    decimals.get(&mint_user).cloned(),
+                );
+                let mint_searcher = transaction_data.accounts.mint_searcher;
+                let searcher_token_notional_usd_value = calculate_notional_value(
+                    prices.get(&mint_searcher).cloned(),
+                    transaction_data.data.amount_searcher,
+                    decimals.get(&mint_searcher).cloned(),
+                );
+
                 let SwapV2Args {
                     fee_token,
                     amount_searcher,
@@ -96,13 +117,11 @@ impl Repository {
 
                     searcher_token_mint: mint_searcher.to_string(),
                     searcher_token_amount: amount_searcher,
-                    // TODO Fill this in
-                    searcher_token_usd_price: None,
+                    searcher_token_notional_usd_value,
 
                     user_token_mint: mint_user.to_string(),
                     user_token_amount: amount_user,
-                    // TODO Fill this in
-                    user_token_usd_price: None,
+                    user_token_notional_usd_value,
 
                     status: serde_json::to_string(&Svm::convert_bid_status(&bid.status))?,
                     status_reason: status_reason

auction-server/src/auction/repository/models.rs

@@ -849,13 +849,13 @@ pub struct BidAnalyticsSwap {
     #[serde(with = "clickhouse::serde::time::datetime64::micros::option")]
     pub conclusion_time: Option<OffsetDateTime>,
 
-    pub searcher_token_mint:      String,
-    pub searcher_token_amount:    u64,
-    pub searcher_token_usd_price: Option<f64>,
+    pub searcher_token_mint:               String,
+    pub searcher_token_amount:             u64,
+    pub searcher_token_notional_usd_value: Option<f64>,
 
-    pub user_token_mint:      String,
-    pub user_token_amount:    u64,
-    pub user_token_usd_price: Option<f64>,
+    pub user_token_mint:               String,
+    pub user_token_amount:             u64,
+    pub user_token_notional_usd_value: Option<f64>,
 
     pub status:        String,
     pub status_reason: Option<String>,

auction-server/src/auction/service/mod.rs

@@ -313,6 +313,7 @@ pub mod tests {
                 secret_key:    "test".to_string(),
                 access_tokens: RwLock::new(HashMap::new()),
                 privileges:    RwLock::new(HashMap::new()),
+                prices:        RwLock::new(HashMap::new()),
             });
             Service(Arc::new(ServiceInner {
                 store,

auction-server/src/auction/service/update_bid_status.rs

@@ -9,8 +9,10 @@ use {
             RestError,
         },
         auction::entities,
+        opportunity::service::get_token_mint::GetTokenMintInput,
     },
     express_relay_api_types::bid::BidStatusWithId,
+    std::collections::HashMap,
 };
 
 pub struct UpdateBidStatusInput {
@@ -19,6 +21,49 @@ pub struct UpdateBidStatusInput {
 }
 
 impl Service {
+    async fn add_bid_analytics(&self, bid: entities::Bid) -> Result<(), RestError> {
+        let transaction_data = self
+            .get_bid_transaction_data(GetBidTransactionDataInput { bid: bid.clone() })
+            .await?;
+        let decimals = match transaction_data.clone() {
+            entities::BidTransactionData::SubmitBid(_) => HashMap::new(),
+            entities::BidTransactionData::Swap(data) => {
+                let searcher_mint = self
+                    .opportunity_service
+                    .get_token_mint(GetTokenMintInput {
+                        chain_id: self.config.chain_id.clone(),
+                        mint:     data.accounts.mint_searcher,
+                    })
+                    .await?;
+
+                let user_mint = self
+                    .opportunity_service
+                    .get_token_mint(GetTokenMintInput {
+                        chain_id: self.config.chain_id.clone(),
+                        mint:     data.accounts.mint_user,
+                    })
+                    .await?;
+
+                HashMap::from([
+                    (data.accounts.mint_searcher, searcher_mint.decimals),
+                    (data.accounts.mint_user, user_mint.decimals),
+                ])
+            }
+        };
+        self.repo
+            .add_bid_analytics(
+                bid.clone(),
+                transaction_data,
+                self.store.prices.read().await.clone(),
+                decimals,
+            )
+            .await
+            .map_err(|e| {
+                tracing::error!(error = ?e, "Failed to add bid analytics");
+                RestError::TemporarilyUnavailable
+            })
+    }
+
     #[tracing::instrument(skip_all, fields(bid_id, status), err(level = tracing::Level::TRACE))]
     pub async fn update_bid_status(&self, input: UpdateBidStatusInput) -> Result<bool, RestError> {
         tracing::Span::current().record("bid_id", input.bid.id.to_string());
@@ -44,22 +89,8 @@ impl Service {
                 let (service, mut bid) = (self.clone(), input.bid.clone());
                 bid.status = input.new_status.clone();
                 async move {
-                    match service
-                        .get_bid_transaction_data(GetBidTransactionDataInput { bid: bid.clone() })
-                        .await
-                    {
-                        Ok(transaction_data) => {
-                            if let Err(e) = service
-                                .repo
-                                .add_bid_analytics(bid.clone(), transaction_data)
-                                .await
-                            {
-                                tracing::error!(error = ?e, "Failed to add bid analytics");
-                            }
-                        }
-                        Err(e) => {
-                            tracing::error!(error = ?e, "Failed to get bid transaction data");
-                        }
+                    if let Err(e) = service.add_bid_analytics(bid.clone()).await {
+                        tracing::error!(bid = ?bid, error = ?e, "Failed to add bid analytics");
                     }
                 }
             });

auction-server/src/config.rs

@@ -90,9 +90,17 @@ pub struct ConfigOptions {
 
 pub type ChainId = String;
 
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct LazerConfig {
+    /// The list of price feeds to subscribe to.
+    pub price_feeds: Vec<crate::kernel::pyth_lazer::PriceFeed>,
+}
+
+
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct ConfigMap {
     pub chains: HashMap<ChainId, Config>,
+    pub lazer:  LazerConfig,
 }
 
 impl ConfigMap {

auction-server/src/config/server.rs

@@ -61,4 +61,12 @@ pub struct Options {
     /// Clickhouse database config to run the migrations for.
     #[command(flatten)]
     pub clickhouse_config:        ClickhouseConfig,
+    /// The Url for the pyth lazer websocket.
+    #[arg(long = "lazer-url")]
+    #[arg(env = "LAZER_URL")]
+    pub lazer_url:                String,
+    /// The API key for the lazer websocket.
+    #[arg(long = "lazer-api-key")]
+    #[arg(env = "LAZER_API_KEY")]
+    pub lazer_api_key:            String,
 }

auction-server/src/kernel/mod.rs

@@ -1,6 +1,9 @@
 pub mod analytics_db;
 pub mod db;
 pub mod entities;
+pub mod pyth_lazer;
+pub mod workers;
+
 #[cfg(test)]
 pub mod rpc_client_svm_tester;
 pub mod traced_sender_svm;