sol-parser-sdk 0.7.2

//! Yellowstone `SubscribeUpdateTransaction` 单笔解析（logs ∥ instructions + 去重）。
//! 从 [`super::client`] 抽出，供 crate 内与下游 streamer 复用。

use smallvec::SmallVec;
use solana_sdk::pubkey::Pubkey;
use yellowstone_grpc_proto::prelude::{
    SubscribeUpdateTransaction, Transaction, TransactionStatusMeta,
};

use super::transaction_meta::try_yellowstone_signature;
use super::types::EventTypeFilter;
use crate::DexEvent;

const PROGRAM_DATA_PREFIX: &[u8] = b"Program data: ";

struct ActiveProgram<'a> {
    encoded: &'a str,
    pubkey: Pubkey,
}

/// 解析单笔 Yellowstone 交易更新（含 meta）：并行 logs + enhanced instructions，再 log/ix 去重合并。
#[inline]
pub fn parse_subscribe_update_transaction(
    tx: &SubscribeUpdateTransaction,
    grpc_recv_us: i64,
    block_us: Option<i64>,
    filter: Option<&EventTypeFilter>,
) -> Vec<DexEvent> {
    parse_transaction_core(tx, grpc_recv_us, block_us, filter)
}

#[inline]
pub(crate) fn parse_transaction_core(
    tx: &SubscribeUpdateTransaction,
    grpc_us: i64,
    block_us: Option<i64>,
    filter: Option<&EventTypeFilter>,
) -> Vec<DexEvent> {
    let Some(info) = &tx.transaction else { return Vec::new() };
    let Some(meta) = &info.meta else { return Vec::new() };

    let Some(sig) = try_yellowstone_signature(&info.signature) else {
        return Vec::new();
    };
    let slot = tx.slot;
    let idx = info.index;
    let needs_pumpfun = filter.map(EventTypeFilter::includes_pumpfun).unwrap_or(true);
    let is_created_buy =
        needs_pumpfun && crate::logs::optimized_matcher::detect_pumpfun_create(&meta.log_messages);

    let (log_events, instr_events) = rayon::join(
        || {
            parse_logs(
                meta,
                &info.transaction,
                &meta.log_messages,
                sig,
                slot,
                idx,
                block_us,
                grpc_us,
                filter,
                is_created_buy,
            )
        },
        || {
            parse_instructions(
                meta,
                &info.transaction,
                sig,
                slot,
                idx,
                block_us,
                grpc_us,
                filter,
                is_created_buy,
            )
        },
    );

    let mut events =
        crate::grpc::log_instr_dedup::dedupe_log_instruction_events(log_events, instr_events);
    crate::grpc::transaction_meta::fill_recent_blockhash(&mut events, &info.transaction);
    for event in &mut events {
        crate::core::common_filler::fill_token_balances(event, meta, &info.transaction);
    }
    if let Some(filter) = filter {
        events.into_iter().map(|e| filter.normalize_dex_event(e)).collect()
    } else {
        events
    }
}

/// 单笔交易解析：**顺序**执行 logs → instructions 再合并。
///
/// 与 [`parse_subscribe_update_transaction`]（内部 `rayon::join` 并行）算法一致，但避免工作窃取与线程池调度，
/// 在「单笔极低延迟」场景通常更快；适合嵌入 latency-sensitive 的订阅流水线。
#[inline]
pub fn parse_subscribe_update_transaction_low_latency(
    tx: &SubscribeUpdateTransaction,
    grpc_recv_us: i64,
    block_us: Option<i64>,
    filter: Option<&EventTypeFilter>,
) -> Vec<DexEvent> {
    parse_transaction_core_sequential(tx, grpc_recv_us, block_us, filter)
}

#[inline]
fn parse_transaction_core_sequential(
    tx: &SubscribeUpdateTransaction,
    grpc_us: i64,
    block_us: Option<i64>,
    filter: Option<&EventTypeFilter>,
) -> Vec<DexEvent> {
    let Some(info) = &tx.transaction else {
        return Vec::new();
    };
    let Some(meta) = &info.meta else {
        return Vec::new();
    };

    let Some(sig) = try_yellowstone_signature(&info.signature) else {
        return Vec::new();
    };
    let slot = tx.slot;
    let idx = info.index;
    let needs_pumpfun = filter.map(EventTypeFilter::includes_pumpfun).unwrap_or(true);
    let is_created_buy =
        needs_pumpfun && crate::logs::optimized_matcher::detect_pumpfun_create(&meta.log_messages);

    let log_events = parse_logs(
        meta,
        &info.transaction,
        &meta.log_messages,
        sig,
        slot,
        idx,
        block_us,
        grpc_us,
        filter,
        is_created_buy,
    );
    let instr_events = parse_instructions(
        meta,
        &info.transaction,
        sig,
        slot,
        idx,
        block_us,
        grpc_us,
        filter,
        is_created_buy,
    );

    let mut events =
        crate::grpc::log_instr_dedup::dedupe_log_instruction_events(log_events, instr_events);
    crate::grpc::transaction_meta::fill_recent_blockhash(&mut events, &info.transaction);
    for event in &mut events {
        crate::core::common_filler::fill_token_balances(event, meta, &info.transaction);
    }
    if let Some(filter) = filter {
        events.into_iter().map(|e| filter.normalize_dex_event(e)).collect()
    } else {
        events
    }
}

#[inline]
fn parse_logs(
    meta: &TransactionStatusMeta,
    transaction: &Option<Transaction>,
    logs: &[String],
    sig: solana_sdk::signature::Signature,
    slot: u64,
    tx_idx: u64,
    block_us: Option<i64>,
    grpc_us: i64,
    filter: Option<&EventTypeFilter>,
    is_created_buy: bool,
) -> Vec<DexEvent> {
    let mut outer_idx: i32 = -1;
    let mut inner_idx: i32 = -1;
    let mut invokes = crate::core::invoke_context::InvokeContext::default();
    let mut active_program_stack: SmallVec<[ActiveProgram<'_>; 8]> = SmallVec::new();
    let mut result = Vec::with_capacity(4);

    for log in logs {
        if log.as_bytes().starts_with(PROGRAM_DATA_PREFIX) {
            let current_program = active_program_stack.last().map(|active| &active.pubkey);
            if let Some(mut e) = crate::logs::parse_log_with_program_id(
                log,
                sig,
                slot,
                tx_idx,
                block_us,
                grpc_us,
                filter,
                is_created_buy,
                None,
                current_program,
            ) {
                crate::core::account_dispatcher::fill_accounts_with_invoke_context(
                    &mut e,
                    meta,
                    transaction,
                    &invokes,
                );
                crate::core::common_filler::fill_data_with_invoke_context(
                    &mut e,
                    meta,
                    transaction,
                    &invokes,
                );
                result.push(e);
            }
            continue;
        }

        if let Some((pid, depth)) = crate::logs::optimized_matcher::parse_invoke_info(log) {
            if depth == 1 {
                inner_idx = -1;
                outer_idx += 1;
            } else {
                inner_idx += 1;
            }
            let pk = crate::grpc::program_ids::known_program_id(pid).unwrap_or_default();
            active_program_stack.truncate(depth - 1);
            active_program_stack.push(ActiveProgram { encoded: pid, pubkey: pk });
            if crate::grpc::program_ids::needs_invoke_context(&pk) {
                invokes.push(pk, (outer_idx, inner_idx));
            }
            continue;
        }

        if let Some(pid) = crate::logs::optimized_matcher::parse_program_complete_info(log) {
            if let Some(pos) = active_program_stack.iter().rposition(|active| active.encoded == pid)
            {
                active_program_stack.truncate(pos);
            }
        }
    }
    result
}

#[inline]
fn parse_instructions(
    meta: &TransactionStatusMeta,
    transaction: &Option<Transaction>,
    sig: solana_sdk::signature::Signature,
    slot: u64,
    tx_idx: u64,
    block_us: Option<i64>,
    grpc_us: i64,
    filter: Option<&EventTypeFilter>,
    is_created_buy: bool,
) -> Vec<DexEvent> {
    crate::grpc::instruction_parser::parse_instructions_enhanced_with_created_buy(
        meta,
        transaction,
        sig,
        slot,
        tx_idx,
        block_us,
        grpc_us,
        filter,
        is_created_buy,
    )
}