add nextblock, 0slot support

2025-04-02 13:39:59 +08:00
parent 2dc15b9020
commit ec03ed8edc
59 changed files with 5315 additions and 1855 deletions
@@ -2,8 +2,10 @@ 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::codec::CompressionEncoding;
 use tonic::{transport::channel::ClientTlsConfig, Status};
 use yellowstone_grpc_client::{GeyserGrpcClient, GeyserGrpcClientResult};
+use yellowstone_grpc_proto::geyser::SubscribeUpdateSlot;
 use yellowstone_grpc_proto::geyser::{
    CommitmentLevel, SubscribeRequest, SubscribeRequestFilterTransactions, SubscribeUpdate,
    SubscribeUpdateTransaction, subscribe_update::UpdateOneof, SubscribeRequestPing,
@@ -15,7 +17,9 @@ use solana_transaction_status::{
    option_serializer::OptionSerializer, EncodedTransactionWithStatusMeta, UiTransactionEncoding,
 };

+use crate::common::logs_data::DexInstruction;
 use crate::common::logs_events::PumpfunEvent;
+use crate::common::logs_filters::LogFilter;
 use crate::error::{ClientError, ClientResult};

 type TransactionsFilterMap = HashMap<String, SubscribeRequestFilterTransactions>;
@@ -25,11 +29,13 @@ const CONNECT_TIMEOUT: u64 = 10;
 const REQUEST_TIMEOUT: u64 = 60;
 const CHANNEL_SIZE: usize = 1000;

+#[derive(Clone)]
 pub struct TransactionPretty {
    pub slot: u64,
    pub signature: Signature,
    pub is_vote: bool,
    pub tx: EncodedTransactionWithStatusMeta,
+    // pub transaction: Option<Transaction>,
 }

 impl fmt::Debug for TransactionPretty {
@@ -54,6 +60,7 @@ impl fmt::Debug for TransactionPretty {
 impl From<SubscribeUpdateTransaction> for TransactionPretty {
    fn from(SubscribeUpdateTransaction { transaction, slot }: SubscribeUpdateTransaction) -> Self {
        let tx = transaction.expect("should be defined");
+        // let transaction_info = tx.transaction.clone().unwrap();
        Self {
            slot,
            signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),
@@ -62,10 +69,12 @@ impl From<SubscribeUpdateTransaction> for TransactionPretty {
                .expect("valid tx with meta")
                .encode(UiTransactionEncoding::Base64, Some(u8::MAX), true)
                .expect("failed to encode"),
+            // transaction: Some(transaction_info),
        }
    }
 }

+#[derive(Clone)]
 pub struct YellowstoneGrpc {
    endpoint: String,
 }
@@ -109,43 +118,6 @@ impl YellowstoneGrpc {
        Ok(client.subscribe_with_request(Some(subscribe_request)).await)
    }

-    pub async fn subscribe_pumpfun<F>(&self, callback: F, bot_wallet: Option<Pubkey>) -> ClientResult<()> 
-    where
-        F: Fn(PumpfunEvent) + Send + Sync + 'static,
-    {
-        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?
-        .map_err(|e| ClientError::Other(format!("Failed to subscribe: {:?}", e)))?;
-        let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(CHANNEL_SIZE);
-
-        let callback = Box::new(callback);
-
-        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, &*callback, bot_wallet).await {
-                error!("Error processing transaction: {:?}", e);
-            }
-        }
-        Ok(())
-    }
-
    pub fn get_subscribe_request_filter(
        &self,
        account_include: Vec<String>,
@@ -167,6 +139,43 @@ impl YellowstoneGrpc {
        transactions
    }

+    // pub fn get_subscribe_account_updater_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(),
+    //         SubscribeUpdateAccount {
+    //             account: account_include,
+    //             slot: None,
+    //             is_startup: None,
+    //         },
+    //     );
+    //     transactions
+    // }
+
+    pub fn get_subscribe_update_slot_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(),
+            SubscribeUpdateSlot {
+                slot: 0,
+                parent: None,
+                status: None,
+                dead_error: None,
+            },
+        );
+        transactions
+    }
+
    async fn handle_stream_message(
        msg: SubscribeUpdate,
        tx: &mut mpsc::Sender<TransactionPretty>,
@@ -195,10 +204,62 @@ impl YellowstoneGrpc {
        Ok(())
    }

-    async fn process_transaction<F>(transaction_pretty: TransactionPretty, callback: &F, bot_wallet: Option<Pubkey>) -> ClientResult<()> 
+    // pub async fn subscribe_account_updater<F>(&self, callback: F, bot_wallet: Option<Pubkey>) -> ClientResult<()> 
+    // where
+    //     F: Fn(PumpfunEvent) + Send + Sync + 'static,
+    // {
+    //     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?
+    //     .map_err(|e| ClientError::Other(format!("Failed to subscribe: {:?}", e)))?;
+    //     let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(CHANNEL_SIZE);
+
+    //     let callback = Box::new(callback);
+        
+    // }
+
+    pub async fn subscribe_pumpfun<F>(&self, callback: F, bot_wallet: Option<Pubkey>) -> ClientResult<()> 
+    where
+        F: Fn(PumpfunEvent) + Send + Sync + 'static,
+    {
+        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?
+        .map_err(|e| ClientError::Other(format!("Failed to subscribe: {:?}", e)))?;
+        let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(CHANNEL_SIZE);
+
+        let callback = Box::new(callback);
+
+        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_pumpfun_transaction(transaction_pretty, &*callback, bot_wallet).await {
+                error!("Error processing transaction: {:?}", e);
+            }
+        }
+        Ok(())
+    }
+
+    async fn process_pumpfun_transaction<F>(transaction_pretty: TransactionPretty, callback: &F, bot_wallet: Option<Pubkey>) -> ClientResult<()> 
    where
        F: Fn(PumpfunEvent) + Send + Sync,
    {
+        let slot = transaction_pretty.slot;
        let trade_raw = transaction_pretty.tx;
        let meta = trade_raw.meta.as_ref()
            .ok_or_else(|| ClientError::Other("Missing transaction metadata".to_string()))?;
@@ -213,21 +274,31 @@ impl YellowstoneGrpc {
            &vec![]
        };

-        let (create_event, trade_event) = PumpfunEvent::parse_logs(logs);
-        if let Some(create_event) = create_event {
-            callback(PumpfunEvent::NewToken(create_event));
-        }
-        if let Some(trade_event) = trade_event {
-            if let Some(bot_wallet_pubkey) = bot_wallet {
-                if trade_event.user == bot_wallet_pubkey {
-                    callback(PumpfunEvent::NewBotTrade(trade_event));
-                } else {
-                    callback(PumpfunEvent::NewUserTrade(trade_event));
+        let mut dev_address: Option<Pubkey> = None;
+        let instructions = LogFilter::parse_instruction(logs, bot_wallet).unwrap();
+        for instruction in instructions {
+            match instruction {
+                DexInstruction::CreateToken(mut token_info) => {
+                    token_info.slot = slot;
+                    dev_address = Some(token_info.user);
+                    callback(PumpfunEvent::NewToken(token_info));
                }
-            } else {
-                callback(PumpfunEvent::NewUserTrade(trade_event));
+                DexInstruction::UserTrade(mut trade_info) => {
+                    trade_info.slot = slot;
+                    if Some(trade_info.user) == dev_address {
+                        callback(PumpfunEvent::NewDevTrade(trade_info));
+                    } else {
+                        callback(PumpfunEvent::NewUserTrade(trade_info));
+                    }
+                }
+                DexInstruction::BotTrade(mut trade_info) => {
+                    trade_info.slot = slot;
+                    callback(PumpfunEvent::NewBotTrade(trade_info));
+                }
+                _ => {}
            }
        }
+
        Ok(())
    }
 }