add callback to subscribe

src/common/dex_events.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/common/event.rs

@@ -0,0 +1,158 @@
+use anyhow::anyhow;
+use base64::engine::general_purpose;
+use base64::Engine;
+use borsh::{BorshDeserialize, BorshSerialize};
+use regex::Regex;
+use solana_sdk::pubkey::Pubkey;
+
+use crate::error::AppError;
+
+pub const PROGRAM_DATA: &str = "Program data: ";
+
+#[derive(Debug)]
+pub enum PumpfunEvent {
+    NewToken(CreateEvent),
+    NewUserTrade(TradeEvent),
+    NewBotTrade(TradeEvent),
+    Error(String),
+}
+
+#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)]
+pub struct CreateEvent {
+    pub name: String,
+    pub symbol: String,
+    pub uri: String,
+    pub mint: Pubkey,
+    pub bonding_curve: Pubkey,
+    pub user: Pubkey,
+}
+
+#[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 CreateEvent {
+    fn from_bytes(bytes: &[u8]) -> Result<Self, AppError> {
+        CreateEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string())))
+    }
+}
+
+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(logs: &Vec<String>) -> (Option<CreateEvent>, Option<TradeEvent>) {
+        let mut create_event: Option<CreateEvent> = None;
+        let mut trade_event: Option<TradeEvent> = 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 create_event.is_none() {
+                        if let Ok(e) = CreateEvent::from_bytes(slice) {
+                            create_event = Some(e);
+                            continue;
+                        }
+                    }
+
+                    if trade_event.is_none() {
+                        if let Ok(e) = TradeEvent::from_bytes(slice) {
+                            trade_event = Some(e);
+                        }
+                    }
+                }
+            }
+        }
+        (create_event, trade_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/common/logs_data.rs

@@ -0,0 +1,37 @@
+use serde::{Serialize, Deserialize};
+
+#[derive(Debug)]
+pub enum DexInstruction {
+    CreateToken(CreateTokenInfo),
+    UserTrade(TradeInfo),
+    BotTrade(TradeInfo),
+    Other,
+}
+
+// 添加新的数据结构
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct CreateTokenInfo {
+    pub signature: String,
+    pub name: String,
+    pub symbol: String,
+    pub uri: String,
+    pub mint: String,
+    pub bonding_curve: String,
+    pub user: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct TradeInfo {
+    pub signature: String,
+    pub mint: String,
+    pub bonding_curve: String,
+    pub sol_amount: u64,
+    pub token_amount: u64,
+    pub is_buy: bool,
+    pub user: String,
+    pub timestamp: i64,
+    pub virtual_sol_reserves: u64,
+    pub virtual_token_reserves: u64,
+    pub real_sol_reserves: u64,
+    pub real_token_reserves: u64,
+}

src/common/logs_events.rs

@@ -0,0 +1,11 @@
+use serde::{Serialize, Deserialize};
+use crate::common::logs_data::{CreateTokenInfo, TradeInfo};
+use crate::common::event::{CreateEvent, TradeEvent};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum DexEvent {
+    NewToken(CreateTokenInfo),
+    NewUserTrade(TradeInfo),
+    NewBotTrade(TradeInfo),
+    Error(String),
+}

src/common/logs_filters.rs

@@ -0,0 +1,88 @@
+use crate::common::logs_data::DexInstruction;
+use crate::common::logs_parser::{parse_create_token_data, parse_trade_data};
+use crate::error::ClientResult;
+use solana_sdk::pubkey::Pubkey;
+pub struct LogFilter;
+
+impl LogFilter {
+    const PROGRAM_ID: &'static str = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P";
+    
+    /// Parse transaction logs and return instruction type and data
+    pub fn parse_instruction(logs: &[String], bot_wallet: Option<Pubkey>) -> ClientResult<Vec<DexInstruction>> {
+        let mut current_instruction = None;
+        let mut program_data = String::new();
+        let mut invoke_depth = 0;
+        let mut last_data_len = 0;
+        let mut instructions = Vec::new();
+        for log in logs {
+            // Check program invocation
+            if log.contains(&format!("Program {} invoke", Self::PROGRAM_ID)) {
+                invoke_depth += 1;
+                if invoke_depth == 1 {  // Only reset state at top level call
+                    current_instruction = None;
+                    program_data.clear();
+                    last_data_len = 0;
+                }
+                continue;
+            }
+            
+            // Skip if not in our program
+            if invoke_depth == 0 {
+                continue;
+            }
+            
+            // Identify instruction type (only at top level)
+            if invoke_depth == 1 && log.contains("Program log: Instruction:") {
+                if log.contains("Create") {
+                    current_instruction = Some("create");
+                } else if log.contains("Buy") || log.contains("Sell") {
+                    current_instruction = Some("trade");
+                }
+                continue;
+            }
+            
+            // Collect Program data
+            if log.starts_with("Program data: ") {
+                let data = log.trim_start_matches("Program data: ");
+                if data.len() > last_data_len {
+                    program_data = data.to_string();
+                    last_data_len = data.len();
+                }
+            }
+            
+            // Check if program ends
+            if log.contains(&format!("Program {} success", Self::PROGRAM_ID)) {
+                invoke_depth -= 1;
+                if invoke_depth == 0 {  // Only process data when top level program ends
+                    if let Some(instruction_type) = current_instruction {
+                        if !program_data.is_empty() {
+                            match instruction_type {
+                                "create" => {
+                                    if let Ok(token_info) = parse_create_token_data(&program_data) {
+                                        instructions.push(DexInstruction::CreateToken(token_info));
+                                    }
+                                },
+                                "trade" => {
+                                    if let Ok(trade_info) = parse_trade_data(&program_data) {
+                                        if let Some(bot_wallet_pubkey) = bot_wallet {
+                                            if trade_info.user == bot_wallet_pubkey.to_string() {
+                                                instructions.push(DexInstruction::BotTrade(trade_info));
+                                            } else {
+                                                instructions.push(DexInstruction::UserTrade(trade_info));
+                                            }
+                                        } else {
+                                            instructions.push(DexInstruction::UserTrade(trade_info));
+                                        }
+                                    }
+                                },
+                                _ => {}
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(instructions)
+    }
+}

src/common/logs_parser.rs

@@ -0,0 +1,173 @@
+use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
+
+use crate::error::{ClientError, ClientResult};
+use crate::common::{
+    logs_data::{DexInstruction, CreateTokenInfo, TradeInfo}, 
+    logs_filters::LogFilter
+};
+
+use solana_sdk::pubkey::Pubkey;
+
+pub async fn process_logs<F>(
+    signature: &str,
+    logs: Vec<String>,
+    callback: F,
+    payer: Option<Pubkey>,
+) -> ClientResult<()>
+where
+    F: Fn(&str, DexInstruction) + Send + Sync,
+{
+    let instructions = LogFilter::parse_instruction(&logs, payer)?;
+    for instruction in instructions {
+        callback(signature, instruction);
+    }
+    Ok(())
+}
+
+// Add parsing function
+pub fn parse_create_token_data(data: &str) -> ClientResult<CreateTokenInfo> {
+    // First do base64 decoding
+    let decoded = BASE64.decode(data)
+        .map_err(|e| ClientError::Other(format!("Failed to decode base64: {}", e)))?;
+    
+    // Skip prefix bytes (if any)
+    let mut cursor = if decoded.len() > 8 { 8 } else { 0 };
+    
+    // Read name length and name
+    if cursor + 4 > decoded.len() {
+        return Err(ClientError::Other("Data too short for name length".to_string()));
+    }
+    let name_len = read_u32(&decoded[cursor..]) as usize;
+    cursor += 4;
+    
+    if cursor + name_len > decoded.len() {
+        return Err(ClientError::Other(format!("Data too short for name: need {} bytes", name_len)));
+    }
+    let name = String::from_utf8(decoded[cursor..cursor + name_len].to_vec())
+        .map_err(|e| ClientError::Other(format!("Invalid UTF-8 in name: {}", e)))?;
+    cursor += name_len;
+    
+    // Read symbol length and symbol
+    if cursor + 4 > decoded.len() {
+        return Err(ClientError::Other("Data too short for symbol length".to_string()));
+    }
+    let symbol_len = read_u32(&decoded[cursor..]) as usize;
+    cursor += 4;
+    
+    if cursor + symbol_len > decoded.len() {
+        return Err(ClientError::Other(format!("Data too short for symbol: need {} bytes", symbol_len)));
+    }
+    let symbol = String::from_utf8(decoded[cursor..cursor + symbol_len].to_vec())
+        .map_err(|e| ClientError::Other(format!("Invalid UTF-8 in symbol: {}", e)))?;
+    cursor += symbol_len;
+    
+    // Read URI length and URI
+    if cursor + 4 > decoded.len() {
+        return Err(ClientError::Other("Data too short for URI length".to_string()));
+    }
+    let uri_len = read_u32(&decoded[cursor..]) as usize;
+    cursor += 4;
+    
+    if cursor + uri_len > decoded.len() {
+        return Err(ClientError::Other(format!("Data too short for URI: need {} bytes", uri_len)));
+    }
+    let uri = String::from_utf8(decoded[cursor..cursor + uri_len].to_vec())
+        .map_err(|e| ClientError::Other(format!("Invalid UTF-8 in uri: {}", e)))?;
+    cursor += uri_len;
+    
+    // Make sure there is enough data to read public keys
+    if cursor + 32 * 3 > decoded.len() {
+        return Err(ClientError::Other("Data too short for public keys".to_string()));
+    }
+    
+    // Parse Mint Public Key
+    let mint = bs58::encode(&decoded[cursor..cursor+32]).into_string();
+    cursor += 32;
+
+    // Parse Bonding Curve Public Key
+    let bonding_curve = bs58::encode(&decoded[cursor..cursor+32]).into_string();
+    cursor += 32;
+
+    // Parse User Public Key
+    let user = bs58::encode(&decoded[cursor..cursor+32]).into_string();
+
+    Ok(CreateTokenInfo {
+        signature: String::new(),
+        name,
+        symbol,
+        uri,
+        mint,
+        bonding_curve,
+        user,
+    })
+}
+
+fn read_u32(data: &[u8]) -> u32 {
+    let mut bytes = [0u8; 4];
+    bytes.copy_from_slice(&data[..4]);
+    u32::from_le_bytes(bytes)
+}
+
+pub fn parse_trade_data(data: &str) -> ClientResult<TradeInfo> {
+    let engine = base64::engine::general_purpose::STANDARD;
+    let decoded = engine.decode(data).map_err(|e| 
+        ClientError::Parse(
+            "Failed to decode base64".to_string(),
+            e.to_string()
+        )
+    )?;
+
+    let mut cursor = 8;  // Skip prefix
+
+    // 1. Mint (32 bytes)
+    let mint = bs58::encode(&decoded[cursor..cursor + 32]).into_string();
+    cursor += 32;
+
+    // 2. Sol Amount (8 bytes)
+    let sol_amount = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
+    cursor += 8;
+
+    // 3. Token Amount (8 bytes)
+    let token_amount = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
+    cursor += 8;
+
+    // 4. Is Buy (1 byte)
+    let is_buy = decoded[cursor] != 0;
+    cursor += 1;
+
+    // 5. User (32 bytes)
+    let user = bs58::encode(&decoded[cursor..cursor + 32]).into_string();
+    cursor += 32;
+
+    // 6. Timestamp (8 bytes)
+    let timestamp = i64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
+    cursor += 8;
+
+    // 7. Virtual Sol Reserves (8 bytes)
+    let virtual_sol_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
+    cursor += 8;
+
+    // 8. Virtual Token Reserves (8 bytes)
+    let virtual_token_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
+    cursor += 8;
+
+    let real_sol_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
+    cursor += 8;
+
+    let real_token_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap());
+
+    Ok(TradeInfo {
+        signature: String::new(),
+        mint,
+        bonding_curve: String::new(),
+        sol_amount,
+        token_amount,
+        is_buy,
+        user,
+        timestamp,
+        virtual_sol_reserves,
+        virtual_token_reserves,
+        real_sol_reserves,
+        real_token_reserves,
+    })
+}

src/common/logs_subscribe.rs

@@ -0,0 +1,103 @@
+use solana_client::{
+    nonblocking::pubsub_client::PubsubClient,
+    rpc_config::{RpcTransactionLogsConfig, RpcTransactionLogsFilter}
+};
+
+use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey};
+use std::sync::Arc;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+use futures::StreamExt;
+use crate::{constants, common::{
+    logs_data::DexInstruction, logs_events::DexEvent, logs_filters::LogFilter
+}};
+
+/// Subscription handle containing task and unsubscribe logic
+pub struct SubscriptionHandle {
+    pub task: JoinHandle<()>,
+    pub unsub_fn: Box<dyn Fn() + Send>,
+}
+
+impl SubscriptionHandle {
+    pub async fn shutdown(self) {
+        (self.unsub_fn)();
+        self.task.abort();
+    }
+}
+
+pub async fn create_pubsub_client(ws_url: &str) -> PubsubClient {
+    PubsubClient::new(ws_url).await.unwrap()
+}
+
+/// 启动订阅
+pub async fn tokens_subscription<F>(
+    ws_url: &str,
+    commitment: CommitmentConfig,
+    callback: F,
+    bot_wallet: Option<Pubkey>,
+) -> Result<SubscriptionHandle, Box<dyn std::error::Error>>
+where
+    F: Fn(DexEvent) + Send + Sync + 'static,
+{
+    let program_address = constants::accounts::PUMPFUN.to_string();
+    let logs_filter = RpcTransactionLogsFilter::Mentions(vec![program_address]);
+
+    let logs_config = RpcTransactionLogsConfig {
+        commitment: Some(commitment),
+    };
+
+    // Create PubsubClient
+    let sub_client = Arc::new(PubsubClient::new(ws_url).await.unwrap());
+
+    let sub_client_clone = Arc::clone(&sub_client);
+
+    // Create channel for unsubscribe
+    let (unsub_tx, _) = mpsc::channel(1);
+
+    // Start subscription task
+    let task = tokio::spawn(async move {
+        let (mut stream, _) = sub_client_clone.logs_subscribe(logs_filter, logs_config).await.unwrap();
+
+        loop {
+            let msg = stream.next().await;
+            match msg {
+                Some(msg) => {
+                    if let Some(_err) = msg.value.err {
+                        continue;
+                    }
+
+                    let instructions = LogFilter::parse_instruction(&msg.value.logs, bot_wallet).unwrap();
+                    for instruction in instructions {
+                        match instruction {
+                            DexInstruction::CreateToken(token_info) => {
+                                callback(DexEvent::NewToken(token_info));
+                            }
+                            DexInstruction::UserTrade(trade_info) => {
+                                callback(DexEvent::NewUserTrade(trade_info));
+                            }
+                            DexInstruction::BotTrade(trade_info) => {
+                                callback(DexEvent::NewBotTrade(trade_info));
+                            }
+                            _ => {}
+                        }
+                    }
+                }
+                None => {
+                    println!("Token subscription stream ended");
+                }
+            }   
+        }
+    });
+
+    // Return subscription handle and unsubscribe logic
+    Ok(SubscriptionHandle {
+        task,
+        unsub_fn: Box::new(move || {
+            let _ = unsub_tx.try_send(());
+        }),
+    })
+}
+
+pub async fn stop_subscription(handle: SubscriptionHandle) {
+    handle.shutdown().await;
+}

src/common/mod.rs

@@ -0,0 +1,6 @@
+pub mod logs_data;
+pub mod logs_parser;
+pub mod logs_filters;
+pub mod logs_events;
+pub mod logs_subscribe;
+pub mod event;