sol_parser_sdk/
rpc_parser.rs

1//! RPC Transaction Parser
2//!
3//! 提供独立的 RPC 交易解析功能，不依赖 gRPC streaming
4//! 可以用于测试验证和离线分析
5
6use crate::core::events::DexEvent;
7use crate::grpc::types::EventTypeFilter;
8use crate::instr::read_pubkey_fast;
9use base64::{engine::general_purpose, Engine as _};
10use solana_client::rpc_client::RpcClient;
11use solana_client::rpc_config::RpcTransactionConfig;
12use solana_sdk::pubkey::Pubkey;
13use solana_sdk::signature::Signature;
14use solana_transaction_status::{
15    EncodedConfirmedTransactionWithStatusMeta, EncodedTransaction, UiTransactionEncoding,
16};
17use std::collections::HashMap;
18use std::str::FromStr;
19use yellowstone_grpc_proto::prelude::{
20    CompiledInstruction, InnerInstruction, InnerInstructions, Message, MessageAddressTableLookup,
21    MessageHeader, Transaction, TransactionStatusMeta,
22};
23
24/// Parse a transaction from RPC by signature
25///
26/// # Arguments
27/// * `rpc_client` - RPC client to fetch the transaction
28/// * `signature` - Transaction signature
29/// * `filter` - Optional event type filter
30///
31/// # Returns
32/// Vector of parsed DEX events
33///
34/// # Example
35/// ```no_run
36/// use solana_client::rpc_client::RpcClient;
37/// use solana_sdk::signature::Signature;
38/// use sol_parser_sdk::parse_transaction_from_rpc;
39/// use std::str::FromStr;
40///
41/// let client = RpcClient::new("https://api.mainnet-beta.solana.com".to_string());
42/// let sig = Signature::from_str("your-signature-here").unwrap();
43/// let events = parse_transaction_from_rpc(&client, &sig, None).unwrap();
44/// ```
45pub fn parse_transaction_from_rpc(
46    rpc_client: &RpcClient,
47    signature: &Signature,
48    filter: Option<&EventTypeFilter>,
49) -> Result<Vec<DexEvent>, ParseError> {
50    // Fetch transaction from RPC with V0 transaction support
51    let config = RpcTransactionConfig {
52        encoding: Some(UiTransactionEncoding::Base64),
53        commitment: None,
54        max_supported_transaction_version: Some(0),
55    };
56
57    let rpc_tx = rpc_client.get_transaction_with_config(signature, config).map_err(|e| {
58        let msg = e.to_string();
59        if msg.contains("invalid type: null") && msg.contains("EncodedConfirmedTransactionWithStatusMeta") {
60            ParseError::RpcError(format!(
61                "Transaction not found (RPC returned null). Common causes: 1) Transaction is too old and pruned (use an archive RPC). 2) Wrong network or invalid signature. Try SOLANA_RPC_URL with an archive endpoint (e.g. Helius, QuickNode) or a more recent tx. Original: {}",
62                msg
63            ))
64        } else {
65            ParseError::RpcError(msg)
66        }
67    })?;
68
69    parse_rpc_transaction(&rpc_tx, filter)
70}
71
72/// Parse a RPC transaction structure
73///
74/// # Arguments
75/// * `rpc_tx` - RPC transaction to parse
76/// * `filter` - Optional event type filter
77///
78/// # Returns
79/// Vector of parsed DEX events
80///
81/// # Example
82/// ```no_run
83/// use sol_parser_sdk::parse_rpc_transaction;
84///
85/// // Assuming you have an rpc_tx from RPC
86/// // let events = parse_rpc_transaction(&rpc_tx, None).unwrap();
87/// ```
88pub fn parse_rpc_transaction(
89    rpc_tx: &EncodedConfirmedTransactionWithStatusMeta,
90    filter: Option<&EventTypeFilter>,
91) -> Result<Vec<DexEvent>, ParseError> {
92    // Convert RPC format to gRPC format
93    let (grpc_meta, grpc_tx) = convert_rpc_to_grpc(rpc_tx)?;
94
95    // Extract metadata
96    let signature = extract_signature(rpc_tx)?;
97    let slot = rpc_tx.slot;
98    let block_time_us = rpc_tx.block_time.map(|t| t * 1_000_000);
99    let grpc_recv_us =
100        std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros()
101            as i64;
102
103    // Wrap grpc_tx in Option for reuse
104    let grpc_tx_opt = Some(grpc_tx);
105
106    let mut program_invokes: HashMap<Pubkey, Vec<(i32, i32)>> = HashMap::new();
107
108    if let Some(ref tx) = grpc_tx_opt {
109        if let Some(ref msg) = tx.message {
110            let keys_len = msg.account_keys.len();
111            let writable_len = grpc_meta.loaded_writable_addresses.len();
112            let get_key = |i: usize| -> Option<&Vec<u8>> {
113                if i < keys_len {
114                    msg.account_keys.get(i)
115                } else if i < keys_len + writable_len {
116                    grpc_meta.loaded_writable_addresses.get(i - keys_len)
117                } else {
118                    grpc_meta.loaded_readonly_addresses.get(i - keys_len - writable_len)
119                }
120            };
121
122            for (i, ix) in msg.instructions.iter().enumerate() {
123                let pid = get_key(ix.program_id_index as usize)
124                    .map_or(Pubkey::default(), |k| read_pubkey_fast(k));
125                if crate::grpc::program_ids::needs_invoke_context(&pid) {
126                    program_invokes.entry(pid).or_default().push((i as i32, -1));
127                }
128            }
129
130            for inner in &grpc_meta.inner_instructions {
131                let outer_idx = inner.index as usize;
132                for (j, inner_ix) in inner.instructions.iter().enumerate() {
133                    let pid = get_key(inner_ix.program_id_index as usize)
134                        .map_or(Pubkey::default(), |k| read_pubkey_fast(k));
135                    if crate::grpc::program_ids::needs_invoke_context(&pid) {
136                        program_invokes.entry(pid).or_default().push((outer_idx as i32, j as i32));
137                    }
138                }
139            }
140        }
141    }
142
143    let needs_pumpfun = filter.map(EventTypeFilter::includes_pumpfun).unwrap_or(true);
144    let is_created_buy = needs_pumpfun
145        && crate::logs::optimized_matcher::detect_pumpfun_create(&grpc_meta.log_messages);
146
147    // Parse instructions
148    let instr_events =
149        crate::grpc::instruction_parser::parse_instructions_enhanced_with_created_buy(
150            &grpc_meta,
151            &grpc_tx_opt,
152            signature,
153            slot,
154            0, // tx_idx
155            block_time_us,
156            grpc_recv_us,
157            filter,
158            is_created_buy,
159        );
160
161    // Parse logs (for protocols like PumpFun that emit events in logs)
162    struct ActiveProgram<'a> {
163        encoded: &'a str,
164        pubkey: Pubkey,
165    }
166
167    let mut active_program_stack: Vec<ActiveProgram<'_>> = Vec::with_capacity(8);
168    let mut log_events = Vec::new();
169
170    for log in &grpc_meta.log_messages {
171        if let Some((pid, depth)) = crate::logs::optimized_matcher::parse_invoke_info(log) {
172            let program_id = crate::grpc::program_ids::known_program_id(pid)
173                .or_else(|| Pubkey::from_str(pid).ok());
174            if let Some(pk) = program_id {
175                active_program_stack.truncate(depth.saturating_sub(1));
176                active_program_stack.push(ActiveProgram { encoded: pid, pubkey: pk });
177            }
178        }
179
180        if let Some(mut event) = crate::logs::parse_log_with_program_id(
181            log,
182            signature,
183            slot,
184            0, // tx_index
185            block_time_us,
186            grpc_recv_us,
187            filter,
188            is_created_buy,
189            None,
190            active_program_stack.last().map(|active| &active.pubkey),
191        ) {
192            // Fill account fields - use same function as gRPC parsing
193            crate::core::account_dispatcher::fill_accounts_with_owned_keys(
194                &mut event,
195                &grpc_meta,
196                &grpc_tx_opt,
197                &program_invokes,
198            );
199
200            // Fill additional data fields (e.g., PumpSwap is_pump_pool)
201            crate::core::common_filler::fill_data(
202                &mut event,
203                &grpc_meta,
204                &grpc_tx_opt,
205                &program_invokes,
206            );
207
208            log_events.push(event);
209        }
210
211        if let Some(pid) = crate::logs::optimized_matcher::parse_program_complete_info(log) {
212            if let Some(pos) = active_program_stack.iter().rposition(|active| active.encoded == pid)
213            {
214                active_program_stack.truncate(pos);
215            }
216        }
217    }
218
219    let mut events = merge_log_and_instruction_events(log_events, instr_events);
220    crate::grpc::transaction_meta::fill_recent_blockhash(&mut events, &grpc_tx_opt);
221    Ok(events)
222}
223
224fn merge_log_and_instruction_events(
225    log_events: Vec<DexEvent>,
226    instr_events: Vec<DexEvent>,
227) -> Vec<DexEvent> {
228    crate::grpc::log_instr_dedup::dedupe_log_instruction_events(log_events, instr_events)
229}
230
231/// Parse error types
232#[derive(Debug)]
233pub enum ParseError {
234    RpcError(String),
235    ConversionError(String),
236    MissingField(String),
237}
238
239impl std::fmt::Display for ParseError {
240    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
241        match self {
242            ParseError::RpcError(msg) => write!(f, "RPC error: {}", msg),
243            ParseError::ConversionError(msg) => write!(f, "Conversion error: {}", msg),
244            ParseError::MissingField(msg) => write!(f, "Missing field: {}", msg),
245        }
246    }
247}
248
249impl std::error::Error for ParseError {}
250
251// ============================================================================
252// Internal conversion functions
253// ============================================================================
254
255fn extract_signature(
256    rpc_tx: &EncodedConfirmedTransactionWithStatusMeta,
257) -> Result<Signature, ParseError> {
258    let ui_tx = &rpc_tx.transaction.transaction;
259
260    match ui_tx {
261        EncodedTransaction::Binary(data, _encoding) => {
262            let bytes = general_purpose::STANDARD.decode(data).map_err(|e| {
263                ParseError::ConversionError(format!("Failed to decode base64: {}", e))
264            })?;
265
266            let versioned_tx: solana_sdk::transaction::VersionedTransaction =
267                bincode::deserialize(&bytes).map_err(|e| {
268                    ParseError::ConversionError(format!("Failed to deserialize transaction: {}", e))
269                })?;
270
271            Ok(versioned_tx.signatures[0])
272        }
273        _ => Err(ParseError::ConversionError("Unsupported transaction encoding".to_string())),
274    }
275}
276
277pub fn convert_rpc_to_grpc(
278    rpc_tx: &EncodedConfirmedTransactionWithStatusMeta,
279) -> Result<(TransactionStatusMeta, Transaction), ParseError> {
280    let rpc_meta = rpc_tx
281        .transaction
282        .meta
283        .as_ref()
284        .ok_or_else(|| ParseError::MissingField("meta".to_string()))?;
285
286    // Convert meta
287    let mut grpc_meta = TransactionStatusMeta {
288        err: None,
289        fee: rpc_meta.fee,
290        pre_balances: rpc_meta.pre_balances.clone(),
291        post_balances: rpc_meta.post_balances.clone(),
292        inner_instructions: Vec::new(),
293        log_messages: {
294            let opt: Option<Vec<String>> = rpc_meta.log_messages.clone().into();
295            opt.unwrap_or_default()
296        },
297        pre_token_balances: Vec::new(),
298        post_token_balances: Vec::new(),
299        rewards: Vec::new(),
300        loaded_writable_addresses: {
301            let loaded_opt: Option<solana_transaction_status::UiLoadedAddresses> =
302                rpc_meta.loaded_addresses.clone().into();
303            loaded_opt
304                .map(|addrs| {
305                    addrs
306                        .writable
307                        .iter()
308                        .map(|pk_str| {
309                            use std::str::FromStr;
310                            solana_sdk::pubkey::Pubkey::from_str(pk_str)
311                                .unwrap()
312                                .to_bytes()
313                                .to_vec()
314                        })
315                        .collect()
316                })
317                .unwrap_or_default()
318        },
319        loaded_readonly_addresses: {
320            let loaded_opt: Option<solana_transaction_status::UiLoadedAddresses> =
321                rpc_meta.loaded_addresses.clone().into();
322            loaded_opt
323                .map(|addrs| {
324                    addrs
325                        .readonly
326                        .iter()
327                        .map(|pk_str| {
328                            use std::str::FromStr;
329                            solana_sdk::pubkey::Pubkey::from_str(pk_str)
330                                .unwrap()
331                                .to_bytes()
332                                .to_vec()
333                        })
334                        .collect()
335                })
336                .unwrap_or_default()
337        },
338        return_data: None,
339        compute_units_consumed: rpc_meta.compute_units_consumed.clone().into(),
340
341        inner_instructions_none: {
342            let opt: Option<Vec<_>> = rpc_meta.inner_instructions.clone().into();
343            opt.is_none()
344        },
345        log_messages_none: {
346            let opt: Option<Vec<String>> = rpc_meta.log_messages.clone().into();
347            opt.is_none()
348        },
349        return_data_none: {
350            let opt: Option<solana_transaction_status::UiTransactionReturnData> =
351                rpc_meta.return_data.clone().into();
352            opt.is_none()
353        },
354        cost_units: rpc_meta.compute_units_consumed.clone().into(),
355    };
356
357    // Convert inner instructions
358    let inner_instructions_opt: Option<Vec<_>> = rpc_meta.inner_instructions.clone().into();
359    if let Some(ref inner_instructions) = inner_instructions_opt {
360        for inner in inner_instructions {
361            let mut grpc_inner =
362                InnerInstructions { index: inner.index as u32, instructions: Vec::new() };
363
364            for ix in &inner.instructions {
365                if let solana_transaction_status::UiInstruction::Compiled(compiled) = ix {
366                    // Decode base58 data
367                    let data = bs58::decode(&compiled.data).into_vec().map_err(|e| {
368                        ParseError::ConversionError(format!(
369                            "Failed to decode instruction data: {}",
370                            e
371                        ))
372                    })?;
373
374                    grpc_inner.instructions.push(InnerInstruction {
375                        program_id_index: compiled.program_id_index as u32,
376                        accounts: compiled.accounts.clone(),
377                        data,
378                        stack_height: compiled.stack_height,
379                    });
380                }
381            }
382
383            grpc_meta.inner_instructions.push(grpc_inner);
384        }
385    }
386
387    // Convert transaction
388    let ui_tx = &rpc_tx.transaction.transaction;
389
390    let (message, signatures) = match ui_tx {
391        EncodedTransaction::Binary(data, _encoding) => {
392            // Decode base64
393            let bytes = general_purpose::STANDARD.decode(data).map_err(|e| {
394                ParseError::ConversionError(format!("Failed to decode base64: {}", e))
395            })?;
396
397            // Parse as versioned transaction
398            let versioned_tx: solana_sdk::transaction::VersionedTransaction =
399                bincode::deserialize(&bytes).map_err(|e| {
400                    ParseError::ConversionError(format!("Failed to deserialize transaction: {}", e))
401                })?;
402
403            let sigs: Vec<Vec<u8>> =
404                versioned_tx.signatures.iter().map(|s| s.as_ref().to_vec()).collect();
405
406            let message = match versioned_tx.message {
407                solana_sdk::message::VersionedMessage::Legacy(legacy_msg) => {
408                    convert_legacy_message(&legacy_msg)?
409                }
410                solana_sdk::message::VersionedMessage::V0(v0_msg) => convert_v0_message(&v0_msg)?,
411            };
412
413            (message, sigs)
414        }
415        EncodedTransaction::Json(_) => {
416            return Err(ParseError::ConversionError(
417                "JSON encoded transactions not supported yet".to_string(),
418            ));
419        }
420        _ => {
421            return Err(ParseError::ConversionError(
422                "Unsupported transaction encoding".to_string(),
423            ));
424        }
425    };
426
427    let grpc_tx = Transaction { signatures, message: Some(message) };
428
429    Ok((grpc_meta, grpc_tx))
430}
431
432fn convert_legacy_message(
433    msg: &solana_sdk::message::legacy::Message,
434) -> Result<Message, ParseError> {
435    let account_keys: Vec<Vec<u8>> =
436        msg.account_keys.iter().map(|k| k.to_bytes().to_vec()).collect();
437
438    let instructions: Vec<CompiledInstruction> = msg
439        .instructions
440        .iter()
441        .map(|ix| CompiledInstruction {
442            program_id_index: ix.program_id_index as u32,
443            accounts: ix.accounts.clone(),
444            data: ix.data.clone(),
445        })
446        .collect();
447
448    Ok(Message {
449        header: Some(MessageHeader {
450            num_required_signatures: msg.header.num_required_signatures as u32,
451            num_readonly_signed_accounts: msg.header.num_readonly_signed_accounts as u32,
452            num_readonly_unsigned_accounts: msg.header.num_readonly_unsigned_accounts as u32,
453        }),
454        account_keys,
455        recent_blockhash: msg.recent_blockhash.to_bytes().to_vec(),
456        instructions,
457        versioned: false,
458        address_table_lookups: Vec::new(),
459    })
460}
461
462fn convert_v0_message(msg: &solana_sdk::message::v0::Message) -> Result<Message, ParseError> {
463    let account_keys: Vec<Vec<u8>> =
464        msg.account_keys.iter().map(|k| k.to_bytes().to_vec()).collect();
465
466    let instructions: Vec<CompiledInstruction> = msg
467        .instructions
468        .iter()
469        .map(|ix| CompiledInstruction {
470            program_id_index: ix.program_id_index as u32,
471            accounts: ix.accounts.clone(),
472            data: ix.data.clone(),
473        })
474        .collect();
475
476    Ok(Message {
477        header: Some(MessageHeader {
478            num_required_signatures: msg.header.num_required_signatures as u32,
479            num_readonly_signed_accounts: msg.header.num_readonly_signed_accounts as u32,
480            num_readonly_unsigned_accounts: msg.header.num_readonly_unsigned_accounts as u32,
481        }),
482        account_keys,
483        recent_blockhash: msg.recent_blockhash.to_bytes().to_vec(),
484        instructions,
485        versioned: true,
486        address_table_lookups: msg
487            .address_table_lookups
488            .iter()
489            .map(|lookup| MessageAddressTableLookup {
490                account_key: lookup.account_key.to_bytes().to_vec(),
491                writable_indexes: lookup.writable_indexes.clone(),
492                readonly_indexes: lookup.readonly_indexes.clone(),
493            })
494            .collect(),
495    })
496}
497
498#[cfg(test)]
499mod tests {
500    use super::*;
501    use crate::core::events::{DexEvent, EventMetadata, PumpSwapCreatePoolEvent};
502    use solana_sdk::{pubkey::Pubkey, signature::Signature};
503
504    fn dummy_meta() -> EventMetadata {
505        EventMetadata {
506            signature: Signature::default(),
507            slot: 1,
508            tx_index: 0,
509            block_time_us: 0,
510            grpc_recv_us: 0,
511            recent_blockhash: None,
512        }
513    }
514
515    #[test]
516    fn rpc_merge_keeps_instruction_cashback_for_log_only_pumpswap_create_pool() {
517        let pool = Pubkey::new_unique();
518        let base_mint = Pubkey::new_unique();
519        let quote_mint = Pubkey::new_unique();
520
521        let log_create = PumpSwapCreatePoolEvent {
522            metadata: dummy_meta(),
523            pool,
524            base_mint,
525            quote_mint,
526            is_cashback_coin: false,
527            ..Default::default()
528        };
529        let ix_create = PumpSwapCreatePoolEvent {
530            metadata: dummy_meta(),
531            pool,
532            base_mint,
533            quote_mint,
534            is_cashback_coin: true,
535            ..Default::default()
536        };
537
538        let merged = merge_log_and_instruction_events(
539            vec![DexEvent::PumpSwapCreatePool(log_create)],
540            vec![DexEvent::PumpSwapCreatePool(ix_create)],
541        );
542
543        assert_eq!(merged.len(), 1);
544        match &merged[0] {
545            DexEvent::PumpSwapCreatePool(e) => assert!(e.is_cashback_coin),
546            other => panic!("expected PumpSwapCreatePool, got {other:?}"),
547        }
548    }
549}
sol_parser_sdk/rpc_parser.rs

sol_parser_sdk/
rpc_parser.rs
