feat: bridge streamer to sol-parser-sdk

This commit is contained in:
0xfnzero
2026-05-15 05:09:29 +08:00
parent e29a2e4f5a
commit ab84249f4a
85 changed files with 6350 additions and 1362 deletions
+393 -5
View File
@@ -1,24 +1,412 @@
use crate::streaming::event_parser::common::{
types::EventType, ACCOUNT_EVENT_TYPES, BLOCK_EVENT_TYPES,
};
use crate::streaming::event_parser::DexEvent;
use sol_parser_sdk::grpc::types::EventType as SdkGrpcEventType;
use sol_parser_sdk::grpc::types::EventTypeFilter as SdkGrpcEventTypeFilter;
#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
pub struct EventTypeFilter {
pub include: Vec<EventType>,
pub exclude: Vec<EventType>,
}
impl EventTypeFilter {
#[inline]
pub fn all() -> Self {
Self::default()
}
#[inline]
pub fn include_only(include: impl Into<Vec<EventType>>) -> Self {
Self { include: include.into(), exclude: Vec::new() }
}
#[inline]
pub fn exclude_only(exclude: impl Into<Vec<EventType>>) -> Self {
Self { include: Vec::new(), exclude: exclude.into() }
}
#[inline]
pub fn include_exclude(
include: impl Into<Vec<EventType>>,
exclude: impl Into<Vec<EventType>>,
) -> Self {
Self { include: include.into(), exclude: exclude.into() }
}
pub fn include_transaction_event(&self) -> bool {
self.include
.iter()
.any(|event| !ACCOUNT_EVENT_TYPES.contains(event) && !BLOCK_EVENT_TYPES.contains(event))
if self.include.is_empty() && self.exclude.is_empty() {
return true;
}
if !self.include.is_empty() {
return self.include.iter().any(|event| {
!ACCOUNT_EVENT_TYPES.contains(event) && !BLOCK_EVENT_TYPES.contains(event)
});
}
// With exclude-only filters, keep the stream open and drop matching events locally.
!self.exclude.is_empty()
}
pub fn include_account_event(&self) -> bool {
self.include.iter().any(|event| ACCOUNT_EVENT_TYPES.contains(event))
if self.include.is_empty() && self.exclude.is_empty() {
return true;
}
if !self.include.is_empty() {
return self.include.iter().any(|event| ACCOUNT_EVENT_TYPES.contains(event));
}
!self.exclude.is_empty()
}
pub fn include_block_event(&self) -> bool {
self.include.iter().any(|event| BLOCK_EVENT_TYPES.contains(event))
if self.include.is_empty() && self.exclude.is_empty() {
return true;
}
if !self.include.is_empty() {
return self.include.iter().any(|event| BLOCK_EVENT_TYPES.contains(event));
}
!self.exclude.is_empty()
}
/// Apply `exclude` first, then `include`. Empty `include` means "allow all non-excluded types".
#[inline]
pub fn passes_event_type(&self, et: &EventType) -> bool {
if self.exclude.iter().any(|excluded| event_type_matches(excluded, et)) {
return false;
}
if self.include.is_empty() {
return true;
}
self.include.iter().any(|included| event_type_matches(included, et))
}
#[inline]
pub fn passes_for_event(&self, ev: &DexEvent) -> bool {
self.passes_event_type(&ev.metadata().event_type)
}
}
/// `None` means no filtering; `Some(f)` applies include/exclude event-type semantics.
#[inline]
pub(crate) fn passes_event_type_filter(filter: Option<&EventTypeFilter>, ev: &DexEvent) -> bool {
match filter {
None => true,
Some(f) => f.passes_for_event(ev),
}
}
/// Whether the local pass should parse ComputeBudget instructions.
#[inline]
pub(crate) fn filter_includes_compute_budget_types(filter: Option<&EventTypeFilter>) -> bool {
match filter {
None => true,
Some(f) => {
let limit_excluded = f.exclude.contains(&EventType::SetComputeUnitLimit);
let price_excluded = f.exclude.contains(&EventType::SetComputeUnitPrice);
if limit_excluded && price_excluded {
return false;
}
if f.include.is_empty() {
return true;
}
f.include.iter().any(|t| {
matches!(t, EventType::SetComputeUnitLimit | EventType::SetComputeUnitPrice)
})
}
}
}
/// Map the streamer filter to the SDK gRPC event-type filter used by
/// [`sol_parser_sdk::grpc::parse_subscribe_update_transaction_low_latency`].
///
/// - Empty include/exclude maps to `None`.
/// - Exclude-only maps to SDK `exclude_types` when at least one SDK type is known.
/// - Non-empty include maps to SDK `include_only`; streamer still applies exclude locally.
/// - If any included type cannot map to an SDK type, return `None` to avoid dropping it upstream.
pub(crate) fn build_sdk_parse_event_filter(
filter: Option<&EventTypeFilter>,
) -> Option<SdkGrpcEventTypeFilter> {
let f = filter?;
if !f.exclude.is_empty() && f.include.is_empty() {
let mut raw: Vec<SdkGrpcEventType> = Vec::new();
for et in &f.exclude {
raw.extend(streamer_event_to_sdk_grpc_types(et));
}
dedup_sdk_grpc_event_types(&mut raw);
return (!raw.is_empty()).then(|| SdkGrpcEventTypeFilter::exclude_types(raw));
}
if f.include.is_empty() {
return None;
}
let mut raw: Vec<SdkGrpcEventType> = Vec::new();
for et in &f.include {
let mapped = streamer_event_to_sdk_grpc_types(et);
if mapped.is_empty() {
return None;
}
raw.extend(mapped);
}
dedup_sdk_grpc_event_types(&mut raw);
Some(SdkGrpcEventTypeFilter::include_only(raw))
}
fn dedup_sdk_grpc_event_types(v: &mut Vec<SdkGrpcEventType>) {
let mut i = 0;
while i < v.len() {
if v[..i].contains(&v[i]) {
v.remove(i);
} else {
i += 1;
}
}
}
fn streamer_event_to_sdk_grpc_types(t: &EventType) -> Vec<SdkGrpcEventType> {
use EventType as St;
use SdkGrpcEventType as Sdk;
match t {
St::BlockMeta => vec![Sdk::BlockMeta],
St::PumpFunCreateToken => vec![Sdk::PumpFunCreate],
St::PumpFunCreateV2Token => vec![Sdk::PumpFunCreateV2],
St::PumpFunBuy => vec![Sdk::PumpFunBuy, Sdk::PumpFunBuyExactSolIn],
St::PumpFunBuyExactSolIn => vec![Sdk::PumpFunBuyExactSolIn],
St::PumpFunSell => vec![Sdk::PumpFunSell],
St::PumpFunMigrate => vec![Sdk::PumpFunMigrate],
St::PumpFeesCreateFeeSharingConfig => vec![Sdk::PumpFeesCreateFeeSharingConfig],
St::PumpFeesInitializeFeeConfig => vec![Sdk::PumpFeesInitializeFeeConfig],
St::PumpFeesResetFeeSharingConfig => vec![Sdk::PumpFeesResetFeeSharingConfig],
St::PumpFeesRevokeFeeSharingAuthority => vec![Sdk::PumpFeesRevokeFeeSharingAuthority],
St::PumpFeesTransferFeeSharingAuthority => vec![Sdk::PumpFeesTransferFeeSharingAuthority],
St::PumpFeesUpdateAdmin => vec![Sdk::PumpFeesUpdateAdmin],
St::PumpFeesUpdateFeeConfig => vec![Sdk::PumpFeesUpdateFeeConfig],
St::PumpFeesUpdateFeeShares => vec![Sdk::PumpFeesUpdateFeeShares],
St::PumpFeesUpsertFeeTiers => vec![Sdk::PumpFeesUpsertFeeTiers],
St::PumpFunMigrateBondingCurveCreator => vec![Sdk::PumpFunMigrateBondingCurveCreator],
St::PumpSwapBuy => vec![Sdk::PumpSwapBuy],
St::PumpSwapSell => vec![Sdk::PumpSwapSell],
St::PumpSwapCreatePool => vec![Sdk::PumpSwapCreatePool],
St::PumpSwapDeposit => vec![Sdk::PumpSwapLiquidityAdded],
St::PumpSwapWithdraw => vec![Sdk::PumpSwapLiquidityRemoved],
St::BonkBuyExactIn | St::BonkBuyExactOut | St::BonkSellExactIn | St::BonkSellExactOut => {
vec![Sdk::BonkTrade]
}
St::BonkInitialize | St::BonkInitializeV2 | St::BonkInitializeWithToken2022 => {
vec![Sdk::BonkPoolCreate]
}
St::BonkMigrateToAmm => vec![Sdk::BonkMigrateAmm],
St::MeteoraDammV2Swap | St::MeteoraDammV2Swap2 => vec![Sdk::MeteoraDammV2Swap],
St::MeteoraDammV2AddLiquidity => vec![Sdk::MeteoraDammV2AddLiquidity],
St::MeteoraDammV2RemoveLiquidity => vec![Sdk::MeteoraDammV2RemoveLiquidity],
St::MeteoraDammV2CreatePosition => vec![Sdk::MeteoraDammV2CreatePosition],
St::MeteoraDammV2ClosePosition => vec![Sdk::MeteoraDammV2ClosePosition],
St::TokenAccount => vec![Sdk::TokenAccount],
St::TokenInfo => vec![Sdk::TokenAccount],
St::NonceAccount => vec![Sdk::NonceAccount],
St::AccountPumpFunGlobal => vec![Sdk::AccountPumpFunGlobal],
St::AccountPumpSwapGlobalConfig => vec![Sdk::AccountPumpSwapGlobalConfig],
St::AccountPumpSwapPool => vec![Sdk::AccountPumpSwapPool],
_ => vec![],
}
}
#[inline]
fn event_type_matches(filter_type: &EventType, event_type: &EventType) -> bool {
filter_type == event_type
|| matches!(
(filter_type, event_type),
(EventType::PumpFunBuy, EventType::PumpFunBuyExactSolIn)
| (EventType::TokenAccount, EventType::TokenInfo)
)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::streaming::event_parser::common::types::ProtocolType;
use crate::streaming::event_parser::common::EventMetadata;
use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent;
use crate::streaming::event_parser::protocols::sol_parser_forward::events::ParserSdkErrorEvent;
fn mk_meta(et: EventType) -> EventMetadata {
EventMetadata::new(
Default::default(),
0,
0,
0,
ProtocolType::PumpFun,
et,
Default::default(),
0,
None,
0,
None,
None,
)
}
#[test]
fn passes_for_event_empty_include_is_all_true() {
let f = EventTypeFilter { include: vec![], ..Default::default() };
let ev = DexEvent::ParserSdkErrorEvent(ParserSdkErrorEvent {
metadata: mk_meta(EventType::ParserSdkError),
message: "x".into(),
});
assert!(f.passes_for_event(&ev));
}
#[test]
fn constructors_set_expected_filter_sides() {
assert_eq!(EventTypeFilter::all(), EventTypeFilter::default());
assert_eq!(
EventTypeFilter::include_only([EventType::PumpFunBuy]).include,
vec![EventType::PumpFunBuy]
);
assert_eq!(
EventTypeFilter::exclude_only([EventType::PumpFunSell]).exclude,
vec![EventType::PumpFunSell]
);
let both =
EventTypeFilter::include_exclude([EventType::PumpFunBuy], [EventType::PumpFunSell]);
assert_eq!(both.include, vec![EventType::PumpFunBuy]);
assert_eq!(both.exclude, vec![EventType::PumpFunSell]);
}
#[test]
fn empty_filter_includes_all_subscription_kinds() {
let f = EventTypeFilter::default();
assert!(f.include_transaction_event());
assert!(f.include_account_event());
assert!(f.include_block_event());
}
#[test]
fn exclude_blocks_even_when_include_empty() {
let f = EventTypeFilter {
include: vec![],
exclude: vec![EventType::PumpFunSell],
..Default::default()
};
let ev = DexEvent::ParserSdkErrorEvent(ParserSdkErrorEvent {
metadata: mk_meta(EventType::PumpFunSell),
message: "x".into(),
});
assert!(!f.passes_for_event(&ev));
}
#[test]
fn exclude_blocks_block_meta_when_stream_kept_open() {
let f = EventTypeFilter::exclude_only([EventType::BlockMeta]);
let ev = DexEvent::BlockMetaEvent(BlockMetaEvent {
metadata: mk_meta(EventType::BlockMeta),
slot: 0,
block_hash: String::new(),
});
assert!(f.include_block_event());
assert!(!f.passes_for_event(&ev));
}
#[test]
fn exclude_applies_after_include_allow() {
let f = EventTypeFilter {
include: vec![EventType::PumpFunBuy, EventType::PumpFunSell],
exclude: vec![EventType::PumpFunSell],
..Default::default()
};
let buy = DexEvent::ParserSdkErrorEvent(ParserSdkErrorEvent {
metadata: mk_meta(EventType::PumpFunBuy),
message: "x".into(),
});
let sell = DexEvent::ParserSdkErrorEvent(ParserSdkErrorEvent {
metadata: mk_meta(EventType::PumpFunSell),
message: "x".into(),
});
assert!(f.passes_for_event(&buy));
assert!(!f.passes_for_event(&sell));
}
#[test]
fn build_sdk_filter_exclude_only_pumpfun_sell() {
let f = EventTypeFilter {
include: vec![],
exclude: vec![EventType::PumpFunSell],
..Default::default()
};
let sdk_f = build_sdk_parse_event_filter(Some(&f)).expect("mapped");
assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunBuy));
assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunSell));
}
#[test]
fn build_sdk_filter_pumpfun_buy_only() {
let f = EventTypeFilter { include: vec![EventType::PumpFunBuy], ..Default::default() };
let sdk_f = build_sdk_parse_event_filter(Some(&f)).expect("mapped");
assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunBuy));
assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunBuyExactSolIn));
assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunSell));
}
#[test]
fn pumpfun_buy_filter_matches_exact_sol_in_for_backward_compat() {
let f = EventTypeFilter { include: vec![EventType::PumpFunBuy], ..Default::default() };
assert!(f.passes_event_type(&EventType::PumpFunBuy));
assert!(f.passes_event_type(&EventType::PumpFunBuyExactSolIn));
let f = EventTypeFilter {
include: vec![EventType::PumpFunBuyExactSolIn],
..Default::default()
};
assert!(!f.passes_event_type(&EventType::PumpFunBuy));
assert!(f.passes_event_type(&EventType::PumpFunBuyExactSolIn));
}
#[test]
fn token_account_filter_matches_token_info_for_backward_compat() {
let f = EventTypeFilter { include: vec![EventType::TokenAccount], ..Default::default() };
assert!(f.passes_event_type(&EventType::TokenAccount));
assert!(f.passes_event_type(&EventType::TokenInfo));
let f = EventTypeFilter { include: vec![EventType::TokenInfo], ..Default::default() };
assert!(!f.passes_event_type(&EventType::TokenAccount));
assert!(f.passes_event_type(&EventType::TokenInfo));
}
#[test]
fn build_sdk_filter_pumpfun_exact_sol_in_only() {
let f = EventTypeFilter {
include: vec![EventType::PumpFunBuyExactSolIn],
..Default::default()
};
let sdk_f = build_sdk_parse_event_filter(Some(&f)).expect("mapped");
assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunBuy));
assert!(sdk_f.should_include(SdkGrpcEventType::PumpFunBuyExactSolIn));
assert!(!sdk_f.should_include(SdkGrpcEventType::PumpFunSell));
}
#[test]
fn build_sdk_filter_token_info_maps_to_sdk_token_account() {
let f = EventTypeFilter { include: vec![EventType::TokenInfo], ..Default::default() };
let sdk_f = build_sdk_parse_event_filter(Some(&f)).expect("mapped");
assert!(sdk_f.should_include(SdkGrpcEventType::TokenAccount));
}
#[test]
fn build_sdk_filter_none_when_orca_in_mix() {
let f = EventTypeFilter {
include: vec![EventType::PumpFunBuy, EventType::OrcaWhirlpoolSwap],
..Default::default()
};
assert!(build_sdk_parse_event_filter(Some(&f)).is_none());
}
#[test]
fn build_sdk_filter_exclude_only_none_when_only_unmapped_types() {
let f = EventTypeFilter {
include: vec![],
exclude: vec![EventType::OrcaWhirlpoolSwap],
..Default::default()
};
assert!(build_sdk_parse_event_filter(Some(&f)).is_none());
}
}
@@ -25,12 +25,14 @@ impl HighPerformanceClock {
// 通过多次采样来减少初始化误差
let mut best_offset = i64::MAX;
let mut best_instant = Instant::now();
let mut best_timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as i64;
let mut best_timestamp =
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as i64;
// 进行3次采样,选择延迟最小的
for _ in 0..3 {
let instant_before = Instant::now();
let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as i64;
let timestamp =
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as i64;
let instant_after = Instant::now();
let sample_latency = instant_after.duration_since(instant_before).as_nanos() as i64;
@@ -113,8 +115,7 @@ impl Default for HighPerformanceClock {
}
/// 全局高性能时钟实例
static HIGH_PERF_CLOCK: std::sync::OnceLock<HighPerformanceClock> =
std::sync::OnceLock::new();
static HIGH_PERF_CLOCK: std::sync::OnceLock<HighPerformanceClock> = std::sync::OnceLock::new();
/// 获取全局高性能时钟实例(最简单的实现)
#[inline(always)]
+69 -12
View File
@@ -51,12 +51,24 @@ pub enum ProtocolType {
RaydiumClmm,
RaydiumAmmV4,
MeteoraDammV2,
OrcaWhirlpool,
MeteoraPools,
MeteoraDlmm,
Common,
}
/// Event type enumeration
#[derive(
Debug, Clone, Default, PartialEq, Eq, Hash, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
Debug,
Clone,
Default,
PartialEq,
Eq,
Hash,
Serialize,
Deserialize,
BorshSerialize,
BorshDeserialize,
)]
pub enum EventType {
// PumpSwap events
@@ -71,8 +83,19 @@ pub enum EventType {
PumpFunCreateToken,
PumpFunCreateV2Token,
PumpFunBuy,
PumpFunBuyExactSolIn,
PumpFunSell,
PumpFunMigrate,
PumpFeesCreateFeeSharingConfig,
PumpFeesInitializeFeeConfig,
PumpFeesResetFeeSharingConfig,
PumpFeesRevokeFeeSharingAuthority,
PumpFeesTransferFeeSharingAuthority,
PumpFeesUpdateAdmin,
PumpFeesUpdateFeeConfig,
PumpFeesUpdateFeeShares,
PumpFeesUpsertFeeTiers,
PumpFunMigrateBondingCurveCreator,
// Bonk events
BonkBuyExactIn,
@@ -101,6 +124,7 @@ pub enum EventType {
RaydiumClmmCreatePool,
RaydiumClmmOpenPositionWithToken22Nft,
RaydiumClmmOpenPositionV2,
RaydiumClmmCollectFee,
// Raydium AMM V4 events
RaydiumAmmV4SwapBaseIn,
@@ -116,6 +140,34 @@ pub enum EventType {
MeteoraDammV2InitializePool,
MeteoraDammV2InitializeCustomizablePool,
MeteoraDammV2InitializePoolWithDynamicConfig,
MeteoraDammV2CreatePosition,
MeteoraDammV2ClosePosition,
MeteoraDammV2AddLiquidity,
MeteoraDammV2RemoveLiquidity,
// Orca Whirlpool
OrcaWhirlpoolSwap,
OrcaWhirlpoolLiquidityIncreased,
OrcaWhirlpoolLiquidityDecreased,
OrcaWhirlpoolPoolInitialized,
// Meteora Pools
MeteoraPoolsSwap,
MeteoraPoolsAddLiquidity,
MeteoraPoolsRemoveLiquidity,
MeteoraPoolsBootstrapLiquidity,
MeteoraPoolsPoolCreated,
MeteoraPoolsSetPoolFees,
// Meteora DLMM
MeteoraDlmmSwap,
MeteoraDlmmAddLiquidity,
MeteoraDlmmRemoveLiquidity,
MeteoraDlmmInitializePool,
MeteoraDlmmInitializeBinArray,
MeteoraDlmmCreatePosition,
MeteoraDlmmClosePosition,
MeteoraDlmmClaimFee,
// Account events
AccountRaydiumAmmV4AmmInfo,
@@ -135,11 +187,13 @@ pub enum EventType {
NonceAccount,
TokenAccount,
TokenInfo,
// Common events
BlockMeta,
SetComputeUnitLimit,
SetComputeUnitPrice,
ParserSdkError,
Unknown,
}
@@ -159,6 +213,7 @@ pub const ACCOUNT_EVENT_TYPES: &[EventType] = &[
EventType::AccountRaydiumCpmmAmmConfig,
EventType::AccountRaydiumCpmmPoolState,
EventType::TokenAccount,
EventType::TokenInfo,
EventType::NonceAccount,
];
pub const BLOCK_EVENT_TYPES: &[EventType] = &[EventType::BlockMeta];
@@ -240,13 +295,10 @@ impl EventMetadata {
}
}
static SOL_MINT: std::sync::LazyLock<Pubkey> =
std::sync::LazyLock::new(spl_token::native_mint::id);
static SYSTEM_PROGRAMS: std::sync::LazyLock<[Pubkey; 3]> = std::sync::LazyLock::new(|| [
spl_token::id(),
spl_token_2022::id(),
solana_sdk::pubkey!("11111111111111111111111111111111"),
]);
static SOL_MINT: std::sync::LazyLock<Pubkey> = std::sync::LazyLock::new(spl_token::native_mint::id);
static SYSTEM_PROGRAMS: std::sync::LazyLock<[Pubkey; 3]> = std::sync::LazyLock::new(|| {
[spl_token::id(), spl_token_2022::id(), solana_sdk::pubkey!("11111111111111111111111111111111")]
});
/// Trait abstracting over different inner-instruction types for swap data extraction
pub trait InnerInstructionLike {
@@ -282,11 +334,16 @@ impl InnerInstructionLike for yellowstone_grpc_proto::prelude::InnerInstruction
}
/// Extract event context (mint/token account/vault info) from a DexEvent
fn extract_swap_context(event: &DexEvent) -> (
fn extract_swap_context(
event: &DexEvent,
) -> (
SwapData,
Option<Pubkey>, Option<Pubkey>,
Option<Pubkey>, Option<Pubkey>,
Option<Pubkey>, Option<Pubkey>,
Option<Pubkey>,
Option<Pubkey>,
Option<Pubkey>,
Option<Pubkey>,
Option<Pubkey>,
Option<Pubkey>,
) {
let mut swap_data = SwapData::default();
let mut from_mint: Option<Pubkey> = None;
@@ -92,7 +92,7 @@ impl AccountEventParser {
) {
// 应用事件类型过滤
if let Some(filter) = event_type_filter {
if filter.include.contains(&event.metadata().event_type) {
if filter.passes_event_type(&event.metadata().event_type) {
return Some(event);
}
// 不匹配过滤器,继续尝试其他解析方式
@@ -120,7 +120,7 @@ impl AccountEventParser {
// 尝试解析 Nonce 账户
if let Some(event) = Self::parse_nonce_account_event(&account, metadata.clone()) {
if let Some(filter) = event_type_filter {
if filter.include.contains(&event.metadata().event_type) {
if filter.passes_event_type(&event.metadata().event_type) {
return Some(event);
}
} else {
@@ -131,7 +131,7 @@ impl AccountEventParser {
// 尝试解析 Token 账户
if let Some(event) = Self::parse_token_account_event(&account, metadata) {
if let Some(filter) = event_type_filter {
if filter.include.contains(&event.metadata().event_type) {
if filter.passes_event_type(&event.metadata().event_type) {
return Some(event);
}
} else {
@@ -156,6 +156,8 @@ impl AccountEventParser {
// Spl Token Mint
if account.data.len() >= Mint::LEN {
if let Ok(mint) = Mint::unpack_from_slice(&account.data) {
let mut metadata = metadata.clone();
metadata.event_type = EventType::TokenInfo;
let mut event = TokenInfoEvent {
metadata,
pubkey,
@@ -174,6 +176,8 @@ impl AccountEventParser {
// Spl Token2022 Mint
if account.data.len() >= Account2022::LEN {
if let Ok(mint) = StateWithExtensions::<Mint2022>::unpack(&account.data) {
let mut metadata = metadata.clone();
metadata.event_type = EventType::TokenInfo;
let mut event = TokenInfoEvent {
metadata,
pubkey,
+52 -10
View File
@@ -1,6 +1,9 @@
//! 中心事件解析调度器
//! 事件路由入口(类比 sol-parser-sdk 的 `instr`,区分「原生字节解析」与「sdk 事件对齐」)。
//!
//! 根据协议类型路由到对应的解析函数,替代原有的静态 CONFIGS 数组架构
//! ## 代码去哪找
//! - **`protocols/<协议>/parser.rs`** — Yellowstone / shred 路径下的顶层与 inner 指令解析(手写)。
//! - **`streaming/parser_sdk_bridge/`** — `sol-parser-sdk::DexEvent` → streamer `DexEvent` 字段映射。
//! - **`protocols/sol_parser_forward/native.rs`** — Orca / Meteora Pools & DLMM:调用 sdk `instr` 后再走 bridge。
//!
//! ## 设计原则
//! - **单一职责**: 每个函数只负责一件事(路由、解析、合并分离)
@@ -11,9 +14,10 @@ use crate::streaming::event_parser::{
common::EventMetadata,
core::common_event_parser::{CommonEventParser, COMPUTE_BUDGET_PROGRAM_ID},
protocols::{
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,
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, sol_parser_forward,
},
DexEvent, Protocol,
};
@@ -54,9 +58,21 @@ impl EventDispatcher {
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2,
Protocol::OrcaWhirlpool => ProtocolType::OrcaWhirlpool,
Protocol::MeteoraPools => ProtocolType::MeteoraPools,
Protocol::MeteoraDlmm => ProtocolType::MeteoraDlmm,
};
match protocol {
Protocol::OrcaWhirlpool | Protocol::MeteoraPools | Protocol::MeteoraDlmm => {
sol_parser_forward::native::dispatch_instruction(
protocol.clone(),
instruction_discriminator,
instruction_data,
accounts,
&metadata,
)
}
Protocol::PumpFun => pumpfun::parse_pumpfun_instruction_data(
instruction_discriminator,
instruction_data,
@@ -129,9 +145,20 @@ impl EventDispatcher {
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2,
Protocol::OrcaWhirlpool => ProtocolType::OrcaWhirlpool,
Protocol::MeteoraPools => ProtocolType::MeteoraPools,
Protocol::MeteoraDlmm => ProtocolType::MeteoraDlmm,
};
match protocol {
Protocol::OrcaWhirlpool | Protocol::MeteoraPools | Protocol::MeteoraDlmm => {
sol_parser_forward::native::dispatch_inner_instruction(
protocol.clone(),
inner_instruction_discriminator,
inner_instruction_data,
&metadata,
)
}
Protocol::PumpFun => pumpfun::parse_pumpfun_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
@@ -162,11 +189,13 @@ 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,
),
Protocol::MeteoraDammV2 => {
meteora_damm_v2::parse_meteora_damm_v2_inner_instruction_data(
inner_instruction_discriminator,
inner_instruction_data,
metadata,
)
}
}
}
@@ -187,6 +216,12 @@ impl EventDispatcher {
Some(Protocol::RaydiumAmmV4)
} else if program_id == &meteora_damm_v2::METEORA_DAMM_V2_PROGRAM_ID {
Some(Protocol::MeteoraDammV2)
} else if program_id == &sol_parser_forward::ORCA_WHIRLPOOL_PROGRAM_ID {
Some(Protocol::OrcaWhirlpool)
} else if program_id == &sol_parser_forward::METEORA_POOLS_PROGRAM_ID {
Some(Protocol::MeteoraPools)
} else if program_id == &sol_parser_forward::METEORA_DLMM_PROGRAM_ID {
Some(Protocol::MeteoraDlmm)
} else {
None
}
@@ -225,6 +260,9 @@ impl EventDispatcher {
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,
Protocol::OrcaWhirlpool => sol_parser_forward::ORCA_WHIRLPOOL_PROGRAM_ID,
Protocol::MeteoraPools => sol_parser_forward::METEORA_POOLS_PROGRAM_ID,
Protocol::MeteoraDlmm => sol_parser_forward::METEORA_DLMM_PROGRAM_ID,
}
}
@@ -261,9 +299,13 @@ impl EventDispatcher {
Protocol::RaydiumClmm => ProtocolType::RaydiumClmm,
Protocol::RaydiumAmmV4 => ProtocolType::RaydiumAmmV4,
Protocol::MeteoraDammV2 => ProtocolType::MeteoraDammV2,
Protocol::OrcaWhirlpool => ProtocolType::OrcaWhirlpool,
Protocol::MeteoraPools => ProtocolType::MeteoraPools,
Protocol::MeteoraDlmm => ProtocolType::MeteoraDlmm,
};
match protocol {
Protocol::OrcaWhirlpool | Protocol::MeteoraPools | Protocol::MeteoraDlmm => None,
Protocol::PumpFun => {
pumpfun::parse_pumpfun_account_data(discriminator, account, metadata)
}
@@ -1,740 +0,0 @@
use crate::streaming::event_parser::{
DexEvent, Protocol, common::{
EventMetadata, filter::EventTypeFilter, high_performance_clock::elapsed_micros_since, parse_swap_data_from_next_grpc_instructions, parse_swap_data_from_next_instructions
}, core::{
dispatcher::EventDispatcher,
global_state::{
add_bonk_dev_address, add_dev_address, is_bonk_dev_address_in_signature,
is_dev_address_in_signature,
},
merger_event::merge,
}, protocols::raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID
};
use prost_types::Timestamp;
use solana_sdk::{
message::compiled_instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature,
transaction::VersionedTransaction,
};
use solana_transaction_status::InnerInstructions;
use std::sync::Arc;
use yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo;
pub struct EventParser {}
impl EventParser {
// ================================================================================================
// Public API - Entry Points
// ================================================================================================
/// Parse transaction from gRPC stream
///
/// This is the main entry point for parsing transactions received from gRPC streams.
/// It extracts account keys, inner instructions, and delegates to instruction parsing.
pub async fn parse_grpc_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
grpc_tx: SubscribeUpdateTransactionInfo,
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// 创建适配器回调,将所有权回调转换为引用回调
let adapter_callback = Arc::new(move |event: &DexEvent| {
callback(event.clone());
});
if let Some(transition) = grpc_tx.transaction {
if let Some(message) = &transition.message {
let mut address_table_lookups: Vec<Vec<u8>> = vec![];
let mut inner_instructions: Vec<
yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions,
> = vec![];
if let Some(meta) = grpc_tx.meta {
inner_instructions = meta.inner_instructions;
address_table_lookups.reserve(
meta.loaded_writable_addresses.len() + meta.loaded_readonly_addresses.len(),
);
let loaded_writable_addresses = meta.loaded_writable_addresses;
let loaded_readonly_addresses = meta.loaded_readonly_addresses;
address_table_lookups.extend(
loaded_writable_addresses.into_iter().chain(loaded_readonly_addresses),
);
}
let mut accounts_bytes: Vec<Vec<u8>> =
Vec::with_capacity(message.account_keys.len() + address_table_lookups.len());
accounts_bytes.extend_from_slice(&message.account_keys);
accounts_bytes.extend(address_table_lookups);
// 转换为 Pubkey
let accounts: Vec<Pubkey> = accounts_bytes
.iter()
.filter_map(|account| {
if account.len() == 32 {
Some(Pubkey::try_from(account.as_slice()).unwrap_or_default())
} else {
None
}
})
.collect();
// 解析指令事件
let instructions = &message.instructions;
let recent_blockhash = if message.recent_blockhash.len() != 32 {
None
} else {
Some(solana_sdk::bs58::encode(&message.recent_blockhash).into_string())
};
Self::parse_instruction_events_from_grpc_transaction(
protocols,
event_type_filter,
&instructions,
signature,
slot,
block_time,
recv_us,
&accounts,
&inner_instructions,
bot_wallet,
tx_index,
recent_blockhash,
adapter_callback,
)
.await?;
}
}
Ok(())
}
/// Parse transaction from VersionedTransaction
///
/// This is the entry point for parsing VersionedTransaction objects.
/// It's used when working with RPC responses or historical data.
#[allow(clippy::too_many_arguments)]
pub async fn parse_instruction_events_from_versioned_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
transaction: &VersionedTransaction,
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
accounts: &[Pubkey],
inner_instructions: &[InnerInstructions],
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// 创建适配器回调,将所有权回调转换为引用回调
let adapter_callback = Arc::new(move |event: &DexEvent| {
callback(event.clone());
});
// 获取交易的指令和账户
let compiled_instructions = transaction.message.instructions();
let recent_blockhash = Some(transaction.message.recent_blockhash().to_string());
let mut accounts: Vec<Pubkey> = accounts.to_vec();
// 检查交易中是否包含程序
let has_program = accounts
.iter()
.any(|account| Self::should_handle(protocols, event_type_filter, account));
if has_program {
// 解析每个指令
for (index, instruction) in compiled_instructions.iter().enumerate() {
if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
let program_id = *program_id; // 克隆程序ID,避免借用冲突
let inner_instructions = inner_instructions
.iter()
.find(|inner_instruction| inner_instruction.index == index as u8);
if Self::should_handle(protocols, event_type_filter, &program_id) {
let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
// 补齐accounts(使用Pubkey::default())
if *max_idx as usize >= accounts.len() {
accounts.resize(*max_idx as usize + 1, Pubkey::default());
}
Self::parse_events_from_instruction(
protocols,
event_type_filter,
instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
None,
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
inner_instructions,
adapter_callback.clone(),
)?;
}
// Immediately process inner instructions for correct ordering
if let Some(inner_instructions) = inner_instructions {
for (inner_index, inner_instruction) in
inner_instructions.instructions.iter().enumerate()
{
Self::parse_events_from_instruction(
protocols,
event_type_filter,
&inner_instruction.instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
Some(inner_index as i64),
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
Some(&inner_instructions),
adapter_callback.clone(),
)?;
}
}
}
}
}
Ok(())
}
// ================================================================================================
// gRPC Transaction Processing
// ================================================================================================
/// Parse instruction events from gRPC transaction format
///
/// Iterates through all instructions in a gRPC transaction, checks if they should be handled,
/// and delegates to instruction-level parsing for both outer and inner instructions.
#[allow(clippy::too_many_arguments)]
async fn parse_instruction_events_from_grpc_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
compiled_instructions: &[yellowstone_grpc_proto::prelude::CompiledInstruction],
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
accounts: &[Pubkey],
inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions],
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<String>,
callback: Arc<dyn for<'a> Fn(&'a DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// 获取交易的指令和账户
let mut accounts = accounts.to_vec();
// 检查交易中是否包含程序
let has_program = accounts
.iter()
.any(|account| Self::should_handle(protocols, event_type_filter, account));
if has_program {
// 解析每个指令
for (index, instruction) in compiled_instructions.iter().enumerate() {
if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
let program_id = *program_id; // 克隆程序ID,避免借用冲突
let inner_instructions = inner_instructions
.iter()
.find(|inner_instruction| inner_instruction.index == index as u32);
let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
// 补齐accounts(使用Pubkey::default())
if *max_idx as usize >= accounts.len() {
accounts.resize(*max_idx as usize + 1, Pubkey::default());
}
if Self::should_handle(protocols, event_type_filter, &program_id) {
Self::parse_events_from_grpc_instruction(
protocols,
event_type_filter,
instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
None,
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
inner_instructions,
callback.clone(),
)?;
}
// Immediately process inner instructions for correct ordering
if let Some(inner_instructions) = inner_instructions {
for (inner_index, inner_instruction) in
inner_instructions.instructions.iter().enumerate()
{
let inner_accounts = &inner_instruction.accounts;
let data = &inner_instruction.data;
let instruction =
yellowstone_grpc_proto::prelude::CompiledInstruction {
program_id_index: inner_instruction.program_id_index,
accounts: inner_accounts.to_vec(),
data: data.to_vec(),
};
Self::parse_events_from_grpc_instruction(
protocols,
event_type_filter,
&instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
inner_instructions.index as i64,
Some(inner_index as i64),
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
Some(&inner_instructions),
callback.clone(),
)?;
}
}
}
}
}
Ok(())
}
/// Parse events from gRPC instruction
///
/// Core parsing logic for a single gRPC instruction. Extracts discriminator, dispatches
/// to protocol-specific parsers, handles inner instructions, and processes swap data.
#[allow(clippy::too_many_arguments)]
fn parse_events_from_grpc_instruction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction,
accounts: &[Pubkey],
signature: Signature,
slot: u64,
block_time: Option<Timestamp>,
recv_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<&str>,
inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
callback: Arc<dyn for<'a> Fn(&'a DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// 添加边界检查以防止越界访问
let program_id_index = instruction.program_id_index as usize;
if program_id_index >= accounts.len() {
return Ok(());
}
let program_id = accounts[program_id_index];
if !Self::should_handle(protocols, event_type_filter, &program_id) {
return Ok(());
}
let is_cu_program = EventDispatcher::is_compute_budget_program(&program_id);
let disc_len = match program_id {
RAYDIUM_AMM_V4_PROGRAM_ID => 1,
_ => 8,
};
// 检查指令数据长度(至少需要 disc_len 字节的 discriminator
if !is_cu_program && instruction.data.len() < disc_len {
return Ok(());
}
// 创建元数据
let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000;
let metadata = EventMetadata::new(
signature,
slot,
timestamp.seconds,
block_time_ms,
Default::default(), // protocol will be set by dispatcher
Default::default(), // event_type will be set by dispatcher
program_id,
outer_index,
inner_index,
recv_us,
tx_index,
recent_blockhash.map(|s| s.to_string()),
);
if is_cu_program {
if let Some(event) = EventDispatcher::dispatch_compute_budget_instruction(
&instruction.data,
metadata.clone(),
) {
callback(&event);
}
return Ok(());
}
// 使用 EventDispatcher 匹配协议
let protocol = match EventDispatcher::match_protocol_by_program_id(&program_id) {
Some(p) => p,
None => return Ok(()),
};
// 提取 discriminator 和数据
let instruction_discriminator = &instruction.data[..disc_len];
let instruction_data = &instruction.data[disc_len..];
// 构建账户公钥列表
let account_pubkeys: Vec<Pubkey> = instruction
.accounts
.iter()
.filter_map(|&idx| accounts.get(idx as usize).copied())
.collect();
// 使用 EventDispatcher 解析 instruction 事件
let mut event = match EventDispatcher::dispatch_instruction(
protocol.clone(),
instruction_discriminator,
instruction_data,
&account_pubkeys,
metadata.clone(),
) {
Some(e) => e,
None => return Ok(()),
};
// 处理 inner instructions - 查找对应的 CPI log 进行 merge
// 当 inner_index 有值时,只查找索引大于当前 inner_index 的 CPI log
// 超低延迟:顺序执行,避免 thread::scope 的 spawn/join 开销
let mut inner_instruction_event: Option<DexEvent> = None;
if let Some(inner_instructions_ref) = inner_instructions {
let raw = inner_index.unwrap_or(-1);
let current_inner_idx = raw.clamp(i32::MIN as i64, i32::MAX as i64) as i32;
for (idx, inner_instruction) in inner_instructions_ref.instructions.iter().enumerate() {
if (idx as i32) <= current_inner_idx {
continue;
}
let inner_data = &inner_instruction.data;
if inner_data.len() < 16 {
continue;
}
let inner_discriminator = &inner_data[..16];
let inner_instruction_data = &inner_data[16..];
if let Some(inner_event) = EventDispatcher::dispatch_inner_instruction(
protocol.clone(),
inner_discriminator,
inner_instruction_data,
metadata.clone(),
) {
inner_instruction_event = Some(inner_event);
break;
}
}
if event.metadata().swap_data.is_none() {
if let Some(swap_data) = parse_swap_data_from_next_grpc_instructions(
&event,
inner_instructions_ref,
current_inner_idx,
accounts,
) {
event.metadata_mut().set_swap_data(swap_data);
}
}
}
// PumpFun MIGRATE: 有 CPI 时合并 log;无 CPI 时仍发出仅含指令数据的事件。
// 合并事件
if let Some(inner_instruction_event) = inner_instruction_event {
merge(&mut event, inner_instruction_event);
}
// 设置处理时间(使用高性能时钟)
event.metadata_mut().handle_us = elapsed_micros_since(recv_us);
event = Self::process_event(event, bot_wallet);
callback(&event);
Ok(())
}
// ================================================================================================
// Standard Instruction Processing
// ================================================================================================
/// Parse events from standard Solana instruction
///
/// Similar to gRPC instruction parsing but works with standard CompiledInstruction format.
/// Used when parsing VersionedTransaction or RPC data.
#[allow(clippy::too_many_arguments)]
fn parse_events_from_instruction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
instruction: &CompiledInstruction,
accounts: &[Pubkey],
signature: Signature,
slot: u64,
block_time: Option<Timestamp>,
recv_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<&str>,
inner_instructions: Option<&InnerInstructions>,
callback: Arc<dyn for<'a> Fn(&'a DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// 添加边界检查以防止越界访问
let program_id_index = instruction.program_id_index as usize;
if program_id_index >= accounts.len() {
return Ok(());
}
let program_id = accounts[program_id_index];
if !Self::should_handle(protocols, event_type_filter, &program_id) {
return Ok(());
}
let is_cu_program = EventDispatcher::is_compute_budget_program(&program_id);
let disc_len = match program_id {
RAYDIUM_AMM_V4_PROGRAM_ID => 1,
_ => 8,
};
// 检查指令数据长度(至少需要 8 字节的 discriminator
if !is_cu_program && instruction.data.len() < disc_len {
return Ok(());
}
// 创建元数据
let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000;
let metadata = EventMetadata::new(
signature,
slot,
timestamp.seconds,
block_time_ms,
Default::default(), // protocol will be set by dispatcher
Default::default(), // event_type will be set by dispatcher
program_id,
outer_index,
inner_index,
recv_us,
tx_index,
recent_blockhash.map(|s| s.to_string()),
);
if is_cu_program {
if let Some(event) = EventDispatcher::dispatch_compute_budget_instruction(
&instruction.data,
metadata.clone(),
) {
callback(&event);
}
return Ok(());
}
// 使用 EventDispatcher 匹配协议
let protocol = match EventDispatcher::match_protocol_by_program_id(&program_id) {
Some(p) => p,
None => return Ok(()),
};
// 提取 discriminator 和数据
let instruction_discriminator = &instruction.data[..disc_len];
let instruction_data = &instruction.data[disc_len..];
// 构建账户公钥列表
let account_pubkeys: Vec<Pubkey> = instruction
.accounts
.iter()
.filter_map(|&idx| accounts.get(idx as usize).copied())
.collect();
// 使用 EventDispatcher 解析 instruction 事件
let mut event = match EventDispatcher::dispatch_instruction(
protocol.clone(),
instruction_discriminator,
instruction_data,
&account_pubkeys,
metadata.clone(),
) {
Some(e) => e,
None => return Ok(()),
};
// 处理 inner instructions - 查找对应的 CPI log 进行 merge
// 当 inner_index 有值时,只查找索引大于当前 inner_index 的 CPI log
let mut inner_instruction_event: Option<DexEvent> = None;
if let Some(inner_instructions_ref) = inner_instructions {
let raw = inner_index.unwrap_or(-1);
let current_inner_idx = raw.clamp(i32::MIN as i64, i32::MAX as i64) as i32;
// 并行执行两个任务: 解析 inner event 和提取 swap_data
let (inner_event_result, swap_data_result) = std::thread::scope(|s| {
let inner_event_handle = s.spawn(|| {
for (idx, inner_instruction) in inner_instructions_ref.instructions.iter().enumerate() {
// 只查找索引大于当前 inner_index 的 CPI log
if (idx as i32) <= current_inner_idx {
continue;
}
let inner_data = &inner_instruction.instruction.data;
// 检查长度(需要 16 字节的 discriminator
if inner_data.len() < 16 {
continue;
}
let inner_discriminator = &inner_data[..16];
let inner_instruction_data = &inner_data[16..];
if let Some(inner_event) = EventDispatcher::dispatch_inner_instruction(
protocol.clone(),
inner_discriminator,
inner_instruction_data,
metadata.clone(),
) {
return Some(inner_event);
}
}
None
});
let swap_data_handle = s.spawn(|| {
if event.metadata().swap_data.is_none() {
parse_swap_data_from_next_instructions(
&event,
inner_instructions_ref,
current_inner_idx,
accounts,
)
} else {
None
}
});
// 等待两个任务完成
(inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap())
});
inner_instruction_event = inner_event_result;
if let Some(swap_data) = swap_data_result {
event.metadata_mut().set_swap_data(swap_data);
}
}
// PumpFun MIGRATE: 有 CPI 时合并 log;无 CPI(如 shred)仍发出仅含指令数据的事件。
// 合并事件
if let Some(inner_instruction_event) = inner_instruction_event {
merge(&mut event, inner_instruction_event);
}
// 设置处理时间(使用高性能时钟)
event.metadata_mut().handle_us = elapsed_micros_since(recv_us);
event = Self::process_event(event, bot_wallet);
callback(&event);
Ok(())
}
// ================================================================================================
// Helper Functions
// ================================================================================================
/// Check if instruction should be processed based on protocol filter
///
/// Determines whether a program_id matches any of the protocols we're interested in.
fn should_handle(
protocols: &[Protocol],
_event_type_filter: Option<&EventTypeFilter>,
program_id: &Pubkey,
) -> bool {
// 使用 EventDispatcher 来匹配协议
if let Some(protocol) = EventDispatcher::match_protocol_by_program_id(program_id) {
protocols.contains(&protocol)
} else if EventDispatcher::is_compute_budget_program(program_id) {
return true;
} else {
false
}
}
// ================================================================================================
// Event Post-Processing
// ================================================================================================
/// Process and enrich parsed event with additional context
///
/// Handles protocol-specific post-processing:
/// - PumpFun: Tracks dev addresses and marks dev trades
/// - PumpSwap: Fills swap data amounts
/// - Bonk: Tracks pool creators and marks dev trades
/// - General: Marks bot wallet trades
fn process_event(event: DexEvent, bot_wallet: Option<Pubkey>) -> DexEvent {
let signature = event.metadata().signature; // Copy the signature to avoid borrowing issues
match event {
DexEvent::PumpFunCreateTokenEvent(token_info) => {
add_dev_address(&signature, token_info.user);
if token_info.creator != Pubkey::default() && token_info.creator != token_info.user
{
add_dev_address(&signature, token_info.creator);
}
DexEvent::PumpFunCreateTokenEvent(token_info)
}
DexEvent::PumpFunCreateV2TokenEvent(token_info) => {
add_dev_address(&signature, token_info.user);
if token_info.creator != Pubkey::default() && token_info.creator != token_info.user
{
add_dev_address(&signature, token_info.creator);
}
DexEvent::PumpFunCreateV2TokenEvent(token_info)
}
DexEvent::PumpFunTradeEvent(mut trade_info) => {
trade_info.is_dev_create_token_trade =
is_dev_address_in_signature(&signature, &trade_info.user)
|| is_dev_address_in_signature(&signature, &trade_info.creator);
trade_info.is_bot = Some(trade_info.user) == bot_wallet;
if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() {
swap_data.from_amount = if trade_info.is_buy {
trade_info.sol_amount
} else {
trade_info.token_amount
};
swap_data.to_amount = if trade_info.is_buy {
trade_info.token_amount
} else {
trade_info.sol_amount
};
}
DexEvent::PumpFunTradeEvent(trade_info)
}
DexEvent::PumpSwapBuyEvent(mut trade_info) => {
if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() {
swap_data.from_amount = trade_info.user_quote_amount_in;
swap_data.to_amount = trade_info.base_amount_out;
}
DexEvent::PumpSwapBuyEvent(trade_info)
}
DexEvent::PumpSwapSellEvent(mut trade_info) => {
if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() {
swap_data.from_amount = trade_info.base_amount_in;
swap_data.to_amount = trade_info.user_quote_amount_out;
}
DexEvent::PumpSwapSellEvent(trade_info)
}
DexEvent::BonkPoolCreateEvent(pool_info) => {
add_bonk_dev_address(&signature, pool_info.creator);
DexEvent::BonkPoolCreateEvent(pool_info)
}
DexEvent::BonkTradeEvent(mut trade_info) => {
trade_info.is_dev_create_token_trade =
is_bonk_dev_address_in_signature(&signature, &trade_info.payer);
trade_info.is_bot = Some(trade_info.payer) == bot_wallet;
DexEvent::BonkTradeEvent(trade_info)
}
_ => event,
}
}
}
@@ -0,0 +1,195 @@
//! Single Solana [`CompiledInstruction`] parsing with local inner merge and swap enrichment.
use crate::streaming::event_parser::{
common::{
filter::{passes_event_type_filter, EventTypeFilter},
high_performance_clock::elapsed_micros_since,
parse_swap_data_from_next_instructions, EventMetadata,
},
core::{dispatcher::EventDispatcher, merger_event::merge},
protocols::{
raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID,
sol_parser_forward::METEORA_DLMM_PROGRAM_ID,
},
DexEvent, Protocol,
};
use prost_types::Timestamp;
use solana_sdk::{
message::compiled_instruction::CompiledInstruction, pubkey::Pubkey, signature::Signature,
};
use solana_transaction_status::InnerInstructions;
use std::sync::Arc;
pub(super) fn parse_events_from_instruction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
instruction: &CompiledInstruction,
accounts: &[Pubkey],
signature: Signature,
slot: u64,
block_time: Option<Timestamp>,
recv_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<&str>,
inner_instructions: Option<&InnerInstructions>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// Bounds check before reading the program id index.
let program_id_index = instruction.program_id_index as usize;
if program_id_index >= accounts.len() {
return Ok(());
}
let program_id = accounts[program_id_index];
if !super::super::helpers::should_handle(protocols, event_type_filter, &program_id) {
return Ok(());
}
let is_cu_program = EventDispatcher::is_compute_budget_program(&program_id);
let disc_len = match program_id {
RAYDIUM_AMM_V4_PROGRAM_ID | METEORA_DLMM_PROGRAM_ID => 1,
_ => 8,
};
// Non-ComputeBudget instructions need at least a discriminator.
if !is_cu_program && instruction.data.len() < disc_len {
return Ok(());
}
// Build streamer metadata.
let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000;
let metadata = EventMetadata::new(
signature,
slot,
timestamp.seconds,
block_time_ms,
Default::default(), // protocol will be set by dispatcher
Default::default(), // event_type will be set by dispatcher
program_id,
outer_index,
inner_index,
recv_us,
tx_index,
recent_blockhash.map(|s| s.to_string()),
);
if is_cu_program {
if let Some(event) = EventDispatcher::dispatch_compute_budget_instruction(
&instruction.data,
metadata.clone(),
) {
if passes_event_type_filter(event_type_filter, &event) {
callback(event);
}
}
return Ok(());
}
// Match the parser protocol.
let protocol = match EventDispatcher::match_protocol_by_program_id(&program_id) {
Some(p) => p,
None => return Ok(()),
};
// Split discriminator and instruction payload.
let instruction_discriminator = &instruction.data[..disc_len];
let instruction_data = &instruction.data[disc_len..];
// Build the account pubkey list for this instruction.
let account_pubkeys: Vec<Pubkey> = instruction
.accounts
.iter()
.filter_map(|&idx| accounts.get(idx as usize).copied())
.collect();
// Parse the instruction event.
let mut event = match EventDispatcher::dispatch_instruction(
protocol.clone(),
instruction_discriminator,
instruction_data,
&account_pubkeys,
metadata.clone(),
) {
Some(e) => e,
None => return Ok(()),
};
// Find the next CPI log for merge.
let mut inner_instruction_event: Option<DexEvent> = None;
if let Some(inner_instructions_ref) = inner_instructions {
let raw = inner_index.unwrap_or(-1);
let current_inner_idx = raw.clamp(i32::MIN as i64, i32::MAX as i64) as i32;
// Parse the inner event and swap data in parallel on the compiled local path.
let (inner_event_result, swap_data_result) = std::thread::scope(|s| {
let inner_event_handle = s.spawn(|| {
for (idx, inner_instruction) in
inner_instructions_ref.instructions.iter().enumerate()
{
// Only inspect CPI logs after the current inner instruction.
if (idx as i32) <= current_inner_idx {
continue;
}
let inner_data = &inner_instruction.instruction.data;
// Inner CPI logs use a 16-byte discriminator.
if inner_data.len() < 16 {
continue;
}
let inner_discriminator = &inner_data[..16];
let inner_instruction_data = &inner_data[16..];
if let Some(inner_event) = EventDispatcher::dispatch_inner_instruction(
protocol.clone(),
inner_discriminator,
inner_instruction_data,
metadata.clone(),
) {
return Some(inner_event);
}
}
None
});
let swap_data_handle = s.spawn(|| {
if event.metadata().swap_data.is_none() {
parse_swap_data_from_next_instructions(
&event,
inner_instructions_ref,
current_inner_idx,
accounts,
)
} else {
None
}
});
// Wait for both local tasks.
(inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap())
});
inner_instruction_event = inner_event_result;
if let Some(swap_data) = swap_data_result {
event.metadata_mut().set_swap_data(swap_data);
}
}
// PumpFun MIGRATE emits instruction-only data when no CPI log exists.
// Merge CPI details into the outer event.
if let Some(inner_instruction_event) = inner_instruction_event {
merge(&mut event, inner_instruction_event);
}
// Stamp handling latency using the high-performance clock.
event.metadata_mut().handle_us = elapsed_micros_since(recv_us);
event = super::super::helpers::process_event(event, bot_wallet);
if passes_event_type_filter(event_type_filter, &event) {
callback(event);
}
Ok(())
}
@@ -0,0 +1,87 @@
//! Sequential top-level and inner ix traversal for [`VersionedTransaction`].
use crate::streaming::event_parser::{common::filter::EventTypeFilter, DexEvent, Protocol};
use prost_types::Timestamp;
use solana_sdk::{pubkey::Pubkey, signature::Signature, transaction::VersionedTransaction};
use solana_transaction_status::InnerInstructions;
use std::sync::Arc;
pub(crate) async fn parse_instruction_events_from_versioned_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
transaction: &VersionedTransaction,
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
accounts: &[Pubkey],
inner_instructions: &[InnerInstructions],
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
let compiled_instructions = transaction.message.instructions();
let recent_blockhash = Some(transaction.message.recent_blockhash().to_string());
let mut accounts: Vec<Pubkey> = accounts.to_vec();
let has_program = accounts
.iter()
.any(|account| super::super::helpers::should_handle(protocols, event_type_filter, account));
if has_program {
// Parse each instruction in order.
for (index, instruction) in compiled_instructions.iter().enumerate() {
if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
let program_id = *program_id;
let inner_instructions = inner_instructions
.iter()
.find(|inner_instruction| inner_instruction.index == index as u8);
if super::super::helpers::should_handle(protocols, event_type_filter, &program_id) {
let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
if *max_idx as usize >= accounts.len() {
accounts.resize(*max_idx as usize + 1, Pubkey::default());
}
super::compiled_instruction::parse_events_from_instruction(
protocols,
event_type_filter,
instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
None,
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
inner_instructions,
callback.clone(),
)?;
}
// Immediately process inner instructions for correct ordering
if let Some(inner_instructions) = inner_instructions {
for (inner_index, inner_instruction) in
inner_instructions.instructions.iter().enumerate()
{
super::compiled_instruction::parse_events_from_instruction(
protocols,
event_type_filter,
&inner_instruction.instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
Some(inner_index as i64),
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
Some(&inner_instructions),
callback.clone(),
)?;
}
}
}
}
}
Ok(())
}
@@ -0,0 +1,11 @@
//! Standard [`VersionedTransaction`] / [`CompiledInstruction`] path for RPC and replay.
//!
//! | Module | Responsibility |
//! |--------|------|
//! | [`compiled_transaction`] | top-level ix loop |
//! | [`compiled_instruction`] | single Solana `CompiledInstruction` |
mod compiled_instruction;
mod compiled_transaction;
pub(super) use compiled_transaction::parse_instruction_events_from_versioned_transaction;
@@ -0,0 +1,171 @@
//! Single Yellowstone [`CompiledInstruction`] parsing: dispatch, inner merge, swap enrichment.
use crate::streaming::event_parser::{
common::{
filter::{passes_event_type_filter, EventTypeFilter},
high_performance_clock::elapsed_micros_since,
parse_swap_data_from_next_grpc_instructions, EventMetadata,
},
core::{dispatcher::EventDispatcher, merger_event::merge},
protocols::{
raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID,
sol_parser_forward::METEORA_DLMM_PROGRAM_ID,
},
DexEvent, Protocol,
};
use prost_types::Timestamp;
use solana_sdk::{pubkey::Pubkey, signature::Signature};
use std::sync::Arc;
pub(super) fn parse_events_from_grpc_instruction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction,
accounts: &[Pubkey],
signature: Signature,
slot: u64,
block_time: Option<Timestamp>,
recv_us: i64,
outer_index: i64,
inner_index: Option<i64>,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<&str>,
inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
// Bounds check before reading the program id index.
let program_id_index = instruction.program_id_index as usize;
if program_id_index >= accounts.len() {
return Ok(());
}
let program_id = accounts[program_id_index];
if !super::super::helpers::should_handle(protocols, event_type_filter, &program_id) {
return Ok(());
}
let is_cu_program = EventDispatcher::is_compute_budget_program(&program_id);
let disc_len = match program_id {
RAYDIUM_AMM_V4_PROGRAM_ID | METEORA_DLMM_PROGRAM_ID => 1,
_ => 8,
};
// Non-ComputeBudget instructions need at least a discriminator.
if !is_cu_program && instruction.data.len() < disc_len {
return Ok(());
}
// Build streamer metadata.
let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 });
let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000;
let metadata = EventMetadata::new(
signature,
slot,
timestamp.seconds,
block_time_ms,
Default::default(), // protocol will be set by dispatcher
Default::default(), // event_type will be set by dispatcher
program_id,
outer_index,
inner_index,
recv_us,
tx_index,
recent_blockhash.map(|s| s.to_string()),
);
if is_cu_program {
if let Some(event) = EventDispatcher::dispatch_compute_budget_instruction(
&instruction.data,
metadata.clone(),
) {
if passes_event_type_filter(event_type_filter, &event) {
callback(event);
}
}
return Ok(());
}
// Match the parser protocol.
let protocol = match EventDispatcher::match_protocol_by_program_id(&program_id) {
Some(p) => p,
None => return Ok(()),
};
// Split discriminator and instruction payload.
let instruction_discriminator = &instruction.data[..disc_len];
let instruction_data = &instruction.data[disc_len..];
// Build the account pubkey list for this instruction.
let account_pubkeys: Vec<Pubkey> = instruction
.accounts
.iter()
.filter_map(|&idx| accounts.get(idx as usize).copied())
.collect();
// Parse the instruction event.
let mut event = match EventDispatcher::dispatch_instruction(
protocol.clone(),
instruction_discriminator,
instruction_data,
&account_pubkeys,
metadata.clone(),
) {
Some(e) => e,
None => return Ok(()),
};
// Find the next CPI log for merge. The gRPC hot path stays sequential to avoid
// thread::scope spawn/join overhead.
let mut inner_instruction_event: Option<DexEvent> = None;
if let Some(inner_instructions_ref) = inner_instructions {
let raw = inner_index.unwrap_or(-1);
let current_inner_idx = raw.clamp(i32::MIN as i64, i32::MAX as i64) as i32;
for (idx, inner_instruction) in inner_instructions_ref.instructions.iter().enumerate() {
if (idx as i32) <= current_inner_idx {
continue;
}
let inner_data = &inner_instruction.data;
if inner_data.len() < 16 {
continue;
}
let inner_discriminator = &inner_data[..16];
let inner_instruction_data = &inner_data[16..];
if let Some(inner_event) = EventDispatcher::dispatch_inner_instruction(
protocol.clone(),
inner_discriminator,
inner_instruction_data,
metadata.clone(),
) {
inner_instruction_event = Some(inner_event);
break;
}
}
if event.metadata().swap_data.is_none() {
if let Some(swap_data) = parse_swap_data_from_next_grpc_instructions(
&event,
inner_instructions_ref,
current_inner_idx,
accounts,
) {
event.metadata_mut().set_swap_data(swap_data);
}
}
}
// PumpFun MIGRATE emits instruction-only data when no CPI log exists.
// Merge CPI details into the outer event.
if let Some(inner_instruction_event) = inner_instruction_event {
merge(&mut event, inner_instruction_event);
}
// Stamp handling latency using the high-performance clock.
event.metadata_mut().handle_us = elapsed_micros_since(recv_us);
event = super::super::helpers::process_event(event, bot_wallet);
if passes_event_type_filter(event_type_filter, &event) {
callback(event);
}
Ok(())
}
@@ -0,0 +1,9 @@
//! Top-level ix parsing strategy for the gRPC path.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum GrpcIxParseMode {
/// Parse all subscribed instructions, used when transaction meta is missing.
Full,
/// Parse only ComputeBudget locally; DEX events come from sol-parser-sdk.
ComputeBudgetOnly,
}
@@ -0,0 +1,265 @@
//! Yellowstone transaction parsing: SDK-first DEX parsing plus optional local ix fallback.
//!
//! When `transaction.meta` is missing, the SDK low-latency parser cannot see logs / complete inner
//! instruction context and may return no events. In that case streamer uses the local full ix path.
//!
//! When meta exists, DEX events come from `sol-parser-sdk`; the local second pass is limited to
//! ComputeBudget events when the user asked for them.
use crate::streaming::event_parser::{
common::{
filter::{
build_sdk_parse_event_filter, filter_includes_compute_budget_types, EventTypeFilter,
},
high_performance_clock::elapsed_micros_since,
},
core::dispatcher::EventDispatcher,
DexEvent, Protocol,
};
use prost_types::Timestamp;
use sol_parser_sdk::grpc::parse_subscribe_update_transaction_low_latency;
use solana_sdk::{pubkey::Pubkey, signature::Signature};
use std::sync::Arc;
use yellowstone_grpc_proto::geyser::{SubscribeUpdateTransaction, SubscribeUpdateTransactionInfo};
use super::grpc_ix_mode::GrpcIxParseMode;
pub(crate) async fn parse_grpc_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
mut grpc_tx: SubscribeUpdateTransactionInfo,
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
let slot_u = slot.unwrap_or(0);
let block_us_micro = block_time.map(|t| t.seconds * 1_000_000 + t.nanos as i64 / 1_000);
if grpc_tx.transaction.as_ref().and_then(|tx| tx.message.as_ref()).is_none() {
return Ok(());
}
let use_sol_parser_sdk = grpc_tx.meta.is_some();
let skip_ix_pass =
use_sol_parser_sdk && !filter_includes_compute_budget_types(event_type_filter);
if use_sol_parser_sdk {
let mut update = SubscribeUpdateTransaction {
slot: slot_u,
transaction: Some(grpc_tx),
..Default::default()
};
let sdk_parse_filter = build_sdk_parse_event_filter(event_type_filter);
let pb_events = parse_subscribe_update_transaction_low_latency(
&update,
recv_us,
block_us_micro,
sdk_parse_filter.as_ref(),
);
let adapted = crate::streaming::parser_sdk_bridge::adapt_parser_events_list(
pb_events,
block_time.as_ref(),
recv_us,
protocols,
event_type_filter,
);
for mut ev in adapted {
ev.metadata_mut().handle_us = elapsed_micros_since(recv_us);
ev = super::super::helpers::process_event(ev, bot_wallet);
callback(ev);
}
if skip_ix_pass {
return Ok(());
}
let Some(tx) = update.transaction.take() else {
return Ok(());
};
grpc_tx = tx;
}
let Some(transition) = grpc_tx.transaction.as_ref() else {
return Ok(());
};
let Some(message) = transition.message.as_ref() else {
return Ok(());
};
let ix_mode =
if use_sol_parser_sdk { GrpcIxParseMode::ComputeBudgetOnly } else { GrpcIxParseMode::Full };
let accounts = build_account_keys(message, grpc_tx.meta.as_ref());
let inner_instructions =
grpc_tx.meta.as_ref().map(|meta| meta.inner_instructions.as_slice()).unwrap_or_default();
let recent_blockhash = if message.recent_blockhash.len() == 32 {
Some(solana_sdk::bs58::encode(&message.recent_blockhash).into_string())
} else {
None
};
parse_instruction_events_from_grpc_transaction(
protocols,
event_type_filter,
ix_mode,
&message.instructions,
signature,
slot,
block_time,
recv_us,
&accounts,
inner_instructions,
bot_wallet,
tx_index,
recent_blockhash,
callback,
)
.await?;
Ok(())
}
fn build_account_keys(
message: &yellowstone_grpc_proto::prelude::Message,
meta: Option<&yellowstone_grpc_proto::prelude::TransactionStatusMeta>,
) -> Vec<Pubkey> {
let loaded_len = meta
.map(|m| m.loaded_writable_addresses.len() + m.loaded_readonly_addresses.len())
.unwrap_or(0);
let mut accounts = Vec::with_capacity(message.account_keys.len() + loaded_len);
for account in &message.account_keys {
push_account_key(&mut accounts, account);
}
if let Some(meta) = meta {
for account in
meta.loaded_writable_addresses.iter().chain(meta.loaded_readonly_addresses.iter())
{
push_account_key(&mut accounts, account);
}
}
accounts
}
#[inline]
fn push_account_key(accounts: &mut Vec<Pubkey>, account: &[u8]) {
let pubkey = if account.len() == 32 {
Pubkey::try_from(account).unwrap_or_default()
} else {
Pubkey::default()
};
accounts.push(pubkey);
}
pub(super) async fn parse_instruction_events_from_grpc_transaction(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
ix_mode: GrpcIxParseMode,
compiled_instructions: &[yellowstone_grpc_proto::prelude::CompiledInstruction],
signature: Signature,
slot: Option<u64>,
block_time: Option<Timestamp>,
recv_us: i64,
accounts: &[Pubkey],
inner_instructions: &[yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions],
bot_wallet: Option<Pubkey>,
tx_index: Option<u64>,
recent_blockhash: Option<String>,
callback: Arc<dyn Fn(DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
let mut accounts = accounts.to_vec();
let has_program = match ix_mode {
GrpcIxParseMode::Full => accounts.iter().any(|account| {
super::super::helpers::should_handle(protocols, event_type_filter, account)
}),
GrpcIxParseMode::ComputeBudgetOnly => compiled_instructions.iter().any(|ix| {
accounts
.get(ix.program_id_index as usize)
.map(EventDispatcher::is_compute_budget_program)
.unwrap_or(false)
}),
};
if has_program {
// Parse each instruction in order.
for (index, instruction) in compiled_instructions.iter().enumerate() {
if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
let program_id = *program_id;
let inner_instructions_ref = inner_instructions
.iter()
.find(|inner_instruction| inner_instruction.index == index as u32);
let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
if *max_idx as usize >= accounts.len() {
accounts.resize(*max_idx as usize + 1, Pubkey::default());
}
let handle_outer = match ix_mode {
GrpcIxParseMode::Full => super::super::helpers::should_handle(
protocols,
event_type_filter,
&program_id,
),
GrpcIxParseMode::ComputeBudgetOnly => {
EventDispatcher::is_compute_budget_program(&program_id)
}
};
if handle_outer {
super::grpc_instruction::parse_events_from_grpc_instruction(
protocols,
event_type_filter,
instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
index as i64,
None,
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
inner_instructions_ref,
callback.clone(),
)?;
}
if ix_mode == GrpcIxParseMode::Full {
if let Some(inner_instructions) = inner_instructions_ref {
for (inner_index, inner_instruction) in
inner_instructions.instructions.iter().enumerate()
{
let inner_accounts = &inner_instruction.accounts;
let data = &inner_instruction.data;
let instruction =
yellowstone_grpc_proto::prelude::CompiledInstruction {
program_id_index: inner_instruction.program_id_index,
accounts: inner_accounts.to_vec(),
data: data.to_vec(),
};
super::grpc_instruction::parse_events_from_grpc_instruction(
protocols,
event_type_filter,
&instruction,
&accounts,
signature,
slot.unwrap_or(0),
block_time,
recv_us,
inner_instructions.index as i64,
Some(inner_index as i64),
bot_wallet,
tx_index,
recent_blockhash.as_deref(),
Some(inner_instructions),
callback.clone(),
)?;
}
}
}
}
}
}
Ok(())
}
@@ -0,0 +1,13 @@
//! Yellowstone gRPC path for `SubscribeUpdateTransactionInfo` and SDK aggregate parsing.
//!
//! | Module | Responsibility |
//! |--------|------|
//! | [`grpc_ix_mode`] | `GrpcIxParseMode` |
//! | [`grpc_transaction`] | whole subscription message and top-level ix loop |
//! | [`grpc_instruction`] | single Yellowstone `CompiledInstruction` |
mod grpc_instruction;
mod grpc_ix_mode;
mod grpc_transaction;
pub(super) use grpc_transaction::parse_grpc_transaction;
@@ -0,0 +1,96 @@
//! Protocol filtering and event enrichment for PumpFun / PumpSwap / Bonk / bot flags.
use crate::streaming::event_parser::{
common::filter::{filter_includes_compute_budget_types, EventTypeFilter},
core::dispatcher::EventDispatcher,
core::global_state::{
add_bonk_dev_address, add_dev_address, is_bonk_dev_address_in_signature,
is_dev_address_in_signature,
},
DexEvent, Protocol,
};
use solana_sdk::pubkey::Pubkey;
pub(super) fn should_handle(
protocols: &[Protocol],
event_type_filter: Option<&EventTypeFilter>,
program_id: &Pubkey,
) -> bool {
if EventDispatcher::is_compute_budget_program(program_id) {
return filter_includes_compute_budget_types(event_type_filter);
}
if let Some(protocol) = EventDispatcher::match_protocol_by_program_id(program_id) {
protocols.contains(&protocol)
} else {
false
}
}
// ================================================================================================
// Event Post-Processing
// ================================================================================================
/// Process and enrich parsed event with additional context
///
/// Handles protocol-specific post-processing:
/// - PumpFun: Tracks dev addresses and marks dev trades
/// - PumpSwap: Fills swap data amounts
/// - Bonk: Tracks pool creators and marks dev trades
/// - General: Marks bot wallet trades
pub(crate) fn process_event(event: DexEvent, bot_wallet: Option<Pubkey>) -> DexEvent {
let signature = event.metadata().signature; // Copy the signature to avoid borrowing issues
match event {
DexEvent::PumpFunCreateTokenEvent(token_info) => {
add_dev_address(&signature, token_info.user);
if token_info.creator != Pubkey::default() && token_info.creator != token_info.user {
add_dev_address(&signature, token_info.creator);
}
DexEvent::PumpFunCreateTokenEvent(token_info)
}
DexEvent::PumpFunCreateV2TokenEvent(token_info) => {
add_dev_address(&signature, token_info.user);
if token_info.creator != Pubkey::default() && token_info.creator != token_info.user {
add_dev_address(&signature, token_info.creator);
}
DexEvent::PumpFunCreateV2TokenEvent(token_info)
}
DexEvent::PumpFunTradeEvent(mut trade_info) => {
trade_info.is_dev_create_token_trade =
is_dev_address_in_signature(&signature, &trade_info.user)
|| is_dev_address_in_signature(&signature, &trade_info.creator);
trade_info.is_bot = Some(trade_info.user) == bot_wallet;
if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() {
swap_data.from_amount =
if trade_info.is_buy { trade_info.sol_amount } else { trade_info.token_amount };
swap_data.to_amount =
if trade_info.is_buy { trade_info.token_amount } else { trade_info.sol_amount };
}
DexEvent::PumpFunTradeEvent(trade_info)
}
DexEvent::PumpSwapBuyEvent(mut trade_info) => {
if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() {
swap_data.from_amount = trade_info.user_quote_amount_in;
swap_data.to_amount = trade_info.base_amount_out;
}
DexEvent::PumpSwapBuyEvent(trade_info)
}
DexEvent::PumpSwapSellEvent(mut trade_info) => {
if let Some(swap_data) = trade_info.metadata.swap_data.as_mut() {
swap_data.from_amount = trade_info.base_amount_in;
swap_data.to_amount = trade_info.user_quote_amount_out;
}
DexEvent::PumpSwapSellEvent(trade_info)
}
DexEvent::BonkPoolCreateEvent(pool_info) => {
add_bonk_dev_address(&signature, pool_info.creator);
DexEvent::BonkPoolCreateEvent(pool_info)
}
DexEvent::BonkTradeEvent(mut trade_info) => {
trade_info.is_dev_create_token_trade =
is_bonk_dev_address_in_signature(&signature, &trade_info.payer);
trade_info.is_bot = Some(trade_info.payer) == bot_wallet;
DexEvent::BonkTradeEvent(trade_info)
}
_ => event,
}
}
@@ -0,0 +1,74 @@
//! Transaction parser entry point with separate gRPC and standard ix paths.
//!
//! | Module | Path |
//! |--------|------|
//! | [`grpc_path`] | Yellowstone gRPC |
//! | [`compiled_path`] | standard transaction / RPC replay |
//! | [`helpers`] | `should_handle`、`process_event` |
mod compiled_path;
mod grpc_path;
pub(crate) mod helpers;
pub struct EventParser;
impl EventParser {
pub async fn parse_grpc_transaction(
protocols: &[crate::streaming::event_parser::Protocol],
event_type_filter: Option<&crate::streaming::event_parser::common::filter::EventTypeFilter>,
grpc_tx: yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo,
signature: solana_sdk::signature::Signature,
slot: Option<u64>,
block_time: Option<prost_types::Timestamp>,
recv_us: i64,
bot_wallet: Option<solana_sdk::pubkey::Pubkey>,
tx_index: Option<u64>,
callback: std::sync::Arc<dyn Fn(crate::streaming::event_parser::DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
grpc_path::parse_grpc_transaction(
protocols,
event_type_filter,
grpc_tx,
signature,
slot,
block_time,
recv_us,
bot_wallet,
tx_index,
callback,
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn parse_instruction_events_from_versioned_transaction(
protocols: &[crate::streaming::event_parser::Protocol],
event_type_filter: Option<&crate::streaming::event_parser::common::filter::EventTypeFilter>,
transaction: &solana_sdk::transaction::VersionedTransaction,
signature: solana_sdk::signature::Signature,
slot: Option<u64>,
block_time: Option<prost_types::Timestamp>,
recv_us: i64,
accounts: &[solana_sdk::pubkey::Pubkey],
inner_instructions: &[solana_transaction_status::InnerInstructions],
bot_wallet: Option<solana_sdk::pubkey::Pubkey>,
tx_index: Option<u64>,
callback: std::sync::Arc<dyn Fn(crate::streaming::event_parser::DexEvent) + Send + Sync>,
) -> anyhow::Result<()> {
compiled_path::parse_instruction_events_from_versioned_transaction(
protocols,
event_type_filter,
transaction,
signature,
slot,
block_time,
recv_us,
accounts,
inner_instructions,
bot_wallet,
tx_index,
callback,
)
.await
}
}
+29 -18
View File
@@ -1,8 +1,8 @@
use dashmap::DashMap;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::Signature;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use dashmap::DashMap;
use std::collections::BTreeSet;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
const MAX_SIGNATURES: usize = 1000;
const CLEANUP_BATCH_SIZE: usize = 100;
@@ -45,15 +45,17 @@ impl GlobalState {
// Use CAS to ensure only one thread performs cleanup
let gen = self.generation.load(Ordering::Relaxed);
if self.generation.compare_exchange_weak(gen, gen + 1, Ordering::Acquire, Ordering::Relaxed).is_err() {
if self
.generation
.compare_exchange_weak(gen, gen + 1, Ordering::Acquire, Ordering::Relaxed)
.is_err()
{
return; // Another thread is cleaning up
}
// Collect only the batch we need to remove (avoid allocating full list)
let signatures_to_remove: Vec<Signature> = self.signature_data.iter()
.take(CLEANUP_BATCH_SIZE)
.map(|entry| *entry.key())
.collect();
let signatures_to_remove: Vec<Signature> =
self.signature_data.iter().take(CLEANUP_BATCH_SIZE).map(|entry| *entry.key()).collect();
// Remove old signatures atomically; only decrement count when entry was present
for signature in signatures_to_remove {
@@ -66,8 +68,9 @@ impl GlobalState {
/// Add developer address for a specific signature (lock-free)
pub fn add_dev_address(&self, signature: &Signature, address: Pubkey) {
self.maybe_cleanup();
self.signature_data.entry(*signature)
self.signature_data
.entry(*signature)
.and_modify(|addresses| {
addresses.dev_addresses.insert(address);
})
@@ -82,8 +85,9 @@ impl GlobalState {
/// Add Bonk developer address for a specific signature (lock-free)
pub fn add_bonk_dev_address(&self, signature: &Signature, address: Pubkey) {
self.maybe_cleanup();
self.signature_data.entry(*signature)
self.signature_data
.entry(*signature)
.and_modify(|addresses| {
addresses.bonk_dev_addresses.insert(address);
})
@@ -97,14 +101,20 @@ impl GlobalState {
/// High-performance: Check if address is a developer address in specific signature (O(log m))
pub fn is_dev_address_in_signature(&self, signature: &Signature, address: &Pubkey) -> bool {
self.signature_data.get(signature)
self.signature_data
.get(signature)
.map(|entry| entry.dev_addresses.contains(address))
.unwrap_or(false)
}
/// High-performance: Check if address is a Bonk developer address in specific signature (O(log m))
pub fn is_bonk_dev_address_in_signature(&self, signature: &Signature, address: &Pubkey) -> bool {
self.signature_data.get(signature)
pub fn is_bonk_dev_address_in_signature(
&self,
signature: &Signature,
address: &Pubkey,
) -> bool {
self.signature_data
.get(signature)
.map(|entry| entry.bonk_dev_addresses.contains(address))
.unwrap_or(false)
}
@@ -143,14 +153,16 @@ impl GlobalState {
/// Get developer addresses for a specific signature
pub fn get_dev_addresses_for_signature(&self, signature: &Signature) -> Vec<Pubkey> {
self.signature_data.get(signature)
self.signature_data
.get(signature)
.map(|entry| entry.dev_addresses.iter().copied().collect())
.unwrap_or_default()
}
/// Get Bonk developer addresses for a specific signature
pub fn get_bonk_dev_addresses_for_signature(&self, signature: &Signature) -> Vec<Pubkey> {
self.signature_data.get(signature)
self.signature_data
.get(signature)
.map(|entry| entry.bonk_dev_addresses.iter().copied().collect())
.unwrap_or_default()
}
@@ -175,8 +187,7 @@ impl Default for GlobalState {
}
/// Global state instance
static GLOBAL_STATE: std::sync::LazyLock<GlobalState> =
std::sync::LazyLock::new(GlobalState::new);
static GLOBAL_STATE: std::sync::LazyLock<GlobalState> = std::sync::LazyLock::new(GlobalState::new);
/// Get global state instance
pub fn get_global_state() -> &'static GlobalState {
@@ -1,4 +1,5 @@
use crate::streaming::event_parser::DexEvent;
use solana_sdk::pubkey::Pubkey;
pub fn merge(instruction_event: &mut DexEvent, cpi_log_event: DexEvent) {
match instruction_event {
@@ -437,6 +438,249 @@ pub fn merge(instruction_event: &mut DexEvent, cpi_log_event: DexEvent) {
_ => {}
},
// Orca Whirlpool:外层指令粗字段 + CPI 日志精修
DexEvent::OrcaWhirlpoolSwapEvent(e) => match cpi_log_event {
DexEvent::OrcaWhirlpoolSwapEvent(cpie) => {
if cpie.whirlpool != Pubkey::default() {
e.whirlpool = cpie.whirlpool;
}
if cpie.input_amount != 0 {
e.input_amount = cpie.input_amount;
}
if cpie.output_amount != 0 {
e.output_amount = cpie.output_amount;
}
e.a_to_b = cpie.a_to_b;
if cpie.pre_sqrt_price != 0 {
e.pre_sqrt_price = cpie.pre_sqrt_price;
}
if cpie.post_sqrt_price != 0 {
e.post_sqrt_price = cpie.post_sqrt_price;
}
if cpie.input_transfer_fee != 0 {
e.input_transfer_fee = cpie.input_transfer_fee;
}
if cpie.output_transfer_fee != 0 {
e.output_transfer_fee = cpie.output_transfer_fee;
}
if cpie.lp_fee != 0 {
e.lp_fee = cpie.lp_fee;
}
if cpie.protocol_fee != 0 {
e.protocol_fee = cpie.protocol_fee;
}
}
_ => {}
},
DexEvent::OrcaWhirlpoolLiquidityIncreasedEvent(e) => match cpi_log_event {
DexEvent::OrcaWhirlpoolLiquidityIncreasedEvent(cpie) => {
if cpie.position != Pubkey::default() {
e.position = cpie.position;
}
if cpie.tick_lower_index != 0 || cpie.tick_upper_index != 0 {
e.tick_lower_index = cpie.tick_lower_index;
e.tick_upper_index = cpie.tick_upper_index;
}
if cpie.token_a_amount != 0 {
e.token_a_amount = cpie.token_a_amount;
}
if cpie.token_b_amount != 0 {
e.token_b_amount = cpie.token_b_amount;
}
if cpie.liquidity != 0 {
e.liquidity = cpie.liquidity;
}
if cpie.token_a_transfer_fee != 0 {
e.token_a_transfer_fee = cpie.token_a_transfer_fee;
}
if cpie.token_b_transfer_fee != 0 {
e.token_b_transfer_fee = cpie.token_b_transfer_fee;
}
}
_ => {}
},
DexEvent::OrcaWhirlpoolLiquidityDecreasedEvent(e) => match cpi_log_event {
DexEvent::OrcaWhirlpoolLiquidityDecreasedEvent(cpie) => {
if cpie.position != Pubkey::default() {
e.position = cpie.position;
}
if cpie.tick_lower_index != 0 || cpie.tick_upper_index != 0 {
e.tick_lower_index = cpie.tick_lower_index;
e.tick_upper_index = cpie.tick_upper_index;
}
if cpie.token_a_amount != 0 {
e.token_a_amount = cpie.token_a_amount;
}
if cpie.token_b_amount != 0 {
e.token_b_amount = cpie.token_b_amount;
}
if cpie.liquidity != 0 {
e.liquidity = cpie.liquidity;
}
if cpie.token_a_transfer_fee != 0 {
e.token_a_transfer_fee = cpie.token_a_transfer_fee;
}
if cpie.token_b_transfer_fee != 0 {
e.token_b_transfer_fee = cpie.token_b_transfer_fee;
}
}
_ => {}
},
// Meteora Pools swap:外层 min_out 等与 CPI 实际结算合并
DexEvent::MeteoraPoolsSwapEvent(e) => match cpi_log_event {
DexEvent::MeteoraPoolsSwapEvent(cpie) => {
if cpie.in_amount != 0 {
e.in_amount = cpie.in_amount;
}
if cpie.out_amount != 0 {
e.out_amount = cpie.out_amount;
}
if cpie.trade_fee != 0 {
e.trade_fee = cpie.trade_fee;
}
if cpie.admin_fee != 0 {
e.admin_fee = cpie.admin_fee;
}
if cpie.host_fee != 0 {
e.host_fee = cpie.host_fee;
}
}
_ => {}
},
DexEvent::MeteoraPoolsAddLiquidityEvent(e) => match cpi_log_event {
DexEvent::MeteoraPoolsAddLiquidityEvent(cpie) => {
if cpie.lp_mint_amount != 0 {
e.lp_mint_amount = cpie.lp_mint_amount;
}
if cpie.token_a_amount != 0 {
e.token_a_amount = cpie.token_a_amount;
}
if cpie.token_b_amount != 0 {
e.token_b_amount = cpie.token_b_amount;
}
}
_ => {}
},
DexEvent::MeteoraPoolsRemoveLiquidityEvent(e) => match cpi_log_event {
DexEvent::MeteoraPoolsRemoveLiquidityEvent(cpie) => {
if cpie.lp_unmint_amount != 0 {
e.lp_unmint_amount = cpie.lp_unmint_amount;
}
if cpie.token_a_out_amount != 0 {
e.token_a_out_amount = cpie.token_a_out_amount;
}
if cpie.token_b_out_amount != 0 {
e.token_b_out_amount = cpie.token_b_out_amount;
}
}
_ => {}
},
// Meteora DLMM
DexEvent::MeteoraDlmmSwapEvent(e) => match cpi_log_event {
DexEvent::MeteoraDlmmSwapEvent(cpie) => {
if cpie.pool != Pubkey::default() {
e.pool = cpie.pool;
}
if cpie.from != Pubkey::default() {
e.from = cpie.from;
}
if cpie.start_bin_id != 0 || cpie.end_bin_id != 0 {
e.start_bin_id = cpie.start_bin_id;
e.end_bin_id = cpie.end_bin_id;
}
if cpie.amount_out != 0 {
e.amount_out = cpie.amount_out;
}
if cpie.amount_in != 0 {
e.amount_in = cpie.amount_in;
}
e.swap_for_y = cpie.swap_for_y;
if cpie.fee != 0 {
e.fee = cpie.fee;
}
if cpie.protocol_fee != 0 {
e.protocol_fee = cpie.protocol_fee;
}
if cpie.fee_bps != 0 {
e.fee_bps = cpie.fee_bps;
}
if cpie.host_fee != 0 {
e.host_fee = cpie.host_fee;
}
}
_ => {}
},
DexEvent::MeteoraDlmmAddLiquidityEvent(e) => match cpi_log_event {
DexEvent::MeteoraDlmmAddLiquidityEvent(cpie) => {
if cpie.active_bin_id != 0 {
e.active_bin_id = cpie.active_bin_id;
}
e.amounts = cpie.amounts;
}
_ => {}
},
DexEvent::MeteoraDlmmRemoveLiquidityEvent(e) => match cpi_log_event {
DexEvent::MeteoraDlmmRemoveLiquidityEvent(cpie) => {
if cpie.active_bin_id != 0 {
e.active_bin_id = cpie.active_bin_id;
}
e.amounts = cpie.amounts;
}
_ => {}
},
DexEvent::MeteoraPoolsBootstrapLiquidityEvent(e) => match cpi_log_event {
DexEvent::MeteoraPoolsBootstrapLiquidityEvent(cpie) => {
if cpie.pool != Pubkey::default() {
e.pool = cpie.pool;
}
if cpie.lp_mint_amount != 0 {
e.lp_mint_amount = cpie.lp_mint_amount;
}
if cpie.token_a_amount != 0 {
e.token_a_amount = cpie.token_a_amount;
}
if cpie.token_b_amount != 0 {
e.token_b_amount = cpie.token_b_amount;
}
}
_ => {}
},
DexEvent::MeteoraPoolsPoolCreatedEvent(e) => match cpi_log_event {
DexEvent::MeteoraPoolsPoolCreatedEvent(cpie) => {
if cpie.pool != Pubkey::default() {
e.pool = cpie.pool;
}
if cpie.lp_mint != Pubkey::default() {
e.lp_mint = cpie.lp_mint;
}
if cpie.token_a_mint != Pubkey::default() {
e.token_a_mint = cpie.token_a_mint;
}
if cpie.token_b_mint != Pubkey::default() {
e.token_b_mint = cpie.token_b_mint;
}
if cpie.pool_type != 0 {
e.pool_type = cpie.pool_type;
}
}
_ => {}
},
DexEvent::MeteoraPoolsSetPoolFeesEvent(e) => match cpi_log_event {
DexEvent::MeteoraPoolsSetPoolFeesEvent(cpie) => {
if cpie.pool != Pubkey::default() {
e.pool = cpie.pool;
}
e.trade_fee_numerator = cpie.trade_fee_numerator;
e.trade_fee_denominator = cpie.trade_fee_denominator;
e.owner_trade_fee_numerator = cpie.owner_trade_fee_numerator;
e.owner_trade_fee_denominator = cpie.owner_trade_fee_denominator;
}
_ => {}
},
_ => {}
}
}
+2 -2
View File
@@ -5,8 +5,8 @@ pub mod global_state;
pub mod parser_cache;
pub mod traits;
pub use traits::DexEvent;
pub use dispatcher::EventDispatcher;
pub use traits::DexEvent;
pub mod event_parser;
pub mod merger_event;
pub mod merger_event;
@@ -13,7 +13,7 @@ use crate::streaming::{
event_parser::{
common::{filter::EventTypeFilter, EventMetadata, EventType, ProtocolType},
core::dispatcher::EventDispatcher,
Protocol, DexEvent,
DexEvent, Protocol,
},
grpc::AccountPretty,
};
@@ -53,9 +53,8 @@ impl CacheKey {
}
/// 全局程序ID缓存(使用读写锁保护)
static GLOBAL_PROGRAM_IDS_CACHE: LazyLock<
std::sync::RwLock<HashMap<CacheKey, Arc<Vec<Pubkey>>>>,
> = LazyLock::new(|| std::sync::RwLock::new(HashMap::new()));
static GLOBAL_PROGRAM_IDS_CACHE: LazyLock<std::sync::RwLock<HashMap<CacheKey, Arc<Vec<Pubkey>>>>> =
LazyLock::new(|| std::sync::RwLock::new(HashMap::new()));
/// 获取指定协议的程序ID列表
///
@@ -101,9 +100,7 @@ impl AccountPubkeyCache {
///
/// 预分配32个位置,覆盖大多数交易场景
pub fn new() -> Self {
Self {
cache: Vec::with_capacity(32),
}
Self { cache: Vec::with_capacity(32) }
}
/// 从指令账户索引构建账户公钥向量
@@ -201,4 +198,3 @@ pub struct AccountEventParseConfig {
/// 账户解析器函数
pub account_parser: AccountEventParserFn,
}
+85 -1
View File
@@ -13,6 +13,16 @@ use crate::streaming::event_parser::protocols::pumpswap::events::*;
use crate::streaming::event_parser::protocols::raydium_amm_v4::events::*;
use crate::streaming::event_parser::protocols::raydium_clmm::events::*;
use crate::streaming::event_parser::protocols::raydium_cpmm::events::*;
use crate::streaming::event_parser::protocols::sol_parser_forward::events::{
MeteoraDlmmAddLiquidityEvent, MeteoraDlmmClaimFeeEvent, MeteoraDlmmClosePositionEvent,
MeteoraDlmmCreatePositionEvent, MeteoraDlmmInitializeBinArrayEvent,
MeteoraDlmmInitializePoolEvent, MeteoraDlmmRemoveLiquidityEvent, MeteoraDlmmSwapEvent,
MeteoraPoolsAddLiquidityEvent, MeteoraPoolsBootstrapLiquidityEvent,
MeteoraPoolsPoolCreatedEvent, MeteoraPoolsRemoveLiquidityEvent, MeteoraPoolsSetPoolFeesEvent,
MeteoraPoolsSwapEvent, OrcaWhirlpoolLiquidityDecreasedEvent,
OrcaWhirlpoolLiquidityIncreasedEvent, OrcaWhirlpoolPoolInitializedEvent,
OrcaWhirlpoolSwapEvent, ParserSdkErrorEvent,
};
use serde::{Deserialize, Serialize};
use std::fmt::Debug;
@@ -33,6 +43,16 @@ pub enum DexEvent {
PumpFunCreateV2TokenEvent(PumpFunCreateV2TokenEvent),
PumpFunTradeEvent(PumpFunTradeEvent),
PumpFunMigrateEvent(PumpFunMigrateEvent),
PumpFeesCreateFeeSharingConfigEvent(PumpFeesCreateFeeSharingConfigEvent),
PumpFeesInitializeFeeConfigEvent(PumpFeesInitializeFeeConfigEvent),
PumpFeesResetFeeSharingConfigEvent(PumpFeesResetFeeSharingConfigEvent),
PumpFeesRevokeFeeSharingAuthorityEvent(PumpFeesRevokeFeeSharingAuthorityEvent),
PumpFeesTransferFeeSharingAuthorityEvent(PumpFeesTransferFeeSharingAuthorityEvent),
PumpFeesUpdateAdminEvent(PumpFeesUpdateAdminEvent),
PumpFeesUpdateFeeConfigEvent(PumpFeesUpdateFeeConfigEvent),
PumpFeesUpdateFeeSharesEvent(PumpFeesUpdateFeeSharesEvent),
PumpFeesUpsertFeeTiersEvent(PumpFeesUpsertFeeTiersEvent),
PumpFunMigrateBondingCurveCreatorEvent(PumpFunMigrateBondingCurveCreatorEvent),
PumpFunBondingCurveAccountEvent(PumpFunBondingCurveAccountEvent),
PumpFunGlobalAccountEvent(PumpFunGlobalAccountEvent),
@@ -59,6 +79,7 @@ pub enum DexEvent {
RaydiumClmmClosePositionEvent(RaydiumClmmClosePositionEvent),
RaydiumClmmIncreaseLiquidityV2Event(RaydiumClmmIncreaseLiquidityV2Event),
RaydiumClmmDecreaseLiquidityV2Event(RaydiumClmmDecreaseLiquidityV2Event),
RaydiumClmmCollectFeeEvent(RaydiumClmmCollectFeeEvent),
RaydiumClmmCreatePoolEvent(RaydiumClmmCreatePoolEvent),
RaydiumClmmOpenPositionWithToken22NftEvent(RaydiumClmmOpenPositionWithToken22NftEvent),
RaydiumClmmOpenPositionV2Event(RaydiumClmmOpenPositionV2Event),
@@ -79,7 +100,35 @@ pub enum DexEvent {
MeteoraDammV2Swap2Event(MeteoraDammV2Swap2Event),
MeteoraDammV2InitializePoolEvent(MeteoraDammV2InitializePoolEvent),
MeteoraDammV2InitializeCustomizablePoolEvent(MeteoraDammV2InitializeCustomizablePoolEvent),
MeteoraDammV2InitializePoolWithDynamicConfigEvent(MeteoraDammV2InitializePoolWithDynamicConfigEvent),
MeteoraDammV2InitializePoolWithDynamicConfigEvent(
MeteoraDammV2InitializePoolWithDynamicConfigEvent,
),
MeteoraDammV2AddLiquidityEvent(MeteoraDammV2AddLiquidityEvent),
MeteoraDammV2RemoveLiquidityEvent(MeteoraDammV2RemoveLiquidityEvent),
MeteoraDammV2CreatePositionEvent(MeteoraDammV2CreatePositionEvent),
MeteoraDammV2ClosePositionEvent(MeteoraDammV2ClosePositionEvent),
OrcaWhirlpoolSwapEvent(OrcaWhirlpoolSwapEvent),
OrcaWhirlpoolLiquidityIncreasedEvent(OrcaWhirlpoolLiquidityIncreasedEvent),
OrcaWhirlpoolLiquidityDecreasedEvent(OrcaWhirlpoolLiquidityDecreasedEvent),
OrcaWhirlpoolPoolInitializedEvent(OrcaWhirlpoolPoolInitializedEvent),
MeteoraPoolsSwapEvent(MeteoraPoolsSwapEvent),
MeteoraPoolsAddLiquidityEvent(MeteoraPoolsAddLiquidityEvent),
MeteoraPoolsRemoveLiquidityEvent(MeteoraPoolsRemoveLiquidityEvent),
MeteoraPoolsBootstrapLiquidityEvent(MeteoraPoolsBootstrapLiquidityEvent),
MeteoraPoolsPoolCreatedEvent(MeteoraPoolsPoolCreatedEvent),
MeteoraPoolsSetPoolFeesEvent(MeteoraPoolsSetPoolFeesEvent),
MeteoraDlmmSwapEvent(MeteoraDlmmSwapEvent),
MeteoraDlmmAddLiquidityEvent(MeteoraDlmmAddLiquidityEvent),
MeteoraDlmmRemoveLiquidityEvent(MeteoraDlmmRemoveLiquidityEvent),
MeteoraDlmmInitializePoolEvent(MeteoraDlmmInitializePoolEvent),
MeteoraDlmmInitializeBinArrayEvent(MeteoraDlmmInitializeBinArrayEvent),
MeteoraDlmmCreatePositionEvent(MeteoraDlmmCreatePositionEvent),
MeteoraDlmmClosePositionEvent(MeteoraDlmmClosePositionEvent),
MeteoraDlmmClaimFeeEvent(MeteoraDlmmClaimFeeEvent),
// Common events
TokenAccountEvent(TokenAccountEvent),
@@ -88,6 +137,7 @@ pub enum DexEvent {
BlockMetaEvent(BlockMetaEvent),
SetComputeUnitLimitEvent(SetComputeUnitLimitEvent),
SetComputeUnitPriceEvent(SetComputeUnitPriceEvent),
ParserSdkErrorEvent(ParserSdkErrorEvent),
}
/// Macro to generate metadata accessors for all DexEvent variants
@@ -123,6 +173,16 @@ impl_dex_event_metadata!(
PumpFunCreateV2TokenEvent,
PumpFunTradeEvent,
PumpFunMigrateEvent,
PumpFeesCreateFeeSharingConfigEvent,
PumpFeesInitializeFeeConfigEvent,
PumpFeesResetFeeSharingConfigEvent,
PumpFeesRevokeFeeSharingAuthorityEvent,
PumpFeesTransferFeeSharingAuthorityEvent,
PumpFeesUpdateAdminEvent,
PumpFeesUpdateFeeConfigEvent,
PumpFeesUpdateFeeSharesEvent,
PumpFeesUpsertFeeTiersEvent,
PumpFunMigrateBondingCurveCreatorEvent,
PumpFunBondingCurveAccountEvent,
PumpFunGlobalAccountEvent,
// PumpSwap events
@@ -146,6 +206,7 @@ impl_dex_event_metadata!(
RaydiumClmmClosePositionEvent,
RaydiumClmmIncreaseLiquidityV2Event,
RaydiumClmmDecreaseLiquidityV2Event,
RaydiumClmmCollectFeeEvent,
RaydiumClmmCreatePoolEvent,
RaydiumClmmOpenPositionWithToken22NftEvent,
RaydiumClmmOpenPositionV2Event,
@@ -165,6 +226,28 @@ impl_dex_event_metadata!(
MeteoraDammV2InitializePoolEvent,
MeteoraDammV2InitializeCustomizablePoolEvent,
MeteoraDammV2InitializePoolWithDynamicConfigEvent,
MeteoraDammV2AddLiquidityEvent,
MeteoraDammV2RemoveLiquidityEvent,
MeteoraDammV2CreatePositionEvent,
MeteoraDammV2ClosePositionEvent,
OrcaWhirlpoolSwapEvent,
OrcaWhirlpoolLiquidityIncreasedEvent,
OrcaWhirlpoolLiquidityDecreasedEvent,
OrcaWhirlpoolPoolInitializedEvent,
MeteoraPoolsSwapEvent,
MeteoraPoolsAddLiquidityEvent,
MeteoraPoolsRemoveLiquidityEvent,
MeteoraPoolsBootstrapLiquidityEvent,
MeteoraPoolsPoolCreatedEvent,
MeteoraPoolsSetPoolFeesEvent,
MeteoraDlmmSwapEvent,
MeteoraDlmmAddLiquidityEvent,
MeteoraDlmmRemoveLiquidityEvent,
MeteoraDlmmInitializePoolEvent,
MeteoraDlmmInitializeBinArrayEvent,
MeteoraDlmmCreatePositionEvent,
MeteoraDlmmClosePositionEvent,
MeteoraDlmmClaimFeeEvent,
// Common events
TokenAccountEvent,
NonceAccountEvent,
@@ -172,4 +255,5 @@ impl_dex_event_metadata!(
BlockMetaEvent,
SetComputeUnitLimitEvent,
SetComputeUnitPriceEvent,
ParserSdkErrorEvent,
);
+9 -1
View File
@@ -1,6 +1,14 @@
//! Solana DEX 交易解析:**对外类型** [`DexEvent`]、[`Protocol`]
//! **`common`**(元数据/过滤器)、**`core`**(调度、合并、gRPC 解析入口)、**`protocols`**(按协议的 parser/events)。
//!
//! 与 **sol-parser-sdk** 对照:`protocols/*/parser` ≈ sdk `instr/*``parser_sdk_bridge` ≈ sdk 事件枚举与各实现的胶水层。
//!
//! [`DexEvent`]: crate::streaming::event_parser::DexEvent
//! [`Protocol`]: crate::streaming::event_parser::Protocol
pub mod common;
pub mod core;
pub mod protocols;
pub use core::traits::DexEvent;
pub use protocols::types::Protocol;
pub use protocols::types::Protocol;
@@ -13,12 +13,7 @@ pub struct BlockMetaEvent {
}
impl BlockMetaEvent {
pub fn new(
slot: u64,
block_hash: String,
block_time_ms: i64,
recv_us: i64,
) -> Self {
pub fn new(slot: u64, block_hash: String, block_time_ms: i64, recv_us: i64) -> Self {
let metadata = EventMetadata::new(
Signature::default(),
slot,
@@ -1 +1 @@
pub mod block_meta_event;
pub mod block_meta_event;
@@ -25,24 +25,14 @@ pub fn parse_bonk_instruction_data(
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::BUY_EXACT_IN => {
parse_buy_exact_in_instruction(data, accounts, metadata)
}
discriminators::BUY_EXACT_OUT => {
parse_buy_exact_out_instruction(data, accounts, metadata)
}
discriminators::SELL_EXACT_IN => {
parse_sell_exact_in_instruction(data, accounts, metadata)
}
discriminators::BUY_EXACT_IN => parse_buy_exact_in_instruction(data, accounts, metadata),
discriminators::BUY_EXACT_OUT => parse_buy_exact_out_instruction(data, accounts, metadata),
discriminators::SELL_EXACT_IN => parse_sell_exact_in_instruction(data, accounts, metadata),
discriminators::SELL_EXACT_OUT => {
parse_sell_exact_out_instruction(data, accounts, metadata)
}
discriminators::INITIALIZE => {
parse_initialize_instruction(data, accounts, metadata)
}
discriminators::INITIALIZE_V2 => {
parse_initialize_v2_instruction(data, accounts, metadata)
}
discriminators::INITIALIZE => parse_initialize_instruction(data, accounts, metadata),
discriminators::INITIALIZE_V2 => parse_initialize_v2_instruction(data, accounts, metadata),
discriminators::INITIALIZE_WITH_TOKEN_2022 => {
parse_initialize_with_token_2022_instruction(data, accounts, metadata)
}
@@ -65,12 +55,8 @@ pub fn parse_bonk_inner_instruction_data(
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::TRADE_EVENT => {
parse_trade_inner_instruction(data, metadata)
}
discriminators::POOL_CREATE_EVENT => {
parse_pool_create_inner_instruction(data, metadata)
}
discriminators::TRADE_EVENT => parse_trade_inner_instruction(data, metadata),
discriminators::POOL_CREATE_EVENT => parse_pool_create_inner_instruction(data, metadata),
_ => None,
}
}
@@ -85,23 +71,26 @@ pub fn parse_bonk_account_data(
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::POOL_STATE_ACCOUNT => {
crate::streaming::event_parser::protocols::bonk::types::pool_state_parser(account, metadata)
crate::streaming::event_parser::protocols::bonk::types::pool_state_parser(
account, metadata,
)
}
discriminators::GLOBAL_CONFIG_ACCOUNT => {
crate::streaming::event_parser::protocols::bonk::types::global_config_parser(account, metadata)
crate::streaming::event_parser::protocols::bonk::types::global_config_parser(
account, metadata,
)
}
discriminators::PLATFORM_CONFIG_ACCOUNT => {
crate::streaming::event_parser::protocols::bonk::types::platform_config_parser(account, metadata)
crate::streaming::event_parser::protocols::bonk::types::platform_config_parser(
account, metadata,
)
}
_ => None,
}
}
/// Parse pool creation event
fn parse_pool_create_inner_instruction(
data: &[u8],
metadata: EventMetadata,
) -> Option<DexEvent> {
fn parse_pool_create_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
// Note: event_type will be set by the instruction parser, not here
// Because different initialize instructions have different event types
if let Some(event) = bonk_pool_create_event_log_decode(data) {
@@ -349,7 +349,8 @@ impl Default for PlatformConfig {
}
}
pub const PLATFORM_CONFIG_SIZE: usize = 8 + 32 * 2 + 8 * 4 + 64 + 256 + 256 + 32 + 8 + 32 + 32 + 8 + 32 + 108;
pub const PLATFORM_CONFIG_SIZE: usize =
8 + 32 * 2 + 8 * 4 + 64 + 256 + 256 + 32 + 8 + 32 + 32 + 8 + 32 + 108;
pub fn platform_config_decode(data: &[u8]) -> Option<PlatformConfig> {
if data.len() < PLATFORM_CONFIG_SIZE {
@@ -374,6 +374,68 @@ pub struct MeteoraDammV2InitializePoolWithDynamicConfigEvent {
pub config: Pubkey,
}
/// DAMM v2 Add Liquidityparser-sdk / CPI 日志字段对齐)
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
pub struct MeteoraDammV2AddLiquidityEvent {
#[borsh(skip)]
pub metadata: EventMetadata,
pub pool: Pubkey,
pub position: Pubkey,
pub owner: Pubkey,
pub token_a_amount: u64,
pub token_b_amount: u64,
#[borsh(skip)]
pub liquidity_delta: u128,
#[borsh(skip)]
pub token_a_amount_threshold: u64,
#[borsh(skip)]
pub token_b_amount_threshold: u64,
#[borsh(skip)]
pub total_amount_a: u64,
#[borsh(skip)]
pub total_amount_b: u64,
}
/// DAMM v2 Remove Liquidity
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
pub struct MeteoraDammV2RemoveLiquidityEvent {
#[borsh(skip)]
pub metadata: EventMetadata,
pub pool: Pubkey,
pub position: Pubkey,
pub owner: Pubkey,
pub token_a_amount: u64,
pub token_b_amount: u64,
#[borsh(skip)]
pub liquidity_delta: u128,
#[borsh(skip)]
pub token_a_amount_threshold: u64,
#[borsh(skip)]
pub token_b_amount_threshold: u64,
}
/// DAMM v2 Create Position
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
pub struct MeteoraDammV2CreatePositionEvent {
#[borsh(skip)]
pub metadata: EventMetadata,
pub pool: Pubkey,
pub owner: Pubkey,
pub position: Pubkey,
pub position_nft_mint: Pubkey,
}
/// DAMM v2 Close Position
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
pub struct MeteoraDammV2ClosePositionEvent {
#[borsh(skip)]
pub metadata: EventMetadata,
pub pool: Pubkey,
pub owner: Pubkey,
pub position: Pubkey,
pub position_nft_mint: Pubkey,
}
/// Event discriminators
pub mod discriminators {
// Instruction discriminators
@@ -6,6 +6,7 @@ pub mod pumpswap;
pub mod raydium_amm_v4;
pub mod raydium_clmm;
pub mod raydium_cpmm;
pub mod sol_parser_forward;
pub mod types;
pub use block::block_meta_event::BlockMetaEvent;
pub use types::Protocol;
@@ -323,7 +323,8 @@ pub fn pumpfun_trade_event_log_decode(data: &[u8]) -> Option<PumpFunTradeEvent>
if data.len() < PUMPFUN_TRADE_EVENT_LOG_SIZE {
return None;
}
let mut event = borsh::from_slice::<PumpFunTradeEvent>(&data[..PUMPFUN_TRADE_EVENT_LOG_SIZE]).ok()?;
let mut event =
borsh::from_slice::<PumpFunTradeEvent>(&data[..PUMPFUN_TRADE_EVENT_LOG_SIZE]).ok()?;
let mut offset = PUMPFUN_TRADE_EVENT_LOG_SIZE;
if offset < data.len() {
let (ix_name, inc) = read_borsh_string(data, offset).unwrap_or((String::new(), 0));
@@ -335,7 +336,8 @@ pub fn pumpfun_trade_event_log_decode(data: &[u8]) -> Option<PumpFunTradeEvent>
offset += 1;
}
if offset + 8 <= data.len() {
event.cashback_fee_basis_points = u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?);
event.cashback_fee_basis_points =
u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?);
offset += 8;
}
if offset + 8 <= data.len() {
@@ -415,6 +417,138 @@ pub struct PumpFunMigrateEvent {
pub program: Pubkey,
}
// ---------- pump-fees IDL: `idls/pump_fees.json` (Program `pfeeUx...`) ----------
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesShareholder {
pub address: Pubkey,
pub share_bps: u16,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum PumpFeesConfigStatus {
Paused,
Active,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesFees {
pub lp_fee_bps: u64,
pub protocol_fee_bps: u64,
pub creator_fee_bps: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesFeeTier {
pub market_cap_lamports_threshold: u128,
pub fees: PumpFeesFees,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesCreateFeeSharingConfigEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub mint: Pubkey,
pub bonding_curve: Pubkey,
pub pool: Option<Pubkey>,
pub sharing_config: Pubkey,
pub admin: Pubkey,
pub initial_shareholders: Vec<PumpFeesShareholder>,
pub status: PumpFeesConfigStatus,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesInitializeFeeConfigEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub admin: Pubkey,
pub fee_config: Pubkey,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesResetFeeSharingConfigEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub mint: Pubkey,
pub sharing_config: Pubkey,
pub old_admin: Pubkey,
pub old_shareholders: Vec<PumpFeesShareholder>,
pub new_admin: Pubkey,
pub new_shareholders: Vec<PumpFeesShareholder>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesRevokeFeeSharingAuthorityEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub mint: Pubkey,
pub sharing_config: Pubkey,
pub admin: Pubkey,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesTransferFeeSharingAuthorityEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub mint: Pubkey,
pub sharing_config: Pubkey,
pub old_admin: Pubkey,
pub new_admin: Pubkey,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesUpdateAdminEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub old_admin: Pubkey,
pub new_admin: Pubkey,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesUpdateFeeConfigEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub admin: Pubkey,
pub fee_config: Pubkey,
pub fee_tiers: Vec<PumpFeesFeeTier>,
pub flat_fees: PumpFeesFees,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesUpdateFeeSharesEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub mint: Pubkey,
pub sharing_config: Pubkey,
pub admin: Pubkey,
#[serde(default)]
pub bonding_curve: Pubkey,
#[serde(default)]
pub pump_creator_vault: Pubkey,
pub new_shareholders: Vec<PumpFeesShareholder>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFeesUpsertFeeTiersEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub admin: Pubkey,
pub fee_config: Pubkey,
pub fee_tiers: Vec<PumpFeesFeeTier>,
pub offset: u8,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PumpFunMigrateBondingCurveCreatorEvent {
pub metadata: EventMetadata,
pub timestamp: i64,
pub mint: Pubkey,
pub bonding_curve: Pubkey,
pub sharing_config: Pubkey,
pub old_creator: Pubkey,
pub new_creator: Pubkey,
}
pub const PUMPFUN_MIGRATE_EVENT_LOG_SIZE: usize = 160;
pub fn pumpfun_migrate_event_log_decode(data: &[u8]) -> Option<PumpFunMigrateEvent> {
@@ -28,7 +28,9 @@ pub fn parse_pumpfun_instruction_data(
parse_create_v2_token_instruction(data, accounts, metadata)
}
discriminators::BUY_IX => parse_buy_instruction(data, accounts, metadata),
discriminators::BUY_EXACT_SOL_IN_IX => parse_buy_exact_sol_in_instruction(data, accounts, metadata),
discriminators::BUY_EXACT_SOL_IN_IX => {
parse_buy_exact_sol_in_instruction(data, accounts, metadata)
}
discriminators::SELL_IX => parse_sell_instruction(data, accounts, metadata),
discriminators::MIGRATE_IX => parse_migrate_instruction(data, accounts, metadata),
_ => None,
@@ -320,10 +322,11 @@ fn parse_buy_instruction(
/// Same account layout as buy: 16 fixed + optional 17th (index 16).
/// Args: spendable_sol_in (SOL), min_tokens_out (token).
fn parse_buy_exact_sol_in_instruction(
data: &[u8], accounts: &[Pubkey],
data: &[u8],
accounts: &[Pubkey],
mut metadata: EventMetadata,
) -> Option<DexEvent> {
metadata.event_type = EventType::PumpFunBuy;
metadata.event_type = EventType::PumpFunBuyExactSolIn;
if data.len() < 16 || accounts.len() < 16 {
return None;
@@ -82,7 +82,8 @@ pub struct Global {
pub is_cashback_enabled: bool,
}
pub const GLOBAL_SIZE: usize = 1 + 32 * 2 + 8 * 5 + 32 + 1 + 8 * 2 + 32 * 7 + 32 * 2 + 1 + 32 * 2 + 1 + 32 * 7 + 1;
pub const GLOBAL_SIZE: usize =
1 + 32 * 2 + 8 * 5 + 32 + 1 + 8 * 2 + 32 * 7 + 32 * 2 + 1 + 32 * 2 + 1 + 32 * 7 + 1;
pub fn global_decode(data: &[u8]) -> Option<Global> {
if data.len() < GLOBAL_SIZE {
@@ -93,7 +93,8 @@ pub fn pump_swap_buy_event_log_decode(data: &[u8]) -> Option<PumpSwapBuyEvent> {
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 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)?;
@@ -243,11 +244,13 @@ pub fn pump_swap_sell_event_log_decode(data: &[u8]) -> Option<PumpSwapSellEvent>
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 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 {
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)
@@ -25,11 +25,11 @@ pub fn parse_pumpswap_instruction_data(
) -> Option<DexEvent> {
match discriminator {
discriminators::BUY_IX => parse_buy_instruction(data, accounts, metadata),
discriminators::BUY_EXACT_QUOTE_IN_IX => parse_buy_exact_quote_in_instruction(data, accounts, metadata),
discriminators::SELL_IX => parse_sell_instruction(data, accounts, metadata),
discriminators::CREATE_POOL_IX => {
parse_create_pool_instruction(data, accounts, metadata)
discriminators::BUY_EXACT_QUOTE_IN_IX => {
parse_buy_exact_quote_in_instruction(data, accounts, metadata)
}
discriminators::SELL_IX => parse_sell_instruction(data, accounts, metadata),
discriminators::CREATE_POOL_IX => parse_create_pool_instruction(data, accounts, metadata),
discriminators::DEPOSIT_IX => parse_deposit_instruction(data, accounts, metadata),
discriminators::WITHDRAW_IX => parse_withdraw_instruction(data, accounts, metadata),
_ => None,
@@ -47,16 +47,13 @@ pub fn parse_pumpswap_inner_instruction_data(
match discriminator {
discriminators::BUY_EVENT => parse_buy_inner_instruction(data, metadata),
discriminators::SELL_EVENT => parse_sell_inner_instruction(data, metadata),
discriminators::CREATE_POOL_EVENT => {
parse_create_pool_inner_instruction(data, metadata)
}
discriminators::CREATE_POOL_EVENT => parse_create_pool_inner_instruction(data, metadata),
discriminators::DEPOSIT_EVENT => parse_deposit_inner_instruction(data, metadata),
discriminators::WITHDRAW_EVENT => parse_withdraw_inner_instruction(data, metadata),
_ => None,
}
}
/// 解析 PumpSwap 账户数据
///
/// 根据判别器路由到具体的账户解析函数
@@ -67,10 +64,14 @@ pub fn parse_pumpswap_account_data(
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::GLOBAL_CONFIG_ACCOUNT => {
crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser(account, metadata)
crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser(
account, metadata,
)
}
discriminators::POOL_ACCOUNT => {
crate::streaming::event_parser::protocols::pumpswap::types::pool_parser(account, metadata)
crate::streaming::event_parser::protocols::pumpswap::types::pool_parser(
account, metadata,
)
}
_ => None,
}
@@ -97,10 +98,7 @@ fn parse_sell_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<
}
/// 解析创建池子日志事件
fn parse_create_pool_inner_instruction(
data: &[u8],
metadata: EventMetadata,
) -> Option<DexEvent> {
fn parse_create_pool_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option<DexEvent> {
// Note: event_type will be set by instruction parser
if let Some(event) = pump_swap_create_pool_event_log_decode(data) {
Some(DexEvent::PumpSwapCreatePoolEvent(PumpSwapCreatePoolEvent { metadata, ..event }))
@@ -1,7 +1,5 @@
use crate::streaming::event_parser::common::EventMetadata;
use crate::{
streaming::event_parser::protocols::raydium_amm_v4::types::AmmInfo,
};
use crate::streaming::event_parser::protocols::raydium_amm_v4::types::AmmInfo;
use borsh::BorshDeserialize;
use serde::{Deserialize, Serialize};
use solana_sdk::pubkey::Pubkey;
@@ -22,18 +22,14 @@ pub fn parse_raydium_amm_v4_instruction_data(
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::SWAP_BASE_IN => {
parse_swap_base_input_instruction(data, accounts, metadata)
}
discriminators::SWAP_BASE_IN => parse_swap_base_input_instruction(data, accounts, metadata),
discriminators::SWAP_BASE_OUT => {
parse_swap_base_output_instruction(data, accounts, metadata)
}
discriminators::DEPOSIT => parse_deposit_instruction(data, accounts, metadata),
discriminators::INITIALIZE2 => parse_initialize2_instruction(data, accounts, metadata),
discriminators::WITHDRAW => parse_withdraw_instruction(data, accounts, metadata),
discriminators::WITHDRAW_PNL => {
parse_withdraw_pnl_instruction(data, accounts, metadata)
}
discriminators::WITHDRAW_PNL => parse_withdraw_pnl_instruction(data, accounts, metadata),
_ => None,
}
}
@@ -49,7 +45,6 @@ pub fn parse_raydium_amm_v4_inner_instruction_data(
None
}
/// 解析 Raydium AMM V4 账户数据
///
/// 根据判别器路由到具体的账户解析函数
@@ -60,13 +55,14 @@ pub fn parse_raydium_amm_v4_account_data(
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::AMM_INFO => {
crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser(account, metadata)
crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser(
account, metadata,
)
}
_ => None,
}
}
/// 解析提现指令事件
fn parse_withdraw_pnl_instruction(
_data: &[u8],
@@ -1,8 +1,6 @@
use crate::streaming::event_parser::common::EventMetadata;
use crate::streaming::event_parser::protocols::raydium_clmm::types::AmmConfig;
use crate::streaming::event_parser::protocols::raydium_clmm::types::{PoolState, TickArrayState};
use crate::{
streaming::event_parser::protocols::raydium_clmm::types::AmmConfig,
};
use serde::{Deserialize, Serialize};
use solana_sdk::pubkey::Pubkey;
@@ -27,7 +25,6 @@ pub struct RaydiumClmmSwapEvent {
pub remaining_accounts: Vec<Pubkey>,
}
/// 交易v2
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RaydiumClmmSwapV2Event {
@@ -90,6 +87,16 @@ pub struct RaydiumClmmDecreaseLiquidityV2Event {
pub remaining_accounts: Vec<Pubkey>,
}
/// 收取流动性费用(与 `sol-parser-sdk` 日志事件字段对齐的精简版)
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RaydiumClmmCollectFeeEvent {
pub metadata: EventMetadata,
pub pool_state: Pubkey,
pub position_nft_mint: Pubkey,
pub amount_0: u64,
pub amount_1: u64,
}
/// 创建池
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RaydiumClmmCreatePoolEvent {
@@ -60,7 +60,6 @@ pub fn parse_raydium_clmm_inner_instruction_data(
None
}
/// 解析 Raydium CLMM 账户数据
///
/// 根据判别器路由到具体的账户解析函数
@@ -71,13 +70,19 @@ pub fn parse_raydium_clmm_account_data(
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::AMM_CONFIG => {
crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser(account, metadata)
crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser(
account, metadata,
)
}
discriminators::POOL_STATE => {
crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser(account, metadata)
crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser(
account, metadata,
)
}
discriminators::TICK_ARRAY_STATE => {
crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser(account, metadata)
crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser(
account, metadata,
)
}
_ => None,
}
@@ -1,8 +1,6 @@
use crate::streaming::event_parser::common::EventMetadata;
use crate::streaming::event_parser::protocols::raydium_cpmm::types::AmmConfig;
use crate::streaming::event_parser::protocols::raydium_cpmm::types::PoolState;
use crate::{
streaming::event_parser::protocols::raydium_cpmm::types::AmmConfig,
};
use borsh::BorshDeserialize;
use serde::{Deserialize, Serialize};
use solana_sdk::pubkey::Pubkey;
@@ -31,7 +29,6 @@ pub struct RaydiumCpmmSwapEvent {
pub observation_state: Pubkey,
}
/// 存款
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
pub struct RaydiumCpmmDepositEvent {
@@ -23,9 +23,7 @@ pub fn parse_raydium_cpmm_instruction_data(
metadata: EventMetadata,
) -> Option<DexEvent> {
match discriminator {
discriminators::SWAP_BASE_IN => {
parse_swap_base_input_instruction(data, accounts, metadata)
}
discriminators::SWAP_BASE_IN => parse_swap_base_input_instruction(data, accounts, metadata),
discriminators::SWAP_BASE_OUT => {
parse_swap_base_output_instruction(data, accounts, metadata)
}
@@ -47,7 +45,6 @@ pub fn parse_raydium_cpmm_inner_instruction_data(
None
}
/// 解析 Raydium CPMM 账户数据
///
/// 根据判别器路由到具体的账户解析函数
@@ -58,16 +55,19 @@ pub fn parse_raydium_cpmm_account_data(
) -> Option<crate::streaming::event_parser::DexEvent> {
match discriminator {
discriminators::AMM_CONFIG => {
crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser(account, metadata)
crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser(
account, metadata,
)
}
discriminators::POOL_STATE => {
crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser(account, metadata)
crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser(
account, metadata,
)
}
_ => None,
}
}
/// 解析提款指令事件
fn parse_withdraw_instruction(
data: &[u8],
@@ -0,0 +1,210 @@
use crate::streaming::event_parser::common::EventMetadata;
use serde::{Deserialize, Serialize};
use solana_sdk::pubkey::Pubkey;
/// `sol-parser-sdk` 错误占位(无有效 EventMetadata 字段)
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ParserSdkErrorEvent {
pub metadata: EventMetadata,
pub message: String,
}
// --- Orca Whirlpool ---
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct OrcaWhirlpoolSwapEvent {
pub metadata: EventMetadata,
pub whirlpool: Pubkey,
pub input_amount: u64,
pub output_amount: u64,
pub a_to_b: bool,
pub pre_sqrt_price: u128,
pub post_sqrt_price: u128,
pub input_transfer_fee: u64,
pub output_transfer_fee: u64,
pub lp_fee: u64,
pub protocol_fee: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct OrcaWhirlpoolLiquidityIncreasedEvent {
pub metadata: EventMetadata,
pub whirlpool: Pubkey,
pub liquidity: u128,
pub token_a_amount: u64,
pub token_b_amount: u64,
pub position: Pubkey,
pub tick_lower_index: i32,
pub tick_upper_index: i32,
pub token_a_transfer_fee: u64,
pub token_b_transfer_fee: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct OrcaWhirlpoolLiquidityDecreasedEvent {
pub metadata: EventMetadata,
pub whirlpool: Pubkey,
pub liquidity: u128,
pub token_a_amount: u64,
pub token_b_amount: u64,
pub position: Pubkey,
pub tick_lower_index: i32,
pub tick_upper_index: i32,
pub token_a_transfer_fee: u64,
pub token_b_transfer_fee: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct OrcaWhirlpoolPoolInitializedEvent {
pub metadata: EventMetadata,
pub whirlpool: Pubkey,
pub whirlpools_config: Pubkey,
pub token_mint_a: Pubkey,
pub token_mint_b: Pubkey,
pub tick_spacing: u16,
pub token_program_a: Pubkey,
pub token_program_b: Pubkey,
pub decimals_a: u8,
pub decimals_b: u8,
pub initial_sqrt_price: u128,
}
// --- Meteora Pools ---
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraPoolsSwapEvent {
pub metadata: EventMetadata,
pub in_amount: u64,
pub out_amount: u64,
pub trade_fee: u64,
pub admin_fee: u64,
pub host_fee: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraPoolsAddLiquidityEvent {
pub metadata: EventMetadata,
pub lp_mint_amount: u64,
pub token_a_amount: u64,
pub token_b_amount: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraPoolsRemoveLiquidityEvent {
pub metadata: EventMetadata,
pub lp_unmint_amount: u64,
pub token_a_out_amount: u64,
pub token_b_out_amount: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraPoolsBootstrapLiquidityEvent {
pub metadata: EventMetadata,
pub lp_mint_amount: u64,
pub token_a_amount: u64,
pub token_b_amount: u64,
pub pool: Pubkey,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraPoolsPoolCreatedEvent {
pub metadata: EventMetadata,
pub lp_mint: Pubkey,
pub token_a_mint: Pubkey,
pub token_b_mint: Pubkey,
pub pool_type: u8,
pub pool: Pubkey,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraPoolsSetPoolFeesEvent {
pub metadata: EventMetadata,
pub trade_fee_numerator: u64,
pub trade_fee_denominator: u64,
pub owner_trade_fee_numerator: u64,
pub owner_trade_fee_denominator: u64,
pub pool: Pubkey,
}
// --- Meteora DLMM ---
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraDlmmSwapEvent {
pub metadata: EventMetadata,
pub pool: Pubkey,
pub from: Pubkey,
pub start_bin_id: i32,
pub end_bin_id: i32,
pub amount_in: u64,
pub amount_out: u64,
pub swap_for_y: bool,
pub fee: u64,
pub protocol_fee: u64,
pub fee_bps: u128,
pub host_fee: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraDlmmAddLiquidityEvent {
pub metadata: EventMetadata,
pub pool: Pubkey,
pub from: Pubkey,
pub position: Pubkey,
pub amounts: [u64; 2],
pub active_bin_id: i32,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraDlmmRemoveLiquidityEvent {
pub metadata: EventMetadata,
pub pool: Pubkey,
pub from: Pubkey,
pub position: Pubkey,
pub amounts: [u64; 2],
pub active_bin_id: i32,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraDlmmInitializePoolEvent {
pub metadata: EventMetadata,
pub pool: Pubkey,
pub creator: Pubkey,
pub active_bin_id: i32,
pub bin_step: u16,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraDlmmInitializeBinArrayEvent {
pub metadata: EventMetadata,
pub pool: Pubkey,
pub bin_array: Pubkey,
pub index: i64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraDlmmCreatePositionEvent {
pub metadata: EventMetadata,
pub pool: Pubkey,
pub position: Pubkey,
pub owner: Pubkey,
pub lower_bin_id: i32,
pub width: u32,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraDlmmClosePositionEvent {
pub metadata: EventMetadata,
pub pool: Pubkey,
pub position: Pubkey,
pub owner: Pubkey,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MeteoraDlmmClaimFeeEvent {
pub metadata: EventMetadata,
pub pool: Pubkey,
pub position: Pubkey,
pub owner: Pubkey,
pub fee_x: u64,
pub fee_y: u64,
}
@@ -0,0 +1,12 @@
//! 由 `sol-parser-sdk` 产出、经 `parser_sdk_bridge` 映射的协议事件类型;`native` 将 sdk 指令解析接到 Yellowstone/shred 路径。
pub mod events;
pub mod native;
use solana_sdk::pubkey::Pubkey;
pub const ORCA_WHIRLPOOL_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc");
pub const METEORA_POOLS_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB");
pub const METEORA_DLMM_PROGRAM_ID: Pubkey =
solana_sdk::pubkey!("LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo");
@@ -0,0 +1,98 @@
//! 将 `sol-parser-sdk` 的 Orca Whirlpool / Meteora Pools / DLMM 顶层与 inner 指令解析接到 streamer `DexEvent`。
use crate::streaming::event_parser::common::EventMetadata;
use crate::streaming::event_parser::{DexEvent, Protocol};
use crate::streaming::parser_sdk_bridge::{
block_timestamp_from_stream_meta, convert_parser_event, fuse_streamer_ix_ctx,
};
use sol_parser_sdk::core::events::EventMetadata as PbEventMetadata;
use sol_parser_sdk::instr::all_inner::{
meteora_amm as pools_inner, meteora_dlmm as dlmm_inner, orca as orca_inner,
};
use sol_parser_sdk::instr::{meteora_amm, meteora_dlmm, orca_whirlpool};
use solana_sdk::pubkey::Pubkey;
#[inline]
fn block_time_us_for_sdk(sm: &EventMetadata) -> Option<i64> {
Some(sm.block_time_ms.saturating_mul(1000))
}
#[inline]
fn pb_meta_from_streamer(sm: &EventMetadata) -> PbEventMetadata {
PbEventMetadata {
signature: sm.signature,
slot: sm.slot,
tx_index: sm.tx_index.unwrap_or(0),
block_time_us: sm.block_time_ms.saturating_mul(1000),
grpc_recv_us: sm.recv_us,
recent_blockhash: sm.recent_blockhash.clone(),
}
}
pub fn dispatch_instruction(
protocol: Protocol,
instruction_discriminator: &[u8],
instruction_data: &[u8],
accounts: &[Pubkey],
stream_meta: &EventMetadata,
) -> Option<DexEvent> {
let mut full = Vec::with_capacity(instruction_discriminator.len() + instruction_data.len());
full.extend_from_slice(instruction_discriminator);
full.extend_from_slice(instruction_data);
let tx_index = stream_meta.tx_index.unwrap_or(0);
let bt_us = block_time_us_for_sdk(stream_meta);
let pb = match protocol {
Protocol::OrcaWhirlpool => orca_whirlpool::parse_instruction(
&full,
accounts,
stream_meta.signature,
stream_meta.slot,
tx_index,
bt_us,
)?,
Protocol::MeteoraPools => meteora_amm::parse_instruction(
&full,
accounts,
stream_meta.signature,
stream_meta.slot,
tx_index,
bt_us,
)?,
Protocol::MeteoraDlmm => meteora_dlmm::parse_instruction(
&full,
accounts,
stream_meta.signature,
stream_meta.slot,
tx_index,
bt_us,
)?,
_ => return None,
};
let ts = block_timestamp_from_stream_meta(stream_meta);
let ev = convert_parser_event(pb, Some(&ts), stream_meta.recv_us)?;
Some(fuse_streamer_ix_ctx(ev, stream_meta))
}
pub fn dispatch_inner_instruction(
protocol: Protocol,
inner_instruction_discriminator: &[u8],
inner_instruction_data: &[u8],
stream_meta: &EventMetadata,
) -> Option<DexEvent> {
let disc: [u8; 16] = inner_instruction_discriminator.try_into().ok()?;
let pm = pb_meta_from_streamer(stream_meta);
let pb = match protocol {
Protocol::OrcaWhirlpool => orca_inner::parse(&disc, inner_instruction_data, pm)?,
Protocol::MeteoraPools => pools_inner::parse(&disc, inner_instruction_data, pm)?,
Protocol::MeteoraDlmm => dlmm_inner::parse(&disc, inner_instruction_data, pm)?,
_ => return None,
};
let ts = block_timestamp_from_stream_meta(stream_meta);
let ev = convert_parser_event(pb, Some(&ts), stream_meta.recv_us)?;
Some(fuse_streamer_ix_ctx(ev, stream_meta))
}
+58 -7
View File
@@ -1,8 +1,14 @@
use crate::streaming::event_parser::protocols::{
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,
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,
sol_parser_forward::{
METEORA_DLMM_PROGRAM_ID, METEORA_POOLS_PROGRAM_ID, ORCA_WHIRLPOOL_PROGRAM_ID,
},
};
use anyhow::{anyhow, Result};
use solana_sdk::pubkey::Pubkey;
@@ -17,6 +23,9 @@ pub enum Protocol {
RaydiumClmm,
RaydiumAmmV4,
MeteoraDammV2,
OrcaWhirlpool,
MeteoraPools,
MeteoraDlmm,
}
impl Protocol {
@@ -29,6 +38,9 @@ impl Protocol {
Protocol::RaydiumClmm => vec![RAYDIUM_CLMM_PROGRAM_ID],
Protocol::RaydiumAmmV4 => vec![RAYDIUM_AMM_V4_PROGRAM_ID],
Protocol::MeteoraDammV2 => vec![METEORA_DAMM_V2_PROGRAM_ID],
Protocol::OrcaWhirlpool => vec![ORCA_WHIRLPOOL_PROGRAM_ID],
Protocol::MeteoraPools => vec![METEORA_POOLS_PROGRAM_ID],
Protocol::MeteoraDlmm => vec![METEORA_DLMM_PROGRAM_ID],
}
}
}
@@ -43,6 +55,9 @@ impl std::fmt::Display for Protocol {
Protocol::RaydiumClmm => write!(f, "RaydiumClmm"),
Protocol::RaydiumAmmV4 => write!(f, "RaydiumAmmV4"),
Protocol::MeteoraDammV2 => write!(f, "MeteoraDammV2"),
Protocol::OrcaWhirlpool => write!(f, "OrcaWhirlpool"),
Protocol::MeteoraPools => write!(f, "MeteoraPools"),
Protocol::MeteoraDlmm => write!(f, "MeteoraDlmm"),
}
}
}
@@ -55,11 +70,47 @@ impl std::str::FromStr for Protocol {
"pumpswap" => Ok(Protocol::PumpSwap),
"pumpfun" => Ok(Protocol::PumpFun),
"bonk" => Ok(Protocol::Bonk),
"raydiumcpmm" => Ok(Protocol::RaydiumCpmm),
"raydiumclmm" => Ok(Protocol::RaydiumClmm),
"raydiumammv4" => Ok(Protocol::RaydiumAmmV4),
"meteoradamm_v2" => Ok(Protocol::MeteoraDammV2),
"raydiumcpmm" | "raydium_cpmm" => Ok(Protocol::RaydiumCpmm),
"raydiumclmm" | "raydium_clmm" => Ok(Protocol::RaydiumClmm),
"raydiumammv4" | "raydium_amm_v4" => Ok(Protocol::RaydiumAmmV4),
"meteoradammv2" | "meteoradamm_v2" | "meteora_damm_v2" => Ok(Protocol::MeteoraDammV2),
"orcawhirlpool" | "orca_whirlpool" | "orca" => Ok(Protocol::OrcaWhirlpool),
"meteorapools" | "meteora_pools" => Ok(Protocol::MeteoraPools),
"meteoradlmm" | "meteora_dlmm" => Ok(Protocol::MeteoraDlmm),
_ => Err(anyhow!("Unsupported protocol: {}", s)),
}
}
}
#[cfg(test)]
mod tests {
use super::Protocol;
use std::str::FromStr;
#[test]
fn parses_display_style_protocol_names() {
for protocol in [
Protocol::RaydiumCpmm,
Protocol::RaydiumClmm,
Protocol::RaydiumAmmV4,
Protocol::MeteoraDammV2,
Protocol::OrcaWhirlpool,
Protocol::MeteoraPools,
Protocol::MeteoraDlmm,
] {
let parsed = Protocol::from_str(&protocol.to_string()).unwrap();
assert_eq!(parsed, protocol);
}
}
#[test]
fn parses_snake_case_protocol_aliases() {
assert_eq!(Protocol::from_str("raydium_cpmm").unwrap(), Protocol::RaydiumCpmm);
assert_eq!(Protocol::from_str("raydium_clmm").unwrap(), Protocol::RaydiumClmm);
assert_eq!(Protocol::from_str("raydium_amm_v4").unwrap(), Protocol::RaydiumAmmV4);
assert_eq!(Protocol::from_str("meteora_damm_v2").unwrap(), Protocol::MeteoraDammV2);
assert_eq!(Protocol::from_str("orca_whirlpool").unwrap(), Protocol::OrcaWhirlpool);
assert_eq!(Protocol::from_str("meteora_pools").unwrap(), Protocol::MeteoraPools);
assert_eq!(Protocol::from_str("meteora_dlmm").unwrap(), Protocol::MeteoraDlmm);
}
}