diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index 677e281..2384481 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(); diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs index 22e3d20..0129b4a 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,39 +61,84 @@ 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) { + // 构建临时元数据（具体的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, // 会被具体parser覆盖 + 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 } diff --git a/src/streaming/event_parser/core/dispatcher.rs b/src/streaming/event_parser/core/dispatcher.rs new file mode 100644 index 0000000..412fac6 --- /dev/null +++ b/src/streaming/event_parser/core/dispatcher.rs @@ -0,0 +1,367 @@ +//! 中心事件解析调度器 +//! +//! 根据协议类型路由到对应的解析函数，替代原有的静态 CONFIGS 数组架构 +//! +//! ## 设计原则 +//! - **单一职责**: 每个函数只负责一件事（路由、解析、合并分离） +//! - **灵活性**: 调用方可以选择是否合并，或自定义合并逻辑 +//! - **可测试性**: 每个函数都可以独立测试 + +use crate::streaming::event_parser::{ + common::EventMetadata, + core::merger_event::merge, + 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, + ), + } + } + + /// 解析并合并 instruction 和 inner instruction 事件（便捷方法） + /// + /// # 参数 + /// - `protocol`: 协议类型 + /// - `instruction_discriminator`: 指令判别器 (8 bytes) + /// - `inner_instruction_discriminator`: 内联指令判别器 (16 bytes, 可选) + /// - `instruction_data`: 指令数据 + /// - `inner_instruction_data`: 内联指令数据 (可选) + /// - `accounts`: 账户公钥列表 + /// - `metadata`: 事件元数据 + /// + /// # 返回 + /// 解析成功返回 `Some(DexEvent)`，否则返回 `None` + /// + /// # 逻辑说明 + /// 1. 解析 instruction 事件 + /// 2. 解析 inner instruction 事件（如果有） + /// 3. 在外层合并两个事件 + /// 4. 应用特殊处理逻辑（如 PumpFun MIGRATE） + #[inline] + pub fn dispatch( + protocol: Protocol, + instruction_discriminator: &[u8], + inner_instruction_discriminator: Option<&[u8]>, + instruction_data: &[u8], + inner_instruction_data: Option<&[u8]>, + accounts: &[Pubkey], + metadata: EventMetadata, + ) -> Option { + // 1. 解析 instruction 事件 + let mut instruction_event = Self::dispatch_instruction( + protocol.clone(), + instruction_discriminator, + instruction_data, + accounts, + metadata.clone(), + ); + + // 2. 解析 inner instruction 事件（如果有） + if let Some(inner_disc) = inner_instruction_discriminator { + if let Some(inner_data) = inner_instruction_data { + let inner_event = Self::dispatch_inner_instruction( + protocol.clone(), + inner_disc, + inner_data, + metadata, + ); + + // 3. 在外层合并 + if let Some(inner_event) = inner_event { + if let Some(ref mut inst_event) = instruction_event { + merge(inst_event, inner_event); + } else { + // 如果没有 instruction 事件，直接使用 inner 事件 + instruction_event = Some(inner_event); + } + } + } + } + + // 4. 特殊处理: PumpFun MIGRATE 指令需要 inner instruction data + // 如果没有 inner instruction data，应该返回 None（失败的迁移） + 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_data.is_none() { + return None; + } + } + + instruction_event + } + + /// 遍历多个协议尝试解析 (用于不确定协议类型的场景) + /// + /// 先通过 program_id 快速定位协议类型，再检查是否在请求的协议列表中 + pub fn dispatch_multi( + protocols: &[Protocol], + program_id: &Pubkey, + instruction_discriminator: &[u8], + inner_instruction_discriminator: Option<&[u8]>, + instruction_data: &[u8], + inner_instruction_data: Option<&[u8]>, + accounts: &[Pubkey], + metadata: EventMetadata, + ) -> Option { + // 先通过 program_id 快速过滤协议 + let target_protocol = Self::match_protocol_by_program_id(program_id)?; + + // 如果目标协议在请求列表中，则解析 + if protocols.contains(&target_protocol) { + Self::dispatch( + target_protocol, + instruction_discriminator, + inner_instruction_discriminator, + instruction_data, + inner_instruction_data, + accounts, + metadata, + ) + } else { + None + } + } + + /// 通过 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..e033a2b 100644 --- a/src/streaming/event_parser/core/event_parser.rs +++ b/src/streaming/event_parser/core/event_parser.rs @@ -8,14 +8,14 @@ use crate::streaming::{ 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, parser_cache::{ - build_account_pubkeys_with_cache, get_global_instruction_configs, - get_global_program_ids, GenericEventParseConfig, + build_account_pubkeys_with_cache, + get_global_program_ids, }, }, DexEvent, Protocol, @@ -35,6 +35,32 @@ use yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo; pub struct EventParser {} impl EventParser { + // 辅助函数：确保 accounts 数组有足够的容量 + fn ensure_accounts_capacity(accounts: &mut Vec, instruction_accounts: &[u8]) { + if let Some(&max_idx) = instruction_accounts.iter().max() { + let required_len = max_idx as usize + 1; + if required_len > accounts.len() { + accounts.resize(required_len, Pubkey::default()); + } + } + } + + // 辅助函数：转换 timestamp 为毫秒 + fn timestamp_to_millis(block_time: Option) -> (Timestamp, i64) { + let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); + let millis = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; + (timestamp, millis) + } + + // 辅助函数：创建回调适配器 + fn create_callback_adapter( + callback: Arc, + ) -> Arc Fn(&'a DexEvent) + Send + Sync> { + Arc::new(move |event: &DexEvent| { + callback(event.clone()); + }) + } + #[allow(clippy::too_many_arguments)] async fn parse_instruction_events_from_grpc_transaction( protocols: &[Protocol], @@ -64,11 +90,8 @@ impl EventParser { let inner_instructions = inner_instructions .iter() .find(|inner_instruction| inner_instruction.index == index as u32); - 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::ensure_accounts_capacity(&mut accounts, &instruction.accounts); if Self::should_handle(protocols, event_type_filter, &program_id) { Self::parse_events_from_grpc_instruction( protocols, @@ -155,12 +178,9 @@ impl EventParser { let inner_instructions = inner_instructions .iter() .find(|inner_instruction| inner_instruction.index == index as u8); + // 补齐accounts(使用Pubkey::default()) + Self::ensure_accounts_capacity(&mut accounts, &instruction.accounts); 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, @@ -220,10 +240,7 @@ impl EventParser { inner_instructions: &[InnerInstructions], callback: Arc, ) -> anyhow::Result<()> { - // 创建适配器回调，将所有权回调转换为引用回调 - let adapter_callback = Arc::new(move |event: &DexEvent| { - callback(event.clone()); - }); + let adapter_callback = Self::create_callback_adapter(callback); let accounts = versioned_tx.message.static_account_keys(); Self::parse_instruction_events_from_versioned_transaction( protocols, @@ -254,11 +271,7 @@ impl EventParser { transaction_index: Option, callback: Arc, ) -> anyhow::Result<()> { - // 创建适配器回调，将所有权回调转换为引用回调 - let adapter_callback = Arc::new(move |event: &DexEvent| { - callback(event.clone()); - }); - // 调用原始方法 + let adapter_callback = Self::create_callback_adapter(callback); Self::parse_grpc_transaction( protocols, event_type_filter, @@ -452,184 +465,6 @@ impl EventParser { .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( @@ -648,129 +483,95 @@ impl EventParser { 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(); + // 验证 program_id + let program_id = match Self::validate_program_id( + instruction.program_id_index as u32, + accounts, + protocols, + event_type_filter, + ) { + Some(id) => id, + None => return Ok(()), + }; - // Use SIMD-optimized account indices validation (只需检查一次) - if !SimdUtils::validate_account_indices_simd(&instruction.accounts, accounts.len()) { - return Ok(()); - } + // 验证并构建账户公钥列表 + let account_pubkeys = match Self::validate_and_build_account_pubkeys(&instruction.accounts, accounts) { + Some(pubkeys) => pubkeys, + None => return Ok(()), + }; - // 使用线程局部缓存构建账户公钥列表，避免重复分配 (只需构建一次) - let account_pubkeys = build_account_pubkeys_with_cache(&instruction.accounts, accounts); + // 提取判别器 + let (instruction_discriminator, instruction_data) = + match Self::extract_instruction_discriminator(&instruction.data) { + Some(result) => result, + None => return Ok(()), + }; - // 并行处理所有 (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(); + // 准备 inner instruction 数据（如果有） + let (inner_discriminator, inner_data) = if let Some(inner_instructions_ref) = inner_instructions { + // 尝试从第一个 inner instruction 获取数据 + if let Some(first_inner) = inner_instructions_ref.instructions.first() { + let inner_inst_data = &first_inner.instruction.data; + if inner_inst_data.len() >= 24 { // 16 bytes discriminator + data + (Some(&inner_inst_data[0..16]), Some(&inner_inst_data[16..])) + } else { + (None, None) + } + } else { + (None, None) + } + } else { + (None, None) + }; - for (_disc, config, mut event) in all_results { - // 阻塞处理：原有的同步逻辑 - let mut inner_instruction_event: Option = None; + // 创建元数据（使用默认值，dispatcher会返回带正确元数据的事件） + let (timestamp, block_time_ms) = Self::timestamp_to_millis(block_time); + let metadata = EventMetadata::new( + signature, + slot, + timestamp.seconds, + block_time_ms, + crate::streaming::event_parser::common::ProtocolType::Common, + crate::streaming::event_parser::common::EventType::default(), + program_id, + outer_index, + inner_index, + recv_us, + transaction_index, + ); + + // 使用 dispatcher 解析事件 + if let Some(mut event) = EventDispatcher::dispatch_multi( + protocols, + &program_id, + instruction_discriminator, + inner_discriminator, + instruction_data, + inner_data, + &account_pubkeys, + metadata, + ) { + // 解析 swap_data（如果需要） 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); + if event.metadata().swap_data.is_none() { + if let Some(swap_data) = parse_swap_data_from_next_instructions( + &event, + inner_instructions_ref, + inner_index.unwrap_or(-1_i64) as i8, + accounts, + ) { + 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); + // 完成事件处理 + Self::finalize_event(event, recv_us, bot_wallet, &callback); } Ok(()) } - /// 从指令中解析事件 - /// TODO: - wait refactor + /// 从指令中解析事件 (gRPC版本) #[allow(clippy::too_many_arguments)] fn parse_events_from_grpc_instruction( protocols: &[Protocol], @@ -788,123 +589,90 @@ impl EventParser { inner_instructions: Option<&yellowstone_grpc_proto::prelude::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(); + // 验证 program_id + let program_id = match Self::validate_program_id( + instruction.program_id_index, + accounts, + protocols, + event_type_filter, + ) { + Some(id) => id, + None => return Ok(()), + }; - // Use SIMD-optimized account indices validation (只需检查一次) - if !SimdUtils::validate_account_indices_simd(&instruction.accounts, accounts.len()) { - return Ok(()); - } + // 验证并构建账户公钥列表 + let account_pubkeys = match Self::validate_and_build_account_pubkeys(&instruction.accounts, accounts) { + Some(pubkeys) => pubkeys, + None => return Ok(()), + }; - // 使用线程局部缓存构建账户公钥列表，避免重复分配 (只需构建一次) - let account_pubkeys = build_account_pubkeys_with_cache(&instruction.accounts, accounts); + // 提取判别器 + let (instruction_discriminator, instruction_data) = + match Self::extract_instruction_discriminator(&instruction.data) { + Some(result) => result, + None => return Ok(()), + }; - // 并行处理所有 (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(); + // 准备 inner instruction 数据（如果有） + let (inner_discriminator, inner_data) = if let Some(inner_instructions_ref) = inner_instructions { + // 尝试从第一个 inner instruction 获取数据 + if let Some(first_inner) = inner_instructions_ref.instructions.first() { + let inner_inst_data = &first_inner.data; + if inner_inst_data.len() >= 24 { // 16 bytes discriminator + data + (Some(&inner_inst_data[0..16]), Some(&inner_inst_data[16..])) + } else { + (None, None) + } + } else { + (None, None) + } + } else { + (None, None) + }; - for (_disc, config, mut event) in all_results { - // 阻塞处理：原有的同步逻辑 - let mut inner_instruction_event: Option = None; + // 创建元数据（使用默认值，dispatcher会返回带正确元数据的事件） + let (timestamp, block_time_ms) = Self::timestamp_to_millis(block_time); + let metadata = EventMetadata::new( + signature, + slot, + timestamp.seconds, + block_time_ms, + crate::streaming::event_parser::common::ProtocolType::Common, + crate::streaming::event_parser::common::EventType::default(), + program_id, + outer_index, + inner_index, + recv_us, + transaction_index, + ); + + // 使用 dispatcher 解析事件 + if let Some(mut event) = EventDispatcher::dispatch_multi( + protocols, + &program_id, + instruction_discriminator, + inner_discriminator, + instruction_data, + inner_data, + &account_pubkeys, + metadata, + ) { + // 解析 swap_data（如果需要） 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 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 - } - }); - - // 等待两个任务完成 - (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); + if event.metadata().swap_data.is_none() { + if let Some(swap_data) = parse_swap_data_from_next_grpc_instructions( + &event, + inner_instructions_ref, + inner_index.unwrap_or(-1_i64) as i8, + accounts, + ) { + 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); + // 完成事件处理 + Self::finalize_event(event, recv_us, bot_wallet, &callback); } Ok(()) } @@ -917,6 +685,66 @@ impl EventParser { get_global_program_ids(protocols, event_type_filter).contains(program_id) } + // 辅助函数：设置 swap_data 的 from_amount 和 to_amount + fn set_swap_amounts( + swap_data: &mut crate::streaming::event_parser::common::SwapData, + from_amount: u64, + to_amount: u64, + ) { + swap_data.from_amount = from_amount; + swap_data.to_amount = to_amount; + } + + // 辅助函数：验证 program_id 是否在 accounts 范围内并且应该被处理 + fn validate_program_id( + program_id_index: u32, + accounts: &[Pubkey], + protocols: &[Protocol], + event_type_filter: Option<&EventTypeFilter>, + ) -> Option { + let index = program_id_index as usize; + if index >= accounts.len() { + return None; + } + let program_id = accounts[index]; + if Self::should_handle(protocols, event_type_filter, &program_id) { + Some(program_id) + } else { + None + } + } + + // 辅助函数：验证并构建账户公钥列表 + fn validate_and_build_account_pubkeys( + instruction_accounts: &[u8], + accounts: &[Pubkey], + ) -> Option> { + if !SimdUtils::validate_account_indices_simd(instruction_accounts, accounts.len()) { + return None; + } + Some(build_account_pubkeys_with_cache(instruction_accounts, accounts)) + } + + // 辅助函数：提取指令数据的判别器 + fn extract_instruction_discriminator(instruction_data: &[u8]) -> Option<(&[u8], &[u8])> { + if instruction_data.len() < 8 { + return None; + } + Some((&instruction_data[0..8], &instruction_data[8..])) + } + + // 辅助函数：处理事件的最后步骤（设置时间、处理事件、调用回调） + fn finalize_event( + mut event: DexEvent, + recv_us: i64, + bot_wallet: Option, + callback: &Arc Fn(&'a DexEvent) + Send + Sync>, + ) { + event.metadata_mut().handle_us = elapsed_micros_since(recv_us); + event = Self::process_event(event, bot_wallet); + callback(&event); + } + fn process_event(event: DexEvent, bot_wallet: Option) -> DexEvent { let signature = event.metadata().signature; // Copy the signature to avoid borrowing issues match event { @@ -935,30 +763,32 @@ impl EventParser { trade_info.is_bot = Some(trade_info.user) == bot_wallet; if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() { - swap_data.from_amount = if trade_info.is_buy { - trade_info.sol_amount + let (from, to) = if trade_info.is_buy { + (trade_info.sol_amount, trade_info.token_amount) } else { - trade_info.token_amount - }; - swap_data.to_amount = if trade_info.is_buy { - trade_info.token_amount - } else { - trade_info.sol_amount + (trade_info.token_amount, trade_info.sol_amount) }; + Self::set_swap_amounts(swap_data, from, to); } DexEvent::PumpFunTradeEvent(trade_info) } DexEvent::PumpSwapBuyEvent(mut trade_info) => { if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() { - swap_data.from_amount = trade_info.user_quote_amount_in; - swap_data.to_amount = trade_info.base_amount_out; + Self::set_swap_amounts( + swap_data, + trade_info.user_quote_amount_in, + trade_info.base_amount_out, + ); } DexEvent::PumpSwapBuyEvent(trade_info) } DexEvent::PumpSwapSellEvent(mut trade_info) => { if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() { - swap_data.from_amount = trade_info.base_amount_in; - swap_data.to_amount = trade_info.user_quote_amount_out; + Self::set_swap_amounts( + swap_data, + trade_info.base_amount_in, + trade_info.user_quote_amount_out, + ); } DexEvent::PumpSwapSellEvent(trade_info) } 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/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/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/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_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_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; }