diff --git a/examples/grpc_example.rs b/examples/grpc_example.rs index 73c61d8..98deb5e 100644 --- a/examples/grpc_example.rs +++ b/examples/grpc_example.rs @@ -1,13 +1,13 @@ use solana_streamer_sdk::streaming::{ event_parser::{ protocols::{ - bonk::parser::BONK_PROGRAM_ID, pumpfun::parser::PUMPFUN_PROGRAM_ID, - pumpswap::parser::PUMPSWAP_PROGRAM_ID, + bonk::parser::BONK_PROGRAM_ID, meteora_damm_v2::parser::METEORA_DAMM_V2_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, }, - Protocol, DexEvent, + DexEvent, Protocol, }, grpc::ClientConfig, yellowstone_grpc::{AccountFilter, TransactionFilter}, @@ -46,18 +46,20 @@ async fn test_grpc() -> Result<(), Box> { Protocol::RaydiumCpmm, Protocol::RaydiumClmm, Protocol::RaydiumAmmV4, + Protocol::MeteoraDammV2, ]; println!("Protocols to monitor: {:?}", protocols); // Filter accounts let account_include = vec![ - PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID - PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID - BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID - RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID - RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID - RAYDIUM_AMM_V4_PROGRAM_ID.to_string(), // Listen to raydium_amm_v4 program ID + PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID + PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID + BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID + RAYDIUM_AMM_V4_PROGRAM_ID.to_string(), // Listen to raydium_amm_v4 program ID + METEORA_DAMM_V2_PROGRAM_ID.to_string(), // Listen to meteora_damm_v2 program ID ]; let account_exclude = vec![]; let account_required = vec![]; diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index 91a2007..0afe5f0 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -186,6 +186,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> { Protocol::PumpFun, Protocol::RaydiumCpmm, Protocol::RaydiumAmmV4, + Protocol::MeteoraDammV2, ]; // Create callback diff --git a/src/streaming/common/constants.rs b/src/streaming/common/constants.rs index e54309c..6be8e37 100644 --- a/src/streaming/common/constants.rs +++ b/src/streaming/common/constants.rs @@ -10,3 +10,9 @@ pub const DEFAULT_MAX_DECODING_MESSAGE_SIZE: usize = 1024 * 1024 * 10; pub const DEFAULT_METRICS_WINDOW_SECONDS: u64 = 5; pub const DEFAULT_METRICS_PRINT_INTERVAL_SECONDS: u64 = 10; pub const SLOW_PROCESSING_THRESHOLD_US: f64 = 3000.0; + +// gRPC 延迟监控 +// Solana 不存储毫秒，所以我们用500ms来校准以获得更好的近似值 +pub const SOLANA_BLOCK_TIME_ADJUSTMENT_MS: i64 = 500; +// 默认最大延迟阈值（毫秒） +pub const MAX_LATENCY_THRESHOLD_MS: i64 = 1000; diff --git a/src/streaming/common/event_processor.rs b/src/streaming/common/event_processor.rs index e415807..87e986d 100644 --- a/src/streaming/common/event_processor.rs +++ b/src/streaming/common/event_processor.rs @@ -18,12 +18,19 @@ fn create_metrics_callback( callback: Arc, ) -> Arc { Arc::new(move |event: DexEvent| { - let processing_time_us = event.metadata().handle_us as f64; + let metadata = event.metadata(); + let processing_time_us = metadata.handle_us as f64; + let recv_us = metadata.recv_us; + let block_time_ms = metadata.block_time_ms; + callback(event); - MetricsManager::global().update_metrics( + + update_metrics_with_latency( MetricsEventType::Transaction, 1, processing_time_us, + recv_us, + block_time_ms, ); }) } @@ -144,8 +151,20 @@ pub async fn process_shred_transaction( Ok(()) } -/// Update metrics for event processing +/// Update metrics for event processing (with optional latency check) #[inline] fn update_metrics(ty: MetricsEventType, count: u64, time_us: f64) { MetricsManager::global().update_metrics(ty, count, time_us); } + +/// Update metrics with latency check +#[inline] +fn update_metrics_with_latency( + ty: MetricsEventType, + count: u64, + time_us: f64, + recv_us: i64, + block_time_ms: i64, +) { + MetricsManager::global().update_metrics_with_latency(ty, count, time_us, recv_us, block_time_ms); +} diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index dea8bbf..86e9666 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -369,6 +369,25 @@ impl MetricsManager { } } + /// 检查并警告高延迟 (校准后的 gRPC latency) + /// latency = recv_time - (block_time + 500ms) + #[inline] + pub fn check_and_warn_high_latency(&self, recv_us: i64, block_time_ms: i64) { + let recv_ms = recv_us / 1000; + // 校准延迟: recv_time - (block_time + 500ms) + let adjusted_latency_ms = recv_ms - (block_time_ms + SOLANA_BLOCK_TIME_ADJUSTMENT_MS); + + if adjusted_latency_ms > MAX_LATENCY_THRESHOLD_MS { + log::warn!( + "⚠️ High gRPC latency: {}ms (threshold: {}ms, raw: recv={}ms, block={}ms)", + adjusted_latency_ms, + MAX_LATENCY_THRESHOLD_MS, + recv_ms, + block_time_ms + ); + } + } + /// 获取运行时长 pub fn get_uptime(&self) -> std::time::Duration { std::time::Duration::from_secs_f64(GLOBAL_METRICS.get_uptime_seconds()) @@ -481,6 +500,20 @@ impl MetricsManager { self.log_slow_processing(processing_time_us, events_processed as usize); } + /// 更新指标并检查延迟 + #[inline] + pub fn update_metrics_with_latency( + &self, + event_type: MetricsEventType, + events_processed: u64, + processing_time_us: f64, + recv_us: i64, + block_time_ms: i64, + ) { + self.check_and_warn_high_latency(recv_us, block_time_ms); + self.update_metrics(event_type, events_processed, processing_time_us); + } + /// 增加丢弃事件计数 #[inline] pub fn increment_dropped_events(&self) { diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index c07004c..91fd51d 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -51,6 +51,7 @@ pub enum ProtocolType { RaydiumCpmm, RaydiumClmm, RaydiumAmmV4, + MeteoraDammV2, Common, } @@ -110,6 +111,13 @@ pub enum EventType { RaydiumAmmV4Withdraw, RaydiumAmmV4WithdrawPnl, + // Meteora DAMM v2 events + MeteoraDammV2Swap, + MeteoraDammV2Swap2, + MeteoraDammV2InitializePool, + MeteoraDammV2InitializeCustomizablePool, + MeteoraDammV2InitializePoolWithDynamicConfig, + // Account events AccountRaydiumAmmV4AmmInfo, AccountPumpSwapGlobalConfig, @@ -201,6 +209,11 @@ impl fmt::Display for EventType { EventType::RaydiumAmmV4Initialize2 => write!(f, "RaydiumAmmV4Initialize2"), EventType::RaydiumAmmV4Withdraw => write!(f, "RaydiumAmmV4Withdraw"), EventType::RaydiumAmmV4WithdrawPnl => write!(f, "RaydiumAmmV4WithdrawPnl"), + EventType::MeteoraDammV2Swap => write!(f, "MeteoraDammV2Swap"), + EventType::MeteoraDammV2Swap2 => write!(f, "MeteoraDammV2Swap2"), + EventType::MeteoraDammV2InitializePool => write!(f, "MeteoraDammV2InitializePool"), + EventType::MeteoraDammV2InitializeCustomizablePool => write!(f, "MeteoraDammV2InitializeCustomizablePool"), + EventType::MeteoraDammV2InitializePoolWithDynamicConfig => write!(f, "MeteoraDammV2InitializePoolWithDynamicConfig"), EventType::AccountRaydiumAmmV4AmmInfo => write!(f, "AccountRaydiumAmmV4AmmInfo"), EventType::AccountPumpSwapGlobalConfig => write!(f, "AccountPumpSwapGlobalConfig"), EventType::AccountPumpSwapPool => write!(f, "AccountPumpSwapPool"), diff --git a/src/streaming/event_parser/core/dispatcher.rs b/src/streaming/event_parser/core/dispatcher.rs index f5a5054..6429f3a 100644 --- a/src/streaming/event_parser/core/dispatcher.rs +++ b/src/streaming/event_parser/core/dispatcher.rs @@ -10,9 +10,9 @@ use crate::streaming::event_parser::{ common::EventMetadata, protocols::{ - bonk::parser as bonk, pumpfun::parser as pumpfun, pumpswap::parser as pumpswap, - raydium_amm_v4::parser as raydium_amm_v4, raydium_clmm::parser as raydium_clmm, - raydium_cpmm::parser as raydium_cpmm, + bonk::parser as bonk, meteora_damm_v2::parser as meteora_damm_v2, 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, }; @@ -52,6 +52,7 @@ impl EventDispatcher { Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm, Protocol::RaydiumClmm => ProtocolType::RaydiumClmm, Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4, + Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2, }; match protocol { @@ -91,6 +92,12 @@ impl EventDispatcher { accounts, metadata, ), + Protocol::MeteoraDammV2 => meteora_damm_v2::parse_meteora_damm_v2_instruction_data( + instruction_discriminator, + instruction_data, + accounts, + metadata, + ), } } @@ -120,6 +127,7 @@ impl EventDispatcher { Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm, Protocol::RaydiumClmm => ProtocolType::RaydiumClmm, Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4, + Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2, }; match protocol { @@ -153,6 +161,11 @@ impl EventDispatcher { inner_instruction_data, metadata, ), + Protocol::MeteoraDammV2 => meteora_damm_v2::parse_meteora_damm_v2_inner_instruction_data( + inner_instruction_discriminator, + inner_instruction_data, + metadata, + ), } } @@ -171,6 +184,8 @@ impl EventDispatcher { Some(Protocol::RaydiumClmm) } else if program_id == &raydium_amm_v4::RAYDIUM_AMM_V4_PROGRAM_ID { Some(Protocol::RaydiumAmmV4) + } else if program_id == &meteora_damm_v2::METEORA_DAMM_V2_PROGRAM_ID { + Some(Protocol::MeteoraDammV2) } else { None } @@ -186,6 +201,7 @@ impl EventDispatcher { 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, + Protocol::MeteoraDammV2 => meteora_damm_v2::METEORA_DAMM_V2_PROGRAM_ID, } } @@ -221,6 +237,7 @@ impl EventDispatcher { Protocol::RaydiumCpmm => ProtocolType::RaydiumCpmm, Protocol::RaydiumClmm => ProtocolType::RaydiumClmm, Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4, + Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2, }; match protocol { @@ -240,6 +257,10 @@ impl EventDispatcher { Protocol::RaydiumAmmV4 => { raydium_amm_v4::parse_raydium_amm_v4_account_data(discriminator, account, metadata) } + Protocol::MeteoraDammV2 => { + // Meteora DAMM 目前不需要解析账户数据，返回 None + None + } } } } diff --git a/src/streaming/event_parser/core/merger_event.rs b/src/streaming/event_parser/core/merger_event.rs index faf69ef..8fcfd12 100644 --- a/src/streaming/event_parser/core/merger_event.rs +++ b/src/streaming/event_parser/core/merger_event.rs @@ -238,6 +238,136 @@ pub fn merge(instruction_event: &mut DexEvent, cpi_log_event: DexEvent) { } _ => {} }, + DexEvent::MeteoraDammV2SwapEvent(e) => match cpi_log_event { + DexEvent::MeteoraDammV2SwapEvent(cpie) => { + e.pool = cpie.pool; + e.trade_direction = cpie.trade_direction; + e.collect_fee_mode = cpie.collect_fee_mode; + e.has_referral = cpie.has_referral; + e.amount_0 = cpie.amount_0; + e.amount_1 = cpie.amount_1; + e.swap_mode = cpie.swap_mode; + e.included_fee_input_amount = cpie.included_fee_input_amount; + e.excluded_fee_input_amount = cpie.excluded_fee_input_amount; + e.amount_left = cpie.amount_left; + e.output_amount = cpie.output_amount; + e.next_sqrt_price = cpie.next_sqrt_price; + e.trading_fee = cpie.trading_fee; + e.partner_fee = cpie.partner_fee; + e.referral_fee = cpie.referral_fee; + e.included_transfer_fee_amount_in = cpie.included_transfer_fee_amount_in; + e.included_transfer_fee_amount_out = cpie.included_transfer_fee_amount_out; + e.excluded_transfer_fee_amount_out = cpie.excluded_transfer_fee_amount_out; + e.current_timestamp = cpie.current_timestamp; + e.reserve_a_amount = cpie.reserve_a_amount; + e.reserve_b_amount = cpie.reserve_b_amount; + } + _ => {} + }, + DexEvent::MeteoraDammV2Swap2Event(e) => match cpi_log_event { + DexEvent::MeteoraDammV2SwapEvent(cpie) => { + e.pool = cpie.pool; + e.trade_direction = cpie.trade_direction; + e.collect_fee_mode = cpie.collect_fee_mode; + e.has_referral = cpie.has_referral; + e.amount_0 = cpie.amount_0; + e.amount_1 = cpie.amount_1; + e.swap_mode = cpie.swap_mode; + e.included_fee_input_amount = cpie.included_fee_input_amount; + e.excluded_fee_input_amount = cpie.excluded_fee_input_amount; + e.amount_left = cpie.amount_left; + e.output_amount = cpie.output_amount; + e.next_sqrt_price = cpie.next_sqrt_price; + e.trading_fee = cpie.trading_fee; + e.partner_fee = cpie.partner_fee; + e.referral_fee = cpie.referral_fee; + e.included_transfer_fee_amount_in = cpie.included_transfer_fee_amount_in; + e.included_transfer_fee_amount_out = cpie.included_transfer_fee_amount_out; + e.excluded_transfer_fee_amount_out = cpie.excluded_transfer_fee_amount_out; + e.current_timestamp = cpie.current_timestamp; + e.reserve_a_amount = cpie.reserve_a_amount; + e.reserve_b_amount = cpie.reserve_b_amount; + } + _ => {} + }, + DexEvent::MeteoraDammV2InitializePoolEvent(e) => match cpi_log_event { + DexEvent::MeteoraDammV2InitializePoolEvent(cpie) => { + e.pool = cpie.pool; + e.token_a_mint = cpie.token_a_mint; + e.token_b_mint = cpie.token_b_mint; + e.creator = cpie.creator; + e.payer = cpie.payer; + e.alpha_vault = cpie.alpha_vault; + e.pool_fees = cpie.pool_fees; + e.sqrt_min_price = cpie.sqrt_min_price; + e.sqrt_max_price = cpie.sqrt_max_price; + e.activation_type = cpie.activation_type; + e.collect_fee_mode = cpie.collect_fee_mode; + e.liquidity = cpie.liquidity; + e.sqrt_price = cpie.sqrt_price; + e.activation_point = cpie.activation_point; + e.token_a_flag = cpie.token_a_flag; + e.token_b_flag = cpie.token_b_flag; + e.token_a_amount = cpie.token_a_amount; + e.token_b_amount = cpie.token_b_amount; + e.total_amount_a = cpie.total_amount_a; + e.total_amount_b = cpie.total_amount_b; + e.pool_type = cpie.pool_type; + } + _ => {} + }, + DexEvent::MeteoraDammV2InitializeCustomizablePoolEvent(e) => match cpi_log_event { + DexEvent::MeteoraDammV2InitializePoolEvent(cpie) => { + e.pool = cpie.pool; + e.token_a_mint = cpie.token_a_mint; + e.token_b_mint = cpie.token_b_mint; + e.creator = cpie.creator; + e.payer = cpie.payer; + e.alpha_vault = cpie.alpha_vault; + e.pool_fees = cpie.pool_fees; + e.sqrt_min_price = cpie.sqrt_min_price; + e.sqrt_max_price = cpie.sqrt_max_price; + e.activation_type = cpie.activation_type; + e.collect_fee_mode = cpie.collect_fee_mode; + e.liquidity = cpie.liquidity; + e.sqrt_price = cpie.sqrt_price; + e.activation_point = cpie.activation_point; + e.token_a_flag = cpie.token_a_flag; + e.token_b_flag = cpie.token_b_flag; + e.token_a_amount = cpie.token_a_amount; + e.token_b_amount = cpie.token_b_amount; + e.total_amount_a = cpie.total_amount_a; + e.total_amount_b = cpie.total_amount_b; + e.pool_type = cpie.pool_type; + } + _ => {} + }, + DexEvent::MeteoraDammV2InitializePoolWithDynamicConfigEvent(e) => match cpi_log_event { + DexEvent::MeteoraDammV2InitializePoolEvent(cpie) => { + e.pool = cpie.pool; + e.token_a_mint = cpie.token_a_mint; + e.token_b_mint = cpie.token_b_mint; + e.creator = cpie.creator; + e.payer = cpie.payer; + e.alpha_vault = cpie.alpha_vault; + e.pool_fees = cpie.pool_fees; + e.sqrt_min_price = cpie.sqrt_min_price; + e.sqrt_max_price = cpie.sqrt_max_price; + e.activation_type = cpie.activation_type; + e.collect_fee_mode = cpie.collect_fee_mode; + e.liquidity = cpie.liquidity; + e.sqrt_price = cpie.sqrt_price; + e.activation_point = cpie.activation_point; + e.token_a_flag = cpie.token_a_flag; + e.token_b_flag = cpie.token_b_flag; + e.token_a_amount = cpie.token_a_amount; + e.token_b_amount = cpie.token_b_amount; + e.total_amount_a = cpie.total_amount_a; + e.total_amount_b = cpie.total_amount_b; + e.pool_type = cpie.pool_type; + } + _ => {} + }, _ => {} } diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index 833c182..6fd4c7f 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -4,6 +4,7 @@ use crate::streaming::event_parser::core::account_event_parser::{ }; use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent; use crate::streaming::event_parser::protocols::bonk::events::*; +use crate::streaming::event_parser::protocols::meteora_damm_v2::events::*; use crate::streaming::event_parser::protocols::pumpfun::events::*; use crate::streaming::event_parser::protocols::pumpswap::events::*; use crate::streaming::event_parser::protocols::raydium_amm_v4::events::*; @@ -70,6 +71,13 @@ pub enum DexEvent { RaydiumCpmmAmmConfigAccountEvent(RaydiumCpmmAmmConfigAccountEvent), RaydiumCpmmPoolStateAccountEvent(RaydiumCpmmPoolStateAccountEvent), + // Meteora DAMM v2 events + MeteoraDammV2SwapEvent(MeteoraDammV2SwapEvent), + MeteoraDammV2Swap2Event(MeteoraDammV2Swap2Event), + MeteoraDammV2InitializePoolEvent(MeteoraDammV2InitializePoolEvent), + MeteoraDammV2InitializeCustomizablePoolEvent(MeteoraDammV2InitializeCustomizablePoolEvent), + MeteoraDammV2InitializePoolWithDynamicConfigEvent(MeteoraDammV2InitializePoolWithDynamicConfigEvent), + // Common events TokenAccountEvent(TokenAccountEvent), NonceAccountEvent(NonceAccountEvent), @@ -123,6 +131,11 @@ impl DexEvent { DexEvent::RaydiumCpmmInitializeEvent(e) => &e.metadata, DexEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &e.metadata, DexEvent::RaydiumCpmmPoolStateAccountEvent(e) => &e.metadata, + DexEvent::MeteoraDammV2SwapEvent(e) => &e.metadata, + DexEvent::MeteoraDammV2Swap2Event(e) => &e.metadata, + DexEvent::MeteoraDammV2InitializePoolEvent(e) => &e.metadata, + DexEvent::MeteoraDammV2InitializeCustomizablePoolEvent(e) => &e.metadata, + DexEvent::MeteoraDammV2InitializePoolWithDynamicConfigEvent(e) => &e.metadata, DexEvent::TokenAccountEvent(e) => &e.metadata, DexEvent::NonceAccountEvent(e) => &e.metadata, DexEvent::TokenInfoEvent(e) => &e.metadata, @@ -175,6 +188,11 @@ impl DexEvent { DexEvent::RaydiumCpmmInitializeEvent(e) => &mut e.metadata, DexEvent::RaydiumCpmmAmmConfigAccountEvent(e) => &mut e.metadata, DexEvent::RaydiumCpmmPoolStateAccountEvent(e) => &mut e.metadata, + DexEvent::MeteoraDammV2SwapEvent(e) => &mut e.metadata, + DexEvent::MeteoraDammV2Swap2Event(e) => &mut e.metadata, + DexEvent::MeteoraDammV2InitializePoolEvent(e) => &mut e.metadata, + DexEvent::MeteoraDammV2InitializeCustomizablePoolEvent(e) => &mut e.metadata, + DexEvent::MeteoraDammV2InitializePoolWithDynamicConfigEvent(e) => &mut e.metadata, DexEvent::TokenAccountEvent(e) => &mut e.metadata, DexEvent::NonceAccountEvent(e) => &mut e.metadata, DexEvent::TokenInfoEvent(e) => &mut e.metadata, diff --git a/src/streaming/event_parser/protocols/meteora_damm_v2/events.rs b/src/streaming/event_parser/protocols/meteora_damm_v2/events.rs new file mode 100644 index 0000000..06933b3 --- /dev/null +++ b/src/streaming/event_parser/protocols/meteora_damm_v2/events.rs @@ -0,0 +1,417 @@ +use borsh::BorshDeserialize; +use serde::{Deserialize, Serialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::streaming::event_parser::common::EventMetadata; + +/// Base fee parameters +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct BaseFeeParameters { + pub cliff_fee_numerator: u64, + pub first_factor: u16, + pub second_factor: [u8; 8], + pub third_factor: u64, + pub base_fee_mode: u8, +} + +/// Dynamic fee parameters +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct DynamicFeeParameters { + pub bin_step: u16, + pub bin_step_u128: u128, + pub filter_period: u16, + pub decay_period: u16, + pub reduction_factor: u16, + pub max_volatility_accumulator: u32, + pub variable_fee_control: u32, +} + +/// Pool fee parameters +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PoolFeeParameters { + pub base_fee: BaseFeeParameters, + pub padding: [u8; 3], + pub dynamic_fee: Option, +} + +/// Meteora DAMM v2 Swap Event (对应 swap 指令) +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct MeteoraDammV2SwapEvent { + #[borsh(skip)] + pub metadata: EventMetadata, + + // 来自 CPI Log Event 的数据 + pub pool: Pubkey, + pub trade_direction: u8, // 0 or 1 + pub collect_fee_mode: u8, + pub has_referral: bool, + + // Swap parameters + pub amount_0: u64, // amount0 from params + pub amount_1: u64, // amount1 from params + pub swap_mode: u8, // swapMode from params + + // Swap result + pub included_fee_input_amount: u64, + pub excluded_fee_input_amount: u64, + pub amount_left: u64, + pub output_amount: u64, + pub next_sqrt_price: u128, + pub trading_fee: u64, + pub protocol_fee: u64, + pub partner_fee: u64, + pub referral_fee: u64, + + // Transfer fee amounts + pub included_transfer_fee_amount_in: u64, + pub included_transfer_fee_amount_out: u64, + pub excluded_transfer_fee_amount_out: u64, + + // Additional info + pub current_timestamp: u64, + pub reserve_a_amount: u64, + pub reserve_b_amount: u64, + + // 来自 Input Accounts 的数据 + #[borsh(skip)] + pub pool_authority: Pubkey, + #[borsh(skip)] + pub input_token_account: Pubkey, + #[borsh(skip)] + pub output_token_account: Pubkey, + #[borsh(skip)] + pub token_a_vault: Pubkey, + #[borsh(skip)] + pub token_b_vault: Pubkey, + #[borsh(skip)] + pub token_a_mint: Pubkey, + #[borsh(skip)] + pub token_b_mint: Pubkey, + #[borsh(skip)] + pub payer: Pubkey, + #[borsh(skip)] + pub token_a_program: Pubkey, + #[borsh(skip)] + pub token_b_program: Pubkey, + #[borsh(skip)] + pub referral_token_account: Option, + #[borsh(skip)] + pub event_authority: Pubkey, + #[borsh(skip)] + pub program: Pubkey, +} + +/// Meteora DAMM v2 Swap2 Event (对应 swap2 指令) +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct MeteoraDammV2Swap2Event { + #[borsh(skip)] + pub metadata: EventMetadata, + + // 来自 CPI Log Event 的数据 + pub pool: Pubkey, + pub trade_direction: u8, // 0 or 1 + pub collect_fee_mode: u8, + pub has_referral: bool, + + // Swap parameters + pub amount_0: u64, // amount0 from params + pub amount_1: u64, // amount1 from params + pub swap_mode: u8, // swapMode from params + + // Swap result + pub included_fee_input_amount: u64, + pub excluded_fee_input_amount: u64, + pub amount_left: u64, + pub output_amount: u64, + pub next_sqrt_price: u128, + pub trading_fee: u64, + pub protocol_fee: u64, + pub partner_fee: u64, + pub referral_fee: u64, + + // Transfer fee amounts + pub included_transfer_fee_amount_in: u64, + pub included_transfer_fee_amount_out: u64, + pub excluded_transfer_fee_amount_out: u64, + + // Additional info + pub current_timestamp: u64, + pub reserve_a_amount: u64, + pub reserve_b_amount: u64, + + // 来自 Input Accounts 的数据 + #[borsh(skip)] + pub pool_authority: Pubkey, + #[borsh(skip)] + pub input_token_account: Pubkey, + #[borsh(skip)] + pub output_token_account: Pubkey, + #[borsh(skip)] + pub token_a_vault: Pubkey, + #[borsh(skip)] + pub token_b_vault: Pubkey, + #[borsh(skip)] + pub token_a_mint: Pubkey, + #[borsh(skip)] + pub token_b_mint: Pubkey, + #[borsh(skip)] + pub payer: Pubkey, + #[borsh(skip)] + pub token_a_program: Pubkey, + #[borsh(skip)] + pub token_b_program: Pubkey, + #[borsh(skip)] + pub referral_token_account: Option, + #[borsh(skip)] + pub event_authority: Pubkey, + #[borsh(skip)] + pub program: Pubkey, + #[borsh(skip)] + pub sysvar: Pubkey, +} + +/// Meteora DAMM v2 Initialize Pool Event (对应 initialize_pool 指令) +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct MeteoraDammV2InitializePoolEvent { + #[borsh(skip)] + pub metadata: EventMetadata, + + // 来自 CPI Log Event 的数据 + pub pool: Pubkey, + pub token_a_mint: Pubkey, + pub token_b_mint: Pubkey, + pub creator: Pubkey, + pub payer: Pubkey, + pub alpha_vault: Pubkey, + + // Pool fees + pub pool_fees: PoolFeeParameters, + + // Price and liquidity + pub sqrt_min_price: u128, + pub sqrt_max_price: u128, + pub activation_type: u8, + pub collect_fee_mode: u8, + pub liquidity: u128, + pub sqrt_price: u128, + pub activation_point: u64, + + // Token amounts + pub token_a_flag: u8, + pub token_b_flag: u8, + pub token_a_amount: u64, + pub token_b_amount: u64, + pub total_amount_a: u64, + pub total_amount_b: u64, + pub pool_type: u8, + + // 来自 Input Accounts 的数据 + #[borsh(skip)] + pub position_nft_mint: Pubkey, + #[borsh(skip)] + pub position_nft_account: Pubkey, + #[borsh(skip)] + pub pool_authority: Pubkey, + #[borsh(skip)] + pub position: Pubkey, + #[borsh(skip)] + pub token_a_vault: Pubkey, + #[borsh(skip)] + pub token_b_vault: Pubkey, + #[borsh(skip)] + pub payer_token_a: Pubkey, + #[borsh(skip)] + pub payer_token_b: Pubkey, + #[borsh(skip)] + pub token_a_program: Pubkey, + #[borsh(skip)] + pub token_b_program: Pubkey, + #[borsh(skip)] + pub event_authority: Pubkey, + #[borsh(skip)] + pub program: Pubkey, + #[borsh(skip)] + pub config: Pubkey, + #[borsh(skip)] + pub remaining_accounts: Vec, +} + +/// Meteora DAMM v2 Initialize Customizable Pool Event (对应 initialize_customizable_pool 指令) +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct MeteoraDammV2InitializeCustomizablePoolEvent { + #[borsh(skip)] + pub metadata: EventMetadata, + + // 来自 CPI Log Event 的数据 + pub pool: Pubkey, + pub token_a_mint: Pubkey, + pub token_b_mint: Pubkey, + pub creator: Pubkey, + pub payer: Pubkey, + pub alpha_vault: Pubkey, + + // Pool fees + pub pool_fees: PoolFeeParameters, + + // Price and liquidity + pub sqrt_min_price: u128, + pub sqrt_max_price: u128, + pub activation_type: u8, + pub collect_fee_mode: u8, + pub liquidity: u128, + pub sqrt_price: u128, + pub activation_point: u64, + + // Token amounts + pub token_a_flag: u8, + pub token_b_flag: u8, + pub token_a_amount: u64, + pub token_b_amount: u64, + pub total_amount_a: u64, + pub total_amount_b: u64, + pub pool_type: u8, + + // 来自 Input Accounts 的数据 + #[borsh(skip)] + pub position_nft_mint: Pubkey, + #[borsh(skip)] + pub position_nft_account: Pubkey, + #[borsh(skip)] + pub pool_authority: Pubkey, + #[borsh(skip)] + pub position: Pubkey, + #[borsh(skip)] + pub token_a_vault: Pubkey, + #[borsh(skip)] + pub token_b_vault: Pubkey, + #[borsh(skip)] + pub payer_token_a: Pubkey, + #[borsh(skip)] + pub payer_token_b: Pubkey, + #[borsh(skip)] + pub token_a_program: Pubkey, + #[borsh(skip)] + pub token_b_program: Pubkey, + #[borsh(skip)] + pub token_2022_program: Pubkey, + #[borsh(skip)] + pub system_program: Pubkey, + #[borsh(skip)] + pub event_authority: Pubkey, + #[borsh(skip)] + pub program: Pubkey, + #[borsh(skip)] + pub remaining_accounts: Vec, +} + +/// Meteora DAMM v2 Initialize Pool With Dynamic Config Event (对应 initialize_pool_with_dynamic_config 指令) +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct MeteoraDammV2InitializePoolWithDynamicConfigEvent { + #[borsh(skip)] + pub metadata: EventMetadata, + + // 来自 CPI Log Event 的数据 + pub pool: Pubkey, + pub token_a_mint: Pubkey, + pub token_b_mint: Pubkey, + pub creator: Pubkey, + pub payer: Pubkey, + pub alpha_vault: Pubkey, + + // Pool fees + pub pool_fees: PoolFeeParameters, + + // Price and liquidity + pub sqrt_min_price: u128, + pub sqrt_max_price: u128, + pub activation_type: u8, + pub collect_fee_mode: u8, + pub liquidity: u128, + pub sqrt_price: u128, + pub activation_point: u64, + + // Token amounts + pub token_a_flag: u8, + pub token_b_flag: u8, + pub token_a_amount: u64, + pub token_b_amount: u64, + pub total_amount_a: u64, + pub total_amount_b: u64, + pub pool_type: u8, + + // 来自 Input Accounts 的数据 + #[borsh(skip)] + pub position_nft_mint: Pubkey, + #[borsh(skip)] + pub position_nft_account: Pubkey, + #[borsh(skip)] + pub pool_authority: Pubkey, + #[borsh(skip)] + pub pool_creator_authority: Pubkey, + #[borsh(skip)] + pub position: Pubkey, + #[borsh(skip)] + pub token_a_vault: Pubkey, + #[borsh(skip)] + pub token_b_vault: Pubkey, + #[borsh(skip)] + pub payer_token_a: Pubkey, + #[borsh(skip)] + pub payer_token_b: Pubkey, + #[borsh(skip)] + pub token_a_program: Pubkey, + #[borsh(skip)] + pub token_b_program: Pubkey, + #[borsh(skip)] + pub token_2022_program: Pubkey, + #[borsh(skip)] + pub system_program: Pubkey, + #[borsh(skip)] + pub event_authority: Pubkey, + #[borsh(skip)] + pub program: Pubkey, + #[borsh(skip)] + pub config: Pubkey, +} + +/// Event discriminators +pub mod discriminators { + // Instruction discriminators + // 从文档中提取的 instruction data 第一个 8 bytes + pub const SWAP_IX: &[u8] = &[0xf8, 0xc6, 0x9e, 0x91, 0xe1, 0x75, 0x87, 0xc8]; // swap + pub const SWAP2_IX: &[u8] = &[0x41, 0x4b, 0x3f, 0x4c, 0xeb, 0x5b, 0x5b, 0x88]; // swap2 + pub const INITIALIZE_CUSTOMIZABLE_POOL_IX: &[u8] = + &[0x14, 0xa1, 0xf1, 0x18, 0xbd, 0xdd, 0xb4, 0x02]; // initialize_customizable_pool + pub const INITIALIZE_POOL_IX: &[u8] = &[0x5f, 0xb4, 0x0a, 0xac, 0x54, 0xae, 0xe8, 0x28]; // initialize_pool + pub const INITIALIZE_POOL_WITH_DYNAMIC_CONFIG_IX: &[u8] = + &[0x95, 0x52, 0x48, 0xc5, 0xfd, 0xfc, 0x44, 0x0f]; // initialize_pool_with_dynamic_config + + // Event discriminators (CPI Log Event) + // e445a52e51cb9a1d 是 Meteora 的事件前缀 + // 后面的 8 字节是具体事件类型 + pub const SWAP_EVENT: &[u8] = &[ + 0xe4, 0x45, 0xa5, 0x2e, 0x51, 0xcb, 0x9a, 0x1d, 0xbd, 0x42, 0x33, 0xa8, 0x26, 0x50, 0x75, + 0x99, + ]; // swap event + pub const INITIALIZE_POOL_EVENT: &[u8] = &[ + 0xe4, 0x45, 0xa5, 0x2e, 0x51, 0xcb, 0x9a, 0x1d, 0xe4, 0x32, 0xf6, 0x55, 0xcb, 0x42, 0x86, + 0x25, + ]; // initialize pool event +} + +/// Decode swap event from CPI log +pub const METEORA_DAMM_V2_SWAP_EVENT_LOG_SIZE: usize = 180; +pub fn meteora_damm_v2_swap_event_decode(data: &[u8]) -> Option { + if data.len() < METEORA_DAMM_V2_SWAP_EVENT_LOG_SIZE { + return None; + } + borsh::from_slice::(&data[..METEORA_DAMM_V2_SWAP_EVENT_LOG_SIZE]).ok() +} + +/// Decode initialize pool event from CPI log +/// Note: discriminator (16 bytes) is already removed by the caller +pub fn meteora_damm_v2_initialize_pool_event_decode( + data: &[u8], +) -> Option { + borsh::from_slice::(&data).ok() +} diff --git a/src/streaming/event_parser/protocols/meteora_damm_v2/mod.rs b/src/streaming/event_parser/protocols/meteora_damm_v2/mod.rs new file mode 100644 index 0000000..86156fa --- /dev/null +++ b/src/streaming/event_parser/protocols/meteora_damm_v2/mod.rs @@ -0,0 +1,5 @@ +pub mod events; +pub mod parser; +pub mod types; + +pub use events::*; diff --git a/src/streaming/event_parser/protocols/meteora_damm_v2/parser.rs b/src/streaming/event_parser/protocols/meteora_damm_v2/parser.rs new file mode 100644 index 0000000..ee71399 --- /dev/null +++ b/src/streaming/event_parser/protocols/meteora_damm_v2/parser.rs @@ -0,0 +1,444 @@ +use crate::streaming::event_parser::{ + common::{EventMetadata, EventType}, + protocols::meteora_damm_v2::{ + discriminators, meteora_damm_v2_initialize_pool_event_decode, + meteora_damm_v2_swap_event_decode, MeteoraDammV2InitializeCustomizablePoolEvent, + MeteoraDammV2InitializePoolEvent, MeteoraDammV2InitializePoolWithDynamicConfigEvent, + MeteoraDammV2Swap2Event, MeteoraDammV2SwapEvent, + }, + DexEvent, +}; +use solana_sdk::pubkey::Pubkey; + +/// Meteora DAMM v2 程序ID +pub const METEORA_DAMM_V2_PROGRAM_ID: Pubkey = + solana_sdk::pubkey!("cpamdpZCGKUy5JxQXB4dcpGPiikHawvSWAd6mEn1sGG"); + +/// 解析 Meteora DAMM v2 instruction data +/// +/// 根据判别器路由到具体的 instruction 解析函数 +pub fn parse_meteora_damm_v2_instruction_data( + discriminator: &[u8], + data: &[u8], + accounts: &[Pubkey], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::SWAP_IX => parse_swap_instruction(data, accounts, metadata), + discriminators::SWAP2_IX => parse_swap2_instruction(data, accounts, metadata), + discriminators::INITIALIZE_POOL_IX => { + parse_initialize_pool_instruction(data, accounts, metadata) + } + discriminators::INITIALIZE_CUSTOMIZABLE_POOL_IX => { + parse_initialize_customizable_pool_instruction(data, accounts, metadata) + } + discriminators::INITIALIZE_POOL_WITH_DYNAMIC_CONFIG_IX => { + parse_initialize_pool_with_dynamic_config_instruction(data, accounts, metadata) + } + _ => None, + } +} + +/// 解析 Meteora DAMM v2 inner instruction data (CPI events) +/// +/// 根据判别器路由到具体的 inner instruction 解析函数 +pub fn parse_meteora_damm_v2_inner_instruction_data( + discriminator: &[u8], + data: &[u8], + metadata: EventMetadata, +) -> Option { + match discriminator { + discriminators::SWAP_EVENT => parse_swap_inner_instruction(data, metadata), + discriminators::INITIALIZE_POOL_EVENT => { + parse_initialize_pool_inner_instruction(data, metadata) + } + _ => None, + } +} + +/// 解析 swap 指令 +fn parse_swap_instruction( + data: &[u8], + accounts: &[Pubkey], + mut metadata: EventMetadata, +) -> Option { + metadata.event_type = EventType::MeteoraDammV2Swap; + + if data.len() < 16 || accounts.len() < 14 { + return None; + } + + // 跳过 discriminator (8 bytes) + let amount_in = u64::from_le_bytes(data[0..8].try_into().unwrap()); + let minimum_amount_out = u64::from_le_bytes(data[8..16].try_into().unwrap()); + + Some(DexEvent::MeteoraDammV2SwapEvent(MeteoraDammV2SwapEvent { + metadata, + pool_authority: accounts[0], + pool: accounts[1], + input_token_account: accounts[2], + output_token_account: accounts[3], + token_a_vault: accounts[4], + token_b_vault: accounts[5], + token_a_mint: accounts[6], + token_b_mint: accounts[7], + payer: accounts[8], + token_a_program: accounts[9], + token_b_program: accounts[10], + referral_token_account: Some(accounts[11]), + event_authority: accounts[12], + program: accounts[13], + amount_0: amount_in, + amount_1: minimum_amount_out, + ..Default::default() + })) +} + +/// 解析 swap2 指令 +fn parse_swap2_instruction( + data: &[u8], + accounts: &[Pubkey], + mut metadata: EventMetadata, +) -> Option { + metadata.event_type = EventType::MeteoraDammV2Swap2; + + if data.len() < 16 || accounts.len() < 13 { + return None; + } + + // 跳过 discriminator (8 bytes) + let amount_0 = u64::from_le_bytes(data[0..8].try_into().unwrap()); + let amount_1 = u64::from_le_bytes(data[8..16].try_into().unwrap()); + let swap_mode = data[16]; + + // swap2 可能有 15 个账户(带 referral)或 14 个账户 + let has_referral = accounts.len() >= 15; + + Some(DexEvent::MeteoraDammV2Swap2Event(MeteoraDammV2Swap2Event { + metadata, + pool_authority: accounts[0], + pool: accounts[1], + input_token_account: accounts[2], + output_token_account: accounts[3], + token_a_vault: accounts[4], + token_b_vault: accounts[5], + token_a_mint: accounts[6], + token_b_mint: accounts[7], + payer: accounts[8], + token_a_program: accounts[9], + token_b_program: accounts[10], + referral_token_account: if has_referral && accounts.len() > 11 { + Some(accounts[11]) + } else { + None + }, + event_authority: accounts[if has_referral { 12 } else { 11 }], + program: accounts[if has_referral { 13 } else { 12 }], + sysvar: accounts[if has_referral { 14 } else { 13 }], + amount_0, + amount_1, + swap_mode, + has_referral, + ..Default::default() + })) +} + +/// 解析 initialize_pool 指令 +fn parse_initialize_pool_instruction( + data: &[u8], + accounts: &[Pubkey], + mut metadata: EventMetadata, +) -> Option { + metadata.event_type = EventType::MeteoraDammV2InitializePool; + + if accounts.len() < 20 { + return None; + } + + // 解析 instruction data (不包含 discriminator，已被调用者移除) + // 结构: liquidity (u128 = 16 bytes) + sqrt_price (u128 = 16 bytes) + activation_point (Option = 1 + 8 bytes) + if data.len() < 33 { + return None; + } + + let mut offset = 0; + + // 读取 liquidity (u128) + let liquidity = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 sqrt_price (u128) + let sqrt_price = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 activation_point (Option) + let option_tag = data[offset]; + offset += 1; + let _activation_point = if option_tag == 1 && data.len() >= offset + 8 { + Some(u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?)) + } else { + None + }; + + Some(DexEvent::MeteoraDammV2InitializePoolEvent(MeteoraDammV2InitializePoolEvent { + metadata, + creator: accounts[0], + position_nft_mint: accounts[1], + position_nft_account: accounts[2], + payer: accounts[3], + config: accounts[4], + pool_authority: accounts[5], + pool: accounts[6], + position: accounts[7], + token_a_mint: accounts[8], + token_b_mint: accounts[9], + token_a_vault: accounts[10], + token_b_vault: accounts[11], + payer_token_a: accounts[12], + payer_token_b: accounts[13], + token_a_program: accounts[14], + token_b_program: accounts[15], + event_authority: accounts[18], + program: accounts[19], + remaining_accounts: accounts[20..].to_vec(), + liquidity, + sqrt_price, + ..Default::default() + })) +} + +/// 解析 initialize_customizable_pool 指令 +fn parse_initialize_customizable_pool_instruction( + data: &[u8], + accounts: &[Pubkey], + mut metadata: EventMetadata, +) -> Option { + metadata.event_type = EventType::MeteoraDammV2InitializeCustomizablePool; + + if accounts.len() < 19 { + return None; + } + + // 解析 instruction data (不包含 discriminator) + // 结构: PoolFeeParameters + sqrt_min_price + sqrt_max_price + has_alpha_vault + liquidity + sqrt_price + activation_type + collect_fee_mode + activation_point + if data.len() < 99 { + return None; + } + + let mut offset = 0; + + // 解析 PoolFeeParameters + use crate::streaming::event_parser::protocols::meteora_damm_v2::PoolFeeParameters; + use borsh::BorshDeserialize; + + // PoolFeeParameters size: 8 + 2 + 8 + 8 + 1 + 3 + 1 + (optional DynamicFee) + // 先读取前 31 bytes (不包含 dynamic_fee option tag) + let pool_fees = PoolFeeParameters::deserialize(&mut &data[offset..]).ok()?; + + // 计算 pool_fees 消耗的字节数 + // BaseFee: 8 + 2 + 8 + 8 + 1 = 27 bytes + // padding: 3 bytes + // option tag: 1 byte + // 如果 dynamic_fee 存在: 2 + 16 + 2 + 2 + 2 + 4 + 4 = 32 bytes + let pool_fees_size = 31 + if pool_fees.dynamic_fee.is_some() { 32 } else { 0 }; + offset += pool_fees_size; + + // 读取 sqrt_min_price (u128) + let sqrt_min_price = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 sqrt_max_price (u128) + let sqrt_max_price = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 has_alpha_vault (bool) + let _has_alpha_vault = data[offset]; + offset += 1; + + // 读取 liquidity (u128) + let liquidity = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 sqrt_price (u128) + let sqrt_price = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 activation_type (u8) + let activation_type = data[offset]; + offset += 1; + + // 读取 collect_fee_mode (u8) + let collect_fee_mode = data[offset]; + offset += 1; + + // 读取 activation_point (Option) + let option_tag = data[offset]; + let _activation_point = if option_tag == 1 && data.len() >= offset + 9 { + Some(u64::from_le_bytes(data[offset + 1..offset + 9].try_into().ok()?)) + } else { + None + }; + + Some(DexEvent::MeteoraDammV2InitializeCustomizablePoolEvent( + MeteoraDammV2InitializeCustomizablePoolEvent { + metadata, + creator: accounts[0], + position_nft_mint: accounts[1], + position_nft_account: accounts[2], + payer: accounts[3], + pool_authority: accounts[4], + pool: accounts[5], + position: accounts[6], + token_a_mint: accounts[7], + token_b_mint: accounts[8], + token_a_vault: accounts[9], + token_b_vault: accounts[10], + payer_token_a: accounts[11], + payer_token_b: accounts[12], + token_a_program: accounts[13], + token_b_program: accounts[14], + token_2022_program: accounts[15], + system_program: accounts[16], + event_authority: accounts[17], + program: accounts[18], + remaining_accounts: accounts[19..].to_vec(), + pool_fees, + sqrt_min_price, + sqrt_max_price, + activation_type, + collect_fee_mode, + liquidity, + sqrt_price, + ..Default::default() + }, + )) +} + +/// 解析 initialize_pool_with_dynamic_config 指令 +fn parse_initialize_pool_with_dynamic_config_instruction( + data: &[u8], + accounts: &[Pubkey], + mut metadata: EventMetadata, +) -> Option { + metadata.event_type = EventType::MeteoraDammV2InitializePoolWithDynamicConfig; + + if accounts.len() < 21 { + return None; + } + + if data.len() < 99 { + return None; + } + + let mut offset = 0; + + // 解析 PoolFeeParameters + use crate::streaming::event_parser::protocols::meteora_damm_v2::PoolFeeParameters; + use borsh::BorshDeserialize; + + let pool_fees = PoolFeeParameters::deserialize(&mut &data[offset..]).ok()?; + + // 计算 pool_fees 消耗的字节数 + // BaseFee: 8 + 2 + 8 + 8 + 1 = 27 bytes + // padding: 3 bytes + // option tag: 1 byte + // 如果 dynamic_fee 存在: 2 + 16 + 2 + 2 + 2 + 4 + 4 = 32 bytes + let pool_fees_size = 31 + if pool_fees.dynamic_fee.is_some() { 32 } else { 0 }; + offset += pool_fees_size; + + // 读取 sqrt_min_price (u128) + let sqrt_min_price = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 sqrt_max_price (u128) + let sqrt_max_price = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 has_alpha_vault (bool) + let _has_alpha_vault = data[offset]; + offset += 1; + + // 读取 liquidity (u128) + let liquidity = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 sqrt_price (u128) + let sqrt_price = u128::from_le_bytes(data[offset..offset + 16].try_into().ok()?); + offset += 16; + + // 读取 activation_type (u8) + let activation_type = data[offset]; + offset += 1; + + // 读取 collect_fee_mode (u8) + let collect_fee_mode = data[offset]; + offset += 1; + + // 读取 activation_point (Option) + let option_tag = data[offset]; + let _activation_point = if option_tag == 1 && data.len() >= offset + 9 { + Some(u64::from_le_bytes(data[offset + 1..offset + 9].try_into().ok()?)) + } else { + None + }; + + Some(DexEvent::MeteoraDammV2InitializePoolWithDynamicConfigEvent( + MeteoraDammV2InitializePoolWithDynamicConfigEvent { + metadata, + creator: accounts[0], + position_nft_mint: accounts[1], + position_nft_account: accounts[2], + payer: accounts[3], + pool_creator_authority: accounts[4], + pool_authority: accounts[6], + pool: accounts[7], + position: accounts[8], + token_a_mint: accounts[9], + token_b_mint: accounts[10], + token_a_vault: accounts[11], + token_b_vault: accounts[12], + payer_token_a: accounts[13], + payer_token_b: accounts[14], + token_a_program: accounts[15], + token_b_program: accounts[16], + token_2022_program: accounts[17], + system_program: accounts[18], + event_authority: accounts[19], + program: accounts[20], + config: accounts[5], + pool_fees, + sqrt_min_price, + sqrt_max_price, + activation_type, + collect_fee_mode, + liquidity, + sqrt_price, + ..Default::default() + }, + )) +} + +/// 解析 swap inner instruction (CPI event) +fn parse_swap_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option { + if let Some(event) = meteora_damm_v2_swap_event_decode(data) { + Some(DexEvent::MeteoraDammV2SwapEvent(MeteoraDammV2SwapEvent { metadata, ..event })) + } else { + None + } +} + +/// 解析 initialize pool inner instruction (CPI event) +fn parse_initialize_pool_inner_instruction( + data: &[u8], + mut metadata: EventMetadata, +) -> Option { + metadata.event_type = EventType::MeteoraDammV2InitializePool; + if let Some(event) = meteora_damm_v2_initialize_pool_event_decode(data) { + Some(DexEvent::MeteoraDammV2InitializePoolEvent(MeteoraDammV2InitializePoolEvent { + metadata, + ..event + })) + } else { + None + } +} diff --git a/src/streaming/event_parser/protocols/meteora_damm_v2/types.rs b/src/streaming/event_parser/protocols/meteora_damm_v2/types.rs new file mode 100644 index 0000000..f68018b --- /dev/null +++ b/src/streaming/event_parser/protocols/meteora_damm_v2/types.rs @@ -0,0 +1,2 @@ +// 此文件用于定义 Meteora DAMM v2 的账户数据结构 +// 暂时留空,后续如需要解析 Pool 账户状态时可以在这里添加 diff --git a/src/streaming/event_parser/protocols/mod.rs b/src/streaming/event_parser/protocols/mod.rs index 71e3c64..0d44048 100755 --- a/src/streaming/event_parser/protocols/mod.rs +++ b/src/streaming/event_parser/protocols/mod.rs @@ -1,5 +1,6 @@ pub mod block; pub mod bonk; +pub mod meteora_damm_v2; pub mod pumpfun; pub mod pumpswap; pub mod raydium_amm_v4; diff --git a/src/streaming/event_parser/protocols/types.rs b/src/streaming/event_parser/protocols/types.rs index bc81193..3d864dc 100755 --- a/src/streaming/event_parser/protocols/types.rs +++ b/src/streaming/event_parser/protocols/types.rs @@ -1,7 +1,8 @@ use crate::streaming::event_parser::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, + bonk::parser::BONK_PROGRAM_ID, meteora_damm_v2::parser::METEORA_DAMM_V2_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, }; use anyhow::{anyhow, Result}; use solana_sdk::pubkey::Pubkey; @@ -15,6 +16,7 @@ pub enum Protocol { RaydiumCpmm, RaydiumClmm, RaydiumAmmV4, + MeteoraDammV2, } impl Protocol { @@ -26,6 +28,7 @@ impl Protocol { Protocol::RaydiumCpmm => vec![RAYDIUM_CPMM_PROGRAM_ID], Protocol::RaydiumClmm => vec![RAYDIUM_CLMM_PROGRAM_ID], Protocol::RaydiumAmmV4 => vec![RAYDIUM_AMM_V4_PROGRAM_ID], + Protocol::MeteoraDammV2 => vec![METEORA_DAMM_V2_PROGRAM_ID], } } } @@ -39,6 +42,7 @@ impl std::fmt::Display for Protocol { Protocol::RaydiumCpmm => write!(f, "RaydiumCpmm"), Protocol::RaydiumClmm => write!(f, "RaydiumClmm"), Protocol::RaydiumAmmV4 => write!(f, "RaydiumAmmV4"), + Protocol::MeteoraDammV2 => write!(f, "MeteoraDammV2"), } } } @@ -54,6 +58,7 @@ impl std::str::FromStr for Protocol { "raydiumcpmm" => Ok(Protocol::RaydiumCpmm), "raydiumclmm" => Ok(Protocol::RaydiumClmm), "raydiumammv4" => Ok(Protocol::RaydiumAmmV4), + "meteoradamm_v2" => Ok(Protocol::MeteoraDammV2), _ => Err(anyhow!("Unsupported protocol: {}", s)), } }