From 7615102fd6b1aa5a36de967a9b54d4da7949f0ea Mon Sep 17 00:00:00 2001 From: ysq Date: Mon, 21 Jul 2025 23:55:13 +0800 Subject: [PATCH] feat: enhance event parser with transfer data processing capabilities - Bump version to 0.1.2 - Add TransferData struct for transaction data parsing - Implement parse_transfer_datas_from_next_instructions function to extract transfer data from subsequent instructions - Enhance UnifiedEvent trait with set_transfer_datas method - Optimize transfer data extraction when parsing inner instructions - Support transfer data parsing for multiple protocols (PumpFun, PumpSwap, Bonk, Raydium, etc.) - Improve support for Token Program, Token 2022 Program, and System Program transfer instructions --- Cargo.toml | 2 +- README.md | 4 +- README_CN.md | 4 +- src/streaming/event_parser/common/mod.rs | 4 + src/streaming/event_parser/common/types.rs | 140 +++++++++++++++++++++ src/streaming/event_parser/core/traits.rs | 71 +++++++++-- 6 files changed, 210 insertions(+), 15 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 934a40a..7dfb5af 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.1.1" +version = "0.1.2" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" diff --git a/README.md b/README.md index 043e83f..77fdff7 100755 --- a/README.md +++ b/README.md @@ -33,14 +33,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.1" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.2" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.1.1" +solana-streamer-sdk = "0.1.2" ``` ## Usage Examples diff --git a/README_CN.md b/README_CN.md index cb5bb3e..93a7336 100644 --- a/README_CN.md +++ b/README_CN.md @@ -33,14 +33,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.1" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.1.2" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.1.1" +solana-streamer-sdk = "0.1.2" ``` ## 使用示例 diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs index a070106..d25d73e 100755 --- a/src/streaming/event_parser/common/mod.rs +++ b/src/streaming/event_parser/common/mod.rs @@ -46,6 +46,10 @@ macro_rules! impl_unified_event { )* } } + + fn set_transfer_datas(&mut self, transfer_datas: Vec<$crate::streaming::event_parser::common::types::TransferData>) { + self.metadata.transfer_datas = transfer_datas; + } } }; } diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index 0ad95f9..d5497b8 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -1,6 +1,10 @@ use borsh::{BorshDeserialize, BorshSerialize}; use serde::{Deserialize, Serialize}; +use solana_sdk::instruction::CompiledInstruction; use solana_sdk::pubkey::Pubkey; +use solana_transaction_status::{ + UiCompiledInstruction, UiInstruction, UiTransactionStatusMeta, UiTransactionTokenBalance, +}; use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; @@ -129,6 +133,20 @@ impl ProtocolInfo { } } +/// 交易数据 +#[derive( + Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, +)] +pub struct TransferData { + pub token_program: Pubkey, + pub source: Pubkey, + pub destination: Pubkey, + pub authority: Option, + pub amount: u64, + pub decimals: Option, + pub mint: Option, +} + /// 事件元数据 #[derive( Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, @@ -141,6 +159,7 @@ pub struct EventMetadata { pub protocol: ProtocolType, pub event_type: EventType, pub program_id: Pubkey, + pub transfer_datas: Vec, } impl EventMetadata { @@ -160,6 +179,7 @@ impl EventMetadata { protocol, event_type, program_id, + transfer_datas: vec![], } } pub fn set_id(&mut self, id: String) { @@ -171,3 +191,123 @@ impl EventMetadata { self.id = format!("{:x}", hash_value); } } + +/// 解析接下来指令中的token转账数据 +pub fn parse_transfer_datas_from_next_instructions( + inner_instruction: &solana_transaction_status::UiInnerInstructions, + current_index: i8, + accounts: &[Pubkey], + event_type: EventType, +) -> Vec { + let take = match event_type { + EventType::PumpFunBuy => 4, + EventType::PumpFunSell => 1, + EventType::PumpSwapBuy => 3, + EventType::PumpSwapSell => 3, + EventType::BonkBuyExactIn + | EventType::BonkBuyExactOut + | EventType::BonkSellExactIn + | EventType::BonkSellExactOut => 3, + EventType::RaydiumCpmmSwapBaseInput + | EventType::RaydiumCpmmSwapBaseOutput + | EventType::RaydiumClmmSwap + | EventType::RaydiumClmmSwapV2 => 2, + _ => 0, + }; + if take == 0 { + return vec![]; + } + let mut transfer_datas = vec![]; + // 获取当前指令之后的两个指令 + let next_instructions: Vec<&UiInstruction> = inner_instruction + .instructions + .iter() + .skip((current_index + 1) as usize) + .take(take) + .collect(); + + for instruction in next_instructions { + if let UiInstruction::Compiled(compiled) = instruction { + if let Ok(data) = bs58::decode(compiled.data.clone()).into_vec() { + // Token Program: transferChecked + // Token 2022 Program: transferChecked + if data[0] == 12 { + let account_pubkeys: Vec = compiled + .accounts + .iter() + .map(|a| accounts[*a as usize]) + .collect(); + if account_pubkeys.len() < 4 { + continue; + } + let (source, mint, destination, authority) = ( + account_pubkeys[0], + account_pubkeys[1], + account_pubkeys[2], + account_pubkeys[3], + ); + let amount = u64::from_le_bytes(data[1..9].try_into().unwrap()); + let decimals = data[9]; + let token_program = accounts[compiled.program_id_index as usize]; + transfer_datas.push(TransferData { + amount, + decimals: Some(decimals), + mint: Some(mint), + source, + destination, + authority: Some(authority), + token_program, + }); + } + // Token Program: transfer + else if data[0] == 3 { + let account_pubkeys: Vec = compiled + .accounts + .iter() + .map(|a| accounts[*a as usize]) + .collect(); + if account_pubkeys.len() < 3 { + continue; + } + let (source, destination, authority) = + (account_pubkeys[0], account_pubkeys[1], account_pubkeys[2]); + let amount = u64::from_le_bytes(data[1..9].try_into().unwrap()); + let token_program = accounts[compiled.program_id_index as usize]; + transfer_datas.push(TransferData { + amount, + decimals: None, + mint: None, + source, + destination, + authority: Some(authority), + token_program, + }); + } + //System Program: transfer + else if data[0] == 2 { + let account_pubkeys: Vec = compiled + .accounts + .iter() + .map(|a| accounts[*a as usize]) + .collect(); + if account_pubkeys.len() < 2 { + continue; + } + let (source, destination) = (account_pubkeys[0], account_pubkeys[1]); + let amount = u64::from_le_bytes(data[4..12].try_into().unwrap()); + let token_program = accounts[compiled.program_id_index as usize]; + transfer_datas.push(TransferData { + amount, + decimals: None, + mint: None, + source, + destination, + authority: None, + token_program, + }); + } + } + } + } + transfer_datas +} diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 331fe43..756d0cc 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -3,11 +3,14 @@ use solana_sdk::{ instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction, }; use solana_transaction_status::{ - EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInstruction, + EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInnerInstructions, UiInstruction, }; use std::fmt::Debug; use std::{collections::HashMap, str::FromStr}; +use crate::streaming::event_parser::common::{ + parse_transfer_datas_from_next_instructions, TransferData, +}; use crate::streaming::event_parser::{ common::{utils::*, EventMetadata, EventType, ProtocolType}, protocols::{ @@ -46,6 +49,8 @@ pub trait UnifiedEvent: Debug + Send + Sync { fn merge(&mut self, _other: Box) { // 默认实现：不进行任何合并操作 } + + fn set_transfer_datas(&mut self, transfer_datas: Vec); } /// 事件解析器trait - 定义了事件解析的核心方法 @@ -75,6 +80,7 @@ pub trait EventParser: Send + Sync { signature: &str, slot: Option, accounts: &[Pubkey], + inner_instructions: &[UiInnerInstructions], ) -> Result>> { let mut instruction_events = Vec::new(); // 获取交易的指令和账户 @@ -85,7 +91,7 @@ pub trait EventParser: Send + Sync { let has_program = accounts.iter().any(|account| self.should_handle(account)); if has_program { // 解析每个指令 - for instruction in compiled_instructions { + for (index, instruction) in compiled_instructions.iter().enumerate() { if let Some(program_id) = accounts.get(instruction.program_id_index as usize) { if self.should_handle(program_id) { let max_idx = instruction.accounts.iter().max().unwrap_or(&0); @@ -95,11 +101,29 @@ pub trait EventParser: Send + Sync { accounts.push(Pubkey::default()); } } - if let Ok(events) = self + if let Ok(mut events) = self .parse_instruction(instruction, &accounts, signature, slot) .await { - instruction_events.extend(events); + if events.len() > 0 { + if let Some(inn) = + inner_instructions.iter().find(|inner_instruction| { + inner_instruction.index == index as u8 + }) + { + events.iter_mut().for_each(|event| { + let transfer_datas = + parse_transfer_datas_from_next_instructions( + &inn, + -1 as i8, + &accounts, + event.event_type(), + ); + event.set_transfer_datas(transfer_datas.clone()); + }); + } + instruction_events.extend(events); + } } } } @@ -122,6 +146,7 @@ pub trait EventParser: Send + Sync { signature, slot, &accounts, + &vec![], ) .await .unwrap_or_else(|_e| vec![]); @@ -143,7 +168,9 @@ pub trait EventParser: Send + Sync { .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; let mut address_table_lookups: Vec = vec![]; + let mut inner_instructions: Vec = vec![]; if meta.err.is_none() { + inner_instructions = meta.inner_instructions.as_ref().unwrap().clone(); let loaded_addresses = meta.loaded_addresses.as_ref().unwrap(); for lookup in &loaded_addresses.writable { address_table_lookups.push(Pubkey::from_str(lookup).unwrap()); @@ -167,6 +194,7 @@ pub trait EventParser: Send + Sync { signature, slot, &accounts, + &inner_instructions, ) .await .unwrap_or_else(|_e| vec![]); @@ -178,9 +206,8 @@ pub trait EventParser: Send + Sync { let mut inner_instruction_events = Vec::new(); // 检查交易是否成功 if meta.err.is_none() { - let inner_instructions = meta.inner_instructions.as_ref().unwrap(); for inner_instruction in inner_instructions { - for instruction in &inner_instruction.instructions { + for (index, instruction) in inner_instruction.instructions.iter().enumerate() { match instruction { UiInstruction::Compiled(compiled) => { // 解析嵌套指令 @@ -189,7 +216,7 @@ pub trait EventParser: Send + Sync { accounts: compiled.accounts.clone(), data: bs58::decode(compiled.data.clone()).into_vec().unwrap(), }; - if let Ok(events) = self + if let Ok(mut events) = self .parse_instruction( &compiled_instruction, &accounts, @@ -198,13 +225,37 @@ pub trait EventParser: Send + Sync { ) .await { - instruction_events.extend(events); + if events.len() > 0 { + events.iter_mut().for_each(|event| { + let transfer_datas = + parse_transfer_datas_from_next_instructions( + &inner_instruction, + index as i8, + &accounts, + event.event_type(), + ); + event.set_transfer_datas(transfer_datas.clone()); + }); + instruction_events.extend(events); + } } - if let Ok(events) = self + if let Ok(mut events) = self .parse_inner_instruction(compiled, signature, slot) .await { - inner_instruction_events.extend(events); + if events.len() > 0 { + events.iter_mut().for_each(|event| { + let transfer_datas = + parse_transfer_datas_from_next_instructions( + &inner_instruction, + index as i8, + &accounts, + event.event_type(), + ); + event.set_transfer_datas(transfer_datas.clone()); + }); + inner_instruction_events.extend(events); + } } } _ => {}