diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index 677e281..91a2007 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -9,7 +9,7 @@ use std::sync::Arc; #[tokio::main] async fn main() -> Result<()> { let signatures = vec![ - "5sDWrTTkE69CNc6nrAX7SqPS7FiajJTg8TMog3Gve7KjVfrqYn8YZcX1kAoyKok976S4RTnK1EdCV8hRiDWg68Aj", + "4PsHYajH87x2zJPEGZczZtd2ksibuMCFPonC24jk5mTGZ46hzvjpzM5UZuLz9sRv79MkCBbtDqwJapGPTSkCFKoL", ]; // Validate signature format let mut valid_signatures = Vec::new(); @@ -36,8 +36,9 @@ async fn main() -> Result<()> { /// Get details of a single transaction async fn get_single_transaction_details(signature_str: &str) -> Result<()> { - use solana_sdk::signature::Signature; - use solana_transaction_status::UiTransactionEncoding; + use solana_sdk::{signature::Signature, pubkey::Pubkey, message::compiled_instruction::CompiledInstruction}; + use solana_transaction_status::{UiTransactionEncoding, InnerInstruction, InnerInstructions, UiInstruction}; + use prost_types::Timestamp; let signature = Signature::from_str(signature_str)?; @@ -88,6 +89,96 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { } } } + + // Parse the transaction and extract necessary data + let versioned_tx = match transaction.transaction.transaction.decode() { + Some(tx) => tx, + None => { + println!("Failed to decode transaction"); + return Ok(()); + } + }; + + // Convert inner instructions from meta + let mut inner_instructions_vec: Vec = Vec::new(); + if let Some(meta) = &transaction.transaction.meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some( + ui_inner_insts, + ) = &meta.inner_instructions + { + for ui_inner in ui_inner_insts { + let mut converted_instructions = Vec::new(); + + for ui_instruction in &ui_inner.instructions { + if let UiInstruction::Compiled(ui_compiled) = ui_instruction { + if let Ok(data) = solana_sdk::bs58::decode(&ui_compiled.data).into_vec() { + let compiled_instruction = CompiledInstruction { + program_id_index: ui_compiled.program_id_index, + accounts: ui_compiled.accounts.to_vec(), + data, + }; + + let inner_instruction = InnerInstruction { + instruction: compiled_instruction, + stack_height: ui_compiled.stack_height, + }; + + converted_instructions.push(inner_instruction); + } + } + } + + let inner_instructions = InnerInstructions { + index: ui_inner.index, + instructions: converted_instructions, + }; + + inner_instructions_vec.push(inner_instructions); + } + } + } + + // Extract address table lookups + let meta = transaction.transaction.meta; + let mut address_table_lookups: Vec = vec![]; + if let Some(meta) = meta { + if let solana_transaction_status::option_serializer::OptionSerializer::Some( + loaded_addresses, + ) = &meta.loaded_addresses + { + address_table_lookups + .reserve(loaded_addresses.writable.len() + loaded_addresses.readonly.len()); + address_table_lookups.extend( + loaded_addresses + .writable + .iter() + .filter_map(|s| s.parse::().ok()) + .chain( + loaded_addresses + .readonly + .iter() + .filter_map(|s| s.parse::().ok()), + ), + ); + } + } + + // Build complete accounts list + let mut accounts = Vec::with_capacity( + versioned_tx.message.static_account_keys().len() + address_table_lookups.len(), + ); + accounts.extend_from_slice(versioned_tx.message.static_account_keys()); + accounts.extend(address_table_lookups); + + let slot = transaction.slot; + let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 }); + let recv_us = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_micros() as i64; + let bot_wallet = None; + let transaction_index = None; + let protocols = vec![ Protocol::Bonk, Protocol::RaydiumClmm, @@ -96,14 +187,26 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { Protocol::RaydiumCpmm, Protocol::RaydiumAmmV4, ]; - EventParser::parse_encoded_confirmed_transaction_with_status_meta( + + // Create callback + let callback = Arc::new(move |event: DexEvent| { + println!("{:?}\n", event); + }); + + // Call parse_instruction_events_from_versioned_transaction + EventParser::parse_instruction_events_from_versioned_transaction( &protocols, None, + &versioned_tx, signature, - transaction, - Arc::new(move |event: &DexEvent| { - println!("{:?}\n", event); - }), + Some(slot), + block_time, + recv_us, + &accounts, + &inner_instructions_vec, + bot_wallet, + transaction_index, + callback, ) .await?; } diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index 7e7c81e..e415807 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -64,7 +64,7 @@ pub async fn process_grpc_transaction( let adapter_callback = create_metrics_callback(callback.clone()); - EventParser::parse_grpc_transaction_owned( + EventParser::parse_grpc_transaction( protocols, event_type_filter, grpc_tx, @@ -123,18 +123,20 @@ pub async fn process_shred_transaction( let recv_us = transaction_with_slot.recv_us; let adapter_callback = create_metrics_callback(callback); + let accounts = tx.message.static_account_keys(); - EventParser::parse_versioned_transaction_owned( + EventParser::parse_instruction_events_from_versioned_transaction( protocols, event_type_filter, - tx, + &tx, signature, Some(slot), None, recv_us, + accounts, + &[], bot_wallet, None, - &[], adapter_callback, ) .await?; diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 22e3d20..5711d04 100644 --- a/src/streaming/event_parser/core/account_event_parser.rs +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -1,8 +1,6 @@ -use crate::streaming::common::SimdUtils; use crate::streaming::event_parser::common::filter::EventTypeFilter; use crate::streaming::event_parser::common::high_performance_clock::elapsed_micros_since; -use crate::streaming::event_parser::common::EventMetadata; -use crate::streaming::event_parser::core::parser_cache::get_account_configs; +use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; use crate::streaming::event_parser::core::traits::DexEvent; use crate::streaming::event_parser::Protocol; use crate::streaming::grpc::AccountPretty; @@ -63,46 +61,93 @@ impl AccountEventParser { account: AccountPretty, event_type_filter: Option<&EventTypeFilter>, ) -> Option { - // 直接从 parser_cache 获取配置 - let configs = get_account_configs( - protocols, - event_type_filter, - Self::parse_nonce_account_event, - Self::parse_token_account_event, - ); - for config in configs { - if config.program_id == Pubkey::default() - || (account.owner == config.program_id - && SimdUtils::fast_discriminator_match( - &account.data, - config.account_discriminator, - )) - { - let event = (config.account_parser)( - &account, - EventMetadata { + use crate::streaming::event_parser::core::dispatcher::EventDispatcher; + + // 1. 尝试从账户 discriminator 解析（协议特定账户） + if account.data.len() >= 8 { + let discriminator = &account.data[0..8]; + + // 尝试识别协议类型 + if let Some(protocol) = EventDispatcher::match_protocol_by_program_id(&account.owner) { + // 检查是否在请求的协议列表中 + if protocols.contains(&protocol) { + // 构建临时元数据（protocol会被dispatcher设置，event_type会在parser中设置） + let metadata = EventMetadata { slot: account.slot, signature: account.signature, - protocol: config.protocol_type, - event_type: config.event_type, - program_id: config.program_id, + protocol: ProtocolType::Common, // 会被 EventDispatcher::dispatch_account 设置 + event_type: EventType::default(), // 会被具体 parser 设置 + program_id: account.owner, recv_us: account.recv_us, handle_us: elapsed_micros_since(account.recv_us), ..Default::default() - }, - ); - if event.is_some() { - return event; + }; + + // 使用 dispatcher 解析 + if let Some(event) = EventDispatcher::dispatch_account( + protocol, + discriminator, + &account, + metadata, + ) { + // 应用事件类型过滤 + if let Some(filter) = event_type_filter { + if filter.include.contains(&event.metadata().event_type) { + return Some(event); + } + // 不匹配过滤器，继续尝试其他解析方式 + } else { + return Some(event); + } + } } } } + + // 2. 尝试解析特殊账户类型（Token、Nonce等） + // 这些是通用的，不属于特定协议 + let metadata = EventMetadata { + slot: account.slot, + signature: account.signature, + protocol: ProtocolType::Common, + event_type: EventType::default(), + program_id: account.owner, + recv_us: account.recv_us, + handle_us: elapsed_micros_since(account.recv_us), + ..Default::default() + }; + + // 尝试解析 Nonce 账户 + if let Some(event) = Self::parse_nonce_account_event(&account, metadata.clone()) { + if let Some(filter) = event_type_filter { + if filter.include.contains(&event.metadata().event_type) { + return Some(event); + } + } else { + return Some(event); + } + } + + // 尝试解析 Token 账户 + if let Some(event) = Self::parse_token_account_event(&account, metadata) { + if let Some(filter) = event_type_filter { + if filter.include.contains(&event.metadata().event_type) { + return Some(event); + } + } else { + return Some(event); + } + } + None } pub fn parse_token_account_event( account: &AccountPretty, - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::TokenAccount; + let pubkey = account.pubkey; let executable = account.executable; let lamports = account.lamports; @@ -169,8 +214,10 @@ impl AccountEventParser { pub fn parse_nonce_account_event( account: &AccountPretty, - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::NonceAccount; + if let Ok(info) = parse_nonce(&account.data) { match info { solana_account_decoder::parse_nonce::UiNonceState::Initialized(details) => { diff --git a/src/streaming/event_parser/core/dispatcher.rs b/src/streaming/event_parser/core/dispatcher.rs new file mode 100644 index 0000000..f5a5054 --- /dev/null +++ b/src/streaming/event_parser/core/dispatcher.rs @@ -0,0 +1,245 @@ +//! 中心事件解析调度器 +//! +//! 根据协议类型路由到对应的解析函数，替代原有的静态 CONFIGS 数组架构 +//! +//! ## 设计原则 +//! - **单一职责**: 每个函数只负责一件事（路由、解析、合并分离） +//! - **灵活性**: 调用方可以选择是否合并，或自定义合并逻辑 +//! - **可测试性**: 每个函数都可以独立测试 + +use crate::streaming::event_parser::{ + common::EventMetadata, + protocols::{ + bonk::parser as bonk, pumpfun::parser as pumpfun, pumpswap::parser as pumpswap, + raydium_amm_v4::parser as raydium_amm_v4, raydium_clmm::parser as raydium_clmm, + raydium_cpmm::parser as raydium_cpmm, + }, + DexEvent, Protocol, +}; +use solana_sdk::pubkey::Pubkey; + +/// 中心事件解析调度器 +/// +/// 负责将解析请求路由到对应协议的解析函数 +pub struct EventDispatcher; + +impl EventDispatcher { + /// 解析 instruction 事件（只解析，不合并） + /// + /// # 参数 + /// - `protocol`: 协议类型 + /// - `instruction_discriminator`: 指令判别器 (8 bytes) + /// - `instruction_data`: 指令数据 + /// - `accounts`: 账户公钥列表 + /// - `metadata`: 事件元数据 + /// + /// # 返回 + /// 解析成功返回 `Some(DexEvent)`，否则返回 `None` + #[inline] + pub fn dispatch_instruction( + protocol: Protocol, + instruction_discriminator: &[u8], + instruction_data: &[u8], + accounts: &[Pubkey], + mut metadata: EventMetadata, + ) -> Option { + // 根据协议类型设置 metadata.protocol + use crate::streaming::event_parser::common::ProtocolType; + metadata.protocol = match protocol { + Protocol::PumpFun => ProtocolType::PumpFun, + Protocol::PumpSwap => ProtocolType::PumpSwap, + Protocol::Bonk => ProtocolType::Bonk, + Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm, + Protocol::RaydiumClmm => ProtocolType::RaydiumClmm, + Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4, + }; + + match protocol { + Protocol::PumpFun => pumpfun::parse_pumpfun_instruction_data( + instruction_discriminator, + instruction_data, + accounts, + metadata, + ), + Protocol::PumpSwap => pumpswap::parse_pumpswap_instruction_data( + instruction_discriminator, + instruction_data, + accounts, + metadata, + ), + Protocol::Bonk => bonk::parse_bonk_instruction_data( + instruction_discriminator, + instruction_data, + accounts, + metadata, + ), + Protocol::RaydiumCpmm => raydium_cpmm::parse_raydium_cpmm_instruction_data( + instruction_discriminator, + instruction_data, + accounts, + metadata, + ), + Protocol::RaydiumClmm => raydium_clmm::parse_raydium_clmm_instruction_data( + instruction_discriminator, + instruction_data, + accounts, + metadata, + ), + Protocol::RaydiumAmmV4 => raydium_amm_v4::parse_raydium_amm_v4_instruction_data( + instruction_discriminator, + instruction_data, + accounts, + metadata, + ), + } + } + + /// 解析 inner instruction 事件（只解析，不合并） + /// + /// # 参数 + /// - `protocol`: 协议类型 + /// - `inner_instruction_discriminator`: 内联指令判别器 (16 bytes) + /// - `inner_instruction_data`: 内联指令数据 + /// - `metadata`: 事件元数据 + /// + /// # 返回 + /// 解析成功返回 `Some(DexEvent)`，否则返回 `None` + #[inline] + pub fn dispatch_inner_instruction( + protocol: Protocol, + inner_instruction_discriminator: &[u8], + inner_instruction_data: &[u8], + mut metadata: EventMetadata, + ) -> Option { + // 根据协议类型设置 metadata.protocol + use crate::streaming::event_parser::common::ProtocolType; + metadata.protocol = match protocol { + Protocol::PumpFun => ProtocolType::PumpFun, + Protocol::PumpSwap => ProtocolType::PumpSwap, + Protocol::Bonk => ProtocolType::Bonk, + Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm, + Protocol::RaydiumClmm => ProtocolType::RaydiumClmm, + Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4, + }; + + match protocol { + Protocol::PumpFun => pumpfun::parse_pumpfun_inner_instruction_data( + inner_instruction_discriminator, + inner_instruction_data, + metadata, + ), + Protocol::PumpSwap => pumpswap::parse_pumpswap_inner_instruction_data( + inner_instruction_discriminator, + inner_instruction_data, + metadata, + ), + Protocol::Bonk => bonk::parse_bonk_inner_instruction_data( + inner_instruction_discriminator, + inner_instruction_data, + metadata, + ), + Protocol::RaydiumCpmm => raydium_cpmm::parse_raydium_cpmm_inner_instruction_data( + inner_instruction_discriminator, + inner_instruction_data, + metadata, + ), + Protocol::RaydiumClmm => raydium_clmm::parse_raydium_clmm_inner_instruction_data( + inner_instruction_discriminator, + inner_instruction_data, + metadata, + ), + Protocol::RaydiumAmmV4 => raydium_amm_v4::parse_raydium_amm_v4_inner_instruction_data( + inner_instruction_discriminator, + inner_instruction_data, + metadata, + ), + } + } + + /// 通过 program_id 匹配协议类型 + #[inline] + pub fn match_protocol_by_program_id(program_id: &Pubkey) -> Option { + if program_id == &pumpfun::PUMPFUN_PROGRAM_ID { + Some(Protocol::PumpFun) + } else if program_id == &pumpswap::PUMPSWAP_PROGRAM_ID { + Some(Protocol::PumpSwap) + } else if program_id == &bonk::BONK_PROGRAM_ID { + Some(Protocol::Bonk) + } else if program_id == &raydium_cpmm::RAYDIUM_CPMM_PROGRAM_ID { + Some(Protocol::RaydiumCpmm) + } else if program_id == &raydium_clmm::RAYDIUM_CLMM_PROGRAM_ID { + Some(Protocol::RaydiumClmm) + } else if program_id == &raydium_amm_v4::RAYDIUM_AMM_V4_PROGRAM_ID { + Some(Protocol::RaydiumAmmV4) + } else { + None + } + } + + /// 获取指定协议的 program_id + #[inline] + pub fn get_program_id(protocol: Protocol) -> Pubkey { + match protocol { + Protocol::PumpFun => pumpfun::PUMPFUN_PROGRAM_ID, + Protocol::PumpSwap => pumpswap::PUMPSWAP_PROGRAM_ID, + Protocol::Bonk => bonk::BONK_PROGRAM_ID, + Protocol::RaydiumCpmm => raydium_cpmm::RAYDIUM_CPMM_PROGRAM_ID, + Protocol::RaydiumClmm => raydium_clmm::RAYDIUM_CLMM_PROGRAM_ID, + Protocol::RaydiumAmmV4 => raydium_amm_v4::RAYDIUM_AMM_V4_PROGRAM_ID, + } + } + + /// 批量获取 program_ids + pub fn get_program_ids(protocols: &[Protocol]) -> Vec { + protocols.iter().map(|p| Self::get_program_id(p.clone())).collect() + } + + /// 解析账户数据 + /// + /// 根据账户的 discriminator 路由到对应协议的账户解析函数 + /// + /// # 参数 + /// - `protocol`: 协议类型 + /// - `discriminator`: 账户判别器 + /// - `account`: 账户信息 + /// - `metadata`: 事件元数据 + /// + /// # 返回 + /// 解析成功返回 `Some(DexEvent)`，否则返回 `None` + pub fn dispatch_account( + protocol: Protocol, + discriminator: &[u8], + account: &crate::streaming::grpc::AccountPretty, + mut metadata: crate::streaming::event_parser::common::EventMetadata, + ) -> Option { + // 根据协议类型设置 metadata.protocol + use crate::streaming::event_parser::common::ProtocolType; + metadata.protocol = match protocol { + Protocol::PumpFun => ProtocolType::PumpFun, + Protocol::PumpSwap => ProtocolType::PumpSwap, + Protocol::Bonk => ProtocolType::Bonk, + Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm, + Protocol::RaydiumClmm => ProtocolType::RaydiumClmm, + Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4, + }; + + match protocol { + Protocol::PumpFun => { + pumpfun::parse_pumpfun_account_data(discriminator, account, metadata) + } + Protocol::PumpSwap => { + pumpswap::parse_pumpswap_account_data(discriminator, account, metadata) + } + Protocol::Bonk => bonk::parse_bonk_account_data(discriminator, account, metadata), + Protocol::RaydiumCpmm => { + raydium_cpmm::parse_raydium_cpmm_account_data(discriminator, account, metadata) + } + Protocol::RaydiumClmm => { + raydium_clmm::parse_raydium_clmm_account_data(discriminator, account, metadata) + } + Protocol::RaydiumAmmV4 => { + raydium_amm_v4::parse_raydium_amm_v4_account_data(discriminator, account, metadata) + } + } + } +} diff --git a/src/streaming/event_parser/core/event_parser.rs b/src/streaming/event_parser/core/event_parser.rs index 605772b..0da6ee1 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -1,40 +1,210 @@ -use crate::streaming::{ - common::SimdUtils, - event_parser::{ - common::{ - filter::EventTypeFilter, - high_performance_clock::{elapsed_micros_since, get_high_perf_clock}, - parse_swap_data_from_next_grpc_instructions, parse_swap_data_from_next_instructions, - EventMetadata, - }, - core::{ - global_state::{ - add_bonk_dev_address, add_dev_address, is_bonk_dev_address_in_signature, - is_dev_address_in_signature, - }, - merger_event::merge, - parser_cache::{ - build_account_pubkeys_with_cache, get_global_instruction_configs, - get_global_program_ids, GenericEventParseConfig, - }, - }, - DexEvent, Protocol, +use crate::streaming::event_parser::{ + common::{ + filter::EventTypeFilter, high_performance_clock::elapsed_micros_since, + parse_swap_data_from_next_grpc_instructions, parse_swap_data_from_next_instructions, + EventMetadata, }, + core::{ + dispatcher::EventDispatcher, + global_state::{ + add_bonk_dev_address, add_dev_address, is_bonk_dev_address_in_signature, + is_dev_address_in_signature, + }, + merger_event::merge, + }, + DexEvent, Protocol, }; use prost_types::Timestamp; use solana_sdk::{ - bs58, message::compiled_instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature, + message::compiled_instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature, transaction::VersionedTransaction, }; -use solana_transaction_status::{ - EncodedConfirmedTransactionWithStatusMeta, InnerInstruction, InnerInstructions, UiInstruction, -}; +use solana_transaction_status::InnerInstructions; use std::sync::Arc; use yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo; pub struct EventParser {} impl EventParser { + // ================================================================================================ + // Public API - Entry Points + // ================================================================================================ + + /// Parse transaction from gRPC stream + /// + /// This is the main entry point for parsing transactions received from gRPC streams. + /// It extracts account keys, inner instructions, and delegates to instruction parsing. + pub async fn parse_grpc_transaction( + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + grpc_tx: SubscribeUpdateTransactionInfo, + signature: Signature, + slot: Option, + block_time: Option, + recv_us: i64, + bot_wallet: Option, + transaction_index: Option, + callback: Arc, + ) -> anyhow::Result<()> { + // 创建适配器回调，将所有权回调转换为引用回调 + let adapter_callback = Arc::new(move |event: &DexEvent| { + callback(event.clone()); + }); + if let Some(transition) = grpc_tx.transaction { + if let Some(message) = &transition.message { + let mut address_table_lookups: Vec> = vec![]; + let mut inner_instructions: Vec< + yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions, + > = vec![]; + + if let Some(meta) = grpc_tx.meta { + inner_instructions = meta.inner_instructions; + address_table_lookups.reserve( + meta.loaded_writable_addresses.len() + meta.loaded_readonly_addresses.len(), + ); + let loaded_writable_addresses = meta.loaded_writable_addresses; + let loaded_readonly_addresses = meta.loaded_readonly_addresses; + address_table_lookups.extend( + loaded_writable_addresses.into_iter().chain(loaded_readonly_addresses), + ); + } + + let mut accounts_bytes: Vec> = + Vec::with_capacity(message.account_keys.len() + address_table_lookups.len()); + accounts_bytes.extend_from_slice(&message.account_keys); + accounts_bytes.extend(address_table_lookups); + // 转换为 Pubkey + let accounts: Vec = accounts_bytes + .iter() + .filter_map(|account| { + if account.len() == 32 { + Some(Pubkey::try_from(account.as_slice()).unwrap_or_default()) + } else { + None + } + }) + .collect(); + // 解析指令事件 + let instructions = &message.instructions; + Self::parse_instruction_events_from_grpc_transaction( + protocols, + event_type_filter, + &instructions, + signature, + slot, + block_time, + recv_us, + &accounts, + &inner_instructions, + bot_wallet, + transaction_index, + adapter_callback, + ) + .await?; + } + } + + Ok(()) + } + + /// Parse transaction from VersionedTransaction + /// + /// This is the entry point for parsing VersionedTransaction objects. + /// It's used when working with RPC responses or historical data. + #[allow(clippy::too_many_arguments)] + pub async fn parse_instruction_events_from_versioned_transaction( + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + transaction: &VersionedTransaction, + signature: Signature, + slot: Option, + block_time: Option, + recv_us: i64, + accounts: &[Pubkey], + inner_instructions: &[InnerInstructions], + bot_wallet: Option, + transaction_index: Option, + callback: Arc, + ) -> anyhow::Result<()> { + // 创建适配器回调，将所有权回调转换为引用回调 + let adapter_callback = Arc::new(move |event: &DexEvent| { + callback(event.clone()); + }); + // 获取交易的指令和账户 + let compiled_instructions = transaction.message.instructions(); + let mut accounts: Vec = accounts.to_vec(); + // 检查交易中是否包含程序 + let has_program = accounts + .iter() + .any(|account| Self::should_handle(protocols, event_type_filter, account)); + if has_program { + // 解析每个指令 + for (index, instruction) in compiled_instructions.iter().enumerate() { + if let Some(program_id) = accounts.get(instruction.program_id_index as usize) { + let program_id = *program_id; // 克隆程序ID，避免借用冲突 + let inner_instructions = inner_instructions + .iter() + .find(|inner_instruction| inner_instruction.index == index as u8); + if Self::should_handle(protocols, event_type_filter, &program_id) { + let max_idx = instruction.accounts.iter().max().unwrap_or(&0); + // 补齐accounts(使用Pubkey::default()) + if *max_idx as usize >= accounts.len() { + accounts.resize(*max_idx as usize + 1, Pubkey::default()); + } + Self::parse_events_from_instruction( + protocols, + event_type_filter, + instruction, + &accounts, + signature, + slot.unwrap_or(0), + block_time, + recv_us, + index as i64, + None, + bot_wallet, + transaction_index, + inner_instructions, + adapter_callback.clone(), + )?; + } + // Immediately process inner instructions for correct ordering + if let Some(inner_instructions) = inner_instructions { + for (inner_index, inner_instruction) in + inner_instructions.instructions.iter().enumerate() + { + Self::parse_events_from_instruction( + protocols, + event_type_filter, + &inner_instruction.instruction, + &accounts, + signature, + slot.unwrap_or(0), + block_time, + recv_us, + index as i64, + Some(inner_index as i64), + bot_wallet, + transaction_index, + Some(&inner_instructions), + adapter_callback.clone(), + )?; + } + } + } + } + } + Ok(()) + } + + // ================================================================================================ + // gRPC Transaction Processing + // ================================================================================================ + + /// Parse instruction events from gRPC transaction format + /// + /// Iterates through all instructions in a gRPC transaction, checks if they should be handled, + /// and delegates to instruction-level parsing for both outer and inner instructions. #[allow(clippy::too_many_arguments)] async fn parse_instruction_events_from_grpc_transaction( protocols: &[Protocol], @@ -124,653 +294,10 @@ impl EventParser { Ok(()) } - /// 从VersionedTransaction中解析指令事件的通用方法 - #[allow(clippy::too_many_arguments)] - async fn parse_instruction_events_from_versioned_transaction( - protocols: &[Protocol], - event_type_filter: Option<&EventTypeFilter>, - transaction: &VersionedTransaction, - signature: Signature, - slot: Option, - block_time: Option, - recv_us: i64, - accounts: &[Pubkey], - inner_instructions: &[InnerInstructions], - bot_wallet: Option, - transaction_index: Option, - callback: Arc Fn(&'a DexEvent) + Send + Sync>, - ) -> anyhow::Result<()> { - // 获取交易的指令和账户 - let compiled_instructions = transaction.message.instructions(); - let mut accounts: Vec = accounts.to_vec(); - // 检查交易中是否包含程序 - let has_program = accounts - .iter() - .any(|account| Self::should_handle(protocols, event_type_filter, account)); - if has_program { - // 解析每个指令 - for (index, instruction) in compiled_instructions.iter().enumerate() { - if let Some(program_id) = accounts.get(instruction.program_id_index as usize) { - let program_id = *program_id; // 克隆程序ID，避免借用冲突 - let inner_instructions = inner_instructions - .iter() - .find(|inner_instruction| inner_instruction.index == index as u8); - if Self::should_handle(protocols, event_type_filter, &program_id) { - let max_idx = instruction.accounts.iter().max().unwrap_or(&0); - // 补齐accounts(使用Pubkey::default()) - if *max_idx as usize >= accounts.len() { - accounts.resize(*max_idx as usize + 1, Pubkey::default()); - } - Self::parse_events_from_instruction( - protocols, - event_type_filter, - instruction, - &accounts, - signature, - slot.unwrap_or(0), - block_time, - recv_us, - index as i64, - None, - bot_wallet, - transaction_index, - inner_instructions, - callback.clone(), - )?; - } - // Immediately process inner instructions for correct ordering - if let Some(inner_instructions) = inner_instructions { - for (inner_index, inner_instruction) in - inner_instructions.instructions.iter().enumerate() - { - Self::parse_events_from_instruction( - protocols, - event_type_filter, - &inner_instruction.instruction, - &accounts, - signature, - slot.unwrap_or(0), - block_time, - recv_us, - index as i64, - Some(inner_index as i64), - bot_wallet, - transaction_index, - Some(&inner_instructions), - callback.clone(), - )?; - } - } - } - } - } - Ok(()) - } - - pub async fn parse_versioned_transaction_owned( - protocols: &[Protocol], - event_type_filter: Option<&EventTypeFilter>, - versioned_tx: VersionedTransaction, - signature: Signature, - slot: Option, - block_time: Option, - recv_us: i64, - bot_wallet: Option, - transaction_index: Option, - inner_instructions: &[InnerInstructions], - callback: Arc, - ) -> anyhow::Result<()> { - // 创建适配器回调，将所有权回调转换为引用回调 - let adapter_callback = Arc::new(move |event: &DexEvent| { - callback(event.clone()); - }); - let accounts = versioned_tx.message.static_account_keys(); - Self::parse_instruction_events_from_versioned_transaction( - protocols, - event_type_filter, - &versioned_tx, - signature, - slot, - block_time, - recv_us, - accounts, - inner_instructions, - bot_wallet, - transaction_index, - adapter_callback, - ) - .await - } - - pub async fn parse_grpc_transaction_owned( - protocols: &[Protocol], - event_type_filter: Option<&EventTypeFilter>, - grpc_tx: SubscribeUpdateTransactionInfo, - signature: Signature, - slot: Option, - block_time: Option, - recv_us: i64, - bot_wallet: Option, - transaction_index: Option, - callback: Arc, - ) -> anyhow::Result<()> { - // 创建适配器回调，将所有权回调转换为引用回调 - let adapter_callback = Arc::new(move |event: &DexEvent| { - callback(event.clone()); - }); - // 调用原始方法 - Self::parse_grpc_transaction( - protocols, - event_type_filter, - grpc_tx, - signature, - slot, - block_time, - recv_us, - bot_wallet, - transaction_index, - adapter_callback, - ) - .await - } - - async fn parse_grpc_transaction( - protocols: &[Protocol], - event_type_filter: Option<&EventTypeFilter>, - grpc_tx: SubscribeUpdateTransactionInfo, - signature: Signature, - slot: Option, - block_time: Option, - recv_us: i64, - bot_wallet: Option, - transaction_index: Option, - callback: Arc Fn(&'a DexEvent) + Send + Sync>, - ) -> anyhow::Result<()> { - if let Some(transition) = grpc_tx.transaction { - if let Some(message) = &transition.message { - let mut address_table_lookups: Vec> = vec![]; - let mut inner_instructions: Vec< - yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions, - > = vec![]; - - if let Some(meta) = grpc_tx.meta { - inner_instructions = meta.inner_instructions; - address_table_lookups.reserve( - meta.loaded_writable_addresses.len() + meta.loaded_readonly_addresses.len(), - ); - let loaded_writable_addresses = meta.loaded_writable_addresses; - let loaded_readonly_addresses = meta.loaded_readonly_addresses; - address_table_lookups.extend( - loaded_writable_addresses.into_iter().chain(loaded_readonly_addresses), - ); - } - - let mut accounts_bytes: Vec> = - Vec::with_capacity(message.account_keys.len() + address_table_lookups.len()); - accounts_bytes.extend_from_slice(&message.account_keys); - accounts_bytes.extend(address_table_lookups); - // 转换为 Pubkey - let accounts: Vec = accounts_bytes - .iter() - .filter_map(|account| { - if account.len() == 32 { - Some(Pubkey::try_from(account.as_slice()).unwrap_or_default()) - } else { - None - } - }) - .collect(); - // 解析指令事件 - let instructions = &message.instructions; - Self::parse_instruction_events_from_grpc_transaction( - protocols, - event_type_filter, - &instructions, - signature, - slot, - block_time, - recv_us, - &accounts, - &inner_instructions, - bot_wallet, - transaction_index, - callback, - ) - .await?; - } - } - - Ok(()) - } - - pub async fn parse_encoded_confirmed_transaction_with_status_meta( - protocols: &[Protocol], - event_type_filter: Option<&EventTypeFilter>, - signature: Signature, - transaction: EncodedConfirmedTransactionWithStatusMeta, - callback: Arc Fn(&'a DexEvent) + Send + Sync>, - ) -> anyhow::Result<()> { - let versioned_tx = match transaction.transaction.transaction.decode() { - Some(tx) => tx, - None => { - return Ok(()); - } - }; - let mut inner_instructions_vec: Vec = Vec::new(); - if let Some(meta) = &transaction.transaction.meta { - // 从meta中获取inner_instructions，处理OptionSerializer类型 - if let solana_transaction_status::option_serializer::OptionSerializer::Some( - ui_inner_insts, - ) = &meta.inner_instructions - { - // 将UiInnerInstructions转换为InnerInstructions - for ui_inner in ui_inner_insts { - let mut converted_instructions = Vec::new(); - - // 转换每个UiInstruction为InnerInstruction - for ui_instruction in &ui_inner.instructions { - if let UiInstruction::Compiled(ui_compiled) = ui_instruction { - // 解码base58编码的data - if let Ok(data) = bs58::decode(&ui_compiled.data).into_vec() { - // base64解码 - let compiled_instruction = CompiledInstruction { - program_id_index: ui_compiled.program_id_index, - accounts: ui_compiled.accounts.to_vec(), - data, - }; - - let inner_instruction = InnerInstruction { - instruction: compiled_instruction, - stack_height: ui_compiled.stack_height, - }; - - converted_instructions.push(inner_instruction); - } - } - } - - let inner_instructions = InnerInstructions { - index: ui_inner.index, - instructions: converted_instructions, - }; - - inner_instructions_vec.push(inner_instructions); - } - } - } - let inner_instructions: &[InnerInstructions] = &inner_instructions_vec; - - let meta = transaction.transaction.meta; - let mut address_table_lookups: Vec = vec![]; - if let Some(meta) = meta { - if let solana_transaction_status::option_serializer::OptionSerializer::Some( - loaded_addresses, - ) = &meta.loaded_addresses - { - address_table_lookups - .reserve(loaded_addresses.writable.len() + loaded_addresses.readonly.len()); - address_table_lookups.extend( - loaded_addresses - .writable - .iter() - .filter_map(|s| s.parse::().ok()) - .chain( - loaded_addresses - .readonly - .iter() - .filter_map(|s| s.parse::().ok()), - ), - ); - } - } - let mut accounts = Vec::with_capacity( - versioned_tx.message.static_account_keys().len() + address_table_lookups.len(), - ); - accounts.extend_from_slice(versioned_tx.message.static_account_keys()); - accounts.extend(address_table_lookups); - - let slot = transaction.slot; - let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 }); - let recv_us = get_high_perf_clock(); - let bot_wallet = None; - let transaction_index = None; - // 解析指令事件 - Self::parse_instruction_events_from_versioned_transaction( - protocols, - event_type_filter, - &versioned_tx, - signature, - Some(slot), - block_time, - recv_us, - &accounts, - inner_instructions, - bot_wallet, - transaction_index, - callback, - ) - .await - } - - /// Helper function to create EventMetadata from common parameters - #[inline] - fn create_metadata( - config: &GenericEventParseConfig, - signature: Signature, - slot: u64, - block_time: Option, - recv_us: i64, - outer_index: i64, - inner_index: Option, - transaction_index: Option, - ) -> EventMetadata { - let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); - let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; - EventMetadata::new( - signature, - slot, - timestamp.seconds, - block_time_ms, - config.protocol_type.clone(), - config.event_type.clone(), - config.program_id, - outer_index, - inner_index, - recv_us, - transaction_index, - ) - } - - /// 通用的内联指令解析方法 - #[allow(clippy::too_many_arguments)] - fn parse_inner_instruction_event( - config: &GenericEventParseConfig, - data: &[u8], - signature: Signature, - slot: u64, - block_time: Option, - recv_us: i64, - outer_index: i64, - inner_index: Option, - transaction_index: Option, - ) -> Option { - config.inner_instruction_parser.and_then(|parser| { - let metadata = Self::create_metadata( - config, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - ); - parser(data, metadata) - }) - } - - /// 通用的指令解析方法 - #[allow(clippy::too_many_arguments)] - fn parse_instruction_event( - config: &GenericEventParseConfig, - data: &[u8], - account_pubkeys: &[Pubkey], - signature: Signature, - slot: u64, - block_time: Option, - recv_us: i64, - outer_index: i64, - inner_index: Option, - transaction_index: Option, - ) -> Option { - config.instruction_parser.and_then(|parser| { - let metadata = Self::create_metadata( - config, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - ); - parser(data, account_pubkeys, metadata) - }) - } - - /// 从内联指令中解析事件数据 - 通用实现 - #[allow(clippy::too_many_arguments)] - fn parse_events_from_inner_instruction_data( - data: &[u8], - signature: Signature, - slot: u64, - block_time: Option, - recv_us: i64, - outer_index: i64, - inner_index: Option, - transaction_index: Option, - config: &GenericEventParseConfig, - ) -> Vec { - // Use SIMD-optimized data validation with correct discriminator length - let discriminator_len = config.inner_instruction_discriminator.len(); - if !SimdUtils::validate_instruction_data_simd(data, 16, discriminator_len) { - return Vec::new(); - } - - // Use SIMD-optimized discriminator matching - if !SimdUtils::fast_discriminator_match(data, config.inner_instruction_discriminator) { - return Vec::new(); - } - - let data = &data[16..]; - Self::parse_inner_instruction_event( - config, - data, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - ) - .into_iter() - .collect() - } - - /// 从内联指令中解析事件数据 - #[allow(clippy::too_many_arguments)] - fn parse_events_from_inner_instruction( - inner_instruction: &CompiledInstruction, - signature: Signature, - slot: u64, - block_time: Option, - recv_us: i64, - outer_index: i64, - inner_index: Option, - transaction_index: Option, - config: &GenericEventParseConfig, - ) -> Vec { - Self::parse_events_from_inner_instruction_data( - &inner_instruction.data, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - config, - ) - } - - /// 从内联指令中解析事件数据 - #[allow(clippy::too_many_arguments)] - fn parse_events_from_grpc_inner_instruction( - inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstruction, - signature: Signature, - slot: u64, - block_time: Option, - recv_us: i64, - outer_index: i64, - inner_index: Option, - transaction_index: Option, - config: &GenericEventParseConfig, - ) -> Vec { - Self::parse_events_from_inner_instruction_data( - &inner_instruction.data, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - config, - ) - } - - /// 从指令中解析事件 - #[allow(clippy::too_many_arguments)] - fn parse_events_from_instruction( - protocols: &[Protocol], - event_type_filter: Option<&EventTypeFilter>, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: Signature, - slot: u64, - block_time: Option, - recv_us: i64, - outer_index: i64, - inner_index: Option, - bot_wallet: Option, - transaction_index: Option, - inner_instructions: Option<&InnerInstructions>, - callback: Arc Fn(&'a DexEvent) + Send + Sync>, - ) -> anyhow::Result<()> { - // 添加边界检查以防止越界访问 - let program_id_index = instruction.program_id_index as usize; - if program_id_index >= accounts.len() { - return Ok(()); - } - let program_id = accounts[program_id_index]; - if !Self::should_handle(protocols, event_type_filter, &program_id) { - return Ok(()); - } - // 一维化并行处理：将所有 (discriminator, config) 组合展开并行处理 - let instruction_configs = get_global_instruction_configs(protocols, event_type_filter); - let all_processing_params: Vec<_> = instruction_configs - .iter() - .filter(|(disc, _)| { - // Use SIMD-optimized data validation and discriminator matching - SimdUtils::validate_instruction_data_simd(&instruction.data, disc.len(), disc.len()) - && SimdUtils::fast_discriminator_match(&instruction.data, disc) - }) - .flat_map(|(disc, configs)| { - configs - .iter() - .filter(|config| config.program_id == program_id) - .map(move |config| (disc, config)) - }) - .collect(); - - // Use SIMD-optimized account indices validation (只需检查一次) - if !SimdUtils::validate_account_indices_simd(&instruction.accounts, accounts.len()) { - return Ok(()); - } - - // 使用线程局部缓存构建账户公钥列表，避免重复分配 (只需构建一次) - let account_pubkeys = build_account_pubkeys_with_cache(&instruction.accounts, accounts); - - // 并行处理所有 (discriminator, config) 组合 - let all_results: Vec<_> = all_processing_params - .iter() - .filter_map(|(disc, config)| { - let data = &instruction.data[disc.len()..]; - Self::parse_instruction_event( - config, - data, - &account_pubkeys, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - ) - .map(|event| (*disc, *config, event)) - }) - .collect(); - - for (_disc, config, mut event) in all_results { - // 阻塞处理：原有的同步逻辑 - let mut inner_instruction_event: Option = None; - if let Some(inner_instructions_ref) = inner_instructions { - // 并行执行两个任务 - let (inner_event_result, swap_data_result) = std::thread::scope(|s| { - let inner_event_handle = s.spawn(|| { - for inner_instruction in inner_instructions_ref.instructions.iter() { - let result = Self::parse_events_from_inner_instruction( - &inner_instruction.instruction, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - &config, - ); - if !result.is_empty() { - return result.into_iter().next(); - } - } - None - }); - - let swap_data_handle = s.spawn(|| { - if event.metadata().swap_data.is_none() { - parse_swap_data_from_next_instructions( - &event, - inner_instructions_ref, - inner_index.unwrap_or(-1_i64) as i8, - accounts, - ) - } else { - None - } - }); - - // 等待两个任务完成 - (inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap()) - }); - - inner_instruction_event = inner_event_result; - if let Some(swap_data) = swap_data_result { - event.metadata_mut().set_swap_data(swap_data); - } - } - - // Skip events that require inner instruction data but don't have it - if config.requires_inner_instruction && inner_instruction_event.is_none() { - continue; - } - - // 合并事件 - if let Some(inner_instruction_event) = inner_instruction_event { - merge(&mut event, inner_instruction_event); - } - // 设置处理时间（使用高性能时钟） - event.metadata_mut().handle_us = elapsed_micros_since(recv_us); - event = Self::process_event(event, bot_wallet); - callback(&event); - } - Ok(()) - } - - /// 从指令中解析事件 - /// TODO: - wait refactor + /// Parse events from gRPC instruction + /// + /// Core parsing logic for a single gRPC instruction. Extracts discriminator, dispatches + /// to protocol-specific parsers, handles inner instructions, and processes swap data. #[allow(clippy::too_many_arguments)] fn parse_events_from_grpc_instruction( protocols: &[Protocol], @@ -797,126 +324,319 @@ impl EventParser { if !Self::should_handle(protocols, event_type_filter, &program_id) { return Ok(()); } - // 一维化并行处理：将所有 (discriminator, config) 组合展开并行处理 - let instruction_configs = get_global_instruction_configs(protocols, event_type_filter); - let all_processing_params: Vec<_> = instruction_configs - .iter() - .filter(|(disc, _)| { - // Use SIMD-optimized data validation and discriminator matching - SimdUtils::validate_instruction_data_simd(&instruction.data, disc.len(), disc.len()) - && SimdUtils::fast_discriminator_match(&instruction.data, disc) - }) - .flat_map(|(disc, configs)| { - configs - .iter() - .filter(|config| config.program_id == program_id) - .map(move |config| (disc, config)) - }) - .collect(); - // Use SIMD-optimized account indices validation (只需检查一次) - if !SimdUtils::validate_account_indices_simd(&instruction.accounts, accounts.len()) { + // 使用 EventDispatcher 匹配协议 + let protocol = match EventDispatcher::match_protocol_by_program_id(&program_id) { + Some(p) => p, + None => return Ok(()), + }; + + // 检查指令数据长度（至少需要 8 字节的 discriminator） + if instruction.data.len() < 8 { return Ok(()); } - // 使用线程局部缓存构建账户公钥列表，避免重复分配 (只需构建一次) - let account_pubkeys = build_account_pubkeys_with_cache(&instruction.accounts, accounts); + // 提取 discriminator 和数据 + let instruction_discriminator = &instruction.data[..8]; + let instruction_data = &instruction.data[8..]; - // 并行处理所有 (discriminator, config) 组合 - let all_results: Vec<_> = all_processing_params + // 构建账户公钥列表 + let account_pubkeys: Vec = instruction + .accounts .iter() - .filter_map(|(disc, config)| { - let data = &instruction.data[disc.len()..]; - Self::parse_instruction_event( - config, - data, - &account_pubkeys, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - ) - .map(|event| (*disc, *config, event)) - }) + .filter_map(|&idx| accounts.get(idx as usize).copied()) .collect(); - for (_disc, config, mut event) in all_results { - // 阻塞处理：原有的同步逻辑 - let mut inner_instruction_event: Option = None; - if let Some(inner_instructions_ref) = inner_instructions { - // 并行执行两个任务 - let (inner_event_result, swap_data_result) = std::thread::scope(|s| { - let inner_event_handle = s.spawn(|| { - for inner_instruction in inner_instructions_ref.instructions.iter() { - let result = Self::parse_events_from_grpc_inner_instruction( - inner_instruction, - signature, - slot, - block_time, - recv_us, - outer_index, - inner_index, - transaction_index, - &config, - ); - if !result.is_empty() { - return result.into_iter().next(); - } - } - None - }); + // 创建元数据 + let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); + let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; + let metadata = EventMetadata::new( + signature, + slot, + timestamp.seconds, + block_time_ms, + Default::default(), // protocol will be set by dispatcher + Default::default(), // event_type will be set by dispatcher + program_id, + outer_index, + inner_index, + recv_us, + transaction_index, + ); - let swap_data_handle = s.spawn(|| { - if event.metadata().swap_data.is_none() { - parse_swap_data_from_next_grpc_instructions( - &event, - inner_instructions_ref, - inner_index.unwrap_or(-1_i64) as i8, - accounts, - ) - } else { - None - } - }); + // 使用 EventDispatcher 解析 instruction 事件 + let mut event = match EventDispatcher::dispatch_instruction( + protocol.clone(), + instruction_discriminator, + instruction_data, + &account_pubkeys, + metadata.clone(), + ) { + Some(e) => e, + None => return Ok(()), + }; - // 等待两个任务完成 - (inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap()) + // 处理 inner instructions + let mut inner_instruction_event: Option = None; + if let Some(inner_instructions_ref) = inner_instructions { + // 并行执行两个任务: 解析 inner event 和提取 swap_data + let (inner_event_result, swap_data_result) = std::thread::scope(|s| { + let inner_event_handle = s.spawn(|| { + for inner_instruction in inner_instructions_ref.instructions.iter() { + let inner_data = &inner_instruction.data; + // 检查长度（需要 16 字节的 discriminator） + if inner_data.len() < 16 { + continue; + } + let inner_discriminator = &inner_data[..16]; + let inner_instruction_data = &inner_data[16..]; + + if let Some(inner_event) = EventDispatcher::dispatch_inner_instruction( + protocol.clone(), + inner_discriminator, + inner_instruction_data, + metadata.clone(), + ) { + return Some(inner_event); + } + } + None }); - inner_instruction_event = inner_event_result; - if let Some(swap_data) = swap_data_result { - event.metadata_mut().set_swap_data(swap_data); - } - } + let swap_data_handle = s.spawn(|| { + if event.metadata().swap_data.is_none() { + parse_swap_data_from_next_grpc_instructions( + &event, + inner_instructions_ref, + inner_index.unwrap_or(-1_i64) as i8, + accounts, + ) + } else { + None + } + }); - // Skip events that require inner instruction data but don't have it - if config.requires_inner_instruction && inner_instruction_event.is_none() { - continue; - } + // 等待两个任务完成 + (inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap()) + }); - // 合并事件 - if let Some(inner_instruction_event) = inner_instruction_event { - merge(&mut event, inner_instruction_event); + inner_instruction_event = inner_event_result; + if let Some(swap_data) = swap_data_result { + event.metadata_mut().set_swap_data(swap_data); } - // 设置处理时间（使用高性能时钟） - event.metadata_mut().handle_us = elapsed_micros_since(recv_us); - event = Self::process_event(event, bot_wallet); - callback(&event); } + + // 特殊处理: PumpFun MIGRATE 指令需要 inner instruction data + if matches!(protocol, Protocol::PumpFun) { + const PUMPFUN_MIGRATE_IX: &[u8] = &[155, 234, 231, 146, 236, 158, 162, 30]; + if instruction_discriminator == PUMPFUN_MIGRATE_IX && inner_instruction_event.is_none() + { + return Ok(()); + } + } + + // 合并事件 + if let Some(inner_instruction_event) = inner_instruction_event { + merge(&mut event, inner_instruction_event); + } + + // 设置处理时间（使用高性能时钟） + event.metadata_mut().handle_us = elapsed_micros_since(recv_us); + event = Self::process_event(event, bot_wallet); + callback(&event); + Ok(()) } - fn should_handle( + // ================================================================================================ + // Standard Instruction Processing + // ================================================================================================ + + /// Parse events from standard Solana instruction + /// + /// Similar to gRPC instruction parsing but works with standard CompiledInstruction format. + /// Used when parsing VersionedTransaction or RPC data. + #[allow(clippy::too_many_arguments)] + fn parse_events_from_instruction( protocols: &[Protocol], event_type_filter: Option<&EventTypeFilter>, - program_id: &Pubkey, - ) -> bool { - get_global_program_ids(protocols, event_type_filter).contains(program_id) + instruction: &CompiledInstruction, + accounts: &[Pubkey], + signature: Signature, + slot: u64, + block_time: Option, + recv_us: i64, + outer_index: i64, + inner_index: Option, + bot_wallet: Option, + transaction_index: Option, + inner_instructions: Option<&InnerInstructions>, + callback: Arc Fn(&'a DexEvent) + Send + Sync>, + ) -> anyhow::Result<()> { + // 添加边界检查以防止越界访问 + let program_id_index = instruction.program_id_index as usize; + if program_id_index >= accounts.len() { + return Ok(()); + } + let program_id = accounts[program_id_index]; + if !Self::should_handle(protocols, event_type_filter, &program_id) { + return Ok(()); + } + + // 使用 EventDispatcher 匹配协议 + let protocol = match EventDispatcher::match_protocol_by_program_id(&program_id) { + Some(p) => p, + None => return Ok(()), + }; + + // 检查指令数据长度（至少需要 8 字节的 discriminator） + if instruction.data.len() < 8 { + return Ok(()); + } + + // 提取 discriminator 和数据 + let instruction_discriminator = &instruction.data[..8]; + let instruction_data = &instruction.data[8..]; + + // 构建账户公钥列表 + let account_pubkeys: Vec = instruction + .accounts + .iter() + .filter_map(|&idx| accounts.get(idx as usize).copied()) + .collect(); + + // 创建元数据 + let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); + let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; + let metadata = EventMetadata::new( + signature, + slot, + timestamp.seconds, + block_time_ms, + Default::default(), // protocol will be set by dispatcher + Default::default(), // event_type will be set by dispatcher + program_id, + outer_index, + inner_index, + recv_us, + transaction_index, + ); + + // 使用 EventDispatcher 解析 instruction 事件 + let mut event = match EventDispatcher::dispatch_instruction( + protocol.clone(), + instruction_discriminator, + instruction_data, + &account_pubkeys, + metadata.clone(), + ) { + Some(e) => e, + None => return Ok(()), + }; + + // 处理 inner instructions + let mut inner_instruction_event: Option = None; + if let Some(inner_instructions_ref) = inner_instructions { + // 并行执行两个任务: 解析 inner event 和提取 swap_data + let (inner_event_result, swap_data_result) = std::thread::scope(|s| { + let inner_event_handle = s.spawn(|| { + for inner_instruction in inner_instructions_ref.instructions.iter() { + let inner_data = &inner_instruction.instruction.data; + // 检查长度（需要 16 字节的 discriminator） + if inner_data.len() < 16 { + continue; + } + let inner_discriminator = &inner_data[..16]; + let inner_instruction_data = &inner_data[16..]; + + if let Some(inner_event) = EventDispatcher::dispatch_inner_instruction( + protocol.clone(), + inner_discriminator, + inner_instruction_data, + metadata.clone(), + ) { + return Some(inner_event); + } + } + None + }); + + let swap_data_handle = s.spawn(|| { + if event.metadata().swap_data.is_none() { + parse_swap_data_from_next_instructions( + &event, + inner_instructions_ref, + inner_index.unwrap_or(-1_i64) as i8, + accounts, + ) + } else { + None + } + }); + + // 等待两个任务完成 + (inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap()) + }); + + inner_instruction_event = inner_event_result; + if let Some(swap_data) = swap_data_result { + event.metadata_mut().set_swap_data(swap_data); + } + } + + // 特殊处理: PumpFun MIGRATE 指令需要 inner instruction data + if matches!(protocol, Protocol::PumpFun) { + const PUMPFUN_MIGRATE_IX: &[u8] = &[155, 234, 231, 146, 236, 158, 162, 30]; + if instruction_discriminator == PUMPFUN_MIGRATE_IX && inner_instruction_event.is_none() + { + return Ok(()); + } + } + + // 合并事件 + if let Some(inner_instruction_event) = inner_instruction_event { + merge(&mut event, inner_instruction_event); + } + + // 设置处理时间（使用高性能时钟） + event.metadata_mut().handle_us = elapsed_micros_since(recv_us); + event = Self::process_event(event, bot_wallet); + callback(&event); + + Ok(()) } + // ================================================================================================ + // Helper Functions + // ================================================================================================ + + /// Check if instruction should be processed based on protocol filter + /// + /// Determines whether a program_id matches any of the protocols we're interested in. + fn should_handle( + protocols: &[Protocol], + _event_type_filter: Option<&EventTypeFilter>, + program_id: &Pubkey, + ) -> bool { + // 使用 EventDispatcher 来匹配协议 + if let Some(protocol) = EventDispatcher::match_protocol_by_program_id(program_id) { + protocols.contains(&protocol) + } else { + false + } + } + + // ================================================================================================ + // Event Post-Processing + // ================================================================================================ + + /// Process and enrich parsed event with additional context + /// + /// Handles protocol-specific post-processing: + /// - PumpFun: Tracks dev addresses and marks dev trades + /// - PumpSwap: Fills swap data amounts + /// - Bonk: Tracks pool creators and marks dev trades + /// - General: Marks bot wallet trades fn process_event(event: DexEvent, bot_wallet: Option) -> DexEvent { let signature = event.metadata().signature; // Copy the signature to avoid borrowing issues match event { diff --git a/src/streaming/event_parser/core/mod.rs b/src/streaming/event_parser/core/mod.rs index 2fcba6a..5f2b3e5 100755 --- a/src/streaming/event_parser/core/mod.rs +++ b/src/streaming/event_parser/core/mod.rs @@ -1,13 +1,12 @@ pub mod account_event_parser; pub mod common_event_parser; +pub mod dispatcher; pub mod global_state; pub mod parser_cache; pub mod traits; + pub use traits::DexEvent; -pub use parser_cache::{ - AccountEventParseConfig, AccountEventParserFn, GenericEventParseConfig, - InnerInstructionEventParser, InstructionEventParser, get_account_configs, -}; +pub use dispatcher::EventDispatcher; pub mod event_parser; pub mod merger_event; \ No newline at end of file diff --git a/src/streaming/event_parser/core/parser_cache.rs b/src/streaming/event_parser/core/parser_cache.rs index d13cb8c..c1f7908 100644 --- a/src/streaming/event_parser/core/parser_cache.rs +++ b/src/streaming/event_parser/core/parser_cache.rs @@ -1,26 +1,18 @@ //! # 事件解析器配置缓存模块 //! -//! 本模块统一管理所有事件解析器的配置和缓存，包括： -//! - 指令事件解析器（Instruction Event Parser） +//! 本模块管理事件解析器的缓存，包括： +//! - 程序ID缓存 //! - 账户事件解析器（Account Event Parser） //! - 高性能缓存工具 //! //! ## 设计目标 -//! - **统一配置管理**：所有协议的解析器配置集中管理 //! - **高性能缓存**：避免重复初始化和内存分配 -//! - **易于扩展**：添加新协议只需修改配置映射 +//! - **易于扩展**：通过 dispatcher 动态派发 use crate::streaming::{ event_parser::{ common::{filter::EventTypeFilter, EventMetadata, EventType, ProtocolType}, - protocols::{ - bonk::parser::BONK_PROGRAM_ID, - pumpfun::parser::PUMPFUN_PROGRAM_ID, - pumpswap::parser::PUMPSWAP_PROGRAM_ID, - raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID, - raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID, - raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID, - }, + core::dispatcher::EventDispatcher, Protocol, DexEvent, }, grpc::AccountPretty, @@ -28,106 +20,14 @@ use crate::streaming::{ use solana_sdk::pubkey::Pubkey; use std::{ collections::HashMap, - sync::{Arc, LazyLock, OnceLock}, + sync::{Arc, LazyLock}, }; // ============================================================================ -// 第一部分：指令事件解析器配置（Instruction Event Parser） +// 第一部分：程序ID缓存 // ============================================================================ -/// 内联指令事件解析器函数类型 -/// -/// 用于解析内联指令（Inner Instruction）生成的事件 -pub type InnerInstructionEventParser = - fn(data: &[u8], metadata: EventMetadata) -> Option; - -/// 指令事件解析器函数类型 -/// -/// 用于解析普通指令（Instruction）生成的事件，需要账户信息 -pub type InstructionEventParser = - fn(data: &[u8], accounts: &[Pubkey], metadata: EventMetadata) -> Option; - -/// 指令事件解析器配置 -/// -/// 定义如何解析特定协议的指令事件 -#[derive(Debug, Clone)] -pub struct GenericEventParseConfig { - /// 程序ID（Program ID） - pub program_id: Pubkey, - /// 协议类型 - pub protocol_type: ProtocolType, - /// 内联指令判别器（8字节） - pub inner_instruction_discriminator: &'static [u8], - /// 普通指令判别器（8字节） - pub instruction_discriminator: &'static [u8], - /// 事件类型 - pub event_type: EventType, - /// 内联指令解析器 - pub inner_instruction_parser: Option, - /// 普通指令解析器 - pub instruction_parser: Option, - /// 是否需要内联指令数据 - pub requires_inner_instruction: bool, -} - -/// 全局指令事件解析器注册表 -/// -/// 存储所有协议的指令解析器配置，启动时初始化一次 -pub static EVENT_PARSERS: LazyLock> = - LazyLock::new(|| { - let mut parsers = HashMap::with_capacity(6); - - // 注册各协议的解析器配置 - parsers.insert( - Protocol::PumpSwap, - ( - PUMPSWAP_PROGRAM_ID, - crate::streaming::event_parser::protocols::pumpswap::parser::CONFIGS, - ), - ); - parsers.insert( - Protocol::PumpFun, - ( - PUMPFUN_PROGRAM_ID, - crate::streaming::event_parser::protocols::pumpfun::parser::CONFIGS, - ), - ); - parsers.insert( - Protocol::Bonk, - (BONK_PROGRAM_ID, crate::streaming::event_parser::protocols::bonk::parser::CONFIGS), - ); - parsers.insert( - Protocol::RaydiumCpmm, - ( - RAYDIUM_CPMM_PROGRAM_ID, - crate::streaming::event_parser::protocols::raydium_cpmm::parser::CONFIGS, - ), - ); - parsers.insert( - Protocol::RaydiumClmm, - ( - RAYDIUM_CLMM_PROGRAM_ID, - crate::streaming::event_parser::protocols::raydium_clmm::parser::CONFIGS, - ), - ); - parsers.insert( - Protocol::RaydiumAmmV4, - ( - RAYDIUM_AMM_V4_PROGRAM_ID, - crate::streaming::event_parser::protocols::raydium_amm_v4::parser::CONFIGS, - ), - ); - - parsers - }); - -// ---------------------------------------------------------------------------- -// 指令解析器缓存工具 -// ---------------------------------------------------------------------------- - /// 缓存键：用于标识不同的协议和过滤器组合 -/// -/// 通过对协议和事件类型排序，确保相同组合生成相同的哈希键 #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct CacheKey { /// 协议列表（已排序） @@ -138,13 +38,6 @@ pub struct CacheKey { impl CacheKey { /// 创建新的缓存键 - /// - /// # 参数 - /// - `protocols`: 协议列表 - /// - `filter`: 事件类型过滤器 - /// - /// # 优化 - /// 使用 `sort_by_cached_key` 避免重复调用 `format!` pub fn new(mut protocols: Vec, filter: Option<&EventTypeFilter>) -> Self { // 排序协议列表，确保相同协议组合生成相同的key protocols.sort_by_cached_key(|p| format!("{:?}", p)); @@ -164,23 +57,9 @@ static GLOBAL_PROGRAM_IDS_CACHE: LazyLock< parking_lot::RwLock>>>, > = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new())); -/// 全局指令配置缓存（使用读写锁保护） -static GLOBAL_INSTRUCTION_CONFIGS_CACHE: LazyLock< - parking_lot::RwLock, Vec>>>>, -> = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new())); - /// 获取指定协议的程序ID列表 /// -/// # 参数 -/// - `protocols`: 协议列表 -/// - `filter`: 事件类型过滤器（可选） -/// -/// # 返回 -/// Arc包装的程序ID向量，支持零成本共享 -/// -/// # 缓存策略 -/// - 快速路径：从缓存读取（读锁） -/// - 慢速路径：构建并缓存（写锁） +/// 使用 EventDispatcher 获取 program_ids，并缓存结果 pub fn get_global_program_ids( protocols: &[Protocol], filter: Option<&EventTypeFilter>, @@ -195,14 +74,8 @@ pub fn get_global_program_ids( } } - // 慢速路径：构建并缓存 - let mut program_ids = Vec::with_capacity(protocols.len()); - for protocol in protocols { - if let Some(parse) = EVENT_PARSERS.get(protocol) { - program_ids.push(parse.0); - } - } - let program_ids = Arc::new(program_ids); + // 慢速路径：通过 EventDispatcher 获取并缓存 + let program_ids = Arc::new(EventDispatcher::get_program_ids(protocols)); // 缓存结果（写锁） GLOBAL_PROGRAM_IDS_CACHE.write().insert(cache_key, program_ids.clone()); @@ -210,58 +83,6 @@ pub fn get_global_program_ids( program_ids } -/// 获取指定协议的指令配置 -/// -/// # 参数 -/// - `protocols`: 协议列表 -/// - `filter`: 事件类型过滤器（可选） -/// -/// # 返回 -/// Arc包装的指令配置映射表，按判别器（Discriminator）索引 -/// -/// # 缓存策略 -/// - 快速路径：从缓存读取（读锁） -/// - 慢速路径：构建并缓存（写锁） -pub fn get_global_instruction_configs( - protocols: &[Protocol], - filter: Option<&EventTypeFilter>, -) -> Arc, Vec>> { - let cache_key = CacheKey::new(protocols.to_vec(), filter); - - // 快速路径：尝试读取缓存 - { - let cache = GLOBAL_INSTRUCTION_CONFIGS_CACHE.read(); - if let Some(configs) = cache.get(&cache_key) { - return configs.clone(); - } - } - - // 慢速路径：构建并缓存 - let mut instruction_configs = HashMap::with_capacity(protocols.len() * 4); - for protocol in protocols { - if let Some(parse) = EVENT_PARSERS.get(protocol) { - parse - .1 - .iter() - .filter(|config| { - filter.as_ref().map(|f| f.include.contains(&config.event_type)).unwrap_or(true) - }) - .for_each(|config| { - instruction_configs - .entry(config.instruction_discriminator.to_vec()) - .or_insert_with(Vec::new) - .push(config.clone()); - }); - } - } - let instruction_configs = Arc::new(instruction_configs); - - // 缓存结果（写锁） - GLOBAL_INSTRUCTION_CONFIGS_CACHE.write().insert(cache_key, instruction_configs.clone()); - - instruction_configs -} - // ============================================================================ // 第二部分：账户公钥缓存工具（Account Pubkey Cache） // ============================================================================ @@ -381,221 +202,3 @@ pub struct AccountEventParseConfig { pub account_parser: AccountEventParserFn, } -/// 协议账户配置缓存 -/// -/// 存储各协议的账户解析器配置 -static PROTOCOL_CONFIGS_CACHE: OnceLock>> = - OnceLock::new(); - -/// Nonce 账户配置缓存 -static NONCE_CONFIG: OnceLock = OnceLock::new(); - -/// 通用 Token 账户配置缓存 -static COMMON_CONFIG: OnceLock = OnceLock::new(); - -/// 获取账户解析器配置列表 -/// -/// # 参数 -/// - `protocols`: 协议列表 -/// - `event_type_filter`: 事件类型过滤器（可选） -/// - `nonce_parser`: Nonce账户解析器 -/// - `token_parser`: Token账户解析器 -/// -/// # 返回 -/// 账户解析器配置向量 -/// -/// # 配置组成 -/// 1. 协议特定配置（根据protocols参数） -/// 2. Nonce账户配置（通用） -/// 3. Token账户配置（通用，作为兜底） -/// -/// # 示例 -/// ```ignore -/// let configs = get_account_configs( -/// &[Protocol::PumpFun, Protocol::Bonk], -/// None, -/// parse_nonce_account, -/// parse_token_account, -/// ); -/// ``` -pub fn get_account_configs( - protocols: &[Protocol], - event_type_filter: Option<&EventTypeFilter>, - nonce_parser: AccountEventParserFn, - token_parser: AccountEventParserFn, -) -> Vec { - // 初始化协议配置缓存（仅在首次调用时执行） - let protocols_map = PROTOCOL_CONFIGS_CACHE.get_or_init(|| { - let mut map = HashMap::new(); - - // PumpSwap 协议 - map.insert(Protocol::PumpSwap, vec![ - AccountEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - event_type: EventType::AccountPumpSwapGlobalConfig, - account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::GLOBAL_CONFIG_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser, - }, - AccountEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - event_type: EventType::AccountPumpSwapPool, - account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::POOL_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::pumpswap::types::pool_parser, - }, - ]); - - // PumpFun 协议 - map.insert(Protocol::PumpFun, vec![ - AccountEventParseConfig { - program_id: PUMPFUN_PROGRAM_ID, - protocol_type: ProtocolType::PumpFun, - event_type: EventType::AccountPumpFunBondingCurve, - account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::BONDING_CURVE_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::pumpfun::types::bonding_curve_parser, - }, - AccountEventParseConfig { - program_id: PUMPFUN_PROGRAM_ID, - protocol_type: ProtocolType::PumpFun, - event_type: EventType::AccountPumpFunGlobal, - account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::GLOBAL_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::pumpfun::types::global_parser, - }, - ]); - - // Bonk 协议 - map.insert(Protocol::Bonk, vec![ - AccountEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - event_type: EventType::AccountBonkPoolState, - account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::POOL_STATE_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::bonk::types::pool_state_parser, - }, - AccountEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - event_type: EventType::AccountBonkGlobalConfig, - account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::GLOBAL_CONFIG_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::bonk::types::global_config_parser, - }, - AccountEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - event_type: EventType::AccountBonkPlatformConfig, - account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::PLATFORM_CONFIG_ACCOUNT, - account_parser: crate::streaming::event_parser::protocols::bonk::types::platform_config_parser, - }, - ]); - - // Raydium CPMM 协议 - map.insert(Protocol::RaydiumCpmm, vec![ - AccountEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - event_type: EventType::AccountRaydiumCpmmAmmConfig, - account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::AMM_CONFIG, - account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser, - }, - AccountEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - event_type: EventType::AccountRaydiumCpmmPoolState, - account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::POOL_STATE, - account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser, - }, - ]); - - // Raydium CLMM 协议 - map.insert(Protocol::RaydiumClmm, vec![ - AccountEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - event_type: EventType::AccountRaydiumClmmAmmConfig, - account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::AMM_CONFIG, - account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser, - }, - AccountEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - event_type: EventType::AccountRaydiumClmmPoolState, - account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::POOL_STATE, - account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser, - }, - AccountEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - event_type: EventType::AccountRaydiumClmmTickArrayState, - account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::TICK_ARRAY_STATE, - account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser, - }, - ]); - - // Raydium AMM V4 协议 - map.insert(Protocol::RaydiumAmmV4, vec![ - AccountEventParseConfig { - program_id: RAYDIUM_AMM_V4_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumAmmV4, - event_type: EventType::AccountRaydiumAmmV4AmmInfo, - account_discriminator: crate::streaming::event_parser::protocols::raydium_amm_v4::discriminators::AMM_INFO, - account_parser: crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser, - }, - ]); - - map - }); - - // 构建配置列表 - let mut configs = Vec::new(); - let empty_vec = Vec::new(); - - // 预估容量（大多数协议有2-3个配置） - let estimated_capacity = protocols.len() * 3; - configs.reserve(estimated_capacity); - - // 1. 添加协议特定配置 - for protocol in protocols { - let protocol_configs = protocols_map.get(protocol).unwrap_or(&empty_vec); - - if event_type_filter.is_none() { - // 无过滤器：添加所有配置 - configs.extend(protocol_configs.iter().cloned()); - } else { - // 有过滤器：仅添加匹配的配置 - let filter = event_type_filter.unwrap(); - configs.extend( - protocol_configs - .iter() - .filter(|config| filter.include.contains(&config.event_type)) - .cloned(), - ); - } - } - - // 2. 添加 Nonce 账户配置（通用配置） - if event_type_filter.is_none() - || event_type_filter.unwrap().include.contains(&EventType::NonceAccount) - { - let nonce_config = NONCE_CONFIG.get_or_init(|| AccountEventParseConfig { - program_id: Pubkey::default(), - protocol_type: ProtocolType::Common, - event_type: EventType::NonceAccount, - account_discriminator: &[1, 0, 0, 0, 1, 0, 0, 0], - account_parser: nonce_parser, - }); - configs.push(nonce_config.clone()); - } - - // 3. 添加通用 Token 账户配置（兜底配置，始终添加） - let common_config = COMMON_CONFIG.get_or_init(|| AccountEventParseConfig { - program_id: Pubkey::default(), - protocol_type: ProtocolType::Common, - event_type: EventType::TokenAccount, - account_discriminator: &[], - account_parser: token_parser, - }); - configs.push(common_config.clone()); - - configs -} diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 07829d6..bc7bcfd 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -1,8 +1,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::event_parser::{ - common::{utils::*, EventMetadata, EventType, ProtocolType}, - core::GenericEventParseConfig, + common::{utils::*, EventMetadata, EventType}, protocols::bonk::{ bonk_pool_create_event_log_decode, bonk_trade_event_log_decode, discriminators, AmmFeeOn, BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, BonkPoolCreateEvent, BonkTradeEvent, @@ -16,104 +15,95 @@ use crate::streaming::event_parser::{ pub const BONK_PROGRAM_ID: Pubkey = solana_sdk::pubkey!("LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj"); -// Configure all event types -pub const CONFIGS: &[GenericEventParseConfig] = &[ - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::BUY_EXACT_IN, - event_type: EventType::BonkBuyExactIn, - inner_instruction_parser: Some(parse_trade_inner_instruction), - instruction_parser: Some(parse_buy_exact_in_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::BUY_EXACT_OUT, - event_type: EventType::BonkBuyExactOut, - inner_instruction_parser: Some(parse_trade_inner_instruction), - instruction_parser: Some(parse_buy_exact_out_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::SELL_EXACT_IN, - event_type: EventType::BonkSellExactIn, - inner_instruction_parser: Some(parse_trade_inner_instruction), - instruction_parser: Some(parse_sell_exact_in_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::SELL_EXACT_OUT, - event_type: EventType::BonkSellExactOut, - inner_instruction_parser: Some(parse_trade_inner_instruction), - instruction_parser: Some(parse_sell_exact_out_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT, - instruction_discriminator: discriminators::INITIALIZE, - event_type: EventType::BonkInitialize, - inner_instruction_parser: Some(parse_pool_create_inner_instruction), - instruction_parser: Some(parse_initialize_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT, - instruction_discriminator: discriminators::INITIALIZE_V2, - event_type: EventType::BonkInitializeV2, - inner_instruction_parser: Some(parse_pool_create_inner_instruction), - instruction_parser: Some(parse_initialize_v2_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT, - instruction_discriminator: discriminators::INITIALIZE_WITH_TOKEN_2022, - event_type: EventType::BonkInitializeWithToken2022, - inner_instruction_parser: Some(parse_pool_create_inner_instruction), - instruction_parser: Some(parse_initialize_with_token_2022_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::MIGRATE_TO_AMM, - event_type: EventType::BonkMigrateToAmm, - inner_instruction_parser: None, - instruction_parser: Some(parse_migrate_to_amm_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: BONK_PROGRAM_ID, - protocol_type: ProtocolType::Bonk, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::MIGRATE_TO_CP_SWAP, - event_type: EventType::BonkMigrateToCpswap, - inner_instruction_parser: None, - instruction_parser: Some(parse_migrate_to_cpswap_instruction), - requires_inner_instruction: false, - }, -]; +/// 解析 Bonk instruction data +/// +/// 根据判别器路由到具体的 instruction 解析函数 +pub fn parse_bonk_instruction_data( + discriminator: &[u8], + data: &[u8], + accounts: &[Pubkey], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::BUY_EXACT_IN => { + parse_buy_exact_in_instruction(data, accounts, metadata) + } + discriminators::BUY_EXACT_OUT => { + parse_buy_exact_out_instruction(data, accounts, metadata) + } + discriminators::SELL_EXACT_IN => { + parse_sell_exact_in_instruction(data, accounts, metadata) + } + discriminators::SELL_EXACT_OUT => { + parse_sell_exact_out_instruction(data, accounts, metadata) + } + discriminators::INITIALIZE => { + parse_initialize_instruction(data, accounts, metadata) + } + discriminators::INITIALIZE_V2 => { + parse_initialize_v2_instruction(data, accounts, metadata) + } + discriminators::INITIALIZE_WITH_TOKEN_2022 => { + parse_initialize_with_token_2022_instruction(data, accounts, metadata) + } + discriminators::MIGRATE_TO_AMM => { + parse_migrate_to_amm_instruction(data, accounts, metadata) + } + discriminators::MIGRATE_TO_CP_SWAP => { + parse_migrate_to_cpswap_instruction(data, accounts, metadata) + } + _ => None, + } +} + +/// 解析 Bonk inner instruction data +/// +/// 根据判别器路由到具体的 inner instruction 解析函数 +pub fn parse_bonk_inner_instruction_data( + discriminator: &[u8], + data: &[u8], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::TRADE_EVENT => { + parse_trade_inner_instruction(data, metadata) + } + discriminators::POOL_CREATE_EVENT => { + parse_pool_create_inner_instruction(data, metadata) + } + _ => None, + } +} + +/// 解析 Bonk 账户数据 +/// +/// 根据判别器路由到具体的账户解析函数 +pub fn parse_bonk_account_data( + discriminator: &[u8], + account: &crate::streaming::grpc::AccountPretty, + metadata: crate::streaming::event_parser::common::EventMetadata, +) -> Option { + match discriminator { + discriminators::POOL_STATE_ACCOUNT => { + crate::streaming::event_parser::protocols::bonk::types::pool_state_parser(account, metadata) + } + discriminators::GLOBAL_CONFIG_ACCOUNT => { + crate::streaming::event_parser::protocols::bonk::types::global_config_parser(account, metadata) + } + discriminators::PLATFORM_CONFIG_ACCOUNT => { + crate::streaming::event_parser::protocols::bonk::types::platform_config_parser(account, metadata) + } + _ => None, + } +} + /// Parse pool creation event fn parse_pool_create_inner_instruction( data: &[u8], metadata: EventMetadata, ) -> Option { + // Note: event_type will be set by the instruction parser, not here + // Because different initialize instructions have different event types if let Some(event) = bonk_pool_create_event_log_decode(data) { Some(DexEvent::BonkPoolCreateEvent(BonkPoolCreateEvent { metadata, ..event })) } else { @@ -146,8 +136,10 @@ fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option fn parse_buy_exact_in_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::BonkBuyExactIn; + if data.len() < 16 || accounts.len() < 18 { return None; } @@ -184,8 +176,10 @@ fn parse_buy_exact_in_instruction( fn parse_buy_exact_out_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::BonkBuyExactOut; + if data.len() < 16 || accounts.len() < 18 { return None; } @@ -222,8 +216,10 @@ fn parse_buy_exact_out_instruction( fn parse_sell_exact_in_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::BonkSellExactIn; + if data.len() < 16 || accounts.len() < 18 { return None; } @@ -260,8 +256,10 @@ fn parse_sell_exact_in_instruction( fn parse_sell_exact_out_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::BonkSellExactOut; + if data.len() < 16 || accounts.len() < 18 { return None; } @@ -299,8 +297,10 @@ fn parse_sell_exact_out_instruction( fn parse_initialize_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::BonkInitialize; + if data.len() < 24 { return None; } @@ -332,8 +332,10 @@ fn parse_initialize_instruction( fn parse_initialize_v2_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::BonkInitializeV2; + if data.len() < 24 { return None; } @@ -371,8 +373,10 @@ fn parse_initialize_v2_instruction( fn parse_initialize_with_token_2022_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::BonkInitializeWithToken2022; + if data.len() < 24 { return None; } @@ -515,8 +519,10 @@ fn parse_vesting_params(data: &[u8], offset: &mut usize) -> Option Option { + metadata.event_type = EventType::BonkMigrateToAmm; + if data.len() < 16 { return None; } @@ -570,8 +576,10 @@ fn parse_migrate_to_amm_instruction( fn parse_migrate_to_cpswap_instruction( _data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::BonkMigrateToCpswap; + Some(DexEvent::BonkMigrateToCpswapEvent(BonkMigrateToCpswapEvent { metadata, payer: accounts[0], diff --git a/src/streaming/event_parser/protocols/bonk/types.rs b/src/streaming/event_parser/protocols/bonk/types.rs index 5386852..a2ddbef 100755 --- a/src/streaming/event_parser/protocols/bonk/types.rs +++ b/src/streaming/event_parser/protocols/bonk/types.rs @@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::{ event_parser::{ - common::EventMetadata, + common::{EventMetadata, EventType}, protocols::bonk::{ BonkGlobalConfigAccountEvent, BonkPlatformConfigAccountEvent, BonkPoolStateAccountEvent, }, @@ -132,7 +132,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..POOL_STATE_SIZE]).ok() } -pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::AccountBonkPoolState; + if account.data.len() < POOL_STATE_SIZE + 8 { return None; } @@ -182,8 +184,10 @@ pub fn global_config_decode(data: &[u8]) -> Option { pub fn global_config_parser( account: &AccountPretty, - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::AccountBonkGlobalConfig; + if account.data.len() < GLOBAL_CONFIG_SIZE + 8 { return None; } @@ -228,8 +232,10 @@ pub fn platform_config_decode(data: &[u8]) -> Option { pub fn platform_config_parser( account: &AccountPretty, - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::AccountBonkPlatformConfig; + if account.data.len() < PLATFORM_CONFIG_SIZE + 8 { return None; } diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index d92decd..d1bad33 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -1,6 +1,5 @@ use crate::streaming::event_parser::{ - common::{EventMetadata, EventType, ProtocolType}, - core::GenericEventParseConfig, + common::{EventMetadata, EventType}, protocols::pumpfun::{ discriminators, pumpfun_create_token_event_log_decode, pumpfun_migrate_event_log_decode, pumpfun_trade_event_log_decode, PumpFunCreateTokenEvent, PumpFunMigrateEvent, @@ -14,53 +13,78 @@ use solana_sdk::pubkey::Pubkey; pub const PUMPFUN_PROGRAM_ID: Pubkey = solana_sdk::pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"); -// 配置所有事件类型 -pub const CONFIGS: &[GenericEventParseConfig] = &[ - GenericEventParseConfig { - program_id: PUMPFUN_PROGRAM_ID, - protocol_type: ProtocolType::PumpFun, - inner_instruction_discriminator: discriminators::CREATE_TOKEN_EVENT, - instruction_discriminator: discriminators::CREATE_TOKEN_IX, - event_type: EventType::PumpFunCreateToken, - inner_instruction_parser: Some(parse_create_token_inner_instruction), - instruction_parser: Some(parse_create_token_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: PUMPFUN_PROGRAM_ID, - protocol_type: ProtocolType::PumpFun, - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::BUY_IX, - event_type: EventType::PumpFunBuy, - inner_instruction_parser: Some(parse_trade_inner_instruction), - instruction_parser: Some(parse_buy_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: PUMPFUN_PROGRAM_ID, - protocol_type: ProtocolType::PumpFun, - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::SELL_IX, - event_type: EventType::PumpFunSell, - inner_instruction_parser: Some(parse_trade_inner_instruction), - instruction_parser: Some(parse_sell_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: PUMPFUN_PROGRAM_ID, - protocol_type: ProtocolType::PumpFun, - inner_instruction_discriminator: discriminators::COMPLETE_PUMP_AMM_MIGRATION_EVENT, - instruction_discriminator: discriminators::MIGRATE_IX, - event_type: EventType::PumpFunMigrate, - inner_instruction_parser: Some(parse_migrate_inner_instruction), - instruction_parser: Some(parse_migrate_instruction), - // Failed migrations lack inner instruction data (typically "Bonding curve already migrated") - requires_inner_instruction: true, - }, -]; +/// 解析 PumpFun instruction data +/// +/// 根据判别器路由到具体的 instruction 解析函数 +pub fn parse_pumpfun_instruction_data( + discriminator: &[u8], + data: &[u8], + accounts: &[Pubkey], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::CREATE_TOKEN_IX => { + parse_create_token_instruction(data, accounts, metadata) + } + discriminators::BUY_IX => { + parse_buy_instruction(data, accounts, metadata) + } + discriminators::SELL_IX => { + parse_sell_instruction(data, accounts, metadata) + } + discriminators::MIGRATE_IX => { + parse_migrate_instruction(data, accounts, metadata) + } + _ => None, + } +} + +/// 解析 PumpFun inner instruction data +/// +/// 根据判别器路由到具体的 inner instruction 解析函数 +pub fn parse_pumpfun_inner_instruction_data( + discriminator: &[u8], + data: &[u8], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::CREATE_TOKEN_EVENT => { + parse_create_token_inner_instruction(data, metadata) + } + discriminators::TRADE_EVENT => { + parse_trade_inner_instruction(data, metadata) + } + discriminators::COMPLETE_PUMP_AMM_MIGRATION_EVENT => { + parse_migrate_inner_instruction(data, metadata) + } + _ => None, + } +} + + +/// 解析 PumpFun 账户数据 +/// +/// 根据判别器路由到具体的账户解析函数 +pub fn parse_pumpfun_account_data( + discriminator: &[u8], + account: &crate::streaming::grpc::AccountPretty, + metadata: crate::streaming::event_parser::common::EventMetadata, +) -> Option { + match discriminator { + discriminators::BONDING_CURVE_ACCOUNT => { + crate::streaming::event_parser::protocols::pumpfun::types::bonding_curve_parser(account, metadata) + } + discriminators::GLOBAL_ACCOUNT => { + crate::streaming::event_parser::protocols::pumpfun::types::global_parser(account, metadata) + } + _ => None, + } +} + /// 解析迁移事件 -fn parse_migrate_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { +fn parse_migrate_inner_instruction(data: &[u8], mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::PumpFunMigrate; if let Some(event) = pumpfun_migrate_event_log_decode(data) { Some(DexEvent::PumpFunMigrateEvent(PumpFunMigrateEvent { metadata, ..event })) } else { @@ -71,8 +95,9 @@ fn parse_migrate_inner_instruction(data: &[u8], metadata: EventMetadata) -> Opti /// 解析创建代币日志事件 fn parse_create_token_inner_instruction( data: &[u8], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpFunCreateToken; if let Some(event) = pumpfun_create_token_event_log_decode(data) { Some(DexEvent::PumpFunCreateTokenEvent(PumpFunCreateTokenEvent { metadata, ..event })) } else { @@ -80,8 +105,10 @@ fn parse_create_token_inner_instruction( } } -/// 解析交易事件 +/// 解析交易事件 (inner instruction 不设置 event_type，因为不知道是 Buy 还是 Sell) fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { + // 注意：inner instruction 的 trade event 不设置 event_type + // 因为它会被合并到 instruction event 中，而 instruction event 已经设置了正确的 event_type if let Some(event) = pumpfun_trade_event_log_decode(data) { Some(DexEvent::PumpFunTradeEvent(PumpFunTradeEvent { metadata, ..event })) } else { @@ -93,8 +120,10 @@ fn parse_trade_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option fn parse_create_token_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpFunCreateToken; + if data.len() < 16 || accounts.len() < 11 { return None; } @@ -154,8 +183,10 @@ fn parse_create_token_instruction( fn parse_buy_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpFunBuy; + if data.len() < 16 || accounts.len() < 13 { return None; } @@ -188,8 +219,10 @@ fn parse_buy_instruction( fn parse_sell_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpFunSell; + if data.len() < 16 || accounts.len() < 11 { return None; } @@ -222,8 +255,10 @@ fn parse_sell_instruction( fn parse_migrate_instruction( _data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpFunMigrate; + if accounts.len() < 24 { return None; } diff --git a/src/streaming/event_parser/protocols/pumpfun/types.rs b/src/streaming/event_parser/protocols/pumpfun/types.rs index 15a0e23..0805fb4 100644 --- a/src/streaming/event_parser/protocols/pumpfun/types.rs +++ b/src/streaming/event_parser/protocols/pumpfun/types.rs @@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::{ event_parser::{ - common::EventMetadata, + common::{EventMetadata, EventType}, protocols::pumpfun::{PumpFunBondingCurveAccountEvent, PumpFunGlobalAccountEvent}, DexEvent, }, @@ -33,8 +33,10 @@ pub fn bonding_curve_decode(data: &[u8]) -> Option { pub fn bonding_curve_parser( account: &AccountPretty, - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::AccountPumpFunBondingCurve; + if account.data.len() < BONDING_CURVE_SIZE + 8 { return None; } @@ -81,7 +83,9 @@ pub fn global_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..GLOBAL_SIZE]).ok() } -pub fn global_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn global_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::AccountPumpFunGlobal; + if account.data.len() < GLOBAL_SIZE + 8 { return None; } diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index 7b9d2e8..215b929 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -1,6 +1,5 @@ use crate::streaming::event_parser::{ - common::{read_u64_le, EventMetadata, EventType, ProtocolType}, - core::GenericEventParseConfig, + common::{read_u64_le, EventMetadata, EventType}, protocols::pumpswap::{ discriminators, pump_swap_buy_event_log_decode, pump_swap_create_pool_event_log_decode, pump_swap_deposit_event_log_decode, pump_swap_sell_event_log_decode, @@ -15,62 +14,70 @@ use solana_sdk::pubkey::Pubkey; pub const PUMPSWAP_PROGRAM_ID: Pubkey = solana_sdk::pubkey!("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA"); -// 配置所有事件类型 -pub const CONFIGS: &[GenericEventParseConfig] = &[ - GenericEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - inner_instruction_discriminator: discriminators::BUY_EVENT, - instruction_discriminator: discriminators::BUY_IX, - event_type: EventType::PumpSwapBuy, - inner_instruction_parser: Some(parse_buy_inner_instruction), - instruction_parser: Some(parse_buy_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - inner_instruction_discriminator: discriminators::SELL_EVENT, - instruction_discriminator: discriminators::SELL_IX, - event_type: EventType::PumpSwapSell, - inner_instruction_parser: Some(parse_sell_inner_instruction), - instruction_parser: Some(parse_sell_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - inner_instruction_discriminator: discriminators::CREATE_POOL_EVENT, - instruction_discriminator: discriminators::CREATE_POOL_IX, - event_type: EventType::PumpSwapCreatePool, - inner_instruction_parser: Some(parse_create_pool_inner_instruction), - instruction_parser: Some(parse_create_pool_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - inner_instruction_discriminator: discriminators::DEPOSIT_EVENT, - instruction_discriminator: discriminators::DEPOSIT_IX, - event_type: EventType::PumpSwapDeposit, - inner_instruction_parser: Some(parse_deposit_inner_instruction), - instruction_parser: Some(parse_deposit_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: PUMPSWAP_PROGRAM_ID, - protocol_type: ProtocolType::PumpSwap, - inner_instruction_discriminator: discriminators::WITHDRAW_EVENT, - instruction_discriminator: discriminators::WITHDRAW_IX, - event_type: EventType::PumpSwapWithdraw, - inner_instruction_parser: Some(parse_withdraw_inner_instruction), - instruction_parser: Some(parse_withdraw_instruction), - requires_inner_instruction: false, - }, -]; +/// 解析 PumpSwap instruction data +/// +/// 根据判别器路由到具体的 instruction 解析函数 +pub fn parse_pumpswap_instruction_data( + discriminator: &[u8], + data: &[u8], + accounts: &[Pubkey], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::BUY_IX => parse_buy_instruction(data, accounts, metadata), + discriminators::SELL_IX => parse_sell_instruction(data, accounts, metadata), + discriminators::CREATE_POOL_IX => { + parse_create_pool_instruction(data, accounts, metadata) + } + discriminators::DEPOSIT_IX => parse_deposit_instruction(data, accounts, metadata), + discriminators::WITHDRAW_IX => parse_withdraw_instruction(data, accounts, metadata), + _ => None, + } +} + +/// 解析 PumpSwap inner instruction data +/// +/// 根据判别器路由到具体的 inner instruction 解析函数 +pub fn parse_pumpswap_inner_instruction_data( + discriminator: &[u8], + data: &[u8], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::BUY_EVENT => parse_buy_inner_instruction(data, metadata), + discriminators::SELL_EVENT => parse_sell_inner_instruction(data, metadata), + discriminators::CREATE_POOL_EVENT => { + parse_create_pool_inner_instruction(data, metadata) + } + discriminators::DEPOSIT_EVENT => parse_deposit_inner_instruction(data, metadata), + discriminators::WITHDRAW_EVENT => parse_withdraw_inner_instruction(data, metadata), + _ => None, + } +} + + +/// 解析 PumpSwap 账户数据 +/// +/// 根据判别器路由到具体的账户解析函数 +pub fn parse_pumpswap_account_data( + discriminator: &[u8], + account: &crate::streaming::grpc::AccountPretty, + metadata: crate::streaming::event_parser::common::EventMetadata, +) -> Option { + match discriminator { + discriminators::GLOBAL_CONFIG_ACCOUNT => { + crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser(account, metadata) + } + discriminators::POOL_ACCOUNT => { + crate::streaming::event_parser::protocols::pumpswap::types::pool_parser(account, metadata) + } + _ => None, + } +} /// 解析买入日志事件 fn parse_buy_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { + // Note: event_type will be set by instruction parser if let Some(event) = pump_swap_buy_event_log_decode(data) { Some(DexEvent::PumpSwapBuyEvent(PumpSwapBuyEvent { metadata, ..event })) } else { @@ -80,6 +87,7 @@ fn parse_buy_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option Option { + // Note: event_type will be set by instruction parser if let Some(event) = pump_swap_sell_event_log_decode(data) { Some(DexEvent::PumpSwapSellEvent(PumpSwapSellEvent { metadata, ..event })) } else { @@ -92,6 +100,7 @@ fn parse_create_pool_inner_instruction( data: &[u8], metadata: EventMetadata, ) -> Option { + // Note: event_type will be set by instruction parser if let Some(event) = pump_swap_create_pool_event_log_decode(data) { Some(DexEvent::PumpSwapCreatePoolEvent(PumpSwapCreatePoolEvent { metadata, ..event })) } else { @@ -101,6 +110,7 @@ fn parse_create_pool_inner_instruction( /// 解析存款日志事件 fn parse_deposit_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { + // Note: event_type will be set by instruction parser if let Some(event) = pump_swap_deposit_event_log_decode(data) { Some(DexEvent::PumpSwapDepositEvent(PumpSwapDepositEvent { metadata, ..event })) } else { @@ -110,6 +120,7 @@ fn parse_deposit_inner_instruction(data: &[u8], metadata: EventMetadata) -> Opti /// 解析提款日志事件 fn parse_withdraw_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { + // Note: event_type will be set by instruction parser if let Some(event) = pump_swap_withdraw_event_log_decode(data) { Some(DexEvent::PumpSwapWithdrawEvent(PumpSwapWithdrawEvent { metadata, ..event })) } else { @@ -121,8 +132,10 @@ fn parse_withdraw_inner_instruction(data: &[u8], metadata: EventMetadata) -> Opt fn parse_buy_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpSwapBuy; + if data.len() < 16 || accounts.len() < 11 { return None; } @@ -156,8 +169,10 @@ fn parse_buy_instruction( fn parse_sell_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpSwapSell; + if data.len() < 16 || accounts.len() < 11 { return None; } @@ -191,8 +206,10 @@ fn parse_sell_instruction( fn parse_create_pool_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpSwapCreatePool; + if data.len() < 18 || accounts.len() < 11 { return None; } @@ -230,8 +247,10 @@ fn parse_create_pool_instruction( fn parse_deposit_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpSwapDeposit; + if data.len() < 24 || accounts.len() < 11 { return None; } @@ -262,8 +281,10 @@ fn parse_deposit_instruction( fn parse_withdraw_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::PumpSwapWithdraw; + if data.len() < 24 || accounts.len() < 11 { return None; } diff --git a/src/streaming/event_parser/protocols/pumpswap/types.rs b/src/streaming/event_parser/protocols/pumpswap/types.rs index e66d007..afa99bc 100644 --- a/src/streaming/event_parser/protocols/pumpswap/types.rs +++ b/src/streaming/event_parser/protocols/pumpswap/types.rs @@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::{ event_parser::{ - common::EventMetadata, + common::{EventMetadata, EventType}, protocols::pumpswap::{PumpSwapGlobalConfigAccountEvent, PumpSwapPoolAccountEvent}, DexEvent, }, @@ -33,8 +33,10 @@ pub fn global_config_decode(data: &[u8]) -> Option { pub fn global_config_parser( account: &AccountPretty, - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::AccountPumpSwapGlobalConfig; + if account.data.len() < GLOBAL_CONFIG_SIZE + 8 { return None; } @@ -76,7 +78,9 @@ pub fn pool_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..POOL_SIZE]).ok() } -pub fn pool_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn pool_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::AccountPumpSwapPool; + if account.data.len() < POOL_SIZE + 8 { return None; } diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs index d19fb2c..76831a4 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs @@ -1,6 +1,5 @@ use crate::streaming::event_parser::{ - common::{read_u64_le, EventMetadata, EventType, ProtocolType}, - core::GenericEventParseConfig, + common::{read_u64_le, EventMetadata, EventType}, protocols::raydium_amm_v4::{ discriminators, RaydiumAmmV4DepositEvent, RaydiumAmmV4Initialize2Event, RaydiumAmmV4SwapEvent, RaydiumAmmV4WithdrawEvent, RaydiumAmmV4WithdrawPnlEvent, @@ -9,80 +8,73 @@ use crate::streaming::event_parser::{ }; use solana_sdk::pubkey::Pubkey; -/// Raydium CPMM程序ID +/// Raydium AMM V4程序ID pub const RAYDIUM_AMM_V4_PROGRAM_ID: Pubkey = solana_sdk::pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"); -// 配置所有事件类型 -pub const CONFIGS: &[GenericEventParseConfig] = &[ - GenericEventParseConfig { - program_id: RAYDIUM_AMM_V4_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumAmmV4, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::SWAP_BASE_IN, - event_type: EventType::RaydiumAmmV4SwapBaseIn, - inner_instruction_parser: None, - instruction_parser: Some(parse_swap_base_input_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_AMM_V4_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumAmmV4, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::SWAP_BASE_OUT, - event_type: EventType::RaydiumAmmV4SwapBaseOut, - inner_instruction_parser: None, - instruction_parser: Some(parse_swap_base_output_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_AMM_V4_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumAmmV4, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::DEPOSIT, - event_type: EventType::RaydiumAmmV4Deposit, - inner_instruction_parser: None, - instruction_parser: Some(parse_deposit_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_AMM_V4_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumAmmV4, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::INITIALIZE2, - event_type: EventType::RaydiumAmmV4Initialize2, - inner_instruction_parser: None, - instruction_parser: Some(parse_initialize2_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_AMM_V4_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumAmmV4, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::WITHDRAW, - event_type: EventType::RaydiumAmmV4Withdraw, - inner_instruction_parser: None, - instruction_parser: Some(parse_withdraw_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_AMM_V4_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumAmmV4, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::WITHDRAW_PNL, - event_type: EventType::RaydiumAmmV4WithdrawPnl, - inner_instruction_parser: None, - instruction_parser: Some(parse_withdraw_pnl_instruction), - requires_inner_instruction: false, - }, -]; +/// 解析 Raydium AMM V4 instruction data +/// +/// 根据判别器路由到具体的 instruction 解析函数 +pub fn parse_raydium_amm_v4_instruction_data( + discriminator: &[u8], + data: &[u8], + accounts: &[Pubkey], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::SWAP_BASE_IN => { + parse_swap_base_input_instruction(data, accounts, metadata) + } + discriminators::SWAP_BASE_OUT => { + parse_swap_base_output_instruction(data, accounts, metadata) + } + discriminators::DEPOSIT => parse_deposit_instruction(data, accounts, metadata), + discriminators::INITIALIZE2 => parse_initialize2_instruction(data, accounts, metadata), + discriminators::WITHDRAW => parse_withdraw_instruction(data, accounts, metadata), + discriminators::WITHDRAW_PNL => { + parse_withdraw_pnl_instruction(data, accounts, metadata) + } + _ => None, + } +} + +/// 解析 Raydium AMM V4 inner instruction data +/// +/// Raydium AMM V4 没有 inner instruction 事件 +pub fn parse_raydium_amm_v4_inner_instruction_data( + _discriminator: &[u8], + _data: &[u8], + _metadata: EventMetadata, +) -> Option { + None +} + + +/// 解析 Raydium AMM V4 账户数据 +/// +/// 根据判别器路由到具体的账户解析函数 +pub fn parse_raydium_amm_v4_account_data( + discriminator: &[u8], + account: &crate::streaming::grpc::AccountPretty, + metadata: crate::streaming::event_parser::common::EventMetadata, +) -> Option { + match discriminator { + discriminators::AMM_INFO => { + crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser(account, metadata) + } + _ => None, + } +} + /// 解析提现指令事件 fn parse_withdraw_pnl_instruction( _data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumAmmV4WithdrawPnl; + if accounts.len() < 17 { return None; } @@ -113,8 +105,10 @@ fn parse_withdraw_pnl_instruction( fn parse_withdraw_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumAmmV4Withdraw; + if data.len() < 8 || accounts.len() < 22 { return None; } @@ -153,8 +147,10 @@ fn parse_withdraw_instruction( fn parse_initialize2_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumAmmV4Initialize2; + if data.len() < 25 || accounts.len() < 21 { return None; } @@ -198,8 +194,10 @@ fn parse_initialize2_instruction( fn parse_deposit_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumAmmV4Deposit; + if data.len() < 24 || accounts.len() < 14 { return None; } @@ -234,8 +232,10 @@ fn parse_deposit_instruction( fn parse_swap_base_output_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumAmmV4SwapBaseOut; + if data.len() < 16 || accounts.len() < 17 { return None; } @@ -281,8 +281,10 @@ fn parse_swap_base_output_instruction( fn parse_swap_base_input_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumAmmV4SwapBaseIn; + if data.len() < 16 || accounts.len() < 17 { return None; } diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs index ba8d7ce..073dac6 100644 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs @@ -4,7 +4,8 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::{ event_parser::{ - common::EventMetadata, protocols::raydium_amm_v4::RaydiumAmmV4AmmInfoAccountEvent, + common::{EventMetadata, EventType}, + protocols::raydium_amm_v4::RaydiumAmmV4AmmInfoAccountEvent, DexEvent, }, grpc::AccountPretty, @@ -86,7 +87,9 @@ pub fn amm_info_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..AMM_INFO_SIZE]).ok() } -pub fn amm_info_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn amm_info_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::AccountRaydiumAmmV4AmmInfo; + if account.data.len() < AMM_INFO_SIZE { return None; } diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index 8d7e9cf..b127644 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -1,9 +1,8 @@ use crate::streaming::event_parser::{ common::{ read_i32_le, read_option_bool, read_u128_le, read_u64_le, read_u8_le, EventMetadata, - EventType, ProtocolType, + EventType, }, - core::GenericEventParseConfig, protocols::raydium_clmm::{ discriminators, RaydiumClmmClosePositionEvent, RaydiumClmmCreatePoolEvent, RaydiumClmmDecreaseLiquidityV2Event, RaydiumClmmIncreaseLiquidityV2Event, @@ -18,96 +17,80 @@ use solana_sdk::pubkey::Pubkey; pub const RAYDIUM_CLMM_PROGRAM_ID: Pubkey = solana_sdk::pubkey!("CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK"); -// 配置所有事件类型 -pub const CONFIGS: &[GenericEventParseConfig] = &[ - GenericEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::SWAP, - event_type: EventType::RaydiumClmmSwap, - inner_instruction_parser: None, - instruction_parser: Some(parse_swap_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::SWAP_V2, - event_type: EventType::RaydiumClmmSwapV2, - inner_instruction_parser: None, - instruction_parser: Some(parse_swap_v2_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::CLOSE_POSITION, - event_type: EventType::RaydiumClmmClosePosition, - inner_instruction_parser: None, - instruction_parser: Some(parse_close_position_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::DECREASE_LIQUIDITY_V2, - event_type: EventType::RaydiumClmmDecreaseLiquidityV2, - inner_instruction_parser: None, - instruction_parser: Some(parse_decrease_liquidity_v2_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::CREATE_POOL, - event_type: EventType::RaydiumClmmCreatePool, - inner_instruction_parser: None, - instruction_parser: Some(parse_create_pool_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::INCREASE_LIQUIDITY_V2, - event_type: EventType::RaydiumClmmIncreaseLiquidityV2, - inner_instruction_parser: None, - instruction_parser: Some(parse_increase_liquidity_v2_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::OPEN_POSITION_WITH_TOKEN_22_NFT, - event_type: EventType::RaydiumClmmOpenPositionWithToken22Nft, - inner_instruction_parser: None, - instruction_parser: Some(parse_open_position_with_token_22_nft_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CLMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumClmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::OPEN_POSITION_V2, - event_type: EventType::RaydiumClmmOpenPositionV2, - inner_instruction_parser: None, - instruction_parser: Some(parse_open_position_v2_instruction), - requires_inner_instruction: false, - }, -]; +/// 解析 Raydium CLMM instruction data +/// +/// 根据判别器路由到具体的 instruction 解析函数 +pub fn parse_raydium_clmm_instruction_data( + discriminator: &[u8], + data: &[u8], + accounts: &[Pubkey], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::SWAP => parse_swap_instruction(data, accounts, metadata), + discriminators::SWAP_V2 => parse_swap_v2_instruction(data, accounts, metadata), + discriminators::CLOSE_POSITION => { + parse_close_position_instruction(data, accounts, metadata) + } + discriminators::DECREASE_LIQUIDITY_V2 => { + parse_decrease_liquidity_v2_instruction(data, accounts, metadata) + } + discriminators::CREATE_POOL => parse_create_pool_instruction(data, accounts, metadata), + discriminators::INCREASE_LIQUIDITY_V2 => { + parse_increase_liquidity_v2_instruction(data, accounts, metadata) + } + discriminators::OPEN_POSITION_WITH_TOKEN_22_NFT => { + parse_open_position_with_token_22_nft_instruction(data, accounts, metadata) + } + discriminators::OPEN_POSITION_V2 => { + parse_open_position_v2_instruction(data, accounts, metadata) + } + _ => None, + } +} + +/// 解析 Raydium CLMM inner instruction data +/// +/// Raydium CLMM 没有 inner instruction 事件 +pub fn parse_raydium_clmm_inner_instruction_data( + _discriminator: &[u8], + _data: &[u8], + _metadata: EventMetadata, +) -> Option { + None +} + + +/// 解析 Raydium CLMM 账户数据 +/// +/// 根据判别器路由到具体的账户解析函数 +pub fn parse_raydium_clmm_account_data( + discriminator: &[u8], + account: &crate::streaming::grpc::AccountPretty, + metadata: crate::streaming::event_parser::common::EventMetadata, +) -> Option { + match discriminator { + discriminators::AMM_CONFIG => { + crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser(account, metadata) + } + discriminators::POOL_STATE => { + crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser(account, metadata) + } + discriminators::TICK_ARRAY_STATE => { + crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser(account, metadata) + } + _ => None, + } +} /// 解析打开仓位V2指令事件 fn parse_open_position_v2_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumClmmOpenPositionV2; + if data.len() < 51 || accounts.len() < 22 { return None; } @@ -152,8 +135,10 @@ fn parse_open_position_v2_instruction( fn parse_open_position_with_token_22_nft_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumClmmOpenPositionWithToken22Nft; + if data.len() < 51 || accounts.len() < 20 { return None; } @@ -197,8 +182,10 @@ fn parse_open_position_with_token_22_nft_instruction( fn parse_increase_liquidity_v2_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumClmmIncreaseLiquidityV2; + if data.len() < 34 || accounts.len() < 15 { return None; } @@ -230,8 +217,10 @@ fn parse_increase_liquidity_v2_instruction( fn parse_create_pool_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumClmmCreatePool; + if data.len() < 24 || accounts.len() < 13 { return None; } @@ -259,8 +248,10 @@ fn parse_create_pool_instruction( fn parse_decrease_liquidity_v2_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumClmmDecreaseLiquidityV2; + if data.len() < 32 || accounts.len() < 16 { return None; } @@ -293,8 +284,10 @@ fn parse_decrease_liquidity_v2_instruction( fn parse_close_position_instruction( _data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumClmmClosePosition; + if accounts.len() < 6 { return None; } @@ -313,8 +306,10 @@ fn parse_close_position_instruction( fn parse_swap_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumClmmSwap; + if data.len() < 33 || accounts.len() < 10 { return None; } @@ -347,8 +342,10 @@ fn parse_swap_instruction( fn parse_swap_v2_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumClmmSwapV2; + if data.len() < 33 || accounts.len() < 13 { return None; } diff --git a/src/streaming/event_parser/protocols/raydium_clmm/types.rs b/src/streaming/event_parser/protocols/raydium_clmm/types.rs index a3731b5..5f3695c 100644 --- a/src/streaming/event_parser/protocols/raydium_clmm/types.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/types.rs @@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::{ event_parser::{ - common::EventMetadata, + common::{EventMetadata, EventType}, protocols::raydium_clmm::{ RaydiumClmmAmmConfigAccountEvent, RaydiumClmmPoolStateAccountEvent, RaydiumClmmTickArrayStateAccountEvent, @@ -37,7 +37,9 @@ pub fn amm_config_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..AMM_CONFIG_SIZE]).ok() } -pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn amm_config_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::AccountRaydiumClmmAmmConfig; + if account.data.len() < AMM_CONFIG_SIZE + 8 { return None; } @@ -122,7 +124,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..POOL_STATE_SIZE]).ok() } -pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::AccountRaydiumClmmPoolState; + if account.data.len() < POOL_STATE_SIZE + 8 { return None; } @@ -202,8 +206,10 @@ pub fn tick_array_state_decode(data: &[u8]) -> Option { pub fn tick_array_state_parser( account: &AccountPretty, - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::AccountRaydiumClmmTickArrayState; + if account.data.len() < TICK_ARRAY_STATE_SIZE + 8 { return None; } diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index 2e59eb1..3e0407a 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -1,8 +1,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::event_parser::{ - common::{read_u64_le, EventMetadata, EventType, ProtocolType}, - core::GenericEventParseConfig, + common::{read_u64_le, EventMetadata, EventType}, protocols::raydium_cpmm::{ discriminators, RaydiumCpmmDepositEvent, RaydiumCpmmInitializeEvent, RaydiumCpmmSwapEvent, RaydiumCpmmWithdrawEvent, @@ -14,66 +13,69 @@ use crate::streaming::event_parser::{ pub const RAYDIUM_CPMM_PROGRAM_ID: Pubkey = solana_sdk::pubkey!("CPMMoo8L3F4NbTegBCKVNunggL7H1ZpdTHKxQB5qKP1C"); -// 配置所有事件类型 -pub const CONFIGS: &[GenericEventParseConfig] = &[ - GenericEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::SWAP_BASE_IN, - event_type: EventType::RaydiumCpmmSwapBaseInput, - inner_instruction_parser: None, - instruction_parser: Some(parse_swap_base_input_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::SWAP_BASE_OUT, - event_type: EventType::RaydiumCpmmSwapBaseOutput, - inner_instruction_parser: None, - instruction_parser: Some(parse_swap_base_output_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::DEPOSIT, - event_type: EventType::RaydiumCpmmDeposit, - inner_instruction_parser: None, - instruction_parser: Some(parse_deposit_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::INITIALIZE, - event_type: EventType::RaydiumCpmmInitialize, - inner_instruction_parser: None, - instruction_parser: Some(parse_initialize_instruction), - requires_inner_instruction: false, - }, - GenericEventParseConfig { - program_id: RAYDIUM_CPMM_PROGRAM_ID, - protocol_type: ProtocolType::RaydiumCpmm, - inner_instruction_discriminator: &[], - instruction_discriminator: discriminators::WITHDRAW, - event_type: EventType::RaydiumCpmmWithdraw, - inner_instruction_parser: None, - instruction_parser: Some(parse_withdraw_instruction), - requires_inner_instruction: false, - }, -]; +/// 解析 Raydium CPMM instruction data +/// +/// 根据判别器路由到具体的 instruction 解析函数 +pub fn parse_raydium_cpmm_instruction_data( + discriminator: &[u8], + data: &[u8], + accounts: &[Pubkey], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::SWAP_BASE_IN => { + parse_swap_base_input_instruction(data, accounts, metadata) + } + discriminators::SWAP_BASE_OUT => { + parse_swap_base_output_instruction(data, accounts, metadata) + } + discriminators::DEPOSIT => parse_deposit_instruction(data, accounts, metadata), + discriminators::INITIALIZE => parse_initialize_instruction(data, accounts, metadata), + discriminators::WITHDRAW => parse_withdraw_instruction(data, accounts, metadata), + _ => None, + } +} + +/// 解析 Raydium CPMM inner instruction data +/// +/// Raydium CPMM 没有 inner instruction 事件 +pub fn parse_raydium_cpmm_inner_instruction_data( + _discriminator: &[u8], + _data: &[u8], + _metadata: EventMetadata, +) -> Option { + None +} + + +/// 解析 Raydium CPMM 账户数据 +/// +/// 根据判别器路由到具体的账户解析函数 +pub fn parse_raydium_cpmm_account_data( + discriminator: &[u8], + account: &crate::streaming::grpc::AccountPretty, + metadata: crate::streaming::event_parser::common::EventMetadata, +) -> Option { + match discriminator { + discriminators::AMM_CONFIG => { + crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser(account, metadata) + } + discriminators::POOL_STATE => { + crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser(account, metadata) + } + _ => None, + } +} + /// 解析提款指令事件 fn parse_withdraw_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumCpmmWithdraw; + if data.len() < 24 || accounts.len() < 14 { return None; } @@ -103,8 +105,10 @@ fn parse_withdraw_instruction( fn parse_initialize_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumCpmmInitialize; + if data.len() < 24 || accounts.len() < 20 { return None; } @@ -140,8 +144,10 @@ fn parse_initialize_instruction( fn parse_deposit_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumCpmmDeposit; + if data.len() < 24 || accounts.len() < 13 { return None; } @@ -170,8 +176,10 @@ fn parse_deposit_instruction( fn parse_swap_base_input_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumCpmmSwapBaseInput; + if data.len() < 16 || accounts.len() < 13 { return None; } @@ -203,8 +211,10 @@ fn parse_swap_base_input_instruction( fn parse_swap_base_output_instruction( data: &[u8], accounts: &[Pubkey], - metadata: EventMetadata, + mut metadata: EventMetadata, ) -> Option { + metadata.event_type = EventType::RaydiumCpmmSwapBaseOutput; + if data.len() < 16 || accounts.len() < 13 { return None; } diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs index 21601a2..5b8e9fe 100644 --- a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs @@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey; use crate::streaming::{ event_parser::{ - common::EventMetadata, + common::{EventMetadata, EventType}, protocols::raydium_cpmm::{ RaydiumCpmmAmmConfigAccountEvent, RaydiumCpmmPoolStateAccountEvent, }, @@ -36,7 +36,9 @@ pub fn amm_config_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..AMM_CONFIG_SIZE]).ok() } -pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn amm_config_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::AccountRaydiumCpmmAmmConfig; + if account.data.len() < AMM_CONFIG_SIZE + 8 { return None; } @@ -91,7 +93,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option { borsh::from_slice::(&data[..POOL_STATE_SIZE]).ok() } -pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option { +pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { + metadata.event_type = EventType::AccountRaydiumCpmmPoolState; + if account.data.len() < POOL_STATE_SIZE + 8 { return None; }