refactor: unify event parsing API with callback ownership transfer

examples/parse_tx_events.rs

@@ -36,8 +36,9 @@ async fn main() -> Result<()> {
 
 /// Get details of a single transaction
 async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
-    use solana_sdk::signature::Signature;
-    use solana_transaction_status::UiTransactionEncoding;
+    use solana_sdk::{signature::Signature, pubkey::Pubkey, message::compiled_instruction::CompiledInstruction};
+    use solana_transaction_status::{UiTransactionEncoding, InnerInstruction, InnerInstructions, UiInstruction};
+    use prost_types::Timestamp;
 
     let signature = Signature::from_str(signature_str)?;
 
@@ -88,6 +89,96 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
                     }
                 }
             }
+
+            // Parse the transaction and extract necessary data
+            let versioned_tx = match transaction.transaction.transaction.decode() {
+                Some(tx) => tx,
+                None => {
+                    println!("Failed to decode transaction");
+                    return Ok(());
+                }
+            };
+
+            // Convert inner instructions from meta
+            let mut inner_instructions_vec: Vec<InnerInstructions> = Vec::new();
+            if let Some(meta) = &transaction.transaction.meta {
+                if let solana_transaction_status::option_serializer::OptionSerializer::Some(
+                    ui_inner_insts,
+                ) = &meta.inner_instructions
+                {
+                    for ui_inner in ui_inner_insts {
+                        let mut converted_instructions = Vec::new();
+
+                        for ui_instruction in &ui_inner.instructions {
+                            if let UiInstruction::Compiled(ui_compiled) = ui_instruction {
+                                if let Ok(data) = solana_sdk::bs58::decode(&ui_compiled.data).into_vec() {
+                                    let compiled_instruction = CompiledInstruction {
+                                        program_id_index: ui_compiled.program_id_index,
+                                        accounts: ui_compiled.accounts.to_vec(),
+                                        data,
+                                    };
+
+                                    let inner_instruction = InnerInstruction {
+                                        instruction: compiled_instruction,
+                                        stack_height: ui_compiled.stack_height,
+                                    };
+
+                                    converted_instructions.push(inner_instruction);
+                                }
+                            }
+                        }
+
+                        let inner_instructions = InnerInstructions {
+                            index: ui_inner.index,
+                            instructions: converted_instructions,
+                        };
+
+                        inner_instructions_vec.push(inner_instructions);
+                    }
+                }
+            }
+
+            // Extract address table lookups
+            let meta = transaction.transaction.meta;
+            let mut address_table_lookups: Vec<Pubkey> = vec![];
+            if let Some(meta) = meta {
+                if let solana_transaction_status::option_serializer::OptionSerializer::Some(
+                    loaded_addresses,
+                ) = &meta.loaded_addresses
+                {
+                    address_table_lookups
+                        .reserve(loaded_addresses.writable.len() + loaded_addresses.readonly.len());
+                    address_table_lookups.extend(
+                        loaded_addresses
+                            .writable
+                            .iter()
+                            .filter_map(|s| s.parse::<Pubkey>().ok())
+                            .chain(
+                                loaded_addresses
+                                    .readonly
+                                    .iter()
+                                    .filter_map(|s| s.parse::<Pubkey>().ok()),
+                            ),
+                    );
+                }
+            }
+
+            // Build complete accounts list
+            let mut accounts = Vec::with_capacity(
+                versioned_tx.message.static_account_keys().len() + address_table_lookups.len(),
+            );
+            accounts.extend_from_slice(versioned_tx.message.static_account_keys());
+            accounts.extend(address_table_lookups);
+
+            let slot = transaction.slot;
+            let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 });
+            let recv_us = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_micros() as i64;
+            let bot_wallet = None;
+            let transaction_index = None;
+
             let protocols = vec![
                 Protocol::Bonk,
                 Protocol::RaydiumClmm,
@@ -96,14 +187,26 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
                 Protocol::RaydiumCpmm,
                 Protocol::RaydiumAmmV4,
             ];
-            EventParser::parse_encoded_confirmed_transaction_with_status_meta(
+
+            // Create callback
+            let callback = Arc::new(move |event: DexEvent| {
+                println!("{:?}\n", event);
+            });
+
+            // Call parse_instruction_events_from_versioned_transaction
+            EventParser::parse_instruction_events_from_versioned_transaction(
                 &protocols,
                 None,
+                &versioned_tx,
                 signature,
-                transaction,
-                Arc::new(move |event: &DexEvent| {
-                    println!("{:?}\n", event);
-                }),
+                Some(slot),
+                block_time,
+                recv_us,
+                &accounts,
+                &inner_instructions_vec,
+                bot_wallet,
+                transaction_index,
+                callback,
             )
             .await?;
         }

src/streaming/common/event_processor.rs

@@ -64,7 +64,7 @@ pub async fn process_grpc_transaction(
 
             let adapter_callback = create_metrics_callback(callback.clone());
 
-            EventParser::parse_grpc_transaction_owned(
+            EventParser::parse_grpc_transaction(
                 protocols,
                 event_type_filter,
                 grpc_tx,
@@ -123,18 +123,20 @@ pub async fn process_shred_transaction(
     let recv_us = transaction_with_slot.recv_us;
 
     let adapter_callback = create_metrics_callback(callback);
+    let accounts = tx.message.static_account_keys();
 
-    EventParser::parse_versioned_transaction_owned(
+    EventParser::parse_instruction_events_from_versioned_transaction(
         protocols,
         event_type_filter,
-        tx,
+        &tx,
         signature,
         Some(slot),
         None,
         recv_us,
+        accounts,
+        &[],
         bot_wallet,
         None,
-        &[],
         adapter_callback,
     )
     .await?;

src/streaming/event_parser/core/account_event_parser.rs

@@ -71,12 +71,12 @@ impl AccountEventParser {
             if let Some(protocol) = EventDispatcher::match_protocol_by_program_id(&account.owner) {
                 // 检查是否在请求的协议列表中
                 if protocols.contains(&protocol) {
-                    // 构建临时元数据（具体的event_type会在parser中设置）
+                    // 构建临时元数据（protocol会被dispatcher设置，event_type会在parser中设置）
                     let metadata = EventMetadata {
                         slot: account.slot,
                         signature: account.signature,
-                        protocol: ProtocolType::Common, // 会被具体parser覆盖
-                        event_type: EventType::default(), // 会被具体parser覆盖
+                        protocol: ProtocolType::Common, // 会被 EventDispatcher::dispatch_account 设置
+                        event_type: EventType::default(), // 会被具体 parser 设置
                         program_id: account.owner,
                         recv_us: account.recv_us,
                         handle_us: elapsed_micros_since(account.recv_us),
@@ -144,8 +144,10 @@ impl AccountEventParser {
 
     pub fn parse_token_account_event(
         account: &AccountPretty,
-        metadata: EventMetadata,
+        mut metadata: EventMetadata,
     ) -> Option<DexEvent> {
+        metadata.event_type = EventType::TokenAccount;
+
         let pubkey = account.pubkey;
         let executable = account.executable;
         let lamports = account.lamports;
@@ -212,8 +214,10 @@ impl AccountEventParser {
 
     pub fn parse_nonce_account_event(
         account: &AccountPretty,
-        metadata: EventMetadata,
+        mut metadata: EventMetadata,
     ) -> Option<DexEvent> {
+        metadata.event_type = EventType::NonceAccount;
+
         if let Ok(info) = parse_nonce(&account.data) {
             match info {
                 solana_account_decoder::parse_nonce::UiNonceState::Initialized(details) => {

src/streaming/event_parser/protocols/bonk/types.rs

@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
 
 use crate::streaming::{
     event_parser::{
-        common::EventMetadata,
+        common::{EventMetadata, EventType},
         protocols::bonk::{
             BonkGlobalConfigAccountEvent, BonkPlatformConfigAccountEvent, BonkPoolStateAccountEvent,
         },
@@ -132,7 +132,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option<PoolState> {
     borsh::from_slice::<PoolState>(&data[..POOL_STATE_SIZE]).ok()
 }
 
-pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
+pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountBonkPoolState;
+
     if account.data.len() < POOL_STATE_SIZE + 8 {
         return None;
     }
@@ -182,8 +184,10 @@ pub fn global_config_decode(data: &[u8]) -> Option<GlobalConfig> {
 
 pub fn global_config_parser(
     account: &AccountPretty,
-    metadata: EventMetadata,
+    mut metadata: EventMetadata,
 ) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountBonkGlobalConfig;
+
     if account.data.len() < GLOBAL_CONFIG_SIZE + 8 {
         return None;
     }
@@ -228,8 +232,10 @@ pub fn platform_config_decode(data: &[u8]) -> Option<PlatformConfig> {
 
 pub fn platform_config_parser(
     account: &AccountPretty,
-    metadata: EventMetadata,
+    mut metadata: EventMetadata,
 ) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountBonkPlatformConfig;
+
     if account.data.len() < PLATFORM_CONFIG_SIZE + 8 {
         return None;
     }

src/streaming/event_parser/protocols/pumpfun/types.rs

@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
 
 use crate::streaming::{
     event_parser::{
-        common::EventMetadata,
+        common::{EventMetadata, EventType},
         protocols::pumpfun::{PumpFunBondingCurveAccountEvent, PumpFunGlobalAccountEvent},
         DexEvent,
     },
@@ -33,8 +33,10 @@ pub fn bonding_curve_decode(data: &[u8]) -> Option<BondingCurve> {
 
 pub fn bonding_curve_parser(
     account: &AccountPretty,
-    metadata: EventMetadata,
+    mut metadata: EventMetadata,
 ) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountPumpFunBondingCurve;
+
     if account.data.len() < BONDING_CURVE_SIZE + 8 {
         return None;
     }
@@ -81,7 +83,9 @@ pub fn global_decode(data: &[u8]) -> Option<Global> {
     borsh::from_slice::<Global>(&data[..GLOBAL_SIZE]).ok()
 }
 
-pub fn global_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
+pub fn global_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountPumpFunGlobal;
+
     if account.data.len() < GLOBAL_SIZE + 8 {
         return None;
     }

src/streaming/event_parser/protocols/pumpswap/types.rs

@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
 
 use crate::streaming::{
     event_parser::{
-        common::EventMetadata,
+        common::{EventMetadata, EventType},
         protocols::pumpswap::{PumpSwapGlobalConfigAccountEvent, PumpSwapPoolAccountEvent},
         DexEvent,
     },
@@ -33,8 +33,10 @@ pub fn global_config_decode(data: &[u8]) -> Option<GlobalConfig> {
 
 pub fn global_config_parser(
     account: &AccountPretty,
-    metadata: EventMetadata,
+    mut metadata: EventMetadata,
 ) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountPumpSwapGlobalConfig;
+
     if account.data.len() < GLOBAL_CONFIG_SIZE + 8 {
         return None;
     }
@@ -76,7 +78,9 @@ pub fn pool_decode(data: &[u8]) -> Option<Pool> {
     borsh::from_slice::<Pool>(&data[..POOL_SIZE]).ok()
 }
 
-pub fn pool_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
+pub fn pool_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountPumpSwapPool;
+
     if account.data.len() < POOL_SIZE + 8 {
         return None;
     }

src/streaming/event_parser/protocols/raydium_amm_v4/types.rs

@@ -4,7 +4,8 @@ use solana_sdk::pubkey::Pubkey;
 
 use crate::streaming::{
     event_parser::{
-        common::EventMetadata, protocols::raydium_amm_v4::RaydiumAmmV4AmmInfoAccountEvent,
+        common::{EventMetadata, EventType},
+        protocols::raydium_amm_v4::RaydiumAmmV4AmmInfoAccountEvent,
         DexEvent,
     },
     grpc::AccountPretty,
@@ -86,7 +87,9 @@ pub fn amm_info_decode(data: &[u8]) -> Option<AmmInfo> {
     borsh::from_slice::<AmmInfo>(&data[..AMM_INFO_SIZE]).ok()
 }
 
-pub fn amm_info_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
+pub fn amm_info_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountRaydiumAmmV4AmmInfo;
+
     if account.data.len() < AMM_INFO_SIZE {
         return None;
     }

src/streaming/event_parser/protocols/raydium_clmm/types.rs

@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
 
 use crate::streaming::{
     event_parser::{
-        common::EventMetadata,
+        common::{EventMetadata, EventType},
         protocols::raydium_clmm::{
             RaydiumClmmAmmConfigAccountEvent, RaydiumClmmPoolStateAccountEvent,
             RaydiumClmmTickArrayStateAccountEvent,
@@ -37,7 +37,9 @@ pub fn amm_config_decode(data: &[u8]) -> Option<AmmConfig> {
     borsh::from_slice::<AmmConfig>(&data[..AMM_CONFIG_SIZE]).ok()
 }
 
-pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
+pub fn amm_config_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountRaydiumClmmAmmConfig;
+
     if account.data.len() < AMM_CONFIG_SIZE + 8 {
         return None;
     }
@@ -122,7 +124,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option<PoolState> {
     borsh::from_slice::<PoolState>(&data[..POOL_STATE_SIZE]).ok()
 }
 
-pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
+pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountRaydiumClmmPoolState;
+
     if account.data.len() < POOL_STATE_SIZE + 8 {
         return None;
     }
@@ -202,8 +206,10 @@ pub fn tick_array_state_decode(data: &[u8]) -> Option<TickArrayState> {
 
 pub fn tick_array_state_parser(
     account: &AccountPretty,
-    metadata: EventMetadata,
+    mut metadata: EventMetadata,
 ) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountRaydiumClmmTickArrayState;
+
     if account.data.len() < TICK_ARRAY_STATE_SIZE + 8 {
         return None;
     }

src/streaming/event_parser/protocols/raydium_cpmm/types.rs

@@ -4,7 +4,7 @@ use solana_sdk::pubkey::Pubkey;
 
 use crate::streaming::{
     event_parser::{
-        common::EventMetadata,
+        common::{EventMetadata, EventType},
         protocols::raydium_cpmm::{
             RaydiumCpmmAmmConfigAccountEvent, RaydiumCpmmPoolStateAccountEvent,
         },
@@ -36,7 +36,9 @@ pub fn amm_config_decode(data: &[u8]) -> Option<AmmConfig> {
     borsh::from_slice::<AmmConfig>(&data[..AMM_CONFIG_SIZE]).ok()
 }
 
-pub fn amm_config_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
+pub fn amm_config_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountRaydiumCpmmAmmConfig;
+
     if account.data.len() < AMM_CONFIG_SIZE + 8 {
         return None;
     }
@@ -91,7 +93,9 @@ pub fn pool_state_decode(data: &[u8]) -> Option<PoolState> {
     borsh::from_slice::<PoolState>(&data[..POOL_STATE_SIZE]).ok()
 }
 
-pub fn pool_state_parser(account: &AccountPretty, metadata: EventMetadata) -> Option<DexEvent> {
+pub fn pool_state_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option<DexEvent> {
+    metadata.event_type = EventType::AccountRaydiumCpmmPoolState;
+
     if account.data.len() < POOL_STATE_SIZE + 8 {
         return None;
     }