add logs process

2026-08-20 12:28:08 +00:00 · 2024-12-31 17:46:05 +08:00
parent 06d1810328
commit 119f4f18cb
8 changed files with 348 additions and 2 deletions
@@ -19,6 +19,11 @@
 //! - `RateLimitExceeded`: Rate limit exceeded.

 use anchor_client::solana_client;
+use serde_json::Error;
+use anchor_client::solana_client::{
+    client_error::ClientError as SolanaClientError, 
+    pubsub_client::PubsubClientError
+};

 #[derive(Debug)]
 pub enum ClientError {
@@ -42,6 +47,26 @@ pub enum ClientError {
    SimulationError(String),
    /// Rate limit exceeded
    RateLimitExceeded,
+
+    Solana(String, String),
+    
+    Parse(String, String),
+
+    Join(String),
+
+    Subscribe(String, String),
+
+    Send(String, String),
+
+    Other(String),
+
+    Timeout(String, String),
+
+    Duplicate(String),
+
+    InvalidEventType,
+
+    ChannelClosed,
 }

 impl std::fmt::Display for ClientError {
@@ -57,6 +82,16 @@ impl std::fmt::Display for ClientError {
            Self::InsufficientFunds => write!(f, "Insufficient funds for transaction"),
            Self::SimulationError(msg) => write!(f, "Transaction simulation failed: {}", msg),
            Self::RateLimitExceeded => write!(f, "Rate limit exceeded"),
+            Self::Solana(msg, details) => write!(f, "Solana error: {}, details: {}", msg, details),
+            Self::Parse(msg, details) => write!(f, "Parse error: {}, details: {}", msg, details),
+            Self::Join(msg) => write!(f, "Task join error: {}", msg),
+            Self::Subscribe(msg, details) => write!(f, "Subscribe error: {}, details: {}", msg, details),
+            Self::Send(msg, details) => write!(f, "Send error: {}, details: {}", msg, details),
+            Self::Other(msg) => write!(f, "Other error: {}", msg),
+            Self::Timeout(msg, details) => write!(f, "Operation timed out: {}, details: {}", msg, details),
+            Self::Duplicate(msg) => write!(f, "Duplicate event: {}", msg),
+            Self::InvalidEventType => write!(f, "Invalid event type"),
+            Self::ChannelClosed => write!(f, "Channel closed"),
        }
    }
 }
@@ -72,3 +107,32 @@ impl std::error::Error for ClientError {
        }
    }
 }
+
+impl From<SolanaClientError> for ClientError {
+    fn from(error: SolanaClientError) -> Self {
+        ClientError::Solana(
+            "Solana client error".to_string(),
+            error.to_string(),
+        )
+    }
+}
+
+impl From<PubsubClientError> for ClientError {
+    fn from(error: PubsubClientError) -> Self {
+        ClientError::Solana(
+            "PubSub client error".to_string(),
+            error.to_string(),
+        )
+    }
+}
+
+impl From<Error> for ClientError {
+    fn from(err: Error) -> Self {
+        ClientError::Parse(
+            "JSON serialization error".to_string(),
+            err.to_string()
+        )
+    }
+}
+
+pub type ClientResult<T> = Result<T, ClientError>;
@@ -0,0 +1,28 @@
+use serde::{Serialize, Deserialize};
+// 添加新的数据结构
+#[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,
+}
@@ -0,0 +1,88 @@
+use crate::instruction::logs_data::{CreateTokenInfo, TradeInfo};
+use crate::instruction::logs_paser::{parse_create_token_data, parse_trade_data};
+use crate::error::ClientResult;
+
+pub struct LogFilter;
+
+#[derive(Debug)]
+pub enum DexInstruction {
+    CreateToken(CreateTokenInfo),
+    Trade(TradeInfo),
+    Other,
+}
+
+impl LogFilter {
+    const PROGRAM_ID: &'static str = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P";
+    
+    /// 解析交易日志并返回具体的指令类型和数据
+    pub fn parse_instruction(logs: &[String]) -> 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 {
+            // println!("log: {:?}", log);
+            // 检查程序调用
+            if log.contains(&format!("Program {} invoke", Self::PROGRAM_ID)) {
+                invoke_depth += 1;
+                if invoke_depth == 1 {  // 只在顶层调用时重置状态
+                    current_instruction = None;
+                    program_data.clear();
+                    last_data_len = 0;
+                }
+                continue;
+            }
+            
+            // 如果不在我们的程序中，跳过
+            if invoke_depth == 0 {
+                continue;
+            }
+            
+            // 识别指令类型（只在顶层调用时）
+            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;
+            }
+            
+            // 收集 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();
+                }
+            }
+            
+            // 检查程序是否结束
+            if log.contains(&format!("Program {} success", Self::PROGRAM_ID)) {
+                invoke_depth -= 1;
+                if invoke_depth == 0 {  // 只在顶层程序结束时处理数据
+                    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) {
+                                        instructions.push(DexInstruction::Trade(trade_info));
+                                    }
+                                },
+                                _ => {}
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(instructions)
+    }
+} 
@@ -0,0 +1,152 @@
+use crate::error::{ClientError, ClientResult};
+use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
+use serde::{Serialize, Deserialize};
+
+use crate::instruction::logs_data::*;
+// 添加解析函数
+pub fn parse_create_token_data(data: &str) -> ClientResult<CreateTokenInfo> {
+    // 首先进行 base64 解码
+    let decoded = BASE64.decode(data)
+        .map_err(|e| ClientError::Other(format!("Failed to decode base64: {}", e)))?;
+    
+    // 跳过前缀字节（如果有的话）
+    let mut cursor = if decoded.len() > 8 { 8 } else { 0 };
+    
+    // 读取名称长度和名称
+    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;
+    
+    // 读取符号长度和符号
+    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;
+    
+    // 读取 URI 长度和 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;
+    
+    // 确保还有足够的数据来读取公钥
+    if cursor + 32 * 3 > decoded.len() {
+        return Err(ClientError::Other("Data too short for public keys".to_string()));
+    }
+    
+    // 解析 Mint Public Key
+    let mint = bs58::encode(&decoded[cursor..cursor+32]).into_string();
+    cursor += 32;
+
+    // 解析 Bonding Curve Public Key
+    let bonding_curve = bs58::encode(&decoded[cursor..cursor+32]).into_string();
+    cursor += 32;
+
+    // 解析 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;  // 跳过前缀
+
+    // 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,
+    })
+}
@@ -10,6 +10,14 @@
 //! - `buy`: Instruction to buy tokens from a bonding curve by providing SOL.
 //! - `sell`: Instruction to sell tokens back to the bonding curve in exchange for SOL.

+pub mod logs_data;
+pub mod logs_paser;
+pub mod logs_filter;
+
+pub use logs_data::*;
+pub use logs_paser::*;
+pub use logs_filter::*;
+
 use crate::{constants, PumpFun};
 use anchor_client::anchor_lang::InstructionData;
 use anchor_spl::associated_token::get_associated_token_address;
@@ -6,6 +6,8 @@ pub mod error;
 pub mod instruction;
 pub mod utils;

+use crate::error::ClientError::*;
+
 use anchor_client::{
    solana_client::rpc_client::RpcClient,
    solana_sdk::{