gavindiaz/solana-streamer
1
0
Fork 0
mirror of https://github.com/0xfnzero/solana-streamer.git synced 2026-08-06 21:57:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge commit '7ff18fa8e7177318018fe1450335c420a1d1e925'

Browse Source
This commit is contained in:
ysq
2025-11-10 23:17:39 +08:00
parent 6e3d2c23d1 7ff18fa8e7
commit d8bd35efe9
15 changed files with 1135 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/grpc_example.rs
+11 -9
View File
@@ -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<dyn std::error::Error>> {
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![];
examples/parse_tx_events.rs
+1
View File
@@ -186,6 +186,7 @@ async fn get_single_transaction_details(signature_str: &str) -> Result<()> {
Protocol::PumpFun,
Protocol::RaydiumCpmm,
Protocol::RaydiumAmmV4,
Protocol::MeteoraDammV2,
];
// Create callback
src/streaming/common/constants.rs
+6
View File
@@ -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;
src/streaming/common/event_processor.rs
+22 -3
View File
@@ -18,12 +18,19 @@ fn create_metrics_callback(
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> Arc<dyn Fn(DexEvent) + Send + Sync> {
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);
}
src/streaming/common/metrics.rs
+33
View File
@@ -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) {
src/streaming/event_parser/common/types.rs
+13
View File
@@ -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"),
src/streaming/event_parser/core/dispatcher.rs
+24 -3
View File
@@ -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
}
}
}
}
src/streaming/event_parser/core/merger_event.rs
+130
View File
@@ -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;
}
_ => {}
},
_ => {}
}
src/streaming/event_parser/core/traits.rs
+18
View File
@@ -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,
src/streaming/event_parser/protocols/meteora_damm_v2/events.rs
+417
View File
@@ -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<DynamicFeeParameters>,
}
/// 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<Pubkey>,
#[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<Pubkey>,
#[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<Pubkey>,
}
/// 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<Pubkey>,
}
/// 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<MeteoraDammV2SwapEvent> {
if data.len() < METEORA_DAMM_V2_SWAP_EVENT_LOG_SIZE {
return None;
}
borsh::from_slice::<MeteoraDammV2SwapEvent>(&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<MeteoraDammV2InitializePoolEvent> {
borsh::from_slice::<MeteoraDammV2InitializePoolEvent>(&data).ok()
}
src/streaming/event_parser/protocols/meteora_damm_v2/mod.rs
+5
View File
@@ -0,0 +1,5 @@
pub mod events;
pub mod parser;
pub mod types;
pub use events::*;
src/streaming/event_parser/protocols/meteora_damm_v2/parser.rs
+444
View File
@@ -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<DexEvent> {
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<DexEvent> {
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<DexEvent> {
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<DexEvent> {
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<DexEvent> {
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<u64> = 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<u64>)
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<DexEvent> {
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<u64>)
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<DexEvent> {
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<u64>)
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<DexEvent> {
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<DexEvent> {
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
}
}
src/streaming/event_parser/protocols/meteora_damm_v2/types.rs
+2
View File
@@ -0,0 +1,2 @@
// 此文件用于定义 Meteora DAMM v2 的账户数据结构
// 暂时留空,后续如需要解析 Pool 账户状态时可以在这里添加
src/streaming/event_parser/protocols/mod.rs
+1
View File
@@ -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;
src/streaming/event_parser/protocols/types.rs
+8 -3
View File
@@ -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)),
}
}