From c98d022ae86e64329c98fce68ce93971f52919c6 Mon Sep 17 00:00:00 2001 From: ysq Date: Sun, 20 Jul 2025 00:56:11 +0800 Subject: [PATCH] refactor: migrate streaming functionality to separate SDK, bump to v0.2.2 --- Cargo.toml | 3 +- README.md | 27 +- README_CN.md | 27 +- src/common/bonding_curve.rs | 3 +- src/lib.rs | 2 +- src/main.rs | 8 +- src/protos/mod.rs | 1 - src/protos/shredstream.rs | 279 ---------- src/streaming/event_parser/common/mod.rs | 54 -- src/streaming/event_parser/common/types.rs | 173 ------- src/streaming/event_parser/common/utils.rs | 111 ---- src/streaming/event_parser/core/mod.rs | 2 - src/streaming/event_parser/core/traits.rs | 485 ------------------ src/streaming/event_parser/factory.rs | 98 ---- src/streaming/event_parser/mod.rs | 41 -- .../event_parser/protocols/bonk/events.rs | 126 ----- .../event_parser/protocols/bonk/mod.rs | 7 - .../event_parser/protocols/bonk/parser.rs | 445 ---------------- .../event_parser/protocols/bonk/types.rs | 69 --- src/streaming/event_parser/protocols/mod.rs | 11 - .../event_parser/protocols/pumpfun/events.rs | 113 ---- .../event_parser/protocols/pumpfun/mod.rs | 5 - .../event_parser/protocols/pumpfun/parser.rs | 250 --------- .../event_parser/protocols/pumpswap/events.rs | 322 ------------ .../event_parser/protocols/pumpswap/mod.rs | 5 - .../event_parser/protocols/pumpswap/parser.rs | 386 -------------- .../protocols/raydium_clmm/events.rs | 59 --- .../protocols/raydium_clmm/mod.rs | 5 - .../protocols/raydium_clmm/parser.rs | 170 ------ .../protocols/raydium_cpmm/events.rs | 35 -- .../protocols/raydium_cpmm/mod.rs | 5 - .../protocols/raydium_cpmm/parser.rs | 159 ------ src/streaming/mod.rs | 8 - src/streaming/shred_stream.rs | 120 ----- src/streaming/yellowstone_grpc.rs | 266 ---------- src/streaming/yellowstone_sub_system.rs | 96 ---- src/trading/core/params.rs | 4 +- src/trading/pumpfun/common.rs | 3 +- src/utils.rs | 2 +- 39 files changed, 39 insertions(+), 3946 deletions(-) delete mode 100755 src/protos/shredstream.rs delete mode 100755 src/streaming/event_parser/common/mod.rs delete mode 100755 src/streaming/event_parser/common/types.rs delete mode 100755 src/streaming/event_parser/common/utils.rs delete mode 100755 src/streaming/event_parser/core/mod.rs delete mode 100755 src/streaming/event_parser/core/traits.rs delete mode 100755 src/streaming/event_parser/factory.rs delete mode 100755 src/streaming/event_parser/mod.rs delete mode 100755 src/streaming/event_parser/protocols/bonk/events.rs delete mode 100755 src/streaming/event_parser/protocols/bonk/mod.rs delete mode 100755 src/streaming/event_parser/protocols/bonk/parser.rs delete mode 100755 src/streaming/event_parser/protocols/bonk/types.rs delete mode 100755 src/streaming/event_parser/protocols/mod.rs delete mode 100755 src/streaming/event_parser/protocols/pumpfun/events.rs delete mode 100755 src/streaming/event_parser/protocols/pumpfun/mod.rs delete mode 100755 src/streaming/event_parser/protocols/pumpfun/parser.rs delete mode 100755 src/streaming/event_parser/protocols/pumpswap/events.rs delete mode 100755 src/streaming/event_parser/protocols/pumpswap/mod.rs delete mode 100755 src/streaming/event_parser/protocols/pumpswap/parser.rs delete mode 100755 src/streaming/event_parser/protocols/raydium_clmm/events.rs delete mode 100755 src/streaming/event_parser/protocols/raydium_clmm/mod.rs delete mode 100755 src/streaming/event_parser/protocols/raydium_clmm/parser.rs delete mode 100755 src/streaming/event_parser/protocols/raydium_cpmm/events.rs delete mode 100755 src/streaming/event_parser/protocols/raydium_cpmm/mod.rs delete mode 100755 src/streaming/event_parser/protocols/raydium_cpmm/parser.rs delete mode 100755 src/streaming/mod.rs delete mode 100755 src/streaming/shred_stream.rs delete mode 100755 src/streaming/yellowstone_grpc.rs delete mode 100755 src/streaming/yellowstone_sub_system.rs diff --git a/Cargo.toml b/Cargo.toml index f2f8c48..07c9801 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "sol-trade-sdk" -version = "0.2.1" +version = "0.2.2" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/sol-trade-sdk" @@ -13,6 +13,7 @@ readme = "README.md" crate-type = ["cdylib", "rlib"] [dependencies] +solana-streamer-sdk = "0.1.1" solana-sdk = "2.1.16" solana-client = "2.1.16" solana-program = "2.1.16" diff --git a/README.md b/README.md index 7113004..b2d4ba7 100755 --- a/README.md +++ b/README.md @@ -18,6 +18,8 @@ A comprehensive Rust SDK for seamless interaction with Solana DEX trading progra ## Installation +### Direct Clone + Clone this project to your project directory: ```bash @@ -29,7 +31,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -sol-trade-sdk = { path = "./sol-trade-sdk", version = "0.2.1" } +sol-trade-sdk = { path = "./sol-trade-sdk", version = "0.2.2" } +``` + +### Use crates.io + +```toml +# Add to your Cargo.toml +sol-trade-sdk = "0.2.2" ``` ## Usage Examples @@ -51,7 +60,7 @@ In PumpSwap, Bonk, and Raydium CPMM trading, the `auto_handle_wsol` parameter is #### 1.1 Subscribe to Events Using Yellowstone gRPC ```rust -use sol_trade_sdk::{ +use sol_trade_sdk::solana_streamer_sdk::{ streaming::{ event_parser::{ protocols::{ @@ -128,7 +137,7 @@ async fn test_grpc() -> Result<(), Box> { #### 1.2 Subscribe to Events Using ShredStream ```rust -use sol_trade_sdk::streaming::ShredStreamGrpc; +use sol_trade_sdk::solana_streamer_sdk::streaming::ShredStreamGrpc; async fn test_shreds() -> Result<(), Box> { // Subscribe to events using ShredStream client @@ -655,18 +664,6 @@ src/ ├── common/ # Common functionality and tools ├── constants/ # Constant definitions ├── instruction/ # Instruction building -├── streaming/ # Event stream processing -│ ├── event_parser/ # Event parsing system -│ │ ├── common/ # Common event parsing tools -│ │ ├── core/ # Core parsing traits and interfaces -│ │ ├── protocols/# Protocol-specific parsers -│ │ │ ├── bonk/ # Bonk event parsing -│ │ │ ├── pumpfun/ # PumpFun event parsing -│ │ │ ├── pumpswap/ # PumpSwap event parsing -│ │ │ └── raydium_cpmm/ # Raydium CPMM event parsing -│ │ └── factory.rs # Parser factory -│ ├── shred_stream.rs # ShredStream client -│ └── yellowstone_grpc.rs # Yellowstone gRPC client ├── swqos/ # MEV service clients ├── trading/ # Unified trading engine │ ├── common/ # Common trading tools diff --git a/README_CN.md b/README_CN.md index accb9d9..82b8dca 100755 --- a/README_CN.md +++ b/README_CN.md @@ -18,6 +18,8 @@ ## 安装 +### 直接克隆 + 将此项目克隆到您的项目目录： ```bash @@ -29,7 +31,14 @@ git clone https://github.com/0xfnzero/sol-trade-sdk ```toml # 添加到您的 Cargo.toml -sol-trade-sdk = { path = "./sol-trade-sdk", version = "0.2.1" } +sol-trade-sdk = { path = "./sol-trade-sdk", version = "0.2.2" } +``` + +### 使用 crates.io + +```toml +# 添加到您的 Cargo.toml +sol-trade-sdk = "0.2.2" ``` ## 使用示例 @@ -51,7 +60,7 @@ sol-trade-sdk = { path = "./sol-trade-sdk", version = "0.2.1" } #### 1.1 使用 Yellowstone gRPC 订阅事件 ```rust -use sol_trade_sdk::{ +use sol_trade_sdk::solana_streamer_sdk::{ streaming::{ event_parser::{ protocols::{ @@ -128,7 +137,7 @@ async fn test_grpc() -> Result<(), Box> { #### 1.2 使用 ShredStream 订阅事件 ```rust -use sol_trade_sdk::streaming::ShredStreamGrpc; +use sol_trade_sdk::solana_streamer_sdk::streaming::ShredStreamGrpc; async fn test_shreds() -> Result<(), Box> { // 使用 ShredStream 客户端订阅事件 @@ -653,18 +662,6 @@ src/ ├── common/ # 通用功能和工具 ├── constants/ # 常量定义 ├── instruction/ # 指令构建 -├── streaming/ # 事件流处理 -│ ├── event_parser/ # 事件解析系统 -│ │ ├── common/ # 通用事件解析工具 -│ │ ├── core/ # 核心解析特征和接口 -│ │ ├── protocols/# 协议特定解析器 -│ │ │ ├── bonk/ # Bonk事件解析 -│ │ │ ├── pumpfun/ # PumpFun事件解析 -│ │ │ ├── pumpswap/ # PumpSwap事件解析 -│ │ │ └── raydium_cpmm/ # Raydium CPMM事件解析 -│ │ └── factory.rs # 解析器工厂 -│ ├── shred_stream.rs # ShredStream客户端 -│ └── yellowstone_grpc.rs # Yellowstone gRPC客户端 ├── swqos/ # MEV服务客户端 ├── trading/ # 统一交易引擎 │ ├── common/ # 通用交易工具 diff --git a/src/common/bonding_curve.rs b/src/common/bonding_curve.rs index 2633623..b2fa95c 100755 --- a/src/common/bonding_curve.rs +++ b/src/common/bonding_curve.rs @@ -28,7 +28,8 @@ use serde::{Serialize, Deserialize}; use solana_sdk::pubkey::Pubkey; -use crate::{constants::pumpfun::global_constants::{INITIAL_REAL_TOKEN_RESERVES, INITIAL_VIRTUAL_SOL_RESERVES, INITIAL_VIRTUAL_TOKEN_RESERVES, TOKEN_TOTAL_SUPPLY}, streaming::event_parser::protocols::pumpfun::PumpFunTradeEvent, trading::pumpfun::common::{get_bonding_curve_pda, get_creator_vault_pda}}; +use crate::{constants::pumpfun::global_constants::{INITIAL_REAL_TOKEN_RESERVES, INITIAL_VIRTUAL_SOL_RESERVES, INITIAL_VIRTUAL_TOKEN_RESERVES, TOKEN_TOTAL_SUPPLY}, trading::pumpfun::common::{get_bonding_curve_pda, get_creator_vault_pda}}; +use crate::solana_streamer_sdk::streaming::event_parser::protocols::pumpfun::PumpFunTradeEvent; /// Represents the global configuration account for token pricing and fees #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/src/lib.rs b/src/lib.rs index 1c64f34..efd7532 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,10 +2,10 @@ pub mod common; pub mod constants; pub mod instruction; pub mod protos; -pub mod streaming; pub mod swqos; pub mod trading; pub mod utils; +pub use solana_streamer_sdk; use crate::swqos::SwqosConfig; use crate::trading::core::params::BonkParams; diff --git a/src/main.rs b/src/main.rs index 13764f7..f9412fc 100755 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,11 @@ use std::{str::FromStr, sync::Arc}; use sol_trade_sdk::{ common::{bonding_curve::BondingCurveAccount, AnyResult, PriorityFee, TradeConfig}, + swqos::{SwqosConfig, SwqosRegion}, + trading::{core::params::{BonkParams, PumpFunParams, RaydiumCpmmParams}, factory::DexType, raydium_cpmm::{common::{get_buy_token_amount, get_sell_sol_amount}}}, + SolanaTrade, +}; +use sol_trade_sdk::solana_streamer_sdk::{ match_event, streaming::{ event_parser::{ @@ -17,9 +22,6 @@ use sol_trade_sdk::{ }, ShredStreamGrpc, YellowstoneGrpc, }, - swqos::{SwqosConfig, SwqosRegion}, - trading::{core::params::{BonkParams, PumpFunParams, RaydiumCpmmParams}, factory::DexType, raydium_cpmm::{common::{get_buy_token_amount, get_sell_sol_amount}}}, - SolanaTrade, }; use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Keypair}; diff --git a/src/protos/mod.rs b/src/protos/mod.rs index c0a3978..0bf6c4d 100755 --- a/src/protos/mod.rs +++ b/src/protos/mod.rs @@ -6,7 +6,6 @@ pub mod packet; pub mod relayer; pub mod searcher; pub mod shared; -pub mod shredstream; pub mod trace_shred; pub mod convert; pub mod nextblock_grpc; diff --git a/src/protos/shredstream.rs b/src/protos/shredstream.rs deleted file mode 100755 index 95a36f5..0000000 --- a/src/protos/shredstream.rs +++ /dev/null @@ -1,279 +0,0 @@ -// This file is @generated by prost-build. -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct Heartbeat { - /// don't trust IP:PORT from tcp header since it can be tampered over the wire - /// `socket.ip` must match incoming packet's ip. this prevents spamming an unwitting destination - #[prost(message, optional, tag = "1")] - pub socket: ::core::option::Option, - /// regions for shredstream proxy to receive shreds from - /// list of valid regions: - #[prost(string, repeated, tag = "2")] - pub regions: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, -} -#[derive(Clone, Copy, PartialEq, ::prost::Message)] -pub struct HeartbeatResponse { - /// client must respond within `ttl_ms` to keep stream alive - #[prost(uint32, tag = "1")] - pub ttl_ms: u32, -} -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct TraceShred { - /// source region, one of: - #[prost(string, tag = "1")] - pub region: ::prost::alloc::string::String, - /// timestamp of creation - #[prost(message, optional, tag = "2")] - pub created_at: ::core::option::Option<::prost_types::Timestamp>, - /// monotonically increases, resets upon service restart - #[prost(uint32, tag = "3")] - pub seq_num: u32, -} -/// tbd: we may want to add filters here -#[derive(Clone, Copy, PartialEq, ::prost::Message)] -pub struct SubscribeEntriesRequest {} -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct Entry { - /// the slot that the entry is from - #[prost(uint64, tag = "1")] - pub slot: u64, - /// Serialized bytes of Vec: - #[prost(bytes = "vec", tag = "2")] - pub entries: ::prost::alloc::vec::Vec, -} -/// Generated client implementations. -pub mod shredstream_client { - #![allow( - unused_variables, - dead_code, - missing_docs, - clippy::wildcard_imports, - clippy::let_unit_value, - )] - use tonic::codegen::*; - use tonic::codegen::http::Uri; - #[derive(Debug, Clone)] - pub struct ShredstreamClient { - inner: tonic::client::Grpc, - } - impl ShredstreamClient { - /// Attempt to create a new client by connecting to a given endpoint. - pub async fn connect(dst: D) -> Result - where - D: TryInto, - D::Error: Into, - { - let conn = tonic::transport::Endpoint::new(dst)?.connect().await?; - Ok(Self::new(conn)) - } - } - impl ShredstreamClient - where - T: tonic::client::GrpcService, - T::Error: Into, - T::ResponseBody: Body + std::marker::Send + 'static, - ::Error: Into + std::marker::Send, - { - pub fn new(inner: T) -> Self { - let inner = tonic::client::Grpc::new(inner); - Self { inner } - } - pub fn with_origin(inner: T, origin: Uri) -> Self { - let inner = tonic::client::Grpc::with_origin(inner, origin); - Self { inner } - } - pub fn with_interceptor( - inner: T, - interceptor: F, - ) -> ShredstreamClient> - where - F: tonic::service::Interceptor, - T::ResponseBody: Default, - T: tonic::codegen::Service< - http::Request, - Response = http::Response< - >::ResponseBody, - >, - >, - , - >>::Error: Into + std::marker::Send + std::marker::Sync, - { - ShredstreamClient::new(InterceptedService::new(inner, interceptor)) - } - /// Compress requests with the given encoding. - /// - /// This requires the server to support it otherwise it might respond with an - /// error. - #[must_use] - pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { - self.inner = self.inner.send_compressed(encoding); - self - } - /// Enable decompressing responses. - #[must_use] - pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { - self.inner = self.inner.accept_compressed(encoding); - self - } - /// Limits the maximum size of a decoded message. - /// - /// Default: `4MB` - #[must_use] - pub fn max_decoding_message_size(mut self, limit: usize) -> Self { - self.inner = self.inner.max_decoding_message_size(limit); - self - } - /// Limits the maximum size of an encoded message. - /// - /// Default: `usize::MAX` - #[must_use] - pub fn max_encoding_message_size(mut self, limit: usize) -> Self { - self.inner = self.inner.max_encoding_message_size(limit); - self - } - /// RPC endpoint to send heartbeats to keep shreds flowing - pub async fn send_heartbeat( - &mut self, - request: impl tonic::IntoRequest, - ) -> std::result::Result< - tonic::Response, - tonic::Status, - > { - self.inner - .ready() - .await - .map_err(|e| { - tonic::Status::unknown( - format!("Service was not ready: {}", e.into()), - ) - })?; - let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static( - "/shredstream.Shredstream/SendHeartbeat", - ); - let mut req = request.into_request(); - req.extensions_mut() - .insert(GrpcMethod::new("shredstream.Shredstream", "SendHeartbeat")); - self.inner.unary(req, path, codec).await - } - } -} -/// Generated client implementations. -pub mod shredstream_proxy_client { - #![allow( - unused_variables, - dead_code, - missing_docs, - clippy::wildcard_imports, - clippy::let_unit_value, - )] - use tonic::codegen::*; - use tonic::codegen::http::Uri; - #[derive(Debug, Clone)] - pub struct ShredstreamProxyClient { - inner: tonic::client::Grpc, - } - impl ShredstreamProxyClient { - /// Attempt to create a new client by connecting to a given endpoint. - pub async fn connect(dst: D) -> Result - where - D: TryInto, - D::Error: Into, - { - let conn = tonic::transport::Endpoint::new(dst)?.connect().await?; - Ok(Self::new(conn)) - } - } - impl ShredstreamProxyClient - where - T: tonic::client::GrpcService, - T::Error: Into, - T::ResponseBody: Body + std::marker::Send + 'static, - ::Error: Into + std::marker::Send, - { - pub fn new(inner: T) -> Self { - let inner = tonic::client::Grpc::new(inner); - Self { inner } - } - pub fn with_origin(inner: T, origin: Uri) -> Self { - let inner = tonic::client::Grpc::with_origin(inner, origin); - Self { inner } - } - pub fn with_interceptor( - inner: T, - interceptor: F, - ) -> ShredstreamProxyClient> - where - F: tonic::service::Interceptor, - T::ResponseBody: Default, - T: tonic::codegen::Service< - http::Request, - Response = http::Response< - >::ResponseBody, - >, - >, - , - >>::Error: Into + std::marker::Send + std::marker::Sync, - { - ShredstreamProxyClient::new(InterceptedService::new(inner, interceptor)) - } - /// Compress requests with the given encoding. - /// - /// This requires the server to support it otherwise it might respond with an - /// error. - #[must_use] - pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { - self.inner = self.inner.send_compressed(encoding); - self - } - /// Enable decompressing responses. - #[must_use] - pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { - self.inner = self.inner.accept_compressed(encoding); - self - } - /// Limits the maximum size of a decoded message. - /// - /// Default: `4MB` - #[must_use] - pub fn max_decoding_message_size(mut self, limit: usize) -> Self { - self.inner = self.inner.max_decoding_message_size(limit); - self - } - /// Limits the maximum size of an encoded message. - /// - /// Default: `usize::MAX` - #[must_use] - pub fn max_encoding_message_size(mut self, limit: usize) -> Self { - self.inner = self.inner.max_encoding_message_size(limit); - self - } - pub async fn subscribe_entries( - &mut self, - request: impl tonic::IntoRequest, - ) -> std::result::Result< - tonic::Response>, - tonic::Status, - > { - self.inner - .ready() - .await - .map_err(|e| { - tonic::Status::unknown( - format!("Service was not ready: {}", e.into()), - ) - })?; - let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static( - "/shredstream.ShredstreamProxy/SubscribeEntries", - ); - let mut req = request.into_request(); - req.extensions_mut() - .insert( - GrpcMethod::new("shredstream.ShredstreamProxy", "SubscribeEntries"), - ); - self.inner.server_streaming(req, path, codec).await - } - } -} diff --git a/src/streaming/event_parser/common/mod.rs b/src/streaming/event_parser/common/mod.rs deleted file mode 100755 index a070106..0000000 --- a/src/streaming/event_parser/common/mod.rs +++ /dev/null @@ -1,54 +0,0 @@ -pub mod types; -pub mod utils; - -/// 自动生成UnifiedEvent trait实现的宏 -#[macro_export] -macro_rules! impl_unified_event { - // 带有自定义ID表达式的版本 - ($struct_name:ident, $($field:ident),*) => { - impl $crate::streaming::event_parser::core::traits::UnifiedEvent for $struct_name { - fn id(&self) -> &str { - &self.metadata.id - } - - fn event_type(&self) -> $crate::streaming::event_parser::common::types::EventType { - self.metadata.event_type.clone() - } - - fn signature(&self) -> &str { - &self.metadata.signature - } - - fn slot(&self) -> u64 { - self.metadata.slot - } - - fn program_received_time_ms(&self) -> i64 { - self.metadata.program_received_time_ms - } - - fn as_any(&self) -> &dyn std::any::Any { - self - } - - fn as_any_mut(&mut self) -> &mut dyn std::any::Any { - self - } - - fn clone_boxed(&self) -> Box { - Box::new(self.clone()) - } - - fn merge(&mut self, other: Box) { - if let Some(e) = other.as_any().downcast_ref::<$struct_name>() { - $( - self.$field = e.$field.clone(); - )* - } - } - } - }; -} - -pub use types::*; -pub use utils::*; diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs deleted file mode 100755 index 0ad95f9..0000000 --- a/src/streaming/event_parser/common/types.rs +++ /dev/null @@ -1,173 +0,0 @@ -use borsh::{BorshDeserialize, BorshSerialize}; -use serde::{Deserialize, Serialize}; -use solana_sdk::pubkey::Pubkey; -use std::collections::hash_map::DefaultHasher; -use std::hash::{Hash, Hasher}; - -#[derive( - Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, -)] -pub enum ProtocolType { - #[default] - PumpSwap, - PumpFun, - Bonk, - RaydiumCpmm, - RaydiumClmm, -} - -/// 事件类型枚举 -#[derive( - Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, -)] -pub enum EventType { - // PumpSwap 事件 - #[default] - PumpSwapBuy, - PumpSwapSell, - PumpSwapCreatePool, - PumpSwapDeposit, - PumpSwapWithdraw, - - // PumpFun 事件 - PumpFunCreateToken, - PumpFunBuy, - PumpFunSell, - - // Bonk 事件 - BonkBuyExactIn, - BonkBuyExactOut, - BonkSellExactIn, - BonkSellExactOut, - BonkInitialize, - - // Raydium CPMM 事件 - RaydiumCpmmSwapBaseInput, - RaydiumCpmmSwapBaseOutput, - - // Raydium CLMM 事件 - RaydiumClmmSwap, - RaydiumClmmSwapV2, - - // 通用事件 - Unknown, -} - -impl EventType { - pub fn to_string(&self) -> String { - match self { - EventType::PumpSwapBuy => "PumpSwapBuy".to_string(), - EventType::PumpSwapSell => "PumpSwapSell".to_string(), - EventType::PumpSwapCreatePool => "PumpSwapCreatePool".to_string(), - EventType::PumpSwapDeposit => "PumpSwapDeposit".to_string(), - EventType::PumpSwapWithdraw => "PumpSwapWithdraw".to_string(), - EventType::PumpFunCreateToken => "PumpFunCreateToken".to_string(), - EventType::PumpFunBuy => "PumpFunBuy".to_string(), - EventType::PumpFunSell => "PumpFunSell".to_string(), - EventType::BonkBuyExactIn => "BonkBuyExactIn".to_string(), - EventType::BonkBuyExactOut => "BonkBuyExactOut".to_string(), - EventType::BonkSellExactIn => "BonkSellExactIn".to_string(), - EventType::BonkSellExactOut => "BonkSellExactOut".to_string(), - EventType::BonkInitialize => "BonkInitialize".to_string(), - EventType::RaydiumCpmmSwapBaseInput => "RaydiumCpmmSwapBaseInput".to_string(), - EventType::RaydiumCpmmSwapBaseOutput => "RaydiumCpmmSwapBaseOutput".to_string(), - EventType::RaydiumClmmSwap => "RaydiumClmmSwap".to_string(), - EventType::RaydiumClmmSwapV2 => "RaydiumClmmSwapV2".to_string(), - EventType::Unknown => "Unknown".to_string(), - } - } -} - -/// 解析结果 -#[derive(Debug, Clone)] -pub struct ParseResult { - pub success: bool, - pub data: Option, - pub error: Option, -} - -impl ParseResult { - pub fn success(data: T) -> Self { - Self { - success: true, - data: Some(data), - error: None, - } - } - - pub fn failure(error: String) -> Self { - Self { - success: false, - data: None, - error: Some(error), - } - } - - pub fn is_success(&self) -> bool { - self.success - } - - pub fn is_failure(&self) -> bool { - !self.success - } -} - -/// 协议信息 -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct ProtocolInfo { - pub name: String, - pub program_ids: Vec, -} - -impl ProtocolInfo { - pub fn new(name: String, program_ids: Vec) -> Self { - Self { name, program_ids } - } - - pub fn supports_program(&self, program_id: &Pubkey) -> bool { - self.program_ids.contains(program_id) - } -} - -/// 事件元数据 -#[derive( - Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize, -)] -pub struct EventMetadata { - pub id: String, - pub signature: String, - pub slot: u64, - pub program_received_time_ms: i64, - pub protocol: ProtocolType, - pub event_type: EventType, - pub program_id: Pubkey, -} - -impl EventMetadata { - pub fn new( - id: String, - signature: String, - slot: u64, - protocol: ProtocolType, - event_type: EventType, - program_id: Pubkey, - ) -> Self { - Self { - id, - signature, - slot, - program_received_time_ms: chrono::Utc::now().timestamp_millis(), - protocol, - event_type, - program_id, - } - } - pub fn set_id(&mut self, id: String) { - let _id = format!("{}-{}-{}", self.signature, self.event_type.to_string(), id); - // 对传入的 id 进行哈希处理 - let mut hasher = DefaultHasher::new(); - _id.hash(&mut hasher); - let hash_value = hasher.finish(); - self.id = format!("{:x}", hash_value); - } -} diff --git a/src/streaming/event_parser/common/utils.rs b/src/streaming/event_parser/common/utils.rs deleted file mode 100755 index 5452cfc..0000000 --- a/src/streaming/event_parser/common/utils.rs +++ /dev/null @@ -1,111 +0,0 @@ -use base64::engine::general_purpose; -use base64::Engine; -use std::time::{SystemTime, UNIX_EPOCH}; - -/// 获取当前时间戳 -pub fn current_timestamp() -> i64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("Time went backwards") - .as_secs() as i64 -} - -/// 从base64字符串解码数据 -pub fn decode_base64(data: &str) -> Result, base64::DecodeError> { - general_purpose::STANDARD.decode(data) -} - -/// 将数据编码为base64字符串 -pub fn encode_base64(data: &[u8]) -> String { - general_purpose::STANDARD.encode(data) -} - -/// 从字节数组中提取鉴别器和剩余数据 -pub fn extract_discriminator(length: usize, data: &[u8]) -> Option<(&[u8], &[u8])> { - if data.len() < length { - return None; - } - Some((&data[..length], &data[length..])) -} - -/// 检查鉴别器是否匹配 -pub fn discriminator_matches(data: &str, expected: &str) -> bool { - if data.len() < expected.len() { - return false; - } - &data[..expected.len()] == expected -} - -/// 从日志中提取程序数据 -pub fn extract_program_data(log: &str) -> Option<&str> { - const PROGRAM_DATA_PREFIX: &str = "Program data: "; - log.strip_prefix(PROGRAM_DATA_PREFIX) -} - -/// 从日志中提取程序日志 -pub fn extract_program_log<'a>(log: &'a str, prefix: &str) -> Option<&'a str> { - log.strip_prefix(prefix) -} - -/// 安全地从字节数组中读取u64 -pub fn read_u64_le(data: &[u8], offset: usize) -> Option { - if data.len() < offset + 8 { - return None; - } - let bytes: [u8; 8] = data[offset..offset + 8].try_into().ok()?; - Some(u64::from_le_bytes(bytes)) -} - -pub fn read_u128_le(data: &[u8], offset: usize) -> Option { - if data.len() < offset + 16 { - return None; - } - let bytes: [u8; 16] = data[offset..offset + 16].try_into().ok()?; - Some(u128::from_le_bytes(bytes)) -} - -pub fn read_u8_le(data: &[u8], offset: usize) -> Option { - if data.len() < offset + 1 { - return None; - } - let bytes: [u8; 1] = data[offset..offset + 1].try_into().ok()?; - Some(u8::from_le_bytes(bytes)) -} - -/// 安全地从字节数组中读取u32 -pub fn read_u32_le(data: &[u8], offset: usize) -> Option { - if data.len() < offset + 4 { - return None; - } - let bytes: [u8; 4] = data[offset..offset + 4].try_into().ok()?; - Some(u32::from_le_bytes(bytes)) -} - -/// 安全地从字节数组中读取u16 -pub fn read_u16_le(data: &[u8], offset: usize) -> Option { - if data.len() < offset + 2 { - return None; - } - let bytes: [u8; 2] = data[offset..offset + 2].try_into().ok()?; - Some(u16::from_le_bytes(bytes)) -} - -/// 安全地从字节数组中读取u8 -pub fn read_u8(data: &[u8], offset: usize) -> Option { - data.get(offset).copied() -} - -/// 验证账户索引的有效性 -pub fn validate_account_indices(indices: &[u8], account_count: usize) -> bool { - indices.iter().all(|&idx| (idx as usize) < account_count) -} - -/// 格式化公钥为短字符串 -pub fn format_pubkey_short(pubkey: &solana_sdk::pubkey::Pubkey) -> String { - let s = pubkey.to_string(); - if s.len() <= 8 { - s - } else { - format!("{}...{}", &s[..4], &s[s.len() - 4..]) - } -} diff --git a/src/streaming/event_parser/core/mod.rs b/src/streaming/event_parser/core/mod.rs deleted file mode 100755 index 663ef85..0000000 --- a/src/streaming/event_parser/core/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -pub mod traits; -pub use traits::{EventParser, UnifiedEvent}; diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs deleted file mode 100755 index 331fe43..0000000 --- a/src/streaming/event_parser/core/traits.rs +++ /dev/null @@ -1,485 +0,0 @@ -use anyhow::Result; -use solana_sdk::{ - instruction::CompiledInstruction, pubkey::Pubkey, transaction::VersionedTransaction, -}; -use solana_transaction_status::{ - EncodedTransactionWithStatusMeta, UiCompiledInstruction, UiInstruction, -}; -use std::fmt::Debug; -use std::{collections::HashMap, str::FromStr}; - -use crate::streaming::event_parser::{ - common::{utils::*, EventMetadata, EventType, ProtocolType}, - protocols::{ - bonk::{BonkPoolCreateEvent, BonkTradeEvent}, - pumpfun::{PumpFunCreateTokenEvent, PumpFunTradeEvent}, - }, -}; - -/// 统一事件接口 - 所有协议的事件都需要实现此trait -pub trait UnifiedEvent: Debug + Send + Sync { - /// 获取事件ID - fn id(&self) -> &str; - - /// 获取事件类型 - fn event_type(&self) -> EventType; - - /// 获取交易签名 - fn signature(&self) -> &str; - - /// 获取槽位号 - fn slot(&self) -> u64; - - /// 获取程序接收的时间戳(毫秒) - fn program_received_time_ms(&self) -> i64; - - /// 将事件转换为Any以便向下转型 - fn as_any(&self) -> &dyn std::any::Any; - - /// 将事件转换为可变Any以便向下转型 - fn as_any_mut(&mut self) -> &mut dyn std::any::Any; - - /// 克隆事件 - fn clone_boxed(&self) -> Box; - - /// 合并事件（可选实现） - fn merge(&mut self, _other: Box) { - // 默认实现：不进行任何合并操作 - } -} - -/// 事件解析器trait - 定义了事件解析的核心方法 -#[async_trait::async_trait] -pub trait EventParser: Send + Sync { - /// 从内联指令中解析事件数据 - fn parse_events_from_inner_instruction( - &self, - instruction: &UiCompiledInstruction, - signature: &str, - slot: u64, - ) -> Vec>; - - /// 从指令中解析事件数据 - fn parse_events_from_instruction( - &self, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: &str, - slot: u64, - ) -> Vec>; - - /// 从VersionedTransaction中解析指令事件的通用方法 - async fn parse_instruction_events_from_versioned_transaction( - &self, - versioned_tx: &VersionedTransaction, - signature: &str, - slot: Option, - accounts: &[Pubkey], - ) -> Result>> { - let mut instruction_events = Vec::new(); - // 获取交易的指令和账户 - let compiled_instructions = versioned_tx.message.instructions(); - let mut accounts: Vec = accounts.to_vec(); - - // 检查交易中是否包含程序 - let has_program = accounts.iter().any(|account| self.should_handle(account)); - if has_program { - // 解析每个指令 - for instruction in compiled_instructions { - if let Some(program_id) = accounts.get(instruction.program_id_index as usize) { - if self.should_handle(program_id) { - let max_idx = instruction.accounts.iter().max().unwrap_or(&0); - // 补齐accounts(使用Pubkey::default()) - if *max_idx as usize > accounts.len() { - for _i in accounts.len()..*max_idx as usize { - accounts.push(Pubkey::default()); - } - } - if let Ok(events) = self - .parse_instruction(instruction, &accounts, signature, slot) - .await - { - instruction_events.extend(events); - } - } - } - } - } - Ok(instruction_events) - } - - async fn parse_versioned_transaction( - &self, - versioned_tx: &VersionedTransaction, - signature: &str, - slot: Option, - bot_wallet: Option, - ) -> Result>> { - let accounts: Vec = versioned_tx.message.static_account_keys().to_vec(); - let events = self - .parse_instruction_events_from_versioned_transaction( - versioned_tx, - signature, - slot, - &accounts, - ) - .await - .unwrap_or_else(|_e| vec![]); - Ok(self.process_events(events, bot_wallet)) - } - - async fn parse_transaction( - &self, - tx: EncodedTransactionWithStatusMeta, - signature: &str, - slot: Option, - bot_wallet: Option, - ) -> Result>> { - let transaction = tx.transaction; - // 检查交易元数据 - let meta = tx - .meta - .as_ref() - .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; - - let mut address_table_lookups: Vec = vec![]; - if meta.err.is_none() { - let loaded_addresses = meta.loaded_addresses.as_ref().unwrap(); - for lookup in &loaded_addresses.writable { - address_table_lookups.push(Pubkey::from_str(lookup).unwrap()); - } - for lookup in &loaded_addresses.readonly { - address_table_lookups.push(Pubkey::from_str(lookup).unwrap()); - } - } - let mut accounts: Vec = vec![]; - - let mut instruction_events = Vec::new(); - - // 解析指令事件 - if let Some(versioned_tx) = transaction.decode() { - accounts = versioned_tx.message.static_account_keys().to_vec(); - accounts.extend(address_table_lookups.clone()); - - instruction_events = self - .parse_instruction_events_from_versioned_transaction( - &versioned_tx, - signature, - slot, - &accounts, - ) - .await - .unwrap_or_else(|_e| vec![]); - } else { - accounts.extend(address_table_lookups.clone()); - } - - // 解析内联指令事件 - let mut inner_instruction_events = Vec::new(); - // 检查交易是否成功 - if meta.err.is_none() { - let inner_instructions = meta.inner_instructions.as_ref().unwrap(); - for inner_instruction in inner_instructions { - for instruction in &inner_instruction.instructions { - match instruction { - UiInstruction::Compiled(compiled) => { - // 解析嵌套指令 - let compiled_instruction = CompiledInstruction { - program_id_index: compiled.program_id_index, - accounts: compiled.accounts.clone(), - data: bs58::decode(compiled.data.clone()).into_vec().unwrap(), - }; - if let Ok(events) = self - .parse_instruction( - &compiled_instruction, - &accounts, - signature, - slot, - ) - .await - { - instruction_events.extend(events); - } - if let Ok(events) = self - .parse_inner_instruction(compiled, signature, slot) - .await - { - inner_instruction_events.extend(events); - } - } - _ => {} - } - } - } - } - - if instruction_events.len() > 0 && inner_instruction_events.len() > 0 { - for instruction_event in &mut instruction_events { - for inner_instruction_event in &inner_instruction_events { - if instruction_event.id() == inner_instruction_event.id() - && instruction_event.event_type() == inner_instruction_event.event_type() - { - instruction_event.merge(inner_instruction_event.clone_boxed()); - break; - } - } - } - } - Ok(self.process_events(instruction_events, bot_wallet)) - } - - fn process_events( - &self, - mut events: Vec>, - bot_wallet: Option, - ) -> Vec> { - let mut dev_address = None; - let mut bonk_dev_address = None; - for event in &mut events { - if let Some(token_info) = event.as_any().downcast_ref::() { - dev_address = Some(token_info.user); - } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() - { - if Some(trade_info.user) == dev_address { - trade_info.is_dev_create_token_trade = true; - } else if Some(trade_info.user) == bot_wallet { - trade_info.is_bot = true; - } else { - trade_info.is_dev_create_token_trade = false; - } - } - if let Some(pool_info) = event.as_any().downcast_ref::() { - bonk_dev_address = Some(pool_info.creator); - } else if let Some(trade_info) = event.as_any_mut().downcast_mut::() { - if Some(trade_info.payer) == bonk_dev_address { - trade_info.is_dev_create_token_trade = true; - } else if Some(trade_info.payer) == bot_wallet { - trade_info.is_bot = true; - } else { - trade_info.is_dev_create_token_trade = false; - } - } - } - events - } - - async fn parse_inner_instruction( - &self, - instruction: &UiCompiledInstruction, - signature: &str, - slot: Option, - ) -> Result>> { - let slot = slot.unwrap_or(0); - let events = self.parse_events_from_inner_instruction(instruction, signature, slot); - Ok(events) - } - - async fn parse_instruction( - &self, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: &str, - slot: Option, - ) -> Result>> { - let slot = slot.unwrap_or(0); - let events = self.parse_events_from_instruction(instruction, accounts, signature, slot); - Ok(events) - } - - /// 检查是否应该处理此程序ID - fn should_handle(&self, program_id: &Pubkey) -> bool; - - /// 获取支持的程序ID列表 - fn supported_program_ids(&self) -> Vec; -} - -// 为Box实现Clone -impl Clone for Box { - fn clone(&self) -> Self { - self.clone_boxed() - } -} - -/// 通用事件解析器配置 -#[derive(Debug, Clone)] -pub struct GenericEventParseConfig { - pub inner_instruction_discriminator: &'static str, - pub instruction_discriminator: &'static [u8], - pub event_type: EventType, - pub inner_instruction_parser: InnerInstructionEventParser, - pub instruction_parser: InstructionEventParser, -} - -/// 内联指令事件解析器 -pub type InnerInstructionEventParser = - fn(data: &[u8], metadata: EventMetadata) -> Option>; - -/// 指令事件解析器 -pub type InstructionEventParser = - fn(data: &[u8], accounts: &[Pubkey], metadata: EventMetadata) -> Option>; - -/// 通用事件解析器基类 -pub struct GenericEventParser { - program_id: Pubkey, - protocol_type: ProtocolType, - inner_instruction_configs: HashMap<&'static str, Vec>, - instruction_configs: HashMap, Vec>, -} - -impl GenericEventParser { - /// 创建新的通用事件解析器 - pub fn new( - program_id: Pubkey, - protocol_type: ProtocolType, - configs: Vec, - ) -> Self { - let mut inner_instruction_configs = HashMap::new(); - let mut instruction_configs = HashMap::new(); - - for config in configs { - inner_instruction_configs - .entry(config.inner_instruction_discriminator) - .or_insert(vec![]) - .push(config.clone()); - instruction_configs - .entry(config.instruction_discriminator.to_vec()) - .or_insert(vec![]) - .push(config); - } - - Self { - program_id, - protocol_type, - inner_instruction_configs, - instruction_configs, - } - } - - /// 通用的内联指令解析方法 - fn parse_inner_instruction_event( - &self, - config: &GenericEventParseConfig, - data: &[u8], - signature: &str, - slot: u64, - ) -> Option> { - let metadata = EventMetadata::new( - signature.to_string(), - signature.to_string(), - slot, - self.protocol_type.clone(), - config.event_type.clone(), - self.program_id, - ); - (config.inner_instruction_parser)(data, metadata) - } - - /// 通用的指令解析方法 - fn parse_instruction_event( - &self, - config: &GenericEventParseConfig, - data: &[u8], - account_pubkeys: &[Pubkey], - signature: &str, - slot: u64, - ) -> Option> { - let metadata = EventMetadata::new( - signature.to_string(), - signature.to_string(), - slot, - self.protocol_type.clone(), - config.event_type.clone(), - self.program_id, - ); - (config.instruction_parser)(data, account_pubkeys, metadata) - } -} - -#[async_trait::async_trait] -impl EventParser for GenericEventParser { - /// 从内联指令中解析事件数据 - fn parse_events_from_inner_instruction( - &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, - slot: u64, - ) -> Vec> { - let inner_instruction_data = inner_instruction.data.clone(); - let inner_instruction_data_decoded = - bs58::decode(inner_instruction_data).into_vec().unwrap(); - if inner_instruction_data_decoded.len() < 16 { - return Vec::new(); - } - let inner_instruction_data_decoded_str = - format!("0x{}", hex::encode(&inner_instruction_data_decoded)); - let data = &inner_instruction_data_decoded[16..]; - let mut events = Vec::new(); - for (disc, configs) in &self.inner_instruction_configs { - if discriminator_matches(&inner_instruction_data_decoded_str, disc) { - for config in configs { - if let Some(event) = - self.parse_inner_instruction_event(config, data, signature, slot) - { - events.push(event); - } - } - } - } - events - } - - /// 从指令中解析事件 - fn parse_events_from_instruction( - &self, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: &str, - slot: u64, - ) -> Vec> { - let program_id = accounts[instruction.program_id_index as usize]; - if !self.should_handle(&program_id) { - return Vec::new(); - } - let mut events = Vec::new(); - for (disc, configs) in &self.instruction_configs { - if instruction.data.len() < disc.len() { - continue; - } - let discriminator = &instruction.data[..disc.len()]; - let data = &instruction.data[disc.len()..]; - if discriminator == disc { - // 验证账户索引 - if !validate_account_indices(&instruction.accounts, accounts.len()) { - continue; - } - - let account_pubkeys: Vec = instruction - .accounts - .iter() - .map(|&idx| accounts[idx as usize]) - .collect(); - for config in configs { - if let Some(event) = self.parse_instruction_event( - config, - data, - &account_pubkeys, - signature, - slot, - ) { - events.push(event); - } - } - } - } - - events - } - - fn should_handle(&self, program_id: &Pubkey) -> bool { - *program_id == self.program_id - } - - fn supported_program_ids(&self) -> Vec { - vec![self.program_id] - } -} diff --git a/src/streaming/event_parser/factory.rs b/src/streaming/event_parser/factory.rs deleted file mode 100755 index b7830d5..0000000 --- a/src/streaming/event_parser/factory.rs +++ /dev/null @@ -1,98 +0,0 @@ -use anyhow::{anyhow, Result}; -use solana_sdk::pubkey::Pubkey; -use std::sync::Arc; - -use crate::streaming::event_parser::protocols::{ - bonk::parser::BONK_PROGRAM_ID, pumpfun::parser::PUMPFUN_PROGRAM_ID, - pumpswap::parser::PUMPSWAP_PROGRAM_ID, raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID, - raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID, BonkEventParser, RaydiumCpmmEventParser, - RaydiumClmmEventParser, -}; - -use super::{ - core::traits::EventParser, - protocols::{pumpfun::PumpFunEventParser, pumpswap::PumpSwapEventParser}, -}; - -/// 支持的协议 -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub enum Protocol { - PumpSwap, - PumpFun, - Bonk, - RaydiumCpmm, - RaydiumClmm, -} - -impl Protocol { - pub fn get_program_id(&self) -> Vec { - match self { - Protocol::PumpSwap => vec![PUMPSWAP_PROGRAM_ID], - Protocol::PumpFun => vec![PUMPFUN_PROGRAM_ID], - Protocol::Bonk => vec![BONK_PROGRAM_ID], - Protocol::RaydiumCpmm => vec![RAYDIUM_CPMM_PROGRAM_ID], - Protocol::RaydiumClmm => vec![RAYDIUM_CLMM_PROGRAM_ID], - } - } -} - -impl std::fmt::Display for Protocol { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Protocol::PumpSwap => write!(f, "PumpSwap"), - Protocol::PumpFun => write!(f, "PumpFun"), - Protocol::Bonk => write!(f, "Bonk"), - Protocol::RaydiumCpmm => write!(f, "RaydiumCpmm"), - Protocol::RaydiumClmm => write!(f, "RaydiumClmm"), - } - } -} - -impl std::str::FromStr for Protocol { - type Err = anyhow::Error; - - fn from_str(s: &str) -> Result { - match s.to_lowercase().as_str() { - "pumpswap" => Ok(Protocol::PumpSwap), - "pumpfun" => Ok(Protocol::PumpFun), - "bonk" => Ok(Protocol::Bonk), - "raydiumcpmm" => Ok(Protocol::RaydiumCpmm), - "raydiumclmm" => Ok(Protocol::RaydiumClmm), - _ => Err(anyhow!("Unsupported protocol: {}", s)), - } - } -} - -/// 事件解析器工厂 - 用于创建不同协议的事件解析器 -pub struct EventParserFactory; - -impl EventParserFactory { - /// 创建指定协议的事件解析器 - pub fn create_parser(protocol: Protocol) -> Arc { - match protocol { - Protocol::PumpSwap => Arc::new(PumpSwapEventParser::new()), - Protocol::PumpFun => Arc::new(PumpFunEventParser::new()), - Protocol::Bonk => Arc::new(BonkEventParser::new()), - Protocol::RaydiumCpmm => Arc::new(RaydiumCpmmEventParser::new()), - Protocol::RaydiumClmm => Arc::new(RaydiumClmmEventParser::new()), - } - } - - /// 创建所有协议的事件解析器 - pub fn create_all_parsers() -> Vec> { - Self::supported_protocols() - .into_iter() - .map(Self::create_parser) - .collect() - } - - /// 获取所有支持的协议 - pub fn supported_protocols() -> Vec { - vec![Protocol::PumpSwap] - } - - /// 检查协议是否支持 - pub fn is_supported(protocol: &Protocol) -> bool { - Self::supported_protocols().contains(protocol) - } -} diff --git a/src/streaming/event_parser/mod.rs b/src/streaming/event_parser/mod.rs deleted file mode 100755 index 99d71eb..0000000 --- a/src/streaming/event_parser/mod.rs +++ /dev/null @@ -1,41 +0,0 @@ -pub mod common; -pub mod core; -pub mod factory; -pub mod protocols; - -pub use core::traits::{EventParser, UnifiedEvent}; -pub use factory::{EventParserFactory, Protocol}; - -/// 宏：简化 downcast_ref 模式匹配 -/// -/// # 使用示例 -/// ``` -/// use sol_trade_sdk::event_parser::match_event; -/// -/// match_event!(event, { -/// PumpSwapCreatePoolEvent => |typed_event| { -/// println!("CreatePool event: {:?}", typed_event); -/// }, -/// PumpSwapDepositEvent => |typed_event| { -/// // 处理存款事件 -/// }, -/// }); -/// ``` -#[macro_export] -macro_rules! match_event { - ($event:expr, { - $($event_type:ty => $handler:expr),* $(,)? - }) => { - $( - if let Some(typed_event) = $event.as_any().downcast_ref::<$event_type>() { - $handler(typed_event.clone()); - } else - )* - { - // 默认情况：什么都不做 - } - }; -} - -// 重新导出宏以便于使用 -pub use match_event; diff --git a/src/streaming/event_parser/protocols/bonk/events.rs b/src/streaming/event_parser/protocols/bonk/events.rs deleted file mode 100755 index 93cea47..0000000 --- a/src/streaming/event_parser/protocols/bonk/events.rs +++ /dev/null @@ -1,126 +0,0 @@ -use crate::streaming::event_parser::protocols::bonk::types::{ - CurveParams, MintParams, PoolStatus, TradeDirection, VestingParams, -}; -use crate::streaming::event_parser::common::EventMetadata; -use crate::impl_unified_event; -use borsh::BorshDeserialize; -use serde::{Deserialize, Serialize}; -use solana_sdk::pubkey::Pubkey; - -/// 买入事件 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct BonkTradeEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub pool_state: Pubkey, - pub total_base_sell: u64, - pub virtual_base: u64, - pub virtual_quote: u64, - pub real_base_before: u64, - pub real_quote_before: u64, - pub real_base_after: u64, - pub real_quote_after: u64, - pub amount_in: u64, - pub amount_out: u64, - pub protocol_fee: u64, - pub platform_fee: u64, - pub share_fee: u64, - pub trade_direction: TradeDirection, - pub pool_status: PoolStatus, - #[borsh(skip)] - pub minimum_amount_out: u64, - #[borsh(skip)] - pub maximum_amount_in: u64, - #[borsh(skip)] - pub share_fee_rate: u64, - #[borsh(skip)] - pub payer: Pubkey, - #[borsh(skip)] - pub user_base_token: Pubkey, - #[borsh(skip)] - pub user_quote_token: Pubkey, - #[borsh(skip)] - pub base_vault: Pubkey, - #[borsh(skip)] - pub quote_vault: Pubkey, - #[borsh(skip)] - pub base_token_mint: Pubkey, - #[borsh(skip)] - pub quote_token_mint: Pubkey, - #[borsh(skip)] - pub is_dev_create_token_trade: bool, - #[borsh(skip)] - pub is_bot: bool, -} - -// 使用宏生成UnifiedEvent实现，指定需要合并的字段 -impl_unified_event!( - BonkTradeEvent, - pool_state, - total_base_sell, - virtual_base, - virtual_quote, - real_base_before, - real_quote_before, - real_base_after, - real_quote_after, - amount_in, - amount_out, - protocol_fee, - platform_fee, - share_fee, - trade_direction, - pool_status -); - -/// 创建池事件 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct BonkPoolCreateEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub pool_state: Pubkey, - pub creator: Pubkey, - pub config: Pubkey, - pub base_mint_param: MintParams, - pub curve_param: CurveParams, - pub vesting_param: VestingParams, - #[borsh(skip)] - pub payer: Pubkey, - #[borsh(skip)] - pub base_mint: Pubkey, - #[borsh(skip)] - pub quote_mint: Pubkey, - #[borsh(skip)] - pub base_vault: Pubkey, - #[borsh(skip)] - pub quote_vault: Pubkey, - #[borsh(skip)] - pub global_config: Pubkey, - #[borsh(skip)] - pub platform_config: Pubkey, -} - -// 使用宏生成UnifiedEvent实现，指定需要合并的字段 -impl_unified_event!( - BonkPoolCreateEvent, - pool_state, - creator, - config, - base_mint_param, - curve_param, - vesting_param -); - -/// 事件鉴别器常量 -pub mod discriminators { - // 事件鉴别器 - pub const TRADE_EVENT: &str = "0xe445a52e51cb9a1dbddb7fd34ee661ee"; - pub const POOL_CREATE_EVENT: &str = "0xe445a52e51cb9a1d97d7e20976a173ae"; - - // 指令鉴别器 - pub const BUY_EXACT_IN: &[u8] = &[250, 234, 13, 123, 213, 156, 19, 236]; - pub const BUY_EXACT_OUT: &[u8] = &[24, 211, 116, 40, 105, 3, 153, 56]; - pub const SELL_EXACT_IN: &[u8] = &[149, 39, 222, 155, 211, 124, 152, 26]; - pub const SELL_EXACT_OUT: &[u8] = &[95, 200, 71, 34, 8, 9, 11, 166]; - pub const INITIALIZE: &[u8] = &[175, 175, 109, 31, 13, 152, 155, 237]; -} diff --git a/src/streaming/event_parser/protocols/bonk/mod.rs b/src/streaming/event_parser/protocols/bonk/mod.rs deleted file mode 100755 index 8c7a629..0000000 --- a/src/streaming/event_parser/protocols/bonk/mod.rs +++ /dev/null @@ -1,7 +0,0 @@ -pub mod events; -pub mod parser; -pub mod types; - -pub use events::*; -pub use parser::BonkEventParser; -pub use types::*; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs deleted file mode 100755 index 4e20a6f..0000000 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ /dev/null @@ -1,445 +0,0 @@ -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; - -use crate::streaming::event_parser::{ - common::{utils::*, EventMetadata, EventType, ProtocolType}, - core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, - protocols::bonk::{ - discriminators, BonkPoolCreateEvent, BonkTradeEvent, ConstantCurve, CurveParams, - FixedCurve, LinearCurve, MintParams, TradeDirection, VestingParams, - }, -}; - -/// Bonk程序ID -pub const BONK_PROGRAM_ID: Pubkey = - solana_sdk::pubkey!("LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj"); - -/// Bonk事件解析器 -pub struct BonkEventParser { - inner: GenericEventParser, -} - -impl BonkEventParser { - pub fn new() -> Self { - // 配置所有事件类型 - let configs = vec![ - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::BUY_EXACT_IN, - event_type: EventType::BonkBuyExactIn, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_buy_exact_in_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::BUY_EXACT_OUT, - event_type: EventType::BonkBuyExactOut, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_buy_exact_out_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::SELL_EXACT_IN, - event_type: EventType::BonkSellExactIn, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_sell_exact_in_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::SELL_EXACT_OUT, - event_type: EventType::BonkSellExactOut, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_sell_exact_out_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT, - instruction_discriminator: discriminators::INITIALIZE, - event_type: EventType::BonkInitialize, - inner_instruction_parser: Self::parse_pool_create_inner_instruction, - instruction_parser: Self::parse_initialize_instruction, - }, - ]; - - let inner = GenericEventParser::new(BONK_PROGRAM_ID, ProtocolType::Bonk, configs); - - Self { inner } - } - - /// 解析创建池事件 - fn parse_pool_create_inner_instruction( - data: &[u8], - metadata: EventMetadata, - ) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!("{}", metadata.signature,)); - Some(Box::new(BonkPoolCreateEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析交易事件 - fn parse_trade_inner_instruction( - data: &[u8], - metadata: EventMetadata, - ) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}", - metadata.signature, - event.pool_state.to_string() - )); - if metadata.event_type == EventType::BonkBuyExactIn - || metadata.event_type == EventType::BonkBuyExactOut - { - if event.trade_direction != TradeDirection::Buy { - return None; - } - } else if metadata.event_type == EventType::BonkSellExactIn - || metadata.event_type == EventType::BonkSellExactOut - { - if event.trade_direction != TradeDirection::Sell { - return None; - } - } - Some(Box::new(BonkTradeEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析买入指令事件 - fn parse_buy_exact_in_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - - let amount_in = read_u64_le(data, 0)?; - let minimum_amount_out = read_u64_le(data, 8)?; - let share_fee_rate = read_u64_le(data, 16)?; - - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, accounts[4])); - - Some(Box::new(BonkTradeEvent { - metadata, - amount_in, - minimum_amount_out, - share_fee_rate, - payer: accounts[0], - pool_state: accounts[4], - user_base_token: accounts[5], - user_quote_token: accounts[6], - base_vault: accounts[7], - quote_vault: accounts[8], - base_token_mint: accounts[9], - quote_token_mint: accounts[10], - trade_direction: TradeDirection::Buy, - ..Default::default() - })) - } - - fn parse_buy_exact_out_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - - let amount_out = read_u64_le(data, 0)?; - let maximum_amount_in = read_u64_le(data, 8)?; - let share_fee_rate = read_u64_le(data, 16)?; - - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, accounts[4])); - - Some(Box::new(BonkTradeEvent { - metadata, - amount_out, - maximum_amount_in, - share_fee_rate, - payer: accounts[0], - pool_state: accounts[4], - user_base_token: accounts[5], - user_quote_token: accounts[6], - base_vault: accounts[7], - quote_vault: accounts[8], - base_token_mint: accounts[9], - quote_token_mint: accounts[10], - trade_direction: TradeDirection::Buy, - ..Default::default() - })) - } - - fn parse_sell_exact_in_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - - let amount_in = read_u64_le(data, 0)?; - let minimum_amount_out = read_u64_le(data, 8)?; - let share_fee_rate = read_u64_le(data, 16)?; - - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, accounts[4])); - - Some(Box::new(BonkTradeEvent { - metadata, - amount_in, - minimum_amount_out, - share_fee_rate, - payer: accounts[0], - pool_state: accounts[4], - user_base_token: accounts[5], - user_quote_token: accounts[6], - base_vault: accounts[7], - quote_vault: accounts[8], - base_token_mint: accounts[9], - quote_token_mint: accounts[10], - trade_direction: TradeDirection::Sell, - ..Default::default() - })) - } - - fn parse_sell_exact_out_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - - let amount_out = read_u64_le(data, 0)?; - let maximum_amount_in = read_u64_le(data, 8)?; - let share_fee_rate = read_u64_le(data, 16)?; - - let mut metadata = metadata; - metadata.set_id(format!("{}-{}", metadata.signature, accounts[4])); - - Some(Box::new(BonkTradeEvent { - metadata, - amount_out, - maximum_amount_in, - share_fee_rate, - payer: accounts[0], - pool_state: accounts[4], - user_base_token: accounts[5], - user_quote_token: accounts[6], - base_vault: accounts[7], - quote_vault: accounts[8], - base_token_mint: accounts[9], - quote_token_mint: accounts[10], - trade_direction: TradeDirection::Sell, - ..Default::default() - })) - } - - /// 解析初始化事件 - fn parse_initialize_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 24 { - return None; - } - - let mut offset = 0; - let base_mint_param = Self::parse_mint_params(data, &mut offset)?; - let curve_param = Self::parse_curve_params(data, &mut offset)?; - let vesting_param = Self::parse_vesting_params(data, &mut offset)?; - - let mut metadata = metadata; - metadata.set_id(format!("{}", metadata.signature)); - - Some(Box::new(BonkPoolCreateEvent { - metadata, - payer: accounts[0], - creator: accounts[1], - global_config: accounts[2], - platform_config: accounts[3], - pool_state: accounts[5], - base_mint: accounts[6], - quote_mint: accounts[7], - base_vault: accounts[8], - quote_vault: accounts[9], - base_mint_param, - curve_param, - vesting_param, - ..Default::default() - })) - } - - /// 解析 MintParams 结构 - fn parse_mint_params(data: &[u8], offset: &mut usize) -> Option { - // 读取decimals (1字节) - let decimals = read_u8(data, *offset)?; - *offset += 1; - - // 读取name字符串长度和内容 - let name_len = read_u32_le(data, *offset)? as usize; - *offset += 4; - if data.len() < *offset + name_len { - return None; - } - let name = String::from_utf8(data[*offset..*offset + name_len].to_vec()).ok()?; - *offset += name_len; - - // 读取symbol字符串长度和内容 - let symbol_len = read_u32_le(data, *offset)? as usize; - *offset += 4; - if data.len() < *offset + symbol_len { - return None; - } - let symbol = String::from_utf8(data[*offset..*offset + symbol_len].to_vec()).ok()?; - *offset += symbol_len; - - // 读取uri字符串长度和内容 - let uri_len = read_u32_le(data, *offset)? as usize; - *offset += 4; - if data.len() < *offset + uri_len { - return None; - } - let uri = String::from_utf8(data[*offset..*offset + uri_len].to_vec()).ok()?; - *offset += uri_len; - - Some(MintParams { - decimals, - name, - symbol, - uri, - }) - } - - /// 解析 CurveParams 结构 - fn parse_curve_params(data: &[u8], offset: &mut usize) -> Option { - // 读取curve类型标识符 (1字节) - let curve_type = read_u8(data, *offset)?; - *offset += 1; - - match curve_type { - 0 => { - // Constant curve - let supply = read_u64_le(data, *offset)?; - *offset += 8; - let total_base_sell = read_u64_le(data, *offset)?; - *offset += 8; - let total_quote_fund_raising = read_u64_le(data, *offset)?; - *offset += 8; - let migrate_type = read_u8(data, *offset)?; - *offset += 1; - - Some(CurveParams::Constant { - data: ConstantCurve { - supply, - total_base_sell, - total_quote_fund_raising, - migrate_type, - }, - }) - } - 1 => { - // Fixed curve - let supply = read_u64_le(data, *offset)?; - *offset += 8; - let total_quote_fund_raising = read_u64_le(data, *offset)?; - *offset += 8; - let migrate_type = read_u8(data, *offset)?; - *offset += 1; - - Some(CurveParams::Fixed { - data: FixedCurve { - supply, - total_quote_fund_raising, - migrate_type, - }, - }) - } - 2 => { - // Linear curve - let supply = read_u64_le(data, *offset)?; - *offset += 8; - let total_quote_fund_raising = read_u64_le(data, *offset)?; - *offset += 8; - let migrate_type = read_u8(data, *offset)?; - *offset += 1; - - Some(CurveParams::Linear { - data: LinearCurve { - supply, - total_quote_fund_raising, - migrate_type, - }, - }) - } - _ => None, - } - } - - /// 解析 VestingParams 结构 - fn parse_vesting_params(data: &[u8], offset: &mut usize) -> Option { - let total_locked_amount = read_u64_le(data, *offset)?; - *offset += 8; - let cliff_period = read_u64_le(data, *offset)?; - *offset += 8; - let unlock_period = read_u64_le(data, *offset)?; - *offset += 8; - - Some(VestingParams { - total_locked_amount, - cliff_period, - unlock_period, - }) - } -} - -#[async_trait::async_trait] -impl EventParser for BonkEventParser { - fn parse_events_from_inner_instruction( - &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot) - } - - fn parse_events_from_instruction( - &self, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot) - } - - fn should_handle(&self, program_id: &Pubkey) -> bool { - self.inner.should_handle(program_id) - } - - fn supported_program_ids(&self) -> Vec { - self.inner.supported_program_ids() - } -} diff --git a/src/streaming/event_parser/protocols/bonk/types.rs b/src/streaming/event_parser/protocols/bonk/types.rs deleted file mode 100755 index 5751f8c..0000000 --- a/src/streaming/event_parser/protocols/bonk/types.rs +++ /dev/null @@ -1,69 +0,0 @@ -use borsh::BorshDeserialize; -use serde::{Deserialize, Serialize}; - -#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub enum TradeDirection { - #[default] - Buy, - Sell, -} - -#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub enum PoolStatus { - #[default] - Fund, - Migrate, - Trade, -} - -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct MintParams { - pub decimals: u8, - pub name: String, - pub symbol: String, - pub uri: String, -} - -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct VestingParams { - pub total_locked_amount: u64, - pub cliff_period: u64, - pub unlock_period: u64, -} - -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct ConstantCurve { - pub supply: u64, - pub total_base_sell: u64, - pub total_quote_fund_raising: u64, - pub migrate_type: u8, -} - -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct FixedCurve { - pub supply: u64, - pub total_quote_fund_raising: u64, - pub migrate_type: u8, -} - -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct LinearCurve { - pub supply: u64, - pub total_quote_fund_raising: u64, - pub migrate_type: u8, -} - -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub enum CurveParams { - Constant { data: ConstantCurve }, - Fixed { data: FixedCurve }, - Linear { data: LinearCurve }, -} - -impl Default for CurveParams { - fn default() -> Self { - Self::Constant { - data: ConstantCurve::default(), - } - } -} diff --git a/src/streaming/event_parser/protocols/mod.rs b/src/streaming/event_parser/protocols/mod.rs deleted file mode 100755 index e0d8aa5..0000000 --- a/src/streaming/event_parser/protocols/mod.rs +++ /dev/null @@ -1,11 +0,0 @@ -pub mod pumpfun; -pub mod pumpswap; -pub mod bonk; -pub mod raydium_cpmm; -pub mod raydium_clmm; - -pub use pumpfun::PumpFunEventParser; -pub use pumpswap::PumpSwapEventParser; -pub use bonk::BonkEventParser; -pub use raydium_cpmm::RaydiumCpmmEventParser; -pub use raydium_clmm::RaydiumClmmEventParser; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/pumpfun/events.rs b/src/streaming/event_parser/protocols/pumpfun/events.rs deleted file mode 100755 index 99c04ac..0000000 --- a/src/streaming/event_parser/protocols/pumpfun/events.rs +++ /dev/null @@ -1,113 +0,0 @@ -use borsh::BorshDeserialize; -use serde::{Deserialize, Serialize}; -use solana_sdk::pubkey::Pubkey; - -use crate::streaming::event_parser::common::EventMetadata; -use crate::impl_unified_event; - -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct PumpFunCreateTokenEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub name: String, - pub symbol: String, - pub uri: String, - pub mint: Pubkey, - pub bonding_curve: Pubkey, - pub user: Pubkey, - pub creator: Pubkey, - pub timestamp: i64, - pub virtual_token_reserves: u64, - pub virtual_sol_reserves: u64, - pub real_token_reserves: u64, - pub token_total_supply: u64, - #[borsh(skip)] - pub mint_authority: Pubkey, - #[borsh(skip)] - pub associated_bonding_curve: Pubkey, -} - -impl_unified_event!( - PumpFunCreateTokenEvent, - mint, - bonding_curve, - user, - creator, - timestamp, - virtual_token_reserves, - virtual_sol_reserves, - real_token_reserves, - token_total_supply -); - -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct PumpFunTradeEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub mint: Pubkey, - pub sol_amount: u64, - pub token_amount: u64, - pub is_buy: bool, - pub user: Pubkey, - pub timestamp: i64, - pub virtual_sol_reserves: u64, - pub virtual_token_reserves: u64, - pub real_sol_reserves: u64, - pub real_token_reserves: u64, - pub fee_recipient: Pubkey, - pub fee_basis_points: u64, - pub fee: u64, - pub creator: Pubkey, - pub creator_fee_basis_points: u64, - pub creator_fee: u64, - #[borsh(skip)] - pub bonding_curve: Pubkey, - #[borsh(skip)] - pub associated_bonding_curve: Pubkey, - #[borsh(skip)] - pub associated_user: Pubkey, - #[borsh(skip)] - pub creator_vault: Pubkey, - #[borsh(skip)] - pub max_sol_cost: u64, - #[borsh(skip)] - pub min_sol_output: u64, - #[borsh(skip)] - pub amount: u64, - #[borsh(skip)] - pub is_bot: bool, - #[borsh(skip)] - pub is_dev_create_token_trade: bool, // 是否是dev创建token的交易 -} - -impl_unified_event!( - PumpFunTradeEvent, - mint, - sol_amount, - token_amount, - is_buy, - user, - timestamp, - virtual_sol_reserves, - virtual_token_reserves, - real_sol_reserves, - real_token_reserves, - fee_recipient, - fee_basis_points, - fee, - creator, - creator_fee_basis_points, - creator_fee -); - -/// 事件鉴别器常量 -pub mod discriminators { - // 事件鉴别器 - pub const CREATE_TOKEN_EVENT: &str = "0xe445a52e51cb9a1d1b72a94ddeeb6376"; - pub const TRADE_EVENT: &str = "0xe445a52e51cb9a1dbddb7fd34ee661ee"; - - // 指令鉴别器 - pub const CREATE_TOKEN_IX: &[u8] = &[24, 30, 200, 40, 5, 28, 7, 119]; - pub const BUY_IX: &[u8] = &[102, 6, 61, 18, 1, 218, 235, 234]; - pub const SELL_IX: &[u8] = &[51, 230, 133, 164, 1, 127, 131, 173]; -} diff --git a/src/streaming/event_parser/protocols/pumpfun/mod.rs b/src/streaming/event_parser/protocols/pumpfun/mod.rs deleted file mode 100755 index eee7acf..0000000 --- a/src/streaming/event_parser/protocols/pumpfun/mod.rs +++ /dev/null @@ -1,5 +0,0 @@ -pub mod events; -pub mod parser; - -pub use events::*; -pub use parser::PumpFunEventParser; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs deleted file mode 100755 index 4a19b7e..0000000 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ /dev/null @@ -1,250 +0,0 @@ -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; - -use crate::streaming::event_parser::{ - common::{EventMetadata, EventType, ProtocolType}, - core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, - protocols::pumpfun::{discriminators, PumpFunCreateTokenEvent, PumpFunTradeEvent}, -}; - -/// PumpFun程序ID -pub const PUMPFUN_PROGRAM_ID: Pubkey = - solana_sdk::pubkey!("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"); - -/// PumpFun事件解析器 -pub struct PumpFunEventParser { - inner: GenericEventParser, -} - -impl PumpFunEventParser { - pub fn new() -> Self { - // 配置所有事件类型 - let configs = vec![ - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::CREATE_TOKEN_EVENT, - instruction_discriminator: discriminators::CREATE_TOKEN_IX, - event_type: EventType::PumpFunCreateToken, - inner_instruction_parser: Self::parse_create_token_inner_instruction, - instruction_parser: Self::parse_create_token_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::BUY_IX, - event_type: EventType::PumpFunBuy, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_buy_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::TRADE_EVENT, - instruction_discriminator: discriminators::SELL_IX, - event_type: EventType::PumpFunSell, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_sell_instruction, - }, - ]; - - let inner = GenericEventParser::new(PUMPFUN_PROGRAM_ID, ProtocolType::PumpFun, configs); - - Self { inner } - } - - /// 解析创建代币日志事件 - fn parse_create_token_inner_instruction( - data: &[u8], - metadata: EventMetadata, - ) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, - event.name, - event.symbol, - event.mint.to_string() - )); - Some(Box::new(PumpFunCreateTokenEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析交易事件 - fn parse_trade_inner_instruction( - data: &[u8], - metadata: EventMetadata, - ) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, - event.mint.to_string(), - event.user.to_string(), - event.is_buy.to_string() - )); - Some(Box::new(PumpFunTradeEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析创建代币指令事件 - fn parse_create_token_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - let mut offset = 0; - let name_len = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap()) as usize; - offset += 4; - let name = String::from_utf8_lossy(&data[offset..offset + name_len]); - offset += name_len; - let symbol_len = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap()) as usize; - offset += 4; - let symbol = String::from_utf8_lossy(&data[offset..offset + symbol_len]); - offset += symbol_len; - let uri_len = u32::from_le_bytes(data[offset..offset + 4].try_into().unwrap()) as usize; - offset += 4; - let uri = String::from_utf8_lossy(&data[offset..offset + uri_len]); - offset += uri_len; - let creator = if offset + 32 <= data.len() { - Pubkey::new_from_array(data[offset..offset + 32].try_into().ok()?) - } else { - Pubkey::default() - }; - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, - name, - symbol, - accounts[0].to_string() - )); - - Some(Box::new(PumpFunCreateTokenEvent { - metadata, - name: name.to_string(), - symbol: symbol.to_string(), - uri: uri.to_string(), - creator, - mint: accounts[0], - mint_authority: accounts[1], - bonding_curve: accounts[2], - associated_bonding_curve: accounts[3], - user: accounts[7], - ..Default::default() - })) - } - - // 解析买入指令事件 - fn parse_buy_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - let amount = u64::from_le_bytes(data[0..8].try_into().unwrap()); - let max_sol_cost = u64::from_le_bytes(data[8..16].try_into().unwrap()); - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, - accounts[2].to_string(), - accounts[6].to_string(), - true.to_string() - )); - Some(Box::new(PumpFunTradeEvent { - metadata, - fee_recipient: accounts[1], - mint: accounts[2], - bonding_curve: accounts[3], - associated_bonding_curve: accounts[4], - associated_user: accounts[5], - user: accounts[6], - creator_vault: accounts[8], - max_sol_cost, - amount, - is_buy: true, - ..Default::default() - })) - } - - // 解析卖出指令事件 - fn parse_sell_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - let amount = u64::from_le_bytes(data[0..8].try_into().unwrap()); - let min_sol_output = u64::from_le_bytes(data[8..16].try_into().unwrap()); - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, - accounts[2].to_string(), - accounts[6].to_string(), - false.to_string() - )); - Some(Box::new(PumpFunTradeEvent { - metadata, - fee_recipient: accounts[1], - mint: accounts[2], - bonding_curve: accounts[3], - associated_bonding_curve: accounts[4], - associated_user: accounts[5], - user: accounts[6], - creator_vault: accounts[8], - min_sol_output, - amount, - is_buy: false, - ..Default::default() - })) - } -} - -#[async_trait::async_trait] -impl EventParser for PumpFunEventParser { - fn parse_events_from_inner_instruction( - &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot) - } - - fn parse_events_from_instruction( - &self, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot) - } - - fn should_handle(&self, program_id: &Pubkey) -> bool { - self.inner.should_handle(program_id) - } - - fn supported_program_ids(&self) -> Vec { - self.inner.supported_program_ids() - } -} diff --git a/src/streaming/event_parser/protocols/pumpswap/events.rs b/src/streaming/event_parser/protocols/pumpswap/events.rs deleted file mode 100755 index f1447df..0000000 --- a/src/streaming/event_parser/protocols/pumpswap/events.rs +++ /dev/null @@ -1,322 +0,0 @@ -use borsh::BorshDeserialize; -use serde::{Deserialize, Serialize}; -use solana_sdk::pubkey::Pubkey; - -use crate::streaming::event_parser::common::EventMetadata; -use crate::impl_unified_event; - -/// 买入事件 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct PumpSwapBuyEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub timestamp: i64, - pub base_amount_out: u64, - pub max_quote_amount_in: u64, - pub user_base_token_reserves: u64, - pub user_quote_token_reserves: u64, - pub pool_base_token_reserves: u64, - pub pool_quote_token_reserves: u64, - pub quote_amount_in: u64, - pub lp_fee_basis_points: u64, - pub lp_fee: u64, - pub protocol_fee_basis_points: u64, - pub protocol_fee: u64, - pub quote_amount_in_with_lp_fee: u64, - pub user_quote_amount_in: u64, - pub pool: Pubkey, - pub user: Pubkey, - pub user_base_token_account: Pubkey, - pub user_quote_token_account: Pubkey, - pub protocol_fee_recipient: Pubkey, - pub protocol_fee_recipient_token_account: Pubkey, - pub coin_creator: Pubkey, - pub coin_creator_fee_basis_points: u64, - pub coin_creator_fee: u64, - #[borsh(skip)] - pub base_mint: Pubkey, - #[borsh(skip)] - pub quote_mint: Pubkey, - #[borsh(skip)] - pub pool_base_token_account: Pubkey, - #[borsh(skip)] - pub pool_quote_token_account: Pubkey, - #[borsh(skip)] - pub coin_creator_vault_ata: Pubkey, - #[borsh(skip)] - pub coin_creator_vault_authority: Pubkey, -} - -// 使用宏生成UnifiedEvent实现，指定需要合并的字段 -impl_unified_event!( - PumpSwapBuyEvent, - timestamp, - base_amount_out, - max_quote_amount_in, - user_base_token_reserves, - user_quote_token_reserves, - pool_base_token_reserves, - pool_quote_token_reserves, - quote_amount_in, - lp_fee_basis_points, - lp_fee, - protocol_fee_basis_points, - protocol_fee, - quote_amount_in_with_lp_fee, - user_quote_amount_in, - pool, - user, - user_base_token_account, - user_quote_token_account, - protocol_fee_recipient, - protocol_fee_recipient_token_account, - coin_creator, - coin_creator_fee_basis_points, - coin_creator_fee -); - -/// 卖出事件 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct PumpSwapSellEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub timestamp: i64, - pub base_amount_in: u64, - pub min_quote_amount_out: u64, - pub user_base_token_reserves: u64, - pub user_quote_token_reserves: u64, - pub pool_base_token_reserves: u64, - pub pool_quote_token_reserves: u64, - pub quote_amount_out: u64, - pub lp_fee_basis_points: u64, - pub lp_fee: u64, - pub protocol_fee_basis_points: u64, - pub protocol_fee: u64, - pub quote_amount_out_without_lp_fee: u64, - pub user_quote_amount_out: u64, - pub pool: Pubkey, - pub user: Pubkey, - pub user_base_token_account: Pubkey, - pub user_quote_token_account: Pubkey, - pub protocol_fee_recipient: Pubkey, - pub protocol_fee_recipient_token_account: Pubkey, - pub coin_creator: Pubkey, - pub coin_creator_fee_basis_points: u64, - pub coin_creator_fee: u64, - #[borsh(skip)] - pub base_mint: Pubkey, - #[borsh(skip)] - pub quote_mint: Pubkey, - #[borsh(skip)] - pub pool_base_token_account: Pubkey, - #[borsh(skip)] - pub pool_quote_token_account: Pubkey, - #[borsh(skip)] - pub coin_creator_vault_ata: Pubkey, - #[borsh(skip)] - pub coin_creator_vault_authority: Pubkey, -} - -// 使用宏生成UnifiedEvent实现，指定需要合并的字段 -impl_unified_event!( - PumpSwapSellEvent, - timestamp, - base_amount_in, - min_quote_amount_out, - user_base_token_reserves, - user_quote_token_reserves, - pool_base_token_reserves, - pool_quote_token_reserves, - quote_amount_out, - lp_fee_basis_points, - lp_fee, - protocol_fee_basis_points, - protocol_fee, - quote_amount_out_without_lp_fee, - user_quote_amount_out, - pool, - user, - user_base_token_account, - user_quote_token_account, - protocol_fee_recipient, - protocol_fee_recipient_token_account, - coin_creator, - coin_creator_fee_basis_points, - coin_creator_fee -); - -/// 创建池子事件 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct PumpSwapCreatePoolEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub timestamp: i64, - pub index: u16, - pub creator: Pubkey, - pub base_mint: Pubkey, - pub quote_mint: Pubkey, - pub base_mint_decimals: u8, - pub quote_mint_decimals: u8, - pub base_amount_in: u64, - pub quote_amount_in: u64, - pub pool_base_amount: u64, - pub pool_quote_amount: u64, - pub minimum_liquidity: u64, - pub initial_liquidity: u64, - pub lp_token_amount_out: u64, - pub pool_bump: u8, - pub pool: Pubkey, - pub lp_mint: Pubkey, - pub user_base_token_account: Pubkey, - pub user_quote_token_account: Pubkey, - pub coin_creator: Pubkey, - #[borsh(skip)] - pub user_pool_token_account: Pubkey, - #[borsh(skip)] - pub pool_base_token_account: Pubkey, - #[borsh(skip)] - pub pool_quote_token_account: Pubkey, -} - -impl_unified_event!( - PumpSwapCreatePoolEvent, - timestamp, - index, - creator, - base_mint, - quote_mint, - base_mint_decimals, - quote_mint_decimals, - base_amount_in, - quote_amount_in, - pool_base_amount, - pool_quote_amount, - minimum_liquidity, - initial_liquidity, - lp_token_amount_out, - pool_bump, - pool, - lp_mint, - user_base_token_account, - user_quote_token_account, - coin_creator -); - -/// 存款事件 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct PumpSwapDepositEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub timestamp: i64, - pub lp_token_amount_out: u64, - pub max_base_amount_in: u64, - pub max_quote_amount_in: u64, - pub user_base_token_reserves: u64, - pub user_quote_token_reserves: u64, - pub pool_base_token_reserves: u64, - pub pool_quote_token_reserves: u64, - pub base_amount_in: u64, - pub quote_amount_in: u64, - pub lp_mint_supply: u64, - pub pool: Pubkey, - pub user: Pubkey, - pub user_base_token_account: Pubkey, - pub user_quote_token_account: Pubkey, - pub user_pool_token_account: Pubkey, - #[borsh(skip)] - pub base_mint: Pubkey, - #[borsh(skip)] - pub quote_mint: Pubkey, - #[borsh(skip)] - pub pool_base_token_account: Pubkey, - #[borsh(skip)] - pub pool_quote_token_account: Pubkey, -} - -impl_unified_event!( - PumpSwapDepositEvent, - timestamp, - lp_token_amount_out, - max_base_amount_in, - max_quote_amount_in, - user_base_token_reserves, - user_quote_token_reserves, - pool_base_token_reserves, - pool_quote_token_reserves, - base_amount_in, - quote_amount_in, - lp_mint_supply, - pool, - user, - user_base_token_account, - user_quote_token_account, - user_pool_token_account -); - -/// 提款事件 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct PumpSwapWithdrawEvent { - #[borsh(skip)] - pub metadata: EventMetadata, - pub timestamp: i64, - pub lp_token_amount_in: u64, - pub min_base_amount_out: u64, - pub min_quote_amount_out: u64, - pub user_base_token_reserves: u64, - pub user_quote_token_reserves: u64, - pub pool_base_token_reserves: u64, - pub pool_quote_token_reserves: u64, - pub base_amount_out: u64, - pub quote_amount_out: u64, - pub lp_mint_supply: u64, - pub pool: Pubkey, - pub user: Pubkey, - pub user_base_token_account: Pubkey, - pub user_quote_token_account: Pubkey, - pub user_pool_token_account: Pubkey, - #[borsh(skip)] - pub base_mint: Pubkey, - #[borsh(skip)] - pub quote_mint: Pubkey, - #[borsh(skip)] - pub pool_base_token_account: Pubkey, - #[borsh(skip)] - pub pool_quote_token_account: Pubkey, -} - -impl_unified_event!( - PumpSwapWithdrawEvent, - timestamp, - lp_token_amount_in, - min_base_amount_out, - min_quote_amount_out, - user_base_token_reserves, - user_quote_token_reserves, - pool_base_token_reserves, - pool_quote_token_reserves, - base_amount_out, - quote_amount_out, - lp_mint_supply, - pool, - user, - user_base_token_account, - user_quote_token_account, - user_pool_token_account -); - -/// 事件鉴别器常量 -pub mod discriminators { - // 事件鉴别器 - pub const BUY_EVENT: &str = "0xe445a52e51cb9a1d67f4521f2cf57777"; - pub const SELL_EVENT: &str = "0xe445a52e51cb9a1d3e2f370aa503dc2a"; - pub const CREATE_POOL_EVENT: &str = "0xe445a52e51cb9a1db1310cd2a076a774"; - pub const DEPOSIT_EVENT: &str = "0xe445a52e51cb9a1d78f83d531f8e6b90"; - pub const WITHDRAW_EVENT: &str = "0xe445a52e51cb9a1d1609851aa02c47c0"; - - // 指令鉴别器 - pub const BUY_IX: &[u8] = &[102, 6, 61, 18, 1, 218, 235, 234]; - pub const SELL_IX: &[u8] = &[51, 230, 133, 164, 1, 127, 131, 173]; - pub const CREATE_POOL_IX: &[u8] = &[233, 146, 209, 142, 207, 104, 64, 188]; - pub const DEPOSIT_IX: &[u8] = &[242, 35, 198, 137, 82, 225, 242, 182]; - pub const WITHDRAW_IX: &[u8] = &[183, 18, 70, 156, 148, 109, 161, 34]; -} diff --git a/src/streaming/event_parser/protocols/pumpswap/mod.rs b/src/streaming/event_parser/protocols/pumpswap/mod.rs deleted file mode 100755 index cfa440f..0000000 --- a/src/streaming/event_parser/protocols/pumpswap/mod.rs +++ /dev/null @@ -1,5 +0,0 @@ -pub mod events; -pub mod parser; - -pub use events::*; -pub use parser::PumpSwapEventParser; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs deleted file mode 100755 index 52e56a2..0000000 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ /dev/null @@ -1,386 +0,0 @@ -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; - -use crate::streaming::event_parser::{ - common::{EventMetadata, EventType, ProtocolType, read_u64_le}, - core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, - protocols::pumpswap::{ - discriminators, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, PumpSwapDepositEvent, - PumpSwapSellEvent, PumpSwapWithdrawEvent, - }, -}; - -/// PumpSwap程序ID -pub const PUMPSWAP_PROGRAM_ID: Pubkey = - solana_sdk::pubkey!("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA"); - -/// PumpSwap事件解析器 -pub struct PumpSwapEventParser { - inner: GenericEventParser, -} - -impl PumpSwapEventParser { - pub fn new() -> Self { - // 配置所有事件类型 - let configs = vec![ - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::BUY_EVENT, - instruction_discriminator: discriminators::BUY_IX, - event_type: EventType::PumpSwapBuy, - inner_instruction_parser: Self::parse_buy_inner_instruction, - instruction_parser: Self::parse_buy_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::SELL_EVENT, - instruction_discriminator: discriminators::SELL_IX, - event_type: EventType::PumpSwapSell, - inner_instruction_parser: Self::parse_sell_inner_instruction, - instruction_parser: Self::parse_sell_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::CREATE_POOL_EVENT, - instruction_discriminator: discriminators::CREATE_POOL_IX, - event_type: EventType::PumpSwapCreatePool, - inner_instruction_parser: Self::parse_create_pool_inner_instruction, - instruction_parser: Self::parse_create_pool_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::DEPOSIT_EVENT, - instruction_discriminator: discriminators::DEPOSIT_IX, - event_type: EventType::PumpSwapDeposit, - inner_instruction_parser: Self::parse_deposit_inner_instruction, - instruction_parser: Self::parse_deposit_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: discriminators::WITHDRAW_EVENT, - instruction_discriminator: discriminators::WITHDRAW_IX, - event_type: EventType::PumpSwapWithdraw, - inner_instruction_parser: Self::parse_withdraw_inner_instruction, - instruction_parser: Self::parse_withdraw_instruction, - }, - ]; - - let inner = GenericEventParser::new(PUMPSWAP_PROGRAM_ID, ProtocolType::PumpSwap, configs); - - Self { inner } - } - - /// 解析买入日志事件 - fn parse_buy_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.user, event.pool, event.base_amount_out - )); - Some(Box::new(PumpSwapBuyEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析卖出日志事件 - fn parse_sell_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.user, event.pool, event.base_amount_in - )); - Some(Box::new(PumpSwapSellEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析创建池子日志事件 - fn parse_create_pool_inner_instruction( - data: &[u8], - metadata: EventMetadata, - ) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.pool, event.creator, event.base_amount_in - )); - Some(Box::new(PumpSwapCreatePoolEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析存款日志事件 - fn parse_deposit_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.pool, event.user, event.lp_token_amount_out - )); - Some(Box::new(PumpSwapDepositEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析提款日志事件 - fn parse_withdraw_inner_instruction(data: &[u8], metadata: EventMetadata) -> Option> { - if let Ok(event) = borsh::from_slice::(data) { - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, event.pool, event.user, event.lp_token_amount_in - )); - Some(Box::new(PumpSwapWithdrawEvent { - metadata: metadata, - ..event - })) - } else { - None - } - } - - /// 解析买入指令事件 - fn parse_buy_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - - let base_amount_out = read_u64_le(data, 0)?; - let max_quote_amount_in = read_u64_le(data, 8)?; - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[1], accounts[0], base_amount_out - )); - - Some(Box::new(PumpSwapBuyEvent { - metadata, - base_amount_out, - max_quote_amount_in, - pool: accounts[0], - user: accounts[1], - base_mint: accounts[3], - quote_mint: accounts[4], - user_base_token_account: accounts[5], - user_quote_token_account: accounts[6], - pool_base_token_account: accounts[7], - pool_quote_token_account: accounts[8], - protocol_fee_recipient: accounts[9], - protocol_fee_recipient_token_account: accounts[10], - coin_creator_vault_ata: accounts.get(17).copied().unwrap_or_default(), - coin_creator_vault_authority: accounts.get(18).copied().unwrap_or_default(), - ..Default::default() - })) - } - - /// 解析卖出指令事件 - fn parse_sell_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 11 { - return None; - } - - let base_amount_in = read_u64_le(data, 0)?; - let min_quote_amount_out = read_u64_le(data, 8)?; - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[1], accounts[0], base_amount_in - )); - - Some(Box::new(PumpSwapSellEvent { - metadata, - base_amount_in, - min_quote_amount_out, - pool: accounts[0], - user: accounts[1], - base_mint: accounts[3], - quote_mint: accounts[4], - user_base_token_account: accounts[5], - user_quote_token_account: accounts[6], - pool_base_token_account: accounts[7], - pool_quote_token_account: accounts[8], - protocol_fee_recipient: accounts[9], - protocol_fee_recipient_token_account: accounts[10], - coin_creator_vault_ata: accounts.get(17).copied().unwrap_or_default(), - coin_creator_vault_authority: accounts.get(18).copied().unwrap_or_default(), - ..Default::default() - })) - } - - /// 解析创建池子指令事件 - fn parse_create_pool_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 18 || accounts.len() < 11 { - return None; - } - - let index = u16::from_le_bytes(data[0..2].try_into().ok()?); - let base_amount_in = u64::from_le_bytes(data[2..10].try_into().ok()?); - let quote_amount_in = u64::from_le_bytes(data[10..18].try_into().ok()?); - let coin_creator = if data.len() >= 50 { - Pubkey::new_from_array(data[18..50].try_into().ok()?) - } else { - Pubkey::default() - }; - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[0], accounts[2], base_amount_in - )); - - Some(Box::new(PumpSwapCreatePoolEvent { - metadata, - index, - base_amount_in, - quote_amount_in, - pool: accounts[0], - creator: accounts[2], - base_mint: accounts[3], - quote_mint: accounts[4], - lp_mint: accounts[5], - user_base_token_account: accounts[6], - user_quote_token_account: accounts[7], - user_pool_token_account: accounts[8], - pool_base_token_account: accounts[9], - pool_quote_token_account: accounts[10], - coin_creator, - ..Default::default() - })) - } - - /// 解析存款指令事件 - fn parse_deposit_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 24 || accounts.len() < 11 { - return None; - } - - let lp_token_amount_out = u64::from_le_bytes(data[0..8].try_into().ok()?); - let max_base_amount_in = u64::from_le_bytes(data[8..16].try_into().ok()?); - let max_quote_amount_in = u64::from_le_bytes(data[16..24].try_into().ok()?); - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[0], accounts[2], lp_token_amount_out - )); - - Some(Box::new(PumpSwapDepositEvent { - metadata, - lp_token_amount_out, - max_base_amount_in, - max_quote_amount_in, - pool: accounts[0], - user: accounts[2], - base_mint: accounts[3], - quote_mint: accounts[4], - user_base_token_account: accounts[6], - user_quote_token_account: accounts[7], - user_pool_token_account: accounts[8], - pool_base_token_account: accounts[9], - pool_quote_token_account: accounts[10], - ..Default::default() - })) - } - - /// 解析提款指令事件 - fn parse_withdraw_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 24 || accounts.len() < 11 { - return None; - } - - let lp_token_amount_in = u64::from_le_bytes(data[0..8].try_into().ok()?); - let min_base_amount_out = u64::from_le_bytes(data[8..16].try_into().ok()?); - let min_quote_amount_out = u64::from_le_bytes(data[16..24].try_into().ok()?); - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[0], accounts[2], lp_token_amount_in - )); - - Some(Box::new(PumpSwapWithdrawEvent { - metadata, - lp_token_amount_in, - min_base_amount_out, - min_quote_amount_out, - pool: accounts[0], - user: accounts[2], - base_mint: accounts[3], - quote_mint: accounts[4], - user_base_token_account: accounts[6], - user_quote_token_account: accounts[7], - user_pool_token_account: accounts[8], - pool_base_token_account: accounts[9], - pool_quote_token_account: accounts[10], - ..Default::default() - })) - } -} - -#[async_trait::async_trait] -impl EventParser for PumpSwapEventParser { - fn parse_events_from_inner_instruction( - &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot) - } - - fn parse_events_from_instruction( - &self, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot) - } - - fn should_handle(&self, program_id: &Pubkey) -> bool { - self.inner.should_handle(program_id) - } - - fn supported_program_ids(&self) -> Vec { - self.inner.supported_program_ids() - } -} diff --git a/src/streaming/event_parser/protocols/raydium_clmm/events.rs b/src/streaming/event_parser/protocols/raydium_clmm/events.rs deleted file mode 100755 index cd0c0b1..0000000 --- a/src/streaming/event_parser/protocols/raydium_clmm/events.rs +++ /dev/null @@ -1,59 +0,0 @@ -use crate::impl_unified_event; -use crate::streaming::event_parser::common::EventMetadata; -// use borsh::BorshDeserialize; -use serde::{Deserialize, Serialize}; -use solana_sdk::pubkey::Pubkey; - -/// 交易 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] -pub struct RaydiumClmmSwapEvent { - pub metadata: EventMetadata, - pub amount: u64, - pub other_amount_threshold: u64, - pub sqrt_price_limit_x64: u128, - pub is_base_input: bool, - pub payer: Pubkey, - pub amm_config: Pubkey, - pub pool_state: Pubkey, - pub input_token_account: Pubkey, - pub output_token_account: Pubkey, - pub input_vault: Pubkey, - pub output_vault: Pubkey, - pub observation_state: Pubkey, - pub token_program: Pubkey, - pub tick_array: Pubkey, - pub remaining_accounts: Vec, -} - -impl_unified_event!(RaydiumClmmSwapEvent,); - -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] -pub struct RaydiumClmmSwapV2Event { - pub metadata: EventMetadata, - pub amount: u64, - pub other_amount_threshold: u64, - pub sqrt_price_limit_x64: u128, - pub is_base_input: bool, - pub payer: Pubkey, - pub amm_config: Pubkey, - pub pool_state: Pubkey, - pub input_token_account: Pubkey, - pub output_token_account: Pubkey, - pub input_vault: Pubkey, - pub output_vault: Pubkey, - pub observation_state: Pubkey, - pub token_program: Pubkey, - pub token_program2022: Pubkey, - pub memo_program: Pubkey, - pub input_vault_mint: Pubkey, - pub output_vault_mint: Pubkey, - pub remaining_accounts: Vec, -} -impl_unified_event!(RaydiumClmmSwapV2Event,); - -/// 事件鉴别器常量 -pub mod discriminators { - // 指令鉴别器 - pub const SWAP: &[u8] = &[248, 198, 158, 145, 225, 117, 135, 200]; - pub const SWAP_V2: &[u8] = &[43, 4, 237, 11, 26, 201, 30, 98]; -} diff --git a/src/streaming/event_parser/protocols/raydium_clmm/mod.rs b/src/streaming/event_parser/protocols/raydium_clmm/mod.rs deleted file mode 100755 index 89b867a..0000000 --- a/src/streaming/event_parser/protocols/raydium_clmm/mod.rs +++ /dev/null @@ -1,5 +0,0 @@ -pub mod events; -pub mod parser; - -pub use events::*; -pub use parser::RaydiumClmmEventParser; diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs deleted file mode 100755 index 09bc0a6..0000000 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ /dev/null @@ -1,170 +0,0 @@ -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; - -use crate::streaming::event_parser::{ - common::{read_u128_le, read_u64_le, read_u8_le, EventMetadata, EventType, ProtocolType}, - core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, - protocols::raydium_clmm::{discriminators, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event}, -}; - -/// Raydium CLMM程序ID -pub const RAYDIUM_CLMM_PROGRAM_ID: Pubkey = - solana_sdk::pubkey!("CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK"); - -/// Raydium CLMM事件解析器 -pub struct RaydiumClmmEventParser { - inner: GenericEventParser, -} - -impl RaydiumClmmEventParser { - pub fn new() -> Self { - // 配置所有事件类型 - let configs = vec![ - GenericEventParseConfig { - inner_instruction_discriminator: "", - instruction_discriminator: discriminators::SWAP, - event_type: EventType::RaydiumClmmSwap, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_swap_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: "", - instruction_discriminator: discriminators::SWAP_V2, - event_type: EventType::RaydiumClmmSwapV2, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_swap_v2_instruction, - }, - ]; - - let inner = - GenericEventParser::new(RAYDIUM_CLMM_PROGRAM_ID, ProtocolType::RaydiumClmm, configs); - - Self { inner } - } - - /// 解析交易事件 - fn parse_trade_inner_instruction( - _data: &[u8], - _metadata: EventMetadata, - ) -> Option> { - None - } - - /// 解析交易指令事件 - fn parse_swap_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 10 { - return None; - } - - let amount = read_u64_le(data, 0)?; - let other_amount_threshold = read_u64_le(data, 8)?; - let sqrt_price_limit_x64 = read_u128_le(data, 16)?; - let is_base_input = read_u8_le(data, 32)?; - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[2], accounts[3], accounts[4] - )); - - Some(Box::new(RaydiumClmmSwapEvent { - metadata, - amount, - other_amount_threshold, - sqrt_price_limit_x64, - is_base_input: is_base_input == 1, - payer: accounts[0], - amm_config: accounts[1], - pool_state: accounts[2], - input_token_account: accounts[3], - output_token_account: accounts[4], - input_vault: accounts[5], - output_vault: accounts[6], - observation_state: accounts[7], - token_program: accounts[8], - tick_array: accounts[9], - remaining_accounts: accounts[10..].to_vec(), - ..Default::default() - })) - } - - fn parse_swap_v2_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 13 { - return None; - } - - let amount = read_u64_le(data, 0)?; - let other_amount_threshold = read_u64_le(data, 8)?; - let sqrt_price_limit_x64 = read_u128_le(data, 16)?; - let is_base_input = read_u8_le(data, 32)?; - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[2], accounts[3], accounts[4] - )); - - Some(Box::new(RaydiumClmmSwapV2Event { - metadata, - amount, - other_amount_threshold, - sqrt_price_limit_x64, - is_base_input: is_base_input == 1, - payer: accounts[0], - amm_config: accounts[1], - pool_state: accounts[2], - input_token_account: accounts[3], - output_token_account: accounts[4], - input_vault: accounts[5], - output_vault: accounts[6], - observation_state: accounts[7], - token_program: accounts[8], - token_program2022: accounts[9], - memo_program: accounts[10], - input_vault_mint: accounts[11], - output_vault_mint: accounts[12], - remaining_accounts: accounts[13..].to_vec(), - ..Default::default() - })) - } -} - -#[async_trait::async_trait] -impl EventParser for RaydiumClmmEventParser { - fn parse_events_from_inner_instruction( - &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot) - } - - fn parse_events_from_instruction( - &self, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot) - } - - fn should_handle(&self, program_id: &Pubkey) -> bool { - self.inner.should_handle(program_id) - } - - fn supported_program_ids(&self) -> Vec { - self.inner.supported_program_ids() - } -} diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/events.rs b/src/streaming/event_parser/protocols/raydium_cpmm/events.rs deleted file mode 100755 index 2c3f770..0000000 --- a/src/streaming/event_parser/protocols/raydium_cpmm/events.rs +++ /dev/null @@ -1,35 +0,0 @@ -use crate::impl_unified_event; -use crate::streaming::event_parser::common::EventMetadata; -use borsh::BorshDeserialize; -use serde::{Deserialize, Serialize}; -use solana_sdk::pubkey::Pubkey; - -/// 交易 -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] -pub struct RaydiumCpmmSwapEvent { - pub metadata: EventMetadata, - pub amount_in: u64, - pub minimum_amount_out: u64, - pub max_amount_in: u64, - pub amount_out: u64, - pub payer: Pubkey, - pub authority: Pubkey, - pub amm_config: Pubkey, - pub pool_state: Pubkey, - pub input_token_account: Pubkey, - pub output_token_account: Pubkey, - pub input_vault: Pubkey, - pub output_vault: Pubkey, - pub input_token_mint: Pubkey, - pub output_token_mint: Pubkey, - pub observation_state: Pubkey, -} - -impl_unified_event!(RaydiumCpmmSwapEvent,); - -/// 事件鉴别器常量 -pub mod discriminators { - // 指令鉴别器 - pub const SWAP_BASE_IN: &[u8] = &[143, 190, 90, 218, 196, 30, 51, 222]; - pub const SWAP_BASE_OUT: &[u8] = &[55, 217, 98, 86, 163, 74, 180, 173]; -} diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/mod.rs b/src/streaming/event_parser/protocols/raydium_cpmm/mod.rs deleted file mode 100755 index e5896a6..0000000 --- a/src/streaming/event_parser/protocols/raydium_cpmm/mod.rs +++ /dev/null @@ -1,5 +0,0 @@ -pub mod events; -pub mod parser; - -pub use events::*; -pub use parser::RaydiumCpmmEventParser; diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs deleted file mode 100755 index 8b6984a..0000000 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ /dev/null @@ -1,159 +0,0 @@ -use solana_sdk::{instruction::CompiledInstruction, pubkey::Pubkey}; -use solana_transaction_status::UiCompiledInstruction; - -use crate::streaming::event_parser::{ - common::{read_u64_le, EventMetadata, EventType, ProtocolType}, - core::traits::{EventParser, GenericEventParseConfig, GenericEventParser, UnifiedEvent}, - protocols::raydium_cpmm::{discriminators, RaydiumCpmmSwapEvent}, -}; - -/// Raydium CPMM程序ID -pub const RAYDIUM_CPMM_PROGRAM_ID: Pubkey = - solana_sdk::pubkey!("CPMMoo8L3F4NbTegBCKVNunggL7H1ZpdTHKxQB5qKP1C"); - -/// Raydium CPMM事件解析器 -pub struct RaydiumCpmmEventParser { - inner: GenericEventParser, -} - -impl RaydiumCpmmEventParser { - pub fn new() -> Self { - // 配置所有事件类型 - let configs = vec![ - GenericEventParseConfig { - inner_instruction_discriminator: "", - instruction_discriminator: discriminators::SWAP_BASE_IN, - event_type: EventType::RaydiumCpmmSwapBaseInput, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_swap_base_input_instruction, - }, - GenericEventParseConfig { - inner_instruction_discriminator: "", - instruction_discriminator: discriminators::SWAP_BASE_OUT, - event_type: EventType::RaydiumCpmmSwapBaseOutput, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_swap_base_output_instruction, - }, - ]; - - let inner = - GenericEventParser::new(RAYDIUM_CPMM_PROGRAM_ID, ProtocolType::RaydiumCpmm, configs); - - Self { inner } - } - - /// 解析交易事件 - fn parse_trade_inner_instruction( - _data: &[u8], - _metadata: EventMetadata, - ) -> Option> { - None - } - - /// 解析买入指令事件 - fn parse_swap_base_input_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 13 { - return None; - } - - let amount_in = read_u64_le(data, 0)?; - let minimum_amount_out = read_u64_le(data, 8)?; - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - - Some(Box::new(RaydiumCpmmSwapEvent { - metadata, - amount_in, - minimum_amount_out, - payer: accounts[0], - authority: accounts[1], - amm_config: accounts[2], - pool_state: accounts[3], - input_token_account: accounts[4], - output_token_account: accounts[5], - input_vault: accounts[6], - output_vault: accounts[7], - input_token_mint: accounts[10], - output_token_mint: accounts[11], - observation_state: accounts[12], - ..Default::default() - })) - } - - fn parse_swap_base_output_instruction( - data: &[u8], - accounts: &[Pubkey], - metadata: EventMetadata, - ) -> Option> { - if data.len() < 16 || accounts.len() < 13 { - return None; - } - - let max_amount_in = read_u64_le(data, 0)?; - let amount_out = read_u64_le(data, 8)?; - - let mut metadata = metadata; - metadata.set_id(format!( - "{}-{}-{}-{}", - metadata.signature, accounts[3], accounts[10], accounts[11] - )); - - Some(Box::new(RaydiumCpmmSwapEvent { - metadata, - max_amount_in, - amount_out, - payer: accounts[0], - authority: accounts[1], - amm_config: accounts[2], - pool_state: accounts[3], - input_token_account: accounts[4], - output_token_account: accounts[5], - input_vault: accounts[6], - output_vault: accounts[7], - input_token_mint: accounts[10], - output_token_mint: accounts[11], - observation_state: accounts[12], - ..Default::default() - })) - } -} - -#[async_trait::async_trait] -impl EventParser for RaydiumCpmmEventParser { - fn parse_events_from_inner_instruction( - &self, - inner_instruction: &UiCompiledInstruction, - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_inner_instruction(inner_instruction, signature, slot) - } - - fn parse_events_from_instruction( - &self, - instruction: &CompiledInstruction, - accounts: &[Pubkey], - signature: &str, - slot: u64, - ) -> Vec> { - self.inner - .parse_events_from_instruction(instruction, accounts, signature, slot) - } - - fn should_handle(&self, program_id: &Pubkey) -> bool { - self.inner.should_handle(program_id) - } - - fn supported_program_ids(&self) -> Vec { - self.inner.supported_program_ids() - } -} diff --git a/src/streaming/mod.rs b/src/streaming/mod.rs deleted file mode 100755 index 019ab5f..0000000 --- a/src/streaming/mod.rs +++ /dev/null @@ -1,8 +0,0 @@ -pub mod yellowstone_grpc; -pub mod yellowstone_sub_system; -pub mod shred_stream; -pub mod event_parser; - -pub use yellowstone_grpc::YellowstoneGrpc; -pub use yellowstone_sub_system::{SystemEvent, TransferInfo}; -pub use shred_stream::ShredStreamGrpc; \ No newline at end of file diff --git a/src/streaming/shred_stream.rs b/src/streaming/shred_stream.rs deleted file mode 100755 index 471cfdb..0000000 --- a/src/streaming/shred_stream.rs +++ /dev/null @@ -1,120 +0,0 @@ -use std::sync::Arc; - -use futures::{channel::mpsc, StreamExt}; -use solana_entry::entry::Entry; -use tonic::transport::Channel; - -use log::error; -use solana_sdk::transaction::VersionedTransaction; - -use crate::common::AnyResult; -use crate::streaming::event_parser::{EventParserFactory, Protocol, UnifiedEvent}; - -use crate::protos::shredstream::shredstream_proxy_client::ShredstreamProxyClient; -use crate::protos::shredstream::SubscribeEntriesRequest; -use solana_sdk::pubkey::Pubkey; - -const CHANNEL_SIZE: usize = 1000; - -pub struct ShredStreamGrpc { - shredstream_client: Arc>, -} - -struct TransactionWithSlot { - transaction: VersionedTransaction, - slot: u64, -} - -impl ShredStreamGrpc { - pub async fn new(endpoint: String) -> AnyResult { - let shredstream_client = ShredstreamProxyClient::connect(endpoint.clone()).await?; - Ok(Self { - shredstream_client: Arc::new(shredstream_client), - }) - } - - pub async fn shredstream_subscribe( - &self, - protocols: Vec, - bot_wallet: Option, - callback: F, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync + 'static, - { - let request = tonic::Request::new(SubscribeEntriesRequest {}); - let mut client = (*self.shredstream_client).clone(); - let mut stream = client.subscribe_entries(request).await?.into_inner(); - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); - let callback = Box::new(callback); - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Ok(entries) = bincode::deserialize::>(&msg.entries) { - for entry in entries { - for transaction in entry.transactions { - let _ = tx.try_send(TransactionWithSlot { - transaction: transaction.clone(), - slot: msg.slot, - }); - } - } - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); - - while let Some(transaction_with_slot) = rx.next().await { - if let Err(e) = Self::process_transaction( - transaction_with_slot, - protocols.clone(), - bot_wallet, - &*callback, - ) - .await - { - error!("Error processing transaction: {:?}", e); - } - } - - Ok(()) - } - - async fn process_transaction( - transaction_with_slot: TransactionWithSlot, - protocols: Vec, - bot_wallet: Option, - callback: &F, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync, - { - let slot = transaction_with_slot.slot; - let versioned_tx = transaction_with_slot.transaction; - let signature = versioned_tx.signatures[0]; - - for protocol in protocols { - let parser = EventParserFactory::create_parser(protocol.clone()); - let events = parser - .parse_versioned_transaction( - &versioned_tx, - &signature.to_string(), - Some(slot), - bot_wallet.clone(), - ) - .await - .unwrap_or_else(|_e| vec![]); - for event in events { - callback(event); - } - } - - Ok(()) - } -} diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs deleted file mode 100755 index 909aebc..0000000 --- a/src/streaming/yellowstone_grpc.rs +++ /dev/null @@ -1,266 +0,0 @@ -use std::{collections::HashMap, fmt, time::Duration}; - -use chrono::Local; -use futures::{channel::mpsc, sink::Sink, SinkExt, Stream, StreamExt}; -use log::{error, info}; -use rustls::crypto::{ring::default_provider, CryptoProvider}; -use solana_sdk::{pubkey::Pubkey, signature::Signature}; -use solana_transaction_status::{EncodedTransactionWithStatusMeta, UiTransactionEncoding}; -use tonic::{transport::channel::ClientTlsConfig, Status}; -use yellowstone_grpc_client::{GeyserGrpcClient, Interceptor}; -use yellowstone_grpc_proto::geyser::{ - subscribe_update::UpdateOneof, CommitmentLevel, SubscribeRequest, - SubscribeRequestFilterTransactions, SubscribeRequestPing, SubscribeUpdate, - SubscribeUpdateTransaction, -}; - -use crate::common::AnyResult; -use crate::streaming::event_parser::{EventParserFactory, Protocol, UnifiedEvent}; - -type TransactionsFilterMap = HashMap; - -const CONNECT_TIMEOUT: u64 = 10; -const REQUEST_TIMEOUT: u64 = 60; -const CHANNEL_SIZE: usize = 1000; -const MAX_DECODING_MESSAGE_SIZE: usize = 1024 * 1024 * 10; - -#[derive(Clone)] -pub struct TransactionPretty { - pub slot: u64, - pub signature: Signature, - pub is_vote: bool, - pub tx: EncodedTransactionWithStatusMeta, -} - -impl fmt::Debug for TransactionPretty { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - struct TxWrap<'a>(&'a EncodedTransactionWithStatusMeta); - impl<'a> fmt::Debug for TxWrap<'a> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let serialized = serde_json::to_string(self.0).expect("failed to serialize"); - fmt::Display::fmt(&serialized, f) - } - } - - f.debug_struct("TransactionPretty") - .field("slot", &self.slot) - .field("signature", &self.signature) - .field("is_vote", &self.is_vote) - .field("tx", &TxWrap(&self.tx)) - .finish() - } -} - -impl From for TransactionPretty { - fn from(SubscribeUpdateTransaction { transaction, slot }: SubscribeUpdateTransaction) -> Self { - let tx = transaction.expect("should be defined"); - Self { - slot, - signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"), - is_vote: tx.is_vote, - tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx) - .expect("valid tx with meta") - .encode(UiTransactionEncoding::Base64, Some(u8::MAX), true) - .expect("failed to encode"), - } - } -} - -pub struct YellowstoneGrpc { - endpoint: String, - x_token: Option, -} - -impl YellowstoneGrpc { - pub fn new(endpoint: String, x_token: Option) -> AnyResult { - if CryptoProvider::get_default().is_none() { - default_provider() - .install_default() - .map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e))?; - } - - Ok(Self { endpoint, x_token }) - } - - pub async fn connect(&self) -> AnyResult> { - let builder = GeyserGrpcClient::build_from_shared(self.endpoint.clone())? - .x_token(self.x_token.clone())? - .tls_config(ClientTlsConfig::new().with_native_roots())? - .max_decoding_message_size(MAX_DECODING_MESSAGE_SIZE) - .connect_timeout(Duration::from_secs(CONNECT_TIMEOUT)) - .timeout(Duration::from_secs(REQUEST_TIMEOUT)); - - Ok(builder.connect().await?) - } - - pub async fn subscribe_with_request( - &self, - transactions: TransactionsFilterMap, - ) -> AnyResult<( - impl Sink, - impl Stream>, - )> { - let subscribe_request = SubscribeRequest { - transactions, - commitment: Some(CommitmentLevel::Processed.into()), - ..Default::default() - }; - - let mut client = self.connect().await?; - let (sink, stream) = client - .subscribe_with_request(Some(subscribe_request)) - .await?; - Ok((sink, stream)) - } - - pub fn get_subscribe_request_filter( - &self, - account_include: Vec, - account_exclude: Vec, - account_required: Vec, - ) -> TransactionsFilterMap { - let mut transactions = HashMap::new(); - transactions.insert( - "client".to_string(), - SubscribeRequestFilterTransactions { - vote: Some(false), - failed: Some(false), - signature: None, - account_include, - account_exclude, - account_required, - }, - ); - transactions - } - - pub async fn handle_stream_message( - msg: SubscribeUpdate, - tx: &mut mpsc::Sender, - subscribe_tx: &mut (impl Sink + Unpin), - ) -> AnyResult<()> { - match msg.update_oneof { - Some(UpdateOneof::Transaction(sut)) => { - let transaction_pretty = TransactionPretty::from(sut); - tx.try_send(transaction_pretty)?; - } - Some(UpdateOneof::Ping(_)) => { - subscribe_tx - .send(SubscribeRequest { - ping: Some(SubscribeRequestPing { id: 1 }), - ..Default::default() - }) - .await?; - info!("service is ping: {}", Local::now()); - } - Some(UpdateOneof::Pong(_)) => { - info!("service is pong: {}", Local::now()); - } - _ => {} - } - Ok(()) - } - - /// 订阅事件 - pub async fn subscribe_events( - &self, - protocols: Vec, - bot_wallet: Option, - account_include: Option>, - account_exclude: Option>, - callback: F, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync + 'static, - { - // 创建过滤器 - let protocol_accounts = protocols - .iter() - .map(|p| p.get_program_id()) - .flatten() - .map(|p| p.to_string()) - .collect::>(); - let mut account_include = account_include.unwrap_or_default(); - let account_exclude = account_exclude.unwrap_or_default(); - account_include.extend(protocol_accounts.clone()); - - let transactions = - self.get_subscribe_request_filter(account_include, account_exclude, vec![]); - - // 订阅事件 - let (mut subscribe_tx, mut stream) = self.subscribe_with_request(transactions).await?; - - // 创建通道 - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); - - // 创建回调函数 - let callback = Box::new(callback); - - // 启动处理流的任务 - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await - { - error!("Error handling message: {:?}", e); - break; - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); - - // 处理交易 - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = Self::process_event_transaction( - transaction_pretty, - &*callback, - bot_wallet, - protocols.clone(), - ) - .await - { - error!("Error processing transaction: {:?}", e); - } - } - - Ok(()) - } - - /// 处理事件交易 - async fn process_event_transaction( - transaction_pretty: TransactionPretty, - callback: &F, - bot_wallet: Option, - protocols: Vec, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync, - { - let slot = transaction_pretty.slot; - let signature = transaction_pretty.signature.to_string(); - for protocol in protocols { - let parser = EventParserFactory::create_parser(protocol); - let events = parser - .parse_transaction( - transaction_pretty.tx.clone(), - &signature, - Some(slot), - bot_wallet.clone(), - ) - .await - .unwrap_or_else(|_e| vec![]); - for event in events { - callback(event); - } - } - - Ok(()) - } -} diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs deleted file mode 100755 index f673e25..0000000 --- a/src/streaming/yellowstone_sub_system.rs +++ /dev/null @@ -1,96 +0,0 @@ -use crate::{common::AnyResult, streaming::yellowstone_grpc::{TransactionPretty, YellowstoneGrpc}}; -use solana_program::pubkey; -use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction}; -use futures::{channel::mpsc, StreamExt}; -use log::error; -use solana_transaction_status::EncodedTransactionWithStatusMeta; - -const SYSTEM_PROGRAM_ID: Pubkey = pubkey!("11111111111111111111111111111111"); -const CHANNEL_SIZE: usize = 1000; - -#[derive(Debug)] -pub enum SystemEvent { - NewTransfer(TransferInfo), - Error(String), -} - -#[derive(Clone, Debug, Default, PartialEq)] -pub struct TransferInfo { - pub slot: u64, - pub signature: String, - pub tx: Option, -} - -impl YellowstoneGrpc { - pub async fn subscribe_system( - &self, - callback: F, - account_include: Option>, - account_exclude: Option>, - ) -> AnyResult<()> - where - F: Fn(SystemEvent) + Send + Sync + 'static, - { - let addrs = vec![SYSTEM_PROGRAM_ID.to_string()]; - let account_include = account_include.unwrap_or_default(); - let account_exclude = account_exclude.unwrap_or_default(); - let transactions = - self.get_subscribe_request_filter(account_include, account_exclude, addrs); - let (mut subscribe_tx, mut stream) = self.subscribe_with_request(transactions).await?; - let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); - - let callback = Box::new(callback); - - tokio::spawn(async move { - while let Some(message) = stream.next().await { - match message { - Ok(msg) => { - if let Err(e) = - Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await - { - error!("Error handling message: {:?}", e); - break; - } - } - Err(error) => { - error!("Stream error: {error:?}"); - break; - } - } - } - }); - - while let Some(transaction_pretty) = rx.next().await { - if let Err(e) = Self::process_system_transaction(transaction_pretty, &*callback).await { - error!("Error processing transaction: {:?}", e); - } - } - Ok(()) - } - - async fn process_system_transaction( - transaction_pretty: TransactionPretty, - callback: &F, - ) -> AnyResult<()> - where - F: Fn(SystemEvent) + Send + Sync, - { - let trade_raw: EncodedTransactionWithStatusMeta = transaction_pretty.tx; - let meta = trade_raw - .meta - .as_ref() - .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; - - if meta.err.is_some() { - return Ok(()); - } - - callback(SystemEvent::NewTransfer(TransferInfo { - slot: transaction_pretty.slot, - signature: transaction_pretty.signature.to_string(), - tx: trade_raw.transaction.decode(), - })); - - Ok(()) - } -} diff --git a/src/trading/core/params.rs b/src/trading/core/params.rs index ca05c55..aa7d7a0 100755 --- a/src/trading/core/params.rs +++ b/src/trading/core/params.rs @@ -6,8 +6,8 @@ use super::traits::ProtocolParams; use crate::common::bonding_curve::BondingCurveAccount; use crate::common::{PriorityFee, SolanaRpcClient}; use crate::constants::bonk::accounts::{PLATFORM_FEE_RATE, PROTOCOL_FEE_RATE, SHARE_FEE_RATE}; -use crate::streaming::event_parser::common::EventType; -use crate::streaming::event_parser::protocols::bonk::BonkTradeEvent; +use crate::solana_streamer_sdk::streaming::event_parser::common::EventType; +use crate::solana_streamer_sdk::streaming::event_parser::protocols::bonk::BonkTradeEvent; use crate::swqos::SwqosClient; use crate::trading::bonk::common::{get_amount_in, get_amount_in_net, get_amount_out}; diff --git a/src/trading/pumpfun/common.rs b/src/trading/pumpfun/common.rs index 95521b3..0425826 100755 --- a/src/trading/pumpfun/common.rs +++ b/src/trading/pumpfun/common.rs @@ -12,8 +12,9 @@ use crate::{ constants::{ self, pumpfun::global_constants::{CREATOR_FEE, FEE_BASIS_POINTS}, trade::trade::DEFAULT_SLIPPAGE }, - streaming::event_parser::protocols::pumpfun::PumpFunTradeEvent, trading::common::calculate_with_slippage_buy + trading::common::calculate_with_slippage_buy }; +use crate::solana_streamer_sdk::streaming::event_parser::protocols::pumpfun::PumpFunTradeEvent; lazy_static::lazy_static! { static ref ACCOUNT_CACHE: RwLock>> = RwLock::new(HashMap::new()); diff --git a/src/utils.rs b/src/utils.rs index 75dfe21..bcdd63e 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,4 +1,4 @@ -use crate::streaming::event_parser::protocols::pumpfun::PumpFunTradeEvent; +use crate::solana_streamer_sdk::streaming::event_parser::protocols::pumpfun::PumpFunTradeEvent; use crate::trading; use crate::SolanaTrade; use solana_sdk::pubkey::Pubkey;