feat(pumpswap): align pool account and events with cashback IDL

- Extend on-chain Pool layout (is_cashback_coin, 7-byte reserved, 244 bytes)
  with backward-compatible decode for 236-byte legacy accounts
- Decode Buy/Sell program logs manually: min_base_amount_out, ix_name,
  cashback fields; optional sell cashback when payload >= 368 bytes
- Set ix_name on buy / buy_exact_quote_in instructions; merge new fields
  in CPI log merger
- Add read_i64_le helper and pool layout unit test

Made-with: Cursor

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<i64> {
+    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<u64> {
     if data.len() < offset + 8 {

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;
             }
             _ => {}
         },

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<PumpSwapBuyEvent> {
-    if data.len() < PUMP_SWAP_BUY_EVENT_LOG_SIZE {
+    if data.len() < PUMP_SWAP_BUY_EVENT_LOG_MIN {
         return None;
     }
-    borsh::from_slice::<PumpSwapBuyEvent>(&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<PumpSwapSellEvent> {
-    if data.len() < PUMP_SWAP_SELL_EVENT_LOG_SIZE {
+    if data.len() < PUMP_SWAP_SELL_EVENT_LOG_MIN {
         return None;
     }
-    borsh::from_slice::<PumpSwapSellEvent>(&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()
+    })
 }
 
 /// 创建池子事件

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],

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<Pool> {
-    if data.len() < POOL_SIZE {
+    if data.len() >= POOL_BODY {
+        return borsh::from_slice::<Pool>(&data[..POOL_BODY]).ok();
+    }
+    if data.len() < POOL_BODY_LEGACY {
         return None;
     }
-    borsh::from_slice::<Pool>(&data[..POOL_SIZE]).ok()
+    let legacy = borsh::from_slice::<PoolLegacy>(&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<PoolLegacy> 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<DexEvent> {
     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<Pubkey> = (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]);
+    }
+}