diff --git a/src/streaming/event_parser/common/utils.rs b/src/streaming/event_parser/common/utils.rs index eb625c0..7ef340f 100755 --- a/src/streaming/event_parser/common/utils.rs +++ b/src/streaming/event_parser/common/utils.rs @@ -24,6 +24,15 @@ pub fn extract_program_log<'a>(log: &'a str, prefix: &str) -> Option<&'a str> { log.strip_prefix(prefix) } +/// 安全地从字节数组中读取 i64 +pub fn read_i64_le(data: &[u8], offset: usize) -> Option { + if data.len() < offset + 8 { + return None; + } + let bytes: [u8; 8] = data[offset..offset + 8].try_into().ok()?; + Some(i64::from_le_bytes(bytes)) +} + /// 安全地从字节数组中读取u64 pub fn read_u64_le(data: &[u8], offset: usize) -> Option { if data.len() < offset + 8 { diff --git a/src/streaming/event_parser/core/merger_event.rs b/src/streaming/event_parser/core/merger_event.rs index 11fc664..7b69f7e 100644 --- a/src/streaming/event_parser/core/merger_event.rs +++ b/src/streaming/event_parser/core/merger_event.rs @@ -200,6 +200,11 @@ pub fn merge(instruction_event: &mut DexEvent, cpi_log_event: DexEvent) { e.total_claimed_tokens = cpie.total_claimed_tokens; e.current_sol_volume = cpie.current_sol_volume; e.last_update_timestamp = cpie.last_update_timestamp; + e.min_base_amount_out = cpie.min_base_amount_out; + e.ix_name = cpie.ix_name.clone(); + e.cashback_fee_basis_points = cpie.cashback_fee_basis_points; + e.cashback = cpie.cashback; + e.is_pump_pool = cpie.is_pump_pool; } _ => {} }, @@ -228,6 +233,9 @@ pub fn merge(instruction_event: &mut DexEvent, cpi_log_event: DexEvent) { e.coin_creator = cpie.coin_creator; e.coin_creator_fee_basis_points = cpie.coin_creator_fee_basis_points; e.coin_creator_fee = cpie.coin_creator_fee; + e.cashback_fee_basis_points = cpie.cashback_fee_basis_points; + e.cashback = cpie.cashback; + e.is_pump_pool = cpie.is_pump_pool; } _ => {} }, diff --git a/src/streaming/event_parser/protocols/pumpswap/events.rs b/src/streaming/event_parser/protocols/pumpswap/events.rs index baaa83d..2694eca 100755 --- a/src/streaming/event_parser/protocols/pumpswap/events.rs +++ b/src/streaming/event_parser/protocols/pumpswap/events.rs @@ -2,6 +2,7 @@ use borsh::BorshDeserialize; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; +use crate::streaming::event_parser::common::utils::{read_i64_le, read_u32_le, read_u64_le}; use crate::streaming::event_parser::common::EventMetadata; use crate::streaming::event_parser::protocols::pumpswap::types::{GlobalConfig, Pool}; @@ -38,6 +39,14 @@ pub struct PumpSwapBuyEvent { pub total_claimed_tokens: u64, pub current_sol_volume: u64, pub last_update_timestamp: i64, + /// Minimum base out (IDL extension; also set on `buy_exact_quote_in` from ix args). + pub min_base_amount_out: u64, + /// Instruction name from event (`buy`, `buy_exact_quote_in`, …). + pub ix_name: String, + pub cashback_fee_basis_points: u64, + pub cashback: u64, + #[borsh(skip)] + pub is_pump_pool: bool, #[borsh(skip)] pub base_mint: Pubkey, #[borsh(skip)] @@ -56,13 +65,104 @@ pub struct PumpSwapBuyEvent { pub quote_token_program: Pubkey, } -pub const PUMP_SWAP_BUY_EVENT_LOG_SIZE: usize = 385; +/// Minimum bytes through `last_update_timestamp` (Anchor/Borsh layout, bool = 1 byte). +pub const PUMP_SWAP_BUY_EVENT_LOG_MIN: usize = 385; +/// Backwards-compatible name for the minimum BuyEvent payload size (legacy `borsh` slice length). +pub const PUMP_SWAP_BUY_EVENT_LOG_SIZE: usize = PUMP_SWAP_BUY_EVENT_LOG_MIN; pub fn pump_swap_buy_event_log_decode(data: &[u8]) -> Option { - if data.len() < PUMP_SWAP_BUY_EVENT_LOG_SIZE { + if data.len() < PUMP_SWAP_BUY_EVENT_LOG_MIN { return None; } - borsh::from_slice::(&data[..PUMP_SWAP_BUY_EVENT_LOG_SIZE]).ok() + let timestamp = read_i64_le(data, 0)?; + let base_amount_out = read_u64_le(data, 8)?; + let max_quote_amount_in = read_u64_le(data, 16)?; + let user_base_token_reserves = read_u64_le(data, 24)?; + let user_quote_token_reserves = read_u64_le(data, 32)?; + let pool_base_token_reserves = read_u64_le(data, 40)?; + let pool_quote_token_reserves = read_u64_le(data, 48)?; + let quote_amount_in = read_u64_le(data, 56)?; + let lp_fee_basis_points = read_u64_le(data, 64)?; + let lp_fee = read_u64_le(data, 72)?; + let protocol_fee_basis_points = read_u64_le(data, 80)?; + let protocol_fee = read_u64_le(data, 88)?; + let quote_amount_in_with_lp_fee = read_u64_le(data, 96)?; + let user_quote_amount_in = read_u64_le(data, 104)?; + let pool = Pubkey::new_from_array(data.get(112..144)?.try_into().ok()?); + let user = Pubkey::new_from_array(data.get(144..176)?.try_into().ok()?); + let user_base_token_account = Pubkey::new_from_array(data.get(176..208)?.try_into().ok()?); + let user_quote_token_account = Pubkey::new_from_array(data.get(208..240)?.try_into().ok()?); + let protocol_fee_recipient = Pubkey::new_from_array(data.get(240..272)?.try_into().ok()?); + let protocol_fee_recipient_token_account = Pubkey::new_from_array(data.get(272..304)?.try_into().ok()?); + let coin_creator = Pubkey::new_from_array(data.get(304..336)?.try_into().ok()?); + let coin_creator_fee_basis_points = read_u64_le(data, 336)?; + let coin_creator_fee = read_u64_le(data, 344)?; + let track_volume = *data.get(352)? != 0; + let total_unclaimed_tokens = read_u64_le(data, 353)?; + let total_claimed_tokens = read_u64_le(data, 361)?; + let current_sol_volume = read_u64_le(data, 369)?; + let last_update_timestamp = read_i64_le(data, 377)?; + + let mut offset = 385usize; + let min_base_amount_out = if data.len() >= offset + 8 { + let v = read_u64_le(data, offset)?; + offset += 8; + v + } else { + 0 + }; + + let ix_name = if data.len() >= offset + 4 { + let slen = read_u32_le(data, offset)? as usize; + let str_start = offset + 4; + if data.len() < str_start + slen { + return None; + } + offset = str_start + slen; + String::from_utf8_lossy(&data[str_start..offset]).into_owned() + } else { + String::new() + }; + + let cashback_fee_basis_points = read_u64_le(data, offset).unwrap_or(0); + let cashback = read_u64_le(data, offset + 8).unwrap_or(0); + + Some(PumpSwapBuyEvent { + metadata: EventMetadata::default(), + timestamp, + base_amount_out, + max_quote_amount_in, + user_base_token_reserves, + user_quote_token_reserves, + pool_base_token_reserves, + pool_quote_token_reserves, + quote_amount_in, + lp_fee_basis_points, + lp_fee, + protocol_fee_basis_points, + protocol_fee, + quote_amount_in_with_lp_fee, + user_quote_amount_in, + pool, + user, + user_base_token_account, + user_quote_token_account, + protocol_fee_recipient, + protocol_fee_recipient_token_account, + coin_creator, + coin_creator_fee_basis_points, + coin_creator_fee, + track_volume, + total_unclaimed_tokens, + total_claimed_tokens, + current_sol_volume, + last_update_timestamp, + min_base_amount_out, + ix_name, + cashback_fee_basis_points, + cashback, + ..Default::default() + }) } /// 卖出事件 @@ -93,6 +193,10 @@ pub struct PumpSwapSellEvent { pub coin_creator: Pubkey, pub coin_creator_fee_basis_points: u64, pub coin_creator_fee: u64, + pub cashback_fee_basis_points: u64, + pub cashback: u64, + #[borsh(skip)] + pub is_pump_pool: bool, #[borsh(skip)] pub base_mint: Pubkey, #[borsh(skip)] @@ -111,13 +215,73 @@ pub struct PumpSwapSellEvent { pub quote_token_program: Pubkey, } -pub const PUMP_SWAP_SELL_EVENT_LOG_SIZE: usize = 352; +pub const PUMP_SWAP_SELL_EVENT_LOG_MIN: usize = 352; +pub const PUMP_SWAP_SELL_EVENT_WITH_CASHBACK: usize = 368; +/// Backwards-compatible name for the pre-cashback SellEvent payload size. +pub const PUMP_SWAP_SELL_EVENT_LOG_SIZE: usize = PUMP_SWAP_SELL_EVENT_LOG_MIN; pub fn pump_swap_sell_event_log_decode(data: &[u8]) -> Option { - if data.len() < PUMP_SWAP_SELL_EVENT_LOG_SIZE { + if data.len() < PUMP_SWAP_SELL_EVENT_LOG_MIN { return None; } - borsh::from_slice::(&data[..PUMP_SWAP_SELL_EVENT_LOG_SIZE]).ok() + let timestamp = read_i64_le(data, 0)?; + let base_amount_in = read_u64_le(data, 8)?; + let min_quote_amount_out = read_u64_le(data, 16)?; + let user_base_token_reserves = read_u64_le(data, 24)?; + let user_quote_token_reserves = read_u64_le(data, 32)?; + let pool_base_token_reserves = read_u64_le(data, 40)?; + let pool_quote_token_reserves = read_u64_le(data, 48)?; + let quote_amount_out = read_u64_le(data, 56)?; + let lp_fee_basis_points = read_u64_le(data, 64)?; + let lp_fee = read_u64_le(data, 72)?; + let protocol_fee_basis_points = read_u64_le(data, 80)?; + let protocol_fee = read_u64_le(data, 88)?; + let quote_amount_out_without_lp_fee = read_u64_le(data, 96)?; + let user_quote_amount_out = read_u64_le(data, 104)?; + let pool = Pubkey::new_from_array(data.get(112..144)?.try_into().ok()?); + let user = Pubkey::new_from_array(data.get(144..176)?.try_into().ok()?); + let user_base_token_account = Pubkey::new_from_array(data.get(176..208)?.try_into().ok()?); + let user_quote_token_account = Pubkey::new_from_array(data.get(208..240)?.try_into().ok()?); + let protocol_fee_recipient = Pubkey::new_from_array(data.get(240..272)?.try_into().ok()?); + let protocol_fee_recipient_token_account = Pubkey::new_from_array(data.get(272..304)?.try_into().ok()?); + let coin_creator = Pubkey::new_from_array(data.get(304..336)?.try_into().ok()?); + let coin_creator_fee_basis_points = read_u64_le(data, 336)?; + let coin_creator_fee = read_u64_le(data, 344)?; + let (cashback_fee_basis_points, cashback) = if data.len() >= PUMP_SWAP_SELL_EVENT_WITH_CASHBACK { + (read_u64_le(data, 352)?, read_u64_le(data, 360)?) + } else { + (0, 0) + }; + + Some(PumpSwapSellEvent { + metadata: EventMetadata::default(), + timestamp, + base_amount_in, + min_quote_amount_out, + user_base_token_reserves, + user_quote_token_reserves, + pool_base_token_reserves, + pool_quote_token_reserves, + quote_amount_out, + lp_fee_basis_points, + lp_fee, + protocol_fee_basis_points, + protocol_fee, + quote_amount_out_without_lp_fee, + user_quote_amount_out, + pool, + user, + user_base_token_account, + user_quote_token_account, + protocol_fee_recipient, + protocol_fee_recipient_token_account, + coin_creator, + coin_creator_fee_basis_points, + coin_creator_fee, + cashback_fee_basis_points, + cashback, + ..Default::default() + }) } /// 创建池子事件 diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index 21400b0..8b7e2f3 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -155,6 +155,7 @@ fn parse_buy_instruction( metadata, base_amount_out, max_quote_amount_in, + ix_name: "buy".to_string(), pool: accounts[0], user: accounts[1], base_mint: accounts[3], @@ -195,6 +196,8 @@ fn parse_buy_exact_quote_in_instruction( metadata, base_amount_out: min_base_amount_out, max_quote_amount_in: spendable_quote_in, + min_base_amount_out, + ix_name: "buy_exact_quote_in".to_string(), pool: accounts[0], user: accounts[1], base_mint: accounts[3], diff --git a/src/streaming/event_parser/protocols/pumpswap/types.rs b/src/streaming/event_parser/protocols/pumpswap/types.rs index a26ff70..b2e2d94 100644 --- a/src/streaming/event_parser/protocols/pumpswap/types.rs +++ b/src/streaming/event_parser/protocols/pumpswap/types.rs @@ -72,24 +72,73 @@ pub struct Pool { pub lp_supply: u64, pub coin_creator: Pubkey, pub is_mayhem_mode: bool, + pub is_cashback_coin: bool, + /// On-chain reserved tail (7 bytes); keep in sync with pump_amm pool account layout. + pub reserved: [u8; 7], } -pub const POOL_SIZE: usize = 1 + 2 + 32 * 6 + 8 + 32 + 1; +/// Legacy pool account body (before `is_cashback_coin` + reserved). +pub const POOL_BODY_LEGACY: usize = 1 + 2 + 32 * 6 + 8 + 32 + 1; +/// Current pool account body including flags and reserved. +pub const POOL_BODY: usize = POOL_BODY_LEGACY + 1 + 7; + +pub const POOL_SIZE: usize = POOL_BODY; pub fn pool_decode(data: &[u8]) -> Option { - if data.len() < POOL_SIZE { + if data.len() >= POOL_BODY { + return borsh::from_slice::(&data[..POOL_BODY]).ok(); + } + if data.len() < POOL_BODY_LEGACY { return None; } - borsh::from_slice::(&data[..POOL_SIZE]).ok() + let legacy = borsh::from_slice::(&data[..POOL_BODY_LEGACY]).ok()?; + Some(legacy.into()) +} + +/// Pre-cashback on-chain layout (Borsh-compatible prefix of `Pool`). +#[derive(Clone, Debug, BorshDeserialize)] +struct PoolLegacy { + pub pool_bump: u8, + pub index: u16, + pub creator: Pubkey, + pub base_mint: Pubkey, + pub quote_mint: Pubkey, + pub lp_mint: Pubkey, + pub pool_base_token_account: Pubkey, + pub pool_quote_token_account: Pubkey, + pub lp_supply: u64, + pub coin_creator: Pubkey, + pub is_mayhem_mode: bool, +} + +impl From for Pool { + fn from(p: PoolLegacy) -> Self { + Pool { + pool_bump: p.pool_bump, + index: p.index, + creator: p.creator, + base_mint: p.base_mint, + quote_mint: p.quote_mint, + lp_mint: p.lp_mint, + pool_base_token_account: p.pool_base_token_account, + pool_quote_token_account: p.pool_quote_token_account, + lp_supply: p.lp_supply, + coin_creator: p.coin_creator, + is_mayhem_mode: p.is_mayhem_mode, + is_cashback_coin: false, + reserved: [0u8; 7], + } + } } pub fn pool_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Option { metadata.event_type = EventType::AccountPumpSwapPool; - if account.data.len() < POOL_SIZE + 8 { + let body = account.data.get(8..)?; + if body.len() < POOL_BODY_LEGACY { return None; } - if let Some(pool) = pool_decode(&account.data[8..POOL_SIZE + 8]) { + if let Some(pool) = pool_decode(body) { Some(DexEvent::PumpSwapPoolAccountEvent(PumpSwapPoolAccountEvent { metadata, pubkey: account.pubkey, @@ -97,9 +146,45 @@ pub fn pool_parser(account: &AccountPretty, mut metadata: EventMetadata) -> Opti lamports: account.lamports, owner: account.owner, rent_epoch: account.rent_epoch, - pool: pool, + pool, })) } else { None } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pool_decode_legacy_and_extended() { + let keys: Vec = (0..6).map(|_| Pubkey::new_unique()).collect(); + let coin = Pubkey::new_unique(); + let mut legacy = Vec::new(); + legacy.push(9u8); + legacy.extend_from_slice(&7u16.to_le_bytes()); + for k in &keys { + legacy.extend_from_slice(k.as_ref()); + } + legacy.extend_from_slice(&99u64.to_le_bytes()); + legacy.extend_from_slice(coin.as_ref()); + legacy.push(1u8); + + let p = pool_decode(&legacy).expect("legacy"); + assert_eq!(p.pool_bump, 9); + assert_eq!(p.index, 7); + assert_eq!(p.lp_supply, 99); + assert!(p.is_mayhem_mode); + assert!(!p.is_cashback_coin); + assert_eq!(p.reserved, [0u8; 7]); + + let mut ext = legacy.clone(); + ext.push(1u8); + ext.extend_from_slice(&[2u8, 3, 4, 5, 6, 7, 8]); + + let p2 = pool_decode(&ext).expect("extended"); + assert!(p2.is_cashback_coin); + assert_eq!(p2.reserved, [2, 3, 4, 5, 6, 7, 8]); + } +}