refactor: replace static CONFIGS with dynamic event dispatcher

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();

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<DexEvent> {
-        // 直接从 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
     }
 

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<DexEvent> {
+        // 根据协议类型设置 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<DexEvent> {
+        // 根据协议类型设置 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<DexEvent> {
+        // 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<DexEvent> {
+        // 先通过 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<Protocol> {
+        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<Pubkey> {
+        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<DexEvent> {
+        // 根据协议类型设置 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,
+            ),
+        }
+    }
+}

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<Pubkey>, 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>) -> (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<dyn Fn(DexEvent) + Send + Sync>,
+    ) -> Arc<dyn for<'a> 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<dyn Fn(DexEvent) + Send + Sync>,
     ) -> 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<u64>,
         callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
     ) -> 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<Timestamp>,
-        recv_us: i64,
-        outer_index: i64,
-        inner_index: Option<i64>,
-        transaction_index: Option<u64>,
-    ) -> 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<Timestamp>,
-        recv_us: i64,
-        outer_index: i64,
-        inner_index: Option<i64>,
-        transaction_index: Option<u64>,
-    ) -> Option<DexEvent> {
-        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<Timestamp>,
-        recv_us: i64,
-        outer_index: i64,
-        inner_index: Option<i64>,
-        transaction_index: Option<u64>,
-    ) -> Option<DexEvent> {
-        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<Timestamp>,
-        recv_us: i64,
-        outer_index: i64,
-        inner_index: Option<i64>,
-        transaction_index: Option<u64>,
-        config: &GenericEventParseConfig,
-    ) -> Vec<DexEvent> {
-        // 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<Timestamp>,
-        recv_us: i64,
-        outer_index: i64,
-        inner_index: Option<i64>,
-        transaction_index: Option<u64>,
-        config: &GenericEventParseConfig,
-    ) -> Vec<DexEvent> {
-        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<Timestamp>,
-        recv_us: i64,
-        outer_index: i64,
-        inner_index: Option<i64>,
-        transaction_index: Option<u64>,
-        config: &GenericEventParseConfig,
-    ) -> Vec<DexEvent> {
-        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<dyn for<'a> 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<DexEvent> = 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<dyn for<'a> 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<DexEvent> = 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<Pubkey> {
+        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<Vec<Pubkey>> {
+        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<Pubkey>,
+        callback: &Arc<dyn for<'a> 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<Pubkey>) -> 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)
             }

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;

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<DexEvent>;
-
-/// 指令事件解析器函数类型
-///
-/// 用于解析普通指令（Instruction）生成的事件，需要账户信息
-pub type InstructionEventParser =
-    fn(data: &[u8], accounts: &[Pubkey], metadata: EventMetadata) -> Option<DexEvent>;
-
-/// 指令事件解析器配置
-///
-/// 定义如何解析特定协议的指令事件
-#[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<InnerInstructionEventParser>,
-    /// 普通指令解析器
-    pub instruction_parser: Option<InstructionEventParser>,
-    /// 是否需要内联指令数据
-    pub requires_inner_instruction: bool,
-}
-
-/// 全局指令事件解析器注册表
-///
-/// 存储所有协议的指令解析器配置，启动时初始化一次
-pub static EVENT_PARSERS: LazyLock<HashMap<Protocol, (Pubkey, &[GenericEventParseConfig])>> =
-    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<Protocol>, 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<HashMap<CacheKey, Arc<Vec<Pubkey>>>>,
 > = LazyLock::new(|| parking_lot::RwLock::new(HashMap::new()));
 
-/// 全局指令配置缓存（使用读写锁保护）
-static GLOBAL_INSTRUCTION_CONFIGS_CACHE: LazyLock<
-    parking_lot::RwLock<HashMap<CacheKey, Arc<HashMap<Vec<u8>, Vec<GenericEventParseConfig>>>>>,
-> = 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<HashMap<Vec<u8>, Vec<GenericEventParseConfig>>> {
-    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<HashMap<Protocol, Vec<AccountEventParseConfig>>> =
-    OnceLock::new();
-
-/// Nonce 账户配置缓存
-static NONCE_CONFIG: OnceLock<AccountEventParseConfig> = OnceLock::new();
-
-/// 通用 Token 账户配置缓存
-static COMMON_CONFIG: OnceLock<AccountEventParseConfig> = 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<AccountEventParseConfig> {
-    // 初始化协议配置缓存（仅在首次调用时执行）
-    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
-}

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<DexEvent> {
+    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<DexEvent> {
+    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<crate::streaming::event_parser::DexEvent> {
+    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<DexEvent> {
+    // 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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<VestingParams
 fn parse_migrate_to_amm_instruction(
     data: &[u8],
     accounts: &[Pubkey],
-    metadata: EventMetadata,
+    mut metadata: EventMetadata,
 ) -> Option<DexEvent> {
+    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<DexEvent> {
+    metadata.event_type = EventType::BonkMigrateToCpswap;
+
     Some(DexEvent::BonkMigrateToCpswapEvent(BonkMigrateToCpswapEvent {
         metadata,
         payer: accounts[0],

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<DexEvent> {
+    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<DexEvent> {
+    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<crate::streaming::event_parser::DexEvent> {
+    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<DexEvent> {
+fn parse_migrate_inner_instruction(data: &[u8], mut metadata: EventMetadata) -> Option<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    // 注意：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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    metadata.event_type = EventType::PumpFunMigrate;
+
     if accounts.len() < 24 {
         return None;
     }

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<DexEvent> {
+    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<DexEvent> {
+    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<crate::streaming::event_parser::DexEvent> {
+    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<DexEvent> {
+    // 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<D
 
 /// 解析卖出日志事件
 fn parse_sell_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
+    // 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<DexEvent> {
+    // 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<DexEvent> {
+    // 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<DexEvent> {
+    // 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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    metadata.event_type = EventType::PumpSwapWithdraw;
+
     if data.len() < 24 || accounts.len() < 11 {
         return None;
     }

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<DexEvent> {
+    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<DexEvent> {
+    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<crate::streaming::event_parser::DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    metadata.event_type = EventType::RaydiumAmmV4SwapBaseIn;
+
     if data.len() < 16 || accounts.len() < 17 {
         return None;
     }

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<DexEvent> {
+    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<DexEvent> {
+    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<crate::streaming::event_parser::DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    metadata.event_type = EventType::RaydiumClmmSwapV2;
+
     if data.len() < 33 || accounts.len() < 13 {
         return None;
     }

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<DexEvent> {
+    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<DexEvent> {
+    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<crate::streaming::event_parser::DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    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<DexEvent> {
+    metadata.event_type = EventType::RaydiumCpmmSwapBaseOutput;
+
     if data.len() < 16 || accounts.len() < 13 {
         return None;
     }