add yellowstone grpc

src/grpc/event.rs

@@ -0,0 +1,123 @@
+use anyhow::anyhow;
+use base64::engine::general_purpose;
+use base64::Engine;
+use borsh::{BorshDeserialize, BorshSerialize};
+use regex::Regex;
+use solana_sdk::pubkey::Pubkey;
+
+use super::myerror::AppError;
+
+pub const PROGRAM_DATA: &str = "Program data: ";
+
+#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)]
+pub struct TradeEvent {
+    pub mint: Pubkey,
+    pub sol_amount: u64,
+    pub token_amount: u64,
+    pub is_buy: bool,
+    pub user: Pubkey,
+    pub timestamp: u64,
+    pub virtual_sol_reserves: u64,
+    pub virtual_token_reserves: u64,
+    pub real_sol_reserves: u64,
+    pub real_token_reserves: u64,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)]
+pub struct CompleteEvent {
+    pub user: Pubkey,
+    pub mint: Pubkey,
+    pub bonding_curve: Pubkey,
+    pub timestamp: u64,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)]
+pub struct SwapBaseInLog {
+    pub log_type: u8,
+    // input
+    pub amount_in: u64,
+    pub minimum_out: u64,
+    pub direction: u64,
+    // user info
+    pub user_source: u64,
+    // pool info
+    pub pool_coin: u64,
+    pub pool_pc: u64,
+    // calc result
+    pub out_amount: u64,
+}
+
+pub trait EventTrait: Sized + std::fmt::Debug {
+    fn from_bytes(bytes: &[u8]) -> Result<Self, AppError>;
+}
+
+impl EventTrait for TradeEvent {
+    fn from_bytes(bytes: &[u8]) -> Result<Self, AppError> {
+        TradeEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string())))
+    }
+}
+
+impl EventTrait for CompleteEvent {
+    fn from_bytes(bytes: &[u8]) -> Result<Self, AppError> {
+        CompleteEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string())))
+    }
+}
+
+impl EventTrait for SwapBaseInLog {
+    fn from_bytes(bytes: &[u8]) -> Result<Self, AppError> {
+        SwapBaseInLog::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string())))
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct PumpEvent {}
+
+impl PumpEvent {
+    pub fn parse_logs<T: EventTrait + Clone>(logs: &Vec<String>) -> Option<T> {
+        let mut event: Option<T> = None;
+        if !logs.is_empty() {
+            let logs_iter = logs.iter().peekable();
+
+            for l in logs_iter.rev() {
+                if let Some(log) = l.strip_prefix(PROGRAM_DATA) {
+                    let borsh_bytes = general_purpose::STANDARD.decode(log).unwrap();
+                    let slice: &[u8] = &borsh_bytes[8..];
+
+                    if let Ok(e) = T::from_bytes(slice) {
+                        event = Some(e);
+                    }
+                }
+            }
+        }
+        event
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct RaydiumEvent {}
+
+impl RaydiumEvent {
+    pub fn parse_logs<T: EventTrait + Clone>(logs: &Vec<String>) -> Option<T> {
+        let mut event: Option<T> = None;
+
+        if !logs.is_empty() {
+            let logs_iter = logs.iter().peekable();
+
+            for l in logs_iter.rev() {
+                let re = Regex::new(r"ray_log: (?P<base64>[A-Za-z0-9+/=]+)").unwrap();
+
+                if let Some(caps) = re.captures(l) {
+                    if let Some(base64) = caps.name("base64") {
+                        let bytes = general_purpose::STANDARD.decode(base64.as_str()).unwrap();
+
+                        if let Ok(e) = T::from_bytes(&bytes) {
+                            event = Some(e);
+                        }
+                    }
+                }
+            }
+        }
+
+        event
+    }
+}

src/grpc/mod.rs

@@ -0,0 +1,7 @@
+
+pub mod event;
+pub mod myerror;
+// pub mod subscribe_logs;
+// pub mod subscribe_tx;
+pub mod yellowstone_grpc;
+

src/grpc/myerror.rs

@@ -0,0 +1,13 @@
+use anyhow::Error;
+#[derive(Debug)]
+#[allow(dead_code)]
+pub struct AppError(Error);
+
+impl<E> From<E> for AppError
+where
+    E: Into<Error>,
+{
+    fn from(err: E) -> Self {
+        Self(err.into())
+    }
+}

src/grpc/subscribe_logs.rs

@@ -0,0 +1,136 @@
+#[cfg(test)]
+mod subscribe_tx_tests {
+    use crate::common::{
+        event::{PumpEvent, RaydiumEvent, SwapBaseInLog, TradeEvent},
+        myerror::AppError,
+        yellowstone_grpc::{TransactionPretty, YellowstoneGrpc},
+    };
+    use anyhow::anyhow;
+    use chrono::Local;
+    use dotenvy::dotenv;
+    use futures::{channel::mpsc, sink::SinkExt, stream::StreamExt};
+    use log::{error, info};
+    use solana_sdk::pubkey;
+    use solana_sdk::pubkey::Pubkey;
+    use solana_transaction_status::{
+        option_serializer::OptionSerializer, EncodedTransactionWithStatusMeta,
+    };
+    use std::env;
+    use tokio::test;
+    use yellowstone_grpc_proto::geyser::{
+        subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing,
+    };
+    const AMM_V4: Pubkey = pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8");
+    const PUMP_PROGRAM_ID: Pubkey = pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P");
+    pub enum SwapType {
+        Pump,
+        Raydium,
+    }
+
+    #[test]
+    async fn test_subscribe_tx() -> Result<(), AppError> {
+        dotenv().ok();
+        pretty_env_logger::init_custom_env("RUST_LOG");
+        let yellowstone_url = env::var("YELLOWSTONE_URL")?;
+
+        info!("::: {:?}", yellowstone_url);
+        let yellowstone_grpc = YellowstoneGrpc::new(yellowstone_url);
+
+        let addrs = vec![
+            "Aa4QWNkS3RLUv7DA9BM1a2Hzm4HDQo5PyRefqDJnpump".to_string(),
+            "BnDssYyGDF9aj5j2N5BwsJFk9YMneQ8P7LQkoYkrpump".to_string(),
+        ];
+        let transactions = yellowstone_grpc.subscribe_transaction(addrs, vec![], vec![]);
+
+        let (mut subscribe_tx, mut stream) = yellowstone_grpc.connect(transactions).await??;
+
+        let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(1000);
+
+        tokio::spawn(async move {
+            while let Some(message) = stream.next().await {
+                match message {
+                    Ok(msg) => {
+                        match msg.update_oneof {
+                            Some(UpdateOneof::Transaction(sut)) => {
+                                let transaction_pretty: TransactionPretty = sut.into();
+                                let _ = tx.try_send(transaction_pretty);
+                            }
+                            Some(UpdateOneof::Ping(_)) => {
+                                // This is necessary to keep load balancers that expect client pings alive. If your load balancer doesn't
+                                // require periodic client pings then this is unnecessary
+                                let _ = subscribe_tx
+                                    .send(SubscribeRequest {
+                                        ping: Some(SubscribeRequestPing { id: 1 }),
+                                        ..Default::default()
+                                    })
+                                    .await;
+                                info!("service is ping: {}", Local::now());
+                            }
+                            Some(UpdateOneof::Pong(_)) => {
+                                info!("service is pong: {}", Local::now());
+                            }
+                            _ => {}
+                        }
+                        continue;
+                    }
+                    Err(error) => {
+                        error!("error: {error:?}");
+                        break;
+                    }
+                }
+            }
+        });
+
+        while let Some(transaction_pretty) = rx.next().await {
+            let trade_raw = transaction_pretty.tx.clone();
+            let meta = &trade_raw.meta.clone().unwrap();
+            if meta.err.is_some() {
+                continue;
+            }
+
+            let logs = if let OptionSerializer::Some(logs) = &meta.log_messages {
+                logs
+            } else {
+                &vec![]
+            };
+
+            if let Ok(swap_type) = get_swap_type(&trade_raw) {
+                match swap_type {
+                    SwapType::Raydium => {
+                        let event = RaydiumEvent::parse_logs::<SwapBaseInLog>(logs);
+                        info!("RaydiumEvent {:#?}", event);
+                    }
+                    SwapType::Pump => {
+                        let event = PumpEvent::parse_logs::<TradeEvent>(logs);
+                        info!("PumpEvent {:#?}", event);
+                    }
+                }
+            };
+        }
+        Ok(())
+    }
+
+    #[allow(dead_code)]
+    pub fn get_swap_type(
+        trade_raw: &EncodedTransactionWithStatusMeta,
+    ) -> Result<SwapType, AppError> {
+        let transaction = &trade_raw.transaction.decode();
+        if let Some(transaction) = transaction {
+            let account_keys = transaction.message.static_account_keys();
+
+            let program_index = account_keys
+                .iter()
+                .position(|item| item == &AMM_V4 || item == &PUMP_PROGRAM_ID)
+                .ok_or(anyhow!("swap type program_id not found"))?;
+
+            let program_id = account_keys[program_index];
+            let _type = match program_id {
+                AMM_V4 => Ok(SwapType::Raydium),
+                PUMP_PROGRAM_ID => Ok(SwapType::Pump),
+                _ => Err(AppError::from(anyhow!("program_id ix not found"))),
+            };
+            return _type;
+        }
+        Err(AppError::from(anyhow!("program_id ix not found")))
+    }
+}

src/grpc/subscribe_tx.rs

@@ -0,0 +1,73 @@
+#[cfg(test)]
+mod subscribe_tx_tests {
+    use crate::common::{
+        myerror::AppError,
+        yellowstone_grpc::{TransactionPretty, YellowstoneGrpc},
+    };
+    use chrono::Local;
+    use dotenvy::dotenv;
+    use futures::{channel::mpsc, sink::SinkExt, stream::StreamExt};
+    use log::{error, info};
+    use std::env;
+    use tokio::test;
+    use yellowstone_grpc_proto::geyser::{
+        subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing,
+    };
+
+    #[test]
+    async fn test_subscribe_tx() -> Result<(), AppError> {
+        dotenv().ok();
+        pretty_env_logger::init_custom_env("RUST_LOG");
+        let yellowstone_url = env::var("YELLOWSTONE_URL")?;
+
+        info!("::: {:?}", yellowstone_url);
+        let yellowstone_grpc = YellowstoneGrpc::new(yellowstone_url);
+
+        let addrs = vec!["6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P".to_string()];
+        let transactions = yellowstone_grpc.subscribe_transaction(addrs, vec![], vec![]);
+
+        let (mut subscribe_tx, mut stream) = yellowstone_grpc.connect(transactions).await??;
+
+        let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(1000);
+
+        tokio::spawn(async move {
+            while let Some(message) = stream.next().await {
+                match message {
+                    Ok(msg) => {
+                        match msg.update_oneof {
+                            Some(UpdateOneof::Transaction(sut)) => {
+                                let transaction_pretty: TransactionPretty = sut.into();
+                                let _ = tx.try_send(transaction_pretty);
+                            }
+                            Some(UpdateOneof::Ping(_)) => {
+                                // This is necessary to keep load balancers that expect client pings alive. If your load balancer doesn't
+                                // require periodic client pings then this is unnecessary
+                                let _ = subscribe_tx
+                                    .send(SubscribeRequest {
+                                        ping: Some(SubscribeRequestPing { id: 1 }),
+                                        ..Default::default()
+                                    })
+                                    .await;
+                                info!("service is ping: {}", Local::now());
+                            }
+                            Some(UpdateOneof::Pong(_)) => {
+                                info!("service is pong: {}", Local::now());
+                            }
+                            _ => {}
+                        }
+                        continue;
+                    }
+                    Err(error) => {
+                        error!("error: {error:?}");
+                        break;
+                    }
+                }
+            }
+        });
+
+        while let Some(event) = rx.next().await {
+            info!("TransactionPretty {:#?}", event);
+        }
+        Ok(())
+    }
+}

src/grpc/yellowstone_grpc.rs

@@ -0,0 +1,284 @@
+use std::{collections::HashMap, fmt, time::Duration};
+
+use futures::{channel::mpsc, sink::Sink, Stream, StreamExt, SinkExt};
+use rustls::crypto::{ring::default_provider, CryptoProvider};
+use tonic::{transport::channel::ClientTlsConfig, Status};
+use yellowstone_grpc_client::{GeyserGrpcClient, GeyserGrpcClientResult};
+use yellowstone_grpc_proto::geyser::{
+    CommitmentLevel, SubscribeRequest, SubscribeRequestFilterTransactions, SubscribeUpdate,
+    SubscribeUpdateTransaction, subscribe_update::UpdateOneof, SubscribeRequestPing,
+};
+use log::{error, info};
+use chrono::Local;
+use solana_sdk::{pubkey, pubkey::Pubkey, signature::Signature};
+use solana_transaction_status::{
+    option_serializer::OptionSerializer, EncodedTransactionWithStatusMeta, UiTransactionEncoding,
+};
+use anyhow::anyhow;
+
+use crate::grpc::event::{PumpEvent, RaydiumEvent, SwapBaseInLog, TradeEvent};
+use crate::grpc::myerror::AppError;
+
+// 类型别名定义
+type TransactionsFilterMap = HashMap<String, SubscribeRequestFilterTransactions>;
+type GrpcStreamResult = GeyserGrpcClientResult<(
+    Box<dyn Sink<SubscribeRequest, Error = mpsc::SendError> + Unpin + Send>,
+    Box<dyn Stream<Item = Result<SubscribeUpdate, Status>> + Unpin + Send>,
+)>;
+
+// 常量定义
+const AMM_V4: Pubkey = pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8");
+const PUMP_PROGRAM_ID: Pubkey = pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P");
+const CONNECT_TIMEOUT: u64 = 10;
+const REQUEST_TIMEOUT: u64 = 60;
+const CHANNEL_SIZE: usize = 1000;
+
+// 枚举定义
+#[derive(Debug)]
+pub enum SwapType {
+    Pump,
+    Raydium,
+}
+
+// 结构体定义
+#[allow(dead_code)]
+pub struct TransactionPretty {
+    pub slot: u64,
+    pub signature: Signature,
+    pub is_vote: bool,
+    pub tx: EncodedTransactionWithStatusMeta,
+}
+
+impl fmt::Debug for TransactionPretty {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        struct TxWrap<'a>(&'a EncodedTransactionWithStatusMeta);
+        impl<'a> fmt::Debug for TxWrap<'a> {
+            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+                let serialized = serde_json::to_string(self.0).expect("failed to serialize");
+                fmt::Display::fmt(&serialized, f)
+            }
+        }
+
+        f.debug_struct("TransactionPretty")
+            .field("slot", &self.slot)
+            .field("signature", &self.signature)
+            .field("is_vote", &self.is_vote)
+            .field("tx", &TxWrap(&self.tx))
+            .finish()
+    }
+}
+
+impl From<SubscribeUpdateTransaction> for TransactionPretty {
+    fn from(SubscribeUpdateTransaction { transaction, slot }: SubscribeUpdateTransaction) -> Self {
+        let tx = transaction.expect("should be defined");
+        Self {
+            slot,
+            signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),
+            is_vote: tx.is_vote,
+            tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx)
+                .expect("valid tx with meta")
+                .encode(UiTransactionEncoding::Base64, Some(u8::MAX), true)
+                .expect("failed to encode"),
+        }
+    }
+}
+
+pub struct YellowstoneGrpc {
+    endpoint: String,
+}
+
+impl YellowstoneGrpc {
+    pub fn new(endpoint: String) -> Self {
+        Self { endpoint }
+    }
+
+    pub async fn connect(
+        &self,
+        transactions: TransactionsFilterMap,
+    ) -> Result<
+        GeyserGrpcClientResult<(
+            impl Sink<SubscribeRequest, Error = mpsc::SendError>,
+            impl Stream<Item = Result<SubscribeUpdate, Status>>,
+        )>,
+        AppError,
+    > {
+        if CryptoProvider::get_default().is_none() {
+            default_provider()
+                .install_default()
+                .map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e))?;
+        }
+
+        let mut client = GeyserGrpcClient::build_from_shared(self.endpoint.clone())?
+            .tls_config(ClientTlsConfig::new().with_native_roots())?
+            .connect_timeout(Duration::from_secs(10))
+            .timeout(Duration::from_secs(60))
+            .connect()
+            .await?;
+
+        let subscribe_request = SubscribeRequest {
+            transactions,
+            commitment: Some(CommitmentLevel::Processed.into()),
+            ..Default::default()
+        };
+
+        Ok(client.subscribe_with_request(Some(subscribe_request)).await)
+    }
+
+    pub async fn subscribe_accounts(&self, accounts: Vec<String>) -> Result<(), AppError> {
+        let transactions = self.get_subscribe_request_filter(accounts, vec![], vec![]);
+        let (mut subscribe_tx, mut stream) = self.connect(transactions).await??;
+        let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(CHANNEL_SIZE);
+
+        tokio::spawn(async move {
+            while let Some(message) = stream.next().await {
+                match message {
+                    Ok(msg) => {
+                        if let Err(e) = Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await {
+                            error!("Error handling message: {:?}", e);
+                            break;
+                        }
+                    }
+                    Err(error) => {
+                        error!("Stream error: {error:?}");
+                        break;
+                    }
+                }
+            }
+        });
+
+        while let Some(transaction_pretty) = rx.next().await {
+            if let Err(e) = Self::process_transaction(transaction_pretty).await {
+                error!("Error processing transaction: {:?}", e);
+            }
+        }
+        Ok(())
+    }
+
+    pub async fn subscribe_pumpfun(&self) -> Result<(), AppError> {
+        let addrs = vec![PUMP_PROGRAM_ID.to_string()];
+        let transactions = self.get_subscribe_request_filter(addrs, vec![], vec![]);
+        let (mut subscribe_tx, mut stream) = self.connect(transactions).await??;
+        let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(CHANNEL_SIZE);
+
+        tokio::spawn(async move {
+            while let Some(message) = stream.next().await {
+                match message {
+                    Ok(msg) => {
+                        if let Err(e) = Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await {
+                            error!("Error handling message: {:?}", e);
+                            break;
+                        }
+                    }
+                    Err(error) => {
+                        error!("Stream error: {error:?}");
+                        break;
+                    }
+                }
+            }
+        });
+
+        while let Some(transaction_pretty) = rx.next().await {
+            if let Err(e) = Self::process_transaction(transaction_pretty).await {
+                error!("Error processing transaction: {:?}", e);
+            }
+        }
+        Ok(())
+    }
+
+    pub fn get_subscribe_request_filter(
+        &self,
+        account_include: Vec<String>,
+        account_exclude: Vec<String>,
+        account_required: Vec<String>,
+    ) -> TransactionsFilterMap {
+        let mut transactions = HashMap::new();
+        transactions.insert(
+            "client".to_string(),
+            SubscribeRequestFilterTransactions {
+                vote: Some(false),
+                failed: Some(false),
+                signature: None,
+                account_include,
+                account_exclude,
+                account_required,
+            },
+        );
+        transactions
+    }
+
+
+    async fn handle_stream_message(
+        msg: SubscribeUpdate,
+        tx: &mut mpsc::Sender<TransactionPretty>,
+        subscribe_tx: &mut (impl Sink<SubscribeRequest, Error = mpsc::SendError> + Unpin),
+    ) -> Result<(), AppError> {
+        match msg.update_oneof {
+            Some(UpdateOneof::Transaction(sut)) => {
+                let transaction_pretty = TransactionPretty::from(sut);
+                tx.try_send(transaction_pretty).map_err(|e| AppError::from(anyhow!("Send error: {:?}", e)))?;
+            }
+            Some(UpdateOneof::Ping(_)) => {
+                subscribe_tx
+                    .send(SubscribeRequest {
+                        ping: Some(SubscribeRequestPing { id: 1 }),
+                        ..Default::default()
+                    })
+                    .await
+                    .map_err(|e| AppError::from(anyhow!("Ping error: {:?}", e)))?;
+                info!("service is ping: {}", Local::now());
+            }
+            Some(UpdateOneof::Pong(_)) => {
+                info!("service is pong: {}", Local::now());
+            }
+            _ => {}
+        }
+        Ok(())
+    }
+
+    async fn process_transaction(transaction_pretty: TransactionPretty) -> Result<(), AppError> {
+        let trade_raw = transaction_pretty.tx;
+        let meta = trade_raw.meta.as_ref()
+            .ok_or_else(|| AppError::from(anyhow!("Missing transaction metadata")))?;
+            
+        if meta.err.is_some() {
+            return Ok(());
+        }
+
+        let logs = if let OptionSerializer::Some(logs) = &meta.log_messages {
+            logs
+        } else {
+            &vec![]
+        };
+
+        if let Ok(swap_type) = Self::get_swap_type(&trade_raw) {
+            match swap_type {
+                SwapType::Raydium => {
+                    let event = RaydiumEvent::parse_logs::<SwapBaseInLog>(logs);
+                    info!("RaydiumEvent {:#?}", event);
+                }
+                SwapType::Pump => {
+                    let event = PumpEvent::parse_logs::<TradeEvent>(logs);
+                    info!("PumpEvent {:#?}", event);
+                }
+            }
+        }
+        Ok(())
+    }
+
+    pub fn get_swap_type(trade_raw: &EncodedTransactionWithStatusMeta) -> Result<SwapType, AppError> {
+        let transaction = trade_raw.transaction.decode()
+            .ok_or_else(|| AppError::from(anyhow!("Failed to decode transaction")))?;
+        
+        let account_keys = transaction.message.static_account_keys();
+        let program_index = account_keys
+            .iter()
+            .position(|item| item == &AMM_V4 || item == &PUMP_PROGRAM_ID)
+            .ok_or_else(|| AppError::from(anyhow!("swap type program_id not found")))?;
+
+        match account_keys[program_index] {
+            AMM_V4 => Ok(SwapType::Raydium),
+            PUMP_PROGRAM_ID => Ok(SwapType::Pump),
+            _ => Err(AppError::from(anyhow!("Invalid program_id")))
+        }
+    }
+}