From 14b77dbdf4236d9d157452cdaf18a3c74b6b11b3 Mon Sep 17 00:00:00 2001 From: William Date: Fri, 24 Jan 2025 14:01:56 +0800 Subject: [PATCH] update events --- Cargo.toml | 2 +- README.md | 2 +- src/common/dex_events.rs | 123 ----------------------------- src/common/event.rs | 158 ------------------------------------- src/common/logs_data.rs | 76 +++++++++++++++--- src/common/logs_events.rs | 84 +++++++++++++++++++- src/common/logs_filters.rs | 2 +- src/common/logs_parser.rs | 15 ++-- src/common/mod.rs | 3 +- src/grpc/mod.rs | 8 +- 10 files changed, 159 insertions(+), 314 deletions(-) delete mode 100755 src/common/dex_events.rs delete mode 100755 src/common/event.rs diff --git a/Cargo.toml b/Cargo.toml index f809cba..f940bd1 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,9 +18,9 @@ solana-client = "2.1.7" solana-transaction-status = "2.1.7" spl-token = "7.0.0" spl-associated-token-account = "6.0.0" +mpl-token-metadata = "5.1.0" borsh = { version = "1.5.3", features = ["derive"] } isahc = "1.7.2" -mpl-token-metadata = "5.1.0" serde = { version = "1.0.215", features = ["derive"] } serde_json = "1.0.134" futures = "0.3.31" diff --git a/README.md b/README.md index 60c2047..f8a9a57 100755 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ mai3-pumpfun-sdk = "2.4.5" ### logs subscription for token create and trade transaction ```rust -use mai3_pumpfun_sdk::instruction::{ +use mai3_pumpfun_sdk::common::{ logs_events::DexEvent, logs_subscribe::{tokens_subscription, stop_subscription} }; diff --git a/src/common/dex_events.rs b/src/common/dex_events.rs deleted file mode 100755 index b69d096..0000000 --- a/src/common/dex_events.rs +++ /dev/null @@ -1,123 +0,0 @@ -use anyhow::anyhow; -use base64::engine::general_purpose; -use base64::Engine; -use borsh::{BorshDeserialize, BorshSerialize}; -use regex::Regex; -use solana_sdk::pubkey::Pubkey; - -use super::myerror::AppError; - -pub const PROGRAM_DATA: &str = "Program data: "; - -#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] -pub struct TradeEvent { - pub mint: Pubkey, - pub sol_amount: u64, - pub token_amount: u64, - pub is_buy: bool, - pub user: Pubkey, - pub timestamp: u64, - pub virtual_sol_reserves: u64, - pub virtual_token_reserves: u64, - pub real_sol_reserves: u64, - pub real_token_reserves: u64, -} - -#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] -pub struct CompleteEvent { - pub user: Pubkey, - pub mint: Pubkey, - pub bonding_curve: Pubkey, - pub timestamp: u64, -} - -#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] -pub struct SwapBaseInLog { - pub log_type: u8, - // input - pub amount_in: u64, - pub minimum_out: u64, - pub direction: u64, - // user info - pub user_source: u64, - // pool info - pub pool_coin: u64, - pub pool_pc: u64, - // calc result - pub out_amount: u64, -} - -pub trait EventTrait: Sized + std::fmt::Debug { - fn from_bytes(bytes: &[u8]) -> Result; -} - -impl EventTrait for TradeEvent { - fn from_bytes(bytes: &[u8]) -> Result { - TradeEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) - } -} - -impl EventTrait for CompleteEvent { - fn from_bytes(bytes: &[u8]) -> Result { - CompleteEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) - } -} - -impl EventTrait for SwapBaseInLog { - fn from_bytes(bytes: &[u8]) -> Result { - SwapBaseInLog::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) - } -} - -#[derive(Debug, Clone, Copy)] -pub struct PumpEvent {} - -impl PumpEvent { - pub fn parse_logs(logs: &Vec) -> Option { - let mut event: Option = None; - if !logs.is_empty() { - let logs_iter = logs.iter().peekable(); - - for l in logs_iter.rev() { - if let Some(log) = l.strip_prefix(PROGRAM_DATA) { - let borsh_bytes = general_purpose::STANDARD.decode(log).unwrap(); - let slice: &[u8] = &borsh_bytes[8..]; - - if let Ok(e) = T::from_bytes(slice) { - event = Some(e); - } - } - } - } - event - } -} - -#[derive(Debug, Clone, Copy)] -pub struct RaydiumEvent {} - -impl RaydiumEvent { - pub fn parse_logs(logs: &Vec) -> Option { - let mut event: Option = None; - - if !logs.is_empty() { - let logs_iter = logs.iter().peekable(); - - for l in logs_iter.rev() { - let re = Regex::new(r"ray_log: (?P[A-Za-z0-9+/=]+)").unwrap(); - - if let Some(caps) = re.captures(l) { - if let Some(base64) = caps.name("base64") { - let bytes = general_purpose::STANDARD.decode(base64.as_str()).unwrap(); - - if let Ok(e) = T::from_bytes(&bytes) { - event = Some(e); - } - } - } - } - } - - event - } -} \ No newline at end of file diff --git a/src/common/event.rs b/src/common/event.rs deleted file mode 100755 index 5c8aa16..0000000 --- a/src/common/event.rs +++ /dev/null @@ -1,158 +0,0 @@ -use anyhow::anyhow; -use base64::engine::general_purpose; -use base64::Engine; -use borsh::{BorshDeserialize, BorshSerialize}; -use regex::Regex; -use solana_sdk::pubkey::Pubkey; - -use crate::error::AppError; - -pub const PROGRAM_DATA: &str = "Program data: "; - -#[derive(Debug)] -pub enum PumpfunEvent { - NewToken(CreateEvent), - NewUserTrade(TradeEvent), - NewBotTrade(TradeEvent), - Error(String), -} - -#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] -pub struct CreateEvent { - pub name: String, - pub symbol: String, - pub uri: String, - pub mint: Pubkey, - pub bonding_curve: Pubkey, - pub user: Pubkey, -} - -#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] -pub struct TradeEvent { - pub mint: Pubkey, - pub sol_amount: u64, - pub token_amount: u64, - pub is_buy: bool, - pub user: Pubkey, - pub timestamp: u64, - pub virtual_sol_reserves: u64, - pub virtual_token_reserves: u64, - pub real_sol_reserves: u64, - pub real_token_reserves: u64, -} - -#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] -pub struct CompleteEvent { - pub user: Pubkey, - pub mint: Pubkey, - pub bonding_curve: Pubkey, - pub timestamp: u64, -} - -#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] -pub struct SwapBaseInLog { - pub log_type: u8, - // input - pub amount_in: u64, - pub minimum_out: u64, - pub direction: u64, - // user info - pub user_source: u64, - // pool info - pub pool_coin: u64, - pub pool_pc: u64, - // calc result - pub out_amount: u64, -} - -pub trait EventTrait: Sized + std::fmt::Debug { - fn from_bytes(bytes: &[u8]) -> Result; -} - -impl EventTrait for CreateEvent { - fn from_bytes(bytes: &[u8]) -> Result { - CreateEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) - } -} - -impl EventTrait for TradeEvent { - fn from_bytes(bytes: &[u8]) -> Result { - TradeEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) - } -} - -impl EventTrait for CompleteEvent { - fn from_bytes(bytes: &[u8]) -> Result { - CompleteEvent::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) - } -} - -impl EventTrait for SwapBaseInLog { - fn from_bytes(bytes: &[u8]) -> Result { - SwapBaseInLog::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) - } -} - -#[derive(Debug, Clone, Copy)] -pub struct PumpEvent {} - -impl PumpEvent { - pub fn parse_logs(logs: &Vec) -> (Option, Option) { - let mut create_event: Option = None; - let mut trade_event: Option = None; - - if !logs.is_empty() { - let logs_iter = logs.iter().peekable(); - - for l in logs_iter.rev() { - if let Some(log) = l.strip_prefix(PROGRAM_DATA) { - let borsh_bytes = general_purpose::STANDARD.decode(log).unwrap(); - let slice: &[u8] = &borsh_bytes[8..]; - - if create_event.is_none() { - if let Ok(e) = CreateEvent::from_bytes(slice) { - create_event = Some(e); - continue; - } - } - - if trade_event.is_none() { - if let Ok(e) = TradeEvent::from_bytes(slice) { - trade_event = Some(e); - } - } - } - } - } - (create_event, trade_event) - } -} - -#[derive(Debug, Clone, Copy)] -pub struct RaydiumEvent {} - -impl RaydiumEvent { - pub fn parse_logs(logs: &Vec) -> Option { - let mut event: Option = None; - - if !logs.is_empty() { - let logs_iter = logs.iter().peekable(); - - for l in logs_iter.rev() { - let re = Regex::new(r"ray_log: (?P[A-Za-z0-9+/=]+)").unwrap(); - - if let Some(caps) = re.captures(l) { - if let Some(base64) = caps.name("base64") { - let bytes = general_purpose::STANDARD.decode(base64.as_str()).unwrap(); - - if let Ok(e) = T::from_bytes(&bytes) { - event = Some(e); - } - } - } - } - } - - event - } -} \ No newline at end of file diff --git a/src/common/logs_data.rs b/src/common/logs_data.rs index 2d868cd..53cac75 100755 --- a/src/common/logs_data.rs +++ b/src/common/logs_data.rs @@ -1,4 +1,8 @@ -use serde::{Serialize, Deserialize}; +use anyhow::anyhow; +use borsh::{BorshDeserialize, BorshSerialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::error::AppError; #[derive(Debug)] pub enum DexInstruction { @@ -8,30 +12,78 @@ pub enum DexInstruction { Other, } -// 添加新的数据结构 -#[derive(Debug, Clone, Serialize, Deserialize, Default)] +#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] pub struct CreateTokenInfo { - pub signature: String, pub name: String, pub symbol: String, pub uri: String, - pub mint: String, - pub bonding_curve: String, - pub user: String, + pub mint: Pubkey, + pub bonding_curve: Pubkey, + pub user: Pubkey, } -#[derive(Debug, Clone, Serialize, Deserialize, Default)] +#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] pub struct TradeInfo { - pub signature: String, - pub mint: String, - pub bonding_curve: String, + pub mint: Pubkey, pub sol_amount: u64, pub token_amount: u64, pub is_buy: bool, - pub user: String, + 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, +} + +#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] +pub struct CompleteInfo { + pub user: Pubkey, + pub mint: Pubkey, + pub bonding_curve: Pubkey, + pub timestamp: u64, +} + +#[derive(Clone, Debug, Default, PartialEq, BorshDeserialize, BorshSerialize)] +pub struct SwapBaseInLog { + pub log_type: u8, + // input + pub amount_in: u64, + pub minimum_out: u64, + pub direction: u64, + // user info + pub user_source: u64, + // pool info + pub pool_coin: u64, + pub pool_pc: u64, + // calc result + pub out_amount: u64, +} + +pub trait EventTrait: Sized + std::fmt::Debug { + fn from_bytes(bytes: &[u8]) -> Result; +} + +impl EventTrait for CreateTokenInfo { + fn from_bytes(bytes: &[u8]) -> Result { + CreateTokenInfo::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) + } +} + +impl EventTrait for TradeInfo { + fn from_bytes(bytes: &[u8]) -> Result { + TradeInfo::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) + } +} + +impl EventTrait for CompleteInfo { + fn from_bytes(bytes: &[u8]) -> Result { + CompleteInfo::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) + } +} + +impl EventTrait for SwapBaseInLog { + fn from_bytes(bytes: &[u8]) -> Result { + SwapBaseInLog::try_from_slice(bytes).map_err(|e| AppError::from(anyhow!(e.to_string()))) + } } \ No newline at end of file diff --git a/src/common/logs_events.rs b/src/common/logs_events.rs index b65c1cc..4b8be65 100755 --- a/src/common/logs_events.rs +++ b/src/common/logs_events.rs @@ -1,11 +1,87 @@ -use serde::{Serialize, Deserialize}; -use crate::common::logs_data::{CreateTokenInfo, TradeInfo}; -use crate::common::event::{CreateEvent, TradeEvent}; +use base64::engine::general_purpose; +use base64::Engine; +use regex::Regex; +use crate::common::logs_data::{CreateTokenInfo, TradeInfo, EventTrait}; -#[derive(Debug, Clone, Serialize, Deserialize)] +pub const PROGRAM_DATA: &str = "Program data: "; + +#[derive(Debug)] +pub enum PumpfunEvent { + NewToken(CreateTokenInfo), + NewUserTrade(TradeInfo), + NewBotTrade(TradeInfo), + Error(String), +} + + +#[derive(Debug)] pub enum DexEvent { NewToken(CreateTokenInfo), NewUserTrade(TradeInfo), NewBotTrade(TradeInfo), Error(String), +} + +// #[derive(Debug, Clone, Copy)] +// pub struct PumpEvent {} + +impl PumpfunEvent { + pub fn parse_logs(logs: &Vec) -> (Option, Option) { + let mut create_info: Option = None; + let mut trade_info: Option = None; + + if !logs.is_empty() { + let logs_iter = logs.iter().peekable(); + + for l in logs_iter.rev() { + if let Some(log) = l.strip_prefix(PROGRAM_DATA) { + let borsh_bytes = general_purpose::STANDARD.decode(log).unwrap(); + let slice: &[u8] = &borsh_bytes[8..]; + + if create_info.is_none() { + if let Ok(e) = CreateTokenInfo::from_bytes(slice) { + create_info = Some(e); + continue; + } + } + + if trade_info.is_none() { + if let Ok(e) = TradeInfo::from_bytes(slice) { + trade_info = Some(e); + } + } + } + } + } + (create_info, trade_info) + } +} + +#[derive(Debug, Clone, Copy)] +pub struct RaydiumEvent {} + +impl RaydiumEvent { + pub fn parse_logs(logs: &Vec) -> Option { + let mut event: Option = None; + + if !logs.is_empty() { + let logs_iter = logs.iter().peekable(); + + for l in logs_iter.rev() { + let re = Regex::new(r"ray_log: (?P[A-Za-z0-9+/=]+)").unwrap(); + + if let Some(caps) = re.captures(l) { + if let Some(base64) = caps.name("base64") { + let bytes = general_purpose::STANDARD.decode(base64.as_str()).unwrap(); + + if let Ok(e) = T::from_bytes(&bytes) { + event = Some(e); + } + } + } + } + } + + event + } } \ No newline at end of file diff --git a/src/common/logs_filters.rs b/src/common/logs_filters.rs index cfcc42f..4ac4963 100755 --- a/src/common/logs_filters.rs +++ b/src/common/logs_filters.rs @@ -65,7 +65,7 @@ impl LogFilter { "trade" => { if let Ok(trade_info) = parse_trade_data(&program_data) { if let Some(bot_wallet_pubkey) = bot_wallet { - if trade_info.user == bot_wallet_pubkey.to_string() { + if trade_info.user.to_string() == bot_wallet_pubkey.to_string() { instructions.push(DexInstruction::BotTrade(trade_info)); } else { instructions.push(DexInstruction::UserTrade(trade_info)); diff --git a/src/common/logs_parser.rs b/src/common/logs_parser.rs index 5ca672c..62ed4d9 100755 --- a/src/common/logs_parser.rs +++ b/src/common/logs_parser.rs @@ -1,3 +1,5 @@ +use std::str::FromStr; + use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use crate::error::{ClientError, ClientResult}; @@ -92,13 +94,12 @@ pub fn parse_create_token_data(data: &str) -> ClientResult { let user = bs58::encode(&decoded[cursor..cursor+32]).into_string(); Ok(CreateTokenInfo { - signature: String::new(), name, symbol, uri, - mint, - bonding_curve, - user, + mint: Pubkey::from_str(&mint).unwrap(), + bonding_curve: Pubkey::from_str(&bonding_curve).unwrap(), + user: Pubkey::from_str(&user).unwrap(), }) } @@ -157,13 +158,11 @@ pub fn parse_trade_data(data: &str) -> ClientResult { let real_token_reserves = u64::from_le_bytes(decoded[cursor..cursor + 8].try_into().unwrap()); Ok(TradeInfo { - signature: String::new(), - mint, - bonding_curve: String::new(), + mint: Pubkey::from_str(&mint).unwrap(), sol_amount, token_amount, is_buy, - user, + user: Pubkey::from_str(&user).unwrap(), timestamp, virtual_sol_reserves, virtual_token_reserves, diff --git a/src/common/mod.rs b/src/common/mod.rs index 322728d..3dd3acf 100644 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -1,6 +1,5 @@ pub mod logs_data; pub mod logs_parser; pub mod logs_filters; -pub mod logs_events; pub mod logs_subscribe; -pub mod event; \ No newline at end of file +pub mod logs_events; diff --git a/src/grpc/mod.rs b/src/grpc/mod.rs index 2599f0a..03028a1 100755 --- a/src/grpc/mod.rs +++ b/src/grpc/mod.rs @@ -16,9 +16,9 @@ use solana_transaction_status::{ }; use anyhow::anyhow; -use crate::common::event::{PumpEvent, RaydiumEvent, SwapBaseInLog, TradeEvent}; +use crate::common::logs_events::{PumpfunEvent, RaydiumEvent}; +use crate::common::logs_data::SwapBaseInLog; use crate::error::AppError; -use crate::common::event::PumpfunEvent; // 类型别名定义 type TransactionsFilterMap = HashMap; @@ -262,7 +262,7 @@ impl YellowstoneGrpc { info!("RaydiumEvent {:#?}", event); } SwapType::Pump => { - let (create_event, trade_event) = PumpEvent::parse_logs(logs); + let (create_event, trade_event) = PumpfunEvent::parse_logs(logs); if let Some(create_event) = create_event { callback(PumpfunEvent::NewToken(create_event)); } @@ -301,7 +301,7 @@ impl YellowstoneGrpc { } } -async fn subscribe_pumpfun() -> Result<(), AppError> { +async fn test_subscribe_pumpfun() -> Result<(), AppError> { // 创建YellowstoneGrpc实例 let endpoint = "https://grpc.mainnet.solana.com".to_string(); let client = YellowstoneGrpc::new(endpoint);