Sasedev/khadhroony-bobobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.7.4

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-04-28 15:35:15 +02:00
parent e6b7b2d419
commit 467e488b36
9 changed files with 1414 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
kb_lib/src/dex.rs
View File

@@ -2,8 +2,20 @@
//! DEX-specific transaction decoders.
mod pump_fun;
mod pump_swap;
mod raydium_amm_v4;
pub use pump_fun::KB_PUMP_FUN_PROGRAM_ID;
pub use pump_fun::KbPumpFunCreateV2TokenDecoded;
pub use pump_fun::KbPumpFunDecodedEvent;
pub use pump_fun::KbPumpFunDecoder;
pub use pump_swap::KB_PUMP_SWAP_PROGRAM_ID;
pub use pump_swap::KbPumpSwapDecodedEvent;
pub use pump_swap::KbPumpSwapDecoder;
pub use pump_swap::KbPumpSwapTradeDecoded;
pub use raydium_amm_v4::KB_RAYDIUM_AMM_V4_PROGRAM_ID;
pub use raydium_amm_v4::KbRaydiumAmmV4DecodedEvent;
pub use raydium_amm_v4::KbRaydiumAmmV4Decoder;

338
kb_lib/src/dex/pump_fun.rs Normal file
View File

@@ -0,0 +1,338 @@
// file: kb_lib/src/dex/pump_fun.rs
//! Pump.fun bonding-curve transaction decoder.
/// Pump.fun program id.
pub const KB_PUMP_FUN_PROGRAM_ID: &str = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P";
/// Decoded Pump.fun `create_v2` token event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct KbPumpFunCreateV2TokenDecoded {
/// Parent transaction id.
pub transaction_id: i64,
/// Parent instruction id.
pub instruction_id: i64,
/// Transaction signature.
pub signature: std::string::String,
/// Program id.
pub program_id: std::string::String,
/// Optional created mint.
pub mint: std::option::Option<std::string::String>,
/// Optional bonding curve account.
pub bonding_curve: std::option::Option<std::string::String>,
/// Optional associated bonding curve account.
pub associated_bonding_curve: std::option::Option<std::string::String>,
/// Optional creator / user account.
pub creator: std::option::Option<std::string::String>,
/// Decoded payload.
pub payload_json: serde_json::Value,
}
/// Decoded Pump.fun event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum KbPumpFunDecodedEvent {
/// `create_v2` token creation.
CreateV2Token(KbPumpFunCreateV2TokenDecoded),
}
/// Pump.fun decoder.
#[derive(Debug, Clone, Default)]
pub struct KbPumpFunDecoder;
impl KbPumpFunDecoder {
/// Creates a new decoder.
pub fn new() -> Self {
Self
}
/// Decodes one projected transaction into zero or more Pump.fun events.
pub fn decode_transaction(
&self,
transaction: &crate::KbChainTransactionDto,
instructions: &[crate::KbChainInstructionDto],
) -> Result<std::vec::Vec<crate::KbPumpFunDecodedEvent>, crate::KbError> {
let transaction_id_option = transaction.id;
let transaction_id = match transaction_id_option {
Some(transaction_id) => transaction_id,
None => {
return Err(crate::KbError::InvalidState(format!(
"chain transaction '{}' has no internal id",
transaction.signature
)));
}
};
let transaction_json_result =
serde_json::from_str::<serde_json::Value>(transaction.transaction_json.as_str());
let transaction_json = match transaction_json_result {
Ok(transaction_json) => transaction_json,
Err(error) => {
return Err(crate::KbError::Json(format!(
"cannot parse transaction_json for signature '{}': {}",
transaction.signature, error
)));
}
};
let log_messages = kb_extract_log_messages(&transaction_json);
let has_create_v2_log = kb_log_messages_contain_keyword(&log_messages, "create_v2")
|| kb_log_messages_contain_keyword(&log_messages, "createv2");
let mut decoded_events = std::vec::Vec::new();
for instruction in instructions {
if instruction.parent_instruction_id.is_some() {
continue;
}
let program_id_option = &instruction.program_id;
let program_id = match program_id_option {
Some(program_id) => program_id,
None => continue,
};
if program_id.as_str() != crate::KB_PUMP_FUN_PROGRAM_ID {
continue;
}
if !has_create_v2_log {
continue;
}
let instruction_id_option = instruction.id;
let instruction_id = match instruction_id_option {
Some(instruction_id) => instruction_id,
None => continue,
};
let accounts_result = kb_parse_accounts_json(instruction.accounts_json.as_str());
let accounts = match accounts_result {
Ok(accounts) => accounts,
Err(error) => return Err(error),
};
if accounts.len() < 6 {
continue;
}
let mint = kb_extract_account(&accounts, 0);
let bonding_curve = kb_extract_account(&accounts, 2);
let associated_bonding_curve = kb_extract_account(&accounts, 3);
let creator = kb_extract_account(&accounts, 5);
let payload_json = serde_json::json!({
"decoder": "pump_fun",
"eventKind": "create_v2_token",
"signature": transaction.signature,
"instructionId": instruction_id,
"instructionIndex": instruction.instruction_index,
"accounts": accounts,
"logMessages": log_messages,
"mint": mint,
"bondingCurve": bonding_curve,
"associatedBondingCurve": associated_bonding_curve,
"creator": creator
});
decoded_events.push(crate::KbPumpFunDecodedEvent::CreateV2Token(
crate::KbPumpFunCreateV2TokenDecoded {
transaction_id,
instruction_id,
signature: transaction.signature.clone(),
program_id: program_id.clone(),
mint,
bonding_curve,
associated_bonding_curve,
creator,
payload_json,
},
));
}
Ok(decoded_events)
}
}
fn kb_extract_log_messages(
transaction_json: &serde_json::Value,
) -> std::vec::Vec<std::string::String> {
let mut messages = std::vec::Vec::new();
let meta_option = transaction_json.get("meta");
let meta = match meta_option {
Some(meta) => meta,
None => return messages,
};
let logs_option = meta.get("logMessages");
let logs = match logs_option {
Some(logs) => logs,
None => return messages,
};
let logs_array_option = logs.as_array();
let logs_array = match logs_array_option {
Some(logs_array) => logs_array,
None => return messages,
};
for value in logs_array {
let text_option = value.as_str();
if let Some(text) = text_option {
messages.push(text.to_string());
}
}
messages
}
fn kb_log_messages_contain_keyword(
log_messages: &[std::string::String],
keyword: &str,
) -> bool {
let keyword_normalized = kb_normalize_log_text(keyword);
for log_message in log_messages {
let log_normalized = kb_normalize_log_text(log_message.as_str());
if log_normalized.contains(keyword_normalized.as_str()) {
return true;
}
}
false
}
fn kb_parse_accounts_json(
accounts_json: &str,
) -> Result<std::vec::Vec<std::string::String>, crate::KbError> {
let values_result = serde_json::from_str::<std::vec::Vec<serde_json::Value>>(accounts_json);
let values = match values_result {
Ok(values) => values,
Err(error) => {
return Err(crate::KbError::Json(format!(
"cannot parse instruction accounts_json '{}': {}",
accounts_json, error
)));
}
};
let mut accounts = std::vec::Vec::new();
for value in values {
let text_option = value.as_str();
if let Some(text) = text_option {
accounts.push(text.to_string());
}
}
Ok(accounts)
}
fn kb_extract_account(
accounts: &[std::string::String],
index: usize,
) -> std::option::Option<std::string::String> {
if index >= accounts.len() {
return None;
}
Some(accounts[index].clone())
}
fn kb_normalize_log_text(value: &str) -> std::string::String {
let mut normalized = std::string::String::new();
for character in value.chars() {
if character.is_ascii_alphanumeric() {
normalized.push(character.to_ascii_lowercase());
}
}
normalized
}
#[cfg(test)]
mod tests {
fn make_transaction() -> crate::KbChainTransactionDto {
let mut dto = crate::KbChainTransactionDto::new(
"sig-pump-fun-test-1".to_string(),
Some(777001),
Some(1779200001),
Some("helius_primary_http".to_string()),
Some("0".to_string()),
None,
None,
serde_json::json!({
"slot": 777001,
"meta": {
"logMessages": [
"Program log: Instruction: CreateV2"
]
},
"transaction": {
"message": {
"instructions": []
}
}
})
.to_string(),
);
dto.id = Some(91);
dto
}
fn make_instruction() -> crate::KbChainInstructionDto {
let mut dto = crate::KbChainInstructionDto::new(
91,
None,
0,
None,
Some(crate::KB_PUMP_FUN_PROGRAM_ID.to_string()),
Some("pump".to_string()),
Some(1),
serde_json::json!([
"MintCreate111",
"MintAuthority111",
"BondingCurve111",
"AssociatedBondingCurve111",
"Global111",
"Creator111",
"System111",
"Token2022Program111",
"AtaProgram111"
])
.to_string(),
None,
None,
None,
);
dto.id = Some(17);
dto
}
#[test]
fn pump_fun_create_v2_is_detected() {
let decoder = crate::KbPumpFunDecoder::new();
let transaction = make_transaction();
let instructions = vec![make_instruction()];
let decoded_result = decoder.decode_transaction(&transaction, &instructions);
let decoded = match decoded_result {
Ok(decoded) => decoded,
Err(error) => panic!("decode must succeed: {}", error),
};
assert_eq!(decoded.len(), 1);
match &decoded[0] {
crate::KbPumpFunDecodedEvent::CreateV2Token(event) => {
assert_eq!(event.transaction_id, 91);
assert_eq!(event.instruction_id, 17);
assert_eq!(event.mint, Some("MintCreate111".to_string()));
assert_eq!(event.bonding_curve, Some("BondingCurve111".to_string()));
assert_eq!(
event.associated_bonding_curve,
Some("AssociatedBondingCurve111".to_string())
);
assert_eq!(event.creator, Some("Creator111".to_string()));
}
}
}
#[test]
fn pump_fun_create_v2_returns_none_without_expected_log() {
let decoder = crate::KbPumpFunDecoder::new();
let mut transaction = make_transaction();
transaction.transaction_json = serde_json::json!({
"slot": 777001,
"meta": {
"logMessages": [
"Program log: Instruction: Buy"
]
},
"transaction": {
"message": {
"instructions": []
}
}
})
.to_string();
let instructions = vec![make_instruction()];
let decoded_result = decoder.decode_transaction(&transaction, &instructions);
let decoded = match decoded_result {
Ok(decoded) => decoded,
Err(error) => panic!("decode must succeed: {}", error),
};
assert_eq!(decoded.len(), 0);
}
}

315
kb_lib/src/dex/pump_swap.rs Normal file
View File

@@ -0,0 +1,315 @@
// file: kb_lib/src/dex/pump_swap.rs
//! PumpSwap AMM transaction decoder.
/// PumpSwap / PumpAMM program id.
pub const KB_PUMP_SWAP_PROGRAM_ID: &str = "pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA";
/// Decoded PumpSwap trade event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct KbPumpSwapTradeDecoded {
/// Parent transaction id.
pub transaction_id: i64,
/// Parent instruction id.
pub instruction_id: i64,
/// Transaction signature.
pub signature: std::string::String,
/// Program id.
pub program_id: std::string::String,
/// Trade side.
pub trade_side: crate::KbSwapTradeSide,
/// Optional heuristic pool account.
pub pool_account: std::option::Option<std::string::String>,
/// Decoded payload.
pub payload_json: serde_json::Value,
}
/// Decoded PumpSwap event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum KbPumpSwapDecodedEvent {
/// Buy trade.
BuyTrade(KbPumpSwapTradeDecoded),
/// Sell trade.
SellTrade(KbPumpSwapTradeDecoded),
}
/// PumpSwap decoder.
#[derive(Debug, Clone, Default)]
pub struct KbPumpSwapDecoder;
impl KbPumpSwapDecoder {
/// Creates a new decoder.
pub fn new() -> Self {
Self
}
/// Decodes one projected transaction into zero or more PumpSwap events.
pub fn decode_transaction(
&self,
transaction: &crate::KbChainTransactionDto,
instructions: &[crate::KbChainInstructionDto],
) -> Result<std::vec::Vec<crate::KbPumpSwapDecodedEvent>, crate::KbError> {
let transaction_id_option = transaction.id;
let transaction_id = match transaction_id_option {
Some(transaction_id) => transaction_id,
None => {
return Err(crate::KbError::InvalidState(format!(
"chain transaction '{}' has no internal id",
transaction.signature
)));
}
};
let transaction_json_result =
serde_json::from_str::<serde_json::Value>(transaction.transaction_json.as_str());
let transaction_json = match transaction_json_result {
Ok(transaction_json) => transaction_json,
Err(error) => {
return Err(crate::KbError::Json(format!(
"cannot parse transaction_json for signature '{}': {}",
transaction.signature, error
)));
}
};
let log_messages = kb_extract_log_messages(&transaction_json);
let mut decoded_events = std::vec::Vec::new();
for instruction in instructions {
if instruction.parent_instruction_id.is_some() {
continue;
}
let program_id_option = &instruction.program_id;
let program_id = match program_id_option {
Some(program_id) => program_id,
None => continue,
};
if program_id.as_str() != crate::KB_PUMP_SWAP_PROGRAM_ID {
continue;
}
let instruction_id_option = instruction.id;
let instruction_id = match instruction_id_option {
Some(instruction_id) => instruction_id,
None => continue,
};
let accounts_result = kb_parse_accounts_json(instruction.accounts_json.as_str());
let accounts = match accounts_result {
Ok(accounts) => accounts,
Err(error) => return Err(error),
};
let pool_account = kb_extract_account(&accounts, 0);
let is_buy = kb_log_messages_contain_keyword(&log_messages, "buy");
let is_sell = kb_log_messages_contain_keyword(&log_messages, "sell");
if !is_buy && !is_sell {
continue;
}
let payload_json = serde_json::json!({
"decoder": "pump_swap",
"signature": transaction.signature,
"instructionId": instruction_id,
"instructionIndex": instruction.instruction_index,
"accounts": accounts,
"logMessages": log_messages,
"poolAccount": pool_account
});
if is_buy {
decoded_events.push(crate::KbPumpSwapDecodedEvent::BuyTrade(
crate::KbPumpSwapTradeDecoded {
transaction_id,
instruction_id,
signature: transaction.signature.clone(),
program_id: program_id.clone(),
trade_side: crate::KbSwapTradeSide::BuyBase,
pool_account: pool_account.clone(),
payload_json: payload_json.clone(),
},
));
}
if is_sell {
decoded_events.push(crate::KbPumpSwapDecodedEvent::SellTrade(
crate::KbPumpSwapTradeDecoded {
transaction_id,
instruction_id,
signature: transaction.signature.clone(),
program_id: program_id.clone(),
trade_side: crate::KbSwapTradeSide::SellBase,
pool_account,
payload_json,
},
));
}
}
Ok(decoded_events)
}
}
fn kb_extract_log_messages(
transaction_json: &serde_json::Value,
) -> std::vec::Vec<std::string::String> {
let mut messages = std::vec::Vec::new();
let meta_option = transaction_json.get("meta");
let meta = match meta_option {
Some(meta) => meta,
None => return messages,
};
let logs_option = meta.get("logMessages");
let logs = match logs_option {
Some(logs) => logs,
None => return messages,
};
let logs_array_option = logs.as_array();
let logs_array = match logs_array_option {
Some(logs_array) => logs_array,
None => return messages,
};
for value in logs_array {
let text_option = value.as_str();
if let Some(text) = text_option {
messages.push(text.to_string());
}
}
messages
}
fn kb_log_messages_contain_keyword(log_messages: &[std::string::String], keyword: &str) -> bool {
let keyword_lower = keyword.to_ascii_lowercase();
for log_message in log_messages {
let log_lower = log_message.to_ascii_lowercase();
if log_lower.contains(keyword_lower.as_str()) {
return true;
}
}
false
}
fn kb_parse_accounts_json(
accounts_json: &str,
) -> Result<std::vec::Vec<std::string::String>, crate::KbError> {
let values_result = serde_json::from_str::<std::vec::Vec<serde_json::Value>>(accounts_json);
let values = match values_result {
Ok(values) => values,
Err(error) => {
return Err(crate::KbError::Json(format!(
"cannot parse instruction accounts_json '{}': {}",
accounts_json, error
)));
}
};
let mut accounts = std::vec::Vec::new();
for value in values {
let text_option = value.as_str();
if let Some(text) = text_option {
accounts.push(text.to_string());
}
}
Ok(accounts)
}
fn kb_extract_account(
accounts: &[std::string::String],
index: usize,
) -> std::option::Option<std::string::String> {
if index >= accounts.len() {
return None;
}
Some(accounts[index].clone())
}
#[cfg(test)]
mod tests {
fn make_transaction_with_buy_log() -> crate::KbChainTransactionDto {
let mut dto = crate::KbChainTransactionDto::new(
"sig-pump-swap-test-1".to_string(),
Some(777002),
Some(1779200002),
Some("helius_primary_http".to_string()),
Some("0".to_string()),
None,
None,
serde_json::json!({
"slot": 777002,
"meta": {
"logMessages": [
"Program log: Instruction: Buy"
]
},
"transaction": {
"message": {
"instructions": []
}
}
})
.to_string(),
);
dto.id = Some(92);
dto
}
fn make_instruction() -> crate::KbChainInstructionDto {
let mut dto = crate::KbChainInstructionDto::new(
92,
None,
0,
None,
Some(crate::KB_PUMP_SWAP_PROGRAM_ID.to_string()),
Some("pump-amm".to_string()),
Some(1),
serde_json::json!(["PumpPool111", "Other1", "Other2"]).to_string(),
None,
None,
None,
);
dto.id = Some(18);
dto
}
#[test]
fn pump_swap_buy_is_detected() {
let decoder = crate::KbPumpSwapDecoder::new();
let transaction = make_transaction_with_buy_log();
let instructions = vec![make_instruction()];
let decoded_result = decoder.decode_transaction(&transaction, &instructions);
let decoded = match decoded_result {
Ok(decoded) => decoded,
Err(error) => panic!("decode must succeed: {}", error),
};
assert_eq!(decoded.len(), 1);
match &decoded[0] {
crate::KbPumpSwapDecodedEvent::BuyTrade(event) => {
assert_eq!(event.transaction_id, 92);
assert_eq!(event.instruction_id, 18);
assert_eq!(event.pool_account, Some("PumpPool111".to_string()));
assert_eq!(event.trade_side, crate::KbSwapTradeSide::BuyBase);
}
crate::KbPumpSwapDecodedEvent::SellTrade(_) => {
panic!("unexpected sell event")
}
}
}
#[test]
fn pump_swap_returns_none_without_buy_or_sell_log() {
let decoder = crate::KbPumpSwapDecoder::new();
let mut transaction = make_transaction_with_buy_log();
transaction.transaction_json = serde_json::json!({
"slot": 777002,
"meta": {
"logMessages": [
"Program log: Instruction: Deposit"
]
},
"transaction": {
"message": {
"instructions": []
}
}
})
.to_string();
let instructions = vec![make_instruction()];
let decoded_result = decoder.decode_transaction(&transaction, &instructions);
let decoded = match decoded_result {
Ok(decoded) => decoded,
Err(error) => panic!("decode must succeed: {}", error),
};
assert_eq!(decoded.len(), 0);
}
}

414
kb_lib/src/dex_decode.rs
View File

@@ -8,6 +8,8 @@ pub struct KbDexDecodeService {
database: std::sync::Arc<crate::KbDatabase>,
persistence: crate::KbDetectionPersistenceService,
raydium_amm_v4_decoder: crate::KbRaydiumAmmV4Decoder,
pump_fun_decoder: crate::KbPumpFunDecoder,
pump_swap_decoder: crate::KbPumpSwapDecoder,
}
impl KbDexDecodeService {
@@ -18,6 +20,8 @@ impl KbDexDecodeService {
database,
persistence,
raydium_amm_v4_decoder: crate::KbRaydiumAmmV4Decoder::new(),
pump_fun_decoder: crate::KbPumpFunDecoder::new(),
pump_swap_decoder: crate::KbPumpSwapDecoder::new(),
}
}
@@ -60,15 +64,15 @@ impl KbDexDecodeService {
Ok(instructions) => instructions,
Err(error) => return Err(error),
};
let decoded_result = self
let mut persisted = std::vec::Vec::new();
let raydium_decoded_result = self
.raydium_amm_v4_decoder
.decode_transaction(&transaction, &instructions);
let decoded = match decoded_result {
Ok(decoded) => decoded,
let raydium_decoded = match raydium_decoded_result {
Ok(raydium_decoded) => raydium_decoded,
Err(error) => return Err(error),
};
let mut persisted = std::vec::Vec::new();
for decoded_event in &decoded {
for decoded_event in &raydium_decoded {
let persist_result = self
.persist_raydium_event(&transaction, decoded_event)
.await;
@@ -78,6 +82,40 @@ impl KbDexDecodeService {
};
persisted.push(persisted_event);
}
let pump_fun_decoded_result = self
.pump_fun_decoder
.decode_transaction(&transaction, &instructions);
let pump_fun_decoded = match pump_fun_decoded_result {
Ok(pump_fun_decoded) => pump_fun_decoded,
Err(error) => return Err(error),
};
for decoded_event in &pump_fun_decoded {
let persist_result = self
.persist_pump_fun_event(&transaction, decoded_event)
.await;
let persisted_event = match persist_result {
Ok(persisted_event) => persisted_event,
Err(error) => return Err(error),
};
persisted.push(persisted_event);
}
let pump_swap_decoded_result = self
.pump_swap_decoder
.decode_transaction(&transaction, &instructions);
let pump_swap_decoded = match pump_swap_decoded_result {
Ok(pump_swap_decoded) => pump_swap_decoded,
Err(error) => return Err(error),
};
for decoded_event in &pump_swap_decoded {
let persist_result = self
.persist_pump_swap_event(&transaction, decoded_event)
.await;
let persisted_event = match persist_result {
Ok(persisted_event) => persisted_event,
Err(error) => return Err(error),
};
persisted.push(persisted_event);
}
Ok(persisted)
}
@@ -183,6 +221,239 @@ impl KbDexDecodeService {
}
}
}
async fn persist_pump_fun_event(
&self,
transaction: &crate::KbChainTransactionDto,
decoded_event: &crate::KbPumpFunDecodedEvent,
) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
match decoded_event {
crate::KbPumpFunDecodedEvent::CreateV2Token(event) => {
let payload_json_result = serde_json::to_string(&event.payload_json);
let payload_json = match payload_json_result {
Ok(payload_json) => payload_json,
Err(error) => {
return Err(crate::KbError::Json(format!(
"cannot serialize decoded pump.fun payload: {}",
error
)));
}
};
let existing_result = crate::get_dex_decoded_event_by_key(
self.database.as_ref(),
event.transaction_id,
Some(event.instruction_id),
"pump_fun.create_v2_token",
)
.await;
let existing_option = match existing_result {
Ok(existing_option) => existing_option,
Err(error) => return Err(error),
};
let already_present = existing_option.is_some();
let dto = crate::KbDexDecodedEventDto::new(
event.transaction_id,
Some(event.instruction_id),
"pump_fun".to_string(),
event.program_id.clone(),
"pump_fun.create_v2_token".to_string(),
event.bonding_curve.clone(),
None,
event.mint.clone(),
Some(crate::WSOL_MINT_ID.to_string()),
event.associated_bonding_curve.clone(),
payload_json,
);
let upsert_result =
crate::upsert_dex_decoded_event(self.database.as_ref(), &dto).await;
if let Err(error) = upsert_result {
return Err(error);
}
let fetched_result = crate::get_dex_decoded_event_by_key(
self.database.as_ref(),
event.transaction_id,
Some(event.instruction_id),
"pump_fun.create_v2_token",
)
.await;
let fetched_option = match fetched_result {
Ok(fetched_option) => fetched_option,
Err(error) => return Err(error),
};
let fetched = match fetched_option {
Some(fetched) => fetched,
None => {
return Err(crate::KbError::InvalidState(
"decoded event disappeared after upsert".to_string(),
));
}
};
if !already_present {
let payload_value = event.payload_json.clone();
let observation_result = self
.persistence
.record_observation(&crate::KbDetectionObservationInput::new(
"dex.pump_fun.create_v2_token".to_string(),
crate::KbObservationSourceKind::HttpRpc,
transaction.source_endpoint_name.clone(),
transaction.signature.clone(),
transaction.slot,
payload_value.clone(),
))
.await;
let observation_id = match observation_result {
Ok(observation_id) => observation_id,
Err(error) => return Err(error),
};
let signal_result = self
.persistence
.record_signal(&crate::KbDetectionSignalInput::new(
"signal.dex.pump_fun.create_v2_token".to_string(),
crate::KbAnalysisSignalSeverity::Low,
transaction.signature.clone(),
Some(observation_id),
None,
payload_value,
))
.await;
if let Err(error) = signal_result {
return Err(error);
}
}
Ok(fetched)
}
}
}
async fn persist_pump_swap_event(
&self,
transaction: &crate::KbChainTransactionDto,
decoded_event: &crate::KbPumpSwapDecodedEvent,
) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
match decoded_event {
crate::KbPumpSwapDecodedEvent::BuyTrade(event) => {
self.persist_pump_swap_trade_event(
transaction,
event,
"pump_swap.buy",
"signal.dex.pump_swap.buy",
"dex.pump_swap.buy",
)
.await
}
crate::KbPumpSwapDecodedEvent::SellTrade(event) => {
self.persist_pump_swap_trade_event(
transaction,
event,
"pump_swap.sell",
"signal.dex.pump_swap.sell",
"dex.pump_swap.sell",
)
.await
}
}
}
async fn persist_pump_swap_trade_event(
&self,
transaction: &crate::KbChainTransactionDto,
event: &crate::KbPumpSwapTradeDecoded,
event_kind: &str,
signal_kind: &str,
observation_kind: &str,
) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
let payload_json_result = serde_json::to_string(&event.payload_json);
let payload_json = match payload_json_result {
Ok(payload_json) => payload_json,
Err(error) => {
return Err(crate::KbError::Json(format!(
"cannot serialize decoded pump swap payload: {}",
error
)));
}
};
let existing_result = crate::get_dex_decoded_event_by_key(
self.database.as_ref(),
event.transaction_id,
Some(event.instruction_id),
event_kind,
)
.await;
let existing_option = match existing_result {
Ok(existing_option) => existing_option,
Err(error) => return Err(error),
};
let already_present = existing_option.is_some();
let dto = crate::KbDexDecodedEventDto::new(
event.transaction_id,
Some(event.instruction_id),
"pump_swap".to_string(),
event.program_id.clone(),
event_kind.to_string(),
event.pool_account.clone(),
None,
None,
None,
None,
payload_json,
);
let upsert_result = crate::upsert_dex_decoded_event(self.database.as_ref(), &dto).await;
if let Err(error) = upsert_result {
return Err(error);
}
let fetched_result = crate::get_dex_decoded_event_by_key(
self.database.as_ref(),
event.transaction_id,
Some(event.instruction_id),
event_kind,
)
.await;
let fetched_option = match fetched_result {
Ok(fetched_option) => fetched_option,
Err(error) => return Err(error),
};
let fetched = match fetched_option {
Some(fetched) => fetched,
None => {
return Err(crate::KbError::InvalidState(
"decoded event disappeared after upsert".to_string(),
));
}
};
if !already_present {
let payload_value = event.payload_json.clone();
let observation_result = self
.persistence
.record_observation(&crate::KbDetectionObservationInput::new(
observation_kind.to_string(),
crate::KbObservationSourceKind::HttpRpc,
transaction.source_endpoint_name.clone(),
transaction.signature.clone(),
transaction.slot,
payload_value.clone(),
))
.await;
let observation_id = match observation_result {
Ok(observation_id) => observation_id,
Err(error) => return Err(error),
};
let signal_result = self
.persistence
.record_signal(&crate::KbDetectionSignalInput::new(
signal_kind.to_string(),
crate::KbAnalysisSignalSeverity::Low,
transaction.signature.clone(),
Some(observation_id),
None,
payload_value,
))
.await;
if let Err(error) = signal_result {
return Err(error);
}
}
Ok(fetched)
}
}
#[cfg(test)]
@@ -214,7 +485,7 @@ mod tests {
std::sync::Arc::new(database)
}
async fn seed_projected_transaction(
async fn seed_projected_raydium_transaction(
database: std::sync::Arc<crate::KbDatabase>,
signature: &str,
) {
@@ -273,10 +544,61 @@ mod tests {
}
}
async fn seed_projected_pump_fun_transaction(
database: std::sync::Arc<crate::KbDatabase>,
signature: &str,
) {
let service = crate::KbTransactionModelService::new(database);
let resolved_transaction = serde_json::json!({
"slot": 999002,
"blockTime": 1779000002,
"version": 0,
"transaction": {
"message": {
"instructions": [
{
"programId": crate::KB_PUMP_FUN_PROGRAM_ID,
"program": "pump",
"stackHeight": 1,
"accounts": [
"MintPF111",
"MintAuthorityPF111",
"BondingCurvePF111",
"AssociatedBondingCurvePF111",
"GlobalPF111",
"CreatorPF111",
"System111",
"Token2022Program111",
"AtaProgram111"
],
"data": "opaque"
}
]
}
},
"meta": {
"err": null,
"logMessages": [
"Program log: Instruction: CreateV2"
]
}
});
let persist_result = service
.persist_resolved_transaction(
signature,
Some("helius_primary_http".to_string()),
&resolved_transaction,
)
.await;
if let Err(error) = persist_result {
panic!("projection must succeed: {}", error);
}
}
#[tokio::test]
async fn decode_transaction_by_signature_persists_decoded_event() {
async fn decode_transaction_by_signature_persists_decoded_raydium_event() {
let database = make_database().await;
seed_projected_transaction(database.clone(), "sig-dex-decode-1").await;
seed_projected_raydium_transaction(database.clone(), "sig-dex-decode-1").await;
let service = crate::KbDexDecodeService::new(database.clone());
let decoded_result = service
.decode_transaction_by_signature("sig-dex-decode-1")
@@ -289,73 +611,27 @@ mod tests {
assert_eq!(decoded[0].protocol_name, "raydium_amm_v4");
assert_eq!(decoded[0].event_kind, "raydium_amm_v4.initialize2_pool");
assert_eq!(decoded[0].pool_account, Some("PoolXYZ".to_string()));
let transaction_result =
crate::get_chain_transaction_by_signature(database.as_ref(), "sig-dex-decode-1").await;
let transaction_option = match transaction_result {
Ok(transaction_option) => transaction_option,
Err(error) => panic!("transaction fetch must succeed: {}", error),
};
let transaction = match transaction_option {
Some(transaction) => transaction,
None => panic!("transaction must exist"),
};
let transaction_id_option = transaction.id;
let transaction_id = match transaction_id_option {
Some(transaction_id) => transaction_id,
None => panic!("transaction id must exist"),
};
let listed_result =
crate::list_dex_decoded_events_by_transaction_id(database.as_ref(), transaction_id)
.await;
let listed = match listed_result {
Ok(listed) => listed,
Err(error) => panic!("dex event list must succeed: {}", error),
};
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].lp_mint, Some("LpMintXYZ".to_string()));
}
#[tokio::test]
async fn decode_transaction_by_signature_is_idempotent_on_same_transaction() {
async fn decode_transaction_by_signature_persists_decoded_pump_fun_event() {
let database = make_database().await;
seed_projected_transaction(database.clone(), "sig-dex-decode-2").await;
seed_projected_pump_fun_transaction(database.clone(), "sig-dex-decode-pump-1").await;
let service = crate::KbDexDecodeService::new(database.clone());
let first_result = service
.decode_transaction_by_signature("sig-dex-decode-2")
let decoded_result = service
.decode_transaction_by_signature("sig-dex-decode-pump-1")
.await;
if let Err(error) = first_result {
panic!("first decode must succeed: {}", error);
}
let second_result = service
.decode_transaction_by_signature("sig-dex-decode-2")
.await;
let second = match second_result {
Ok(second) => second,
Err(error) => panic!("second decode must succeed: {}", error),
let decoded = match decoded_result {
Ok(decoded) => decoded,
Err(error) => panic!("decode must succeed: {}", error),
};
assert_eq!(second.len(), 1);
let transaction_result =
crate::get_chain_transaction_by_signature(database.as_ref(), "sig-dex-decode-2").await;
let transaction_option = match transaction_result {
Ok(transaction_option) => transaction_option,
Err(error) => panic!("transaction fetch must succeed: {}", error),
};
let transaction = match transaction_option {
Some(transaction) => transaction,
None => panic!("transaction must exist"),
};
let transaction_id_option = transaction.id;
let transaction_id = match transaction_id_option {
Some(transaction_id) => transaction_id,
None => panic!("transaction id must exist"),
};
let listed_result =
crate::list_dex_decoded_events_by_transaction_id(database.as_ref(), transaction_id)
.await;
let listed = match listed_result {
Ok(listed) => listed,
Err(error) => panic!("dex event list must succeed: {}", error),
};
assert_eq!(listed.len(), 1);
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].protocol_name, "pump_fun");
assert_eq!(decoded[0].event_kind, "pump_fun.create_v2_token");
assert_eq!(
decoded[0].pool_account,
Some("BondingCurvePF111".to_string())
);
assert_eq!(decoded[0].token_a_mint, Some("MintPF111".to_string()));
}
}

387
kb_lib/src/dex_detect.rs
View File

@@ -93,6 +93,18 @@ impl KbDexDetectService {
};
detection_results.push(detect_result);
}
if decoded_event.protocol_name == "pump_fun"
&& decoded_event.event_kind == "pump_fun.create_v2_token"
{
let detect_result = self
.detect_pump_fun_create_v2_token(&transaction, decoded_event)
.await;
let detect_result = match detect_result {
Ok(detect_result) => detect_result,
Err(error) => return Err(error),
};
detection_results.push(detect_result);
}
}
Ok(detection_results)
}
@@ -147,7 +159,6 @@ impl KbDexDetectService {
}
};
let lp_mint = decoded_event.lp_mint.clone();
let base_is_token_a =
kb_choose_base_quote_order(token_a_mint.as_str(), token_b_mint.as_str());
let base_mint = if base_is_token_a {
@@ -369,6 +380,266 @@ impl KbDexDetectService {
})
}
async fn detect_pump_fun_create_v2_token(
&self,
transaction: &crate::KbChainTransactionDto,
decoded_event: &crate::KbDexDecodedEventDto,
) -> Result<crate::KbDexPoolDetectionResult, crate::KbError> {
let decoded_event_id_option = decoded_event.id;
let decoded_event_id = match decoded_event_id_option {
Some(decoded_event_id) => decoded_event_id,
None => {
return Err(crate::KbError::InvalidState(
"decoded dex event has no internal id".to_string(),
));
}
};
let dex_id_result = self.ensure_pump_fun_dex().await;
let dex_id = match dex_id_result {
Ok(dex_id) => dex_id,
Err(error) => return Err(error),
};
let pool_address_option = decoded_event.pool_account.clone();
let pool_address = match pool_address_option {
Some(pool_address) => pool_address,
None => {
return Err(crate::KbError::InvalidState(format!(
"decoded event '{}' has no pool_account",
decoded_event_id
)));
}
};
let token_mint_option = decoded_event.token_a_mint.clone();
let token_mint = match token_mint_option {
Some(token_mint) => token_mint,
None => {
return Err(crate::KbError::InvalidState(format!(
"decoded event '{}' has no token_a_mint",
decoded_event_id
)));
}
};
let quote_mint = crate::WSOL_MINT_ID.to_string();
let base_is_token_a = kb_choose_base_quote_order(token_mint.as_str(), quote_mint.as_str());
let base_mint = if base_is_token_a {
token_mint.clone()
} else {
quote_mint.clone()
};
let quote_mint_ordered = if base_is_token_a {
quote_mint.clone()
} else {
token_mint.clone()
};
let base_token_id_result = self.ensure_token(base_mint.as_str()).await;
let base_token_id = match base_token_id_result {
Ok(base_token_id) => base_token_id,
Err(error) => return Err(error),
};
let quote_token_id_result = self.ensure_token(quote_mint_ordered.as_str()).await;
let quote_token_id = match quote_token_id_result {
Ok(quote_token_id) => quote_token_id,
Err(error) => return Err(error),
};
let existing_pool_result =
crate::get_pool_by_address(self.database.as_ref(), pool_address.as_str()).await;
let existing_pool_option = match existing_pool_result {
Ok(existing_pool_option) => existing_pool_option,
Err(error) => return Err(error),
};
let created_pool = existing_pool_option.is_none();
let pool_id = match existing_pool_option {
Some(pool) => {
let pool_id_option = pool.id;
match pool_id_option {
Some(pool_id) => pool_id,
None => {
return Err(crate::KbError::InvalidState(format!(
"pool '{}' has no internal id",
pool.address
)));
}
}
}
None => {
let pool_dto = crate::KbPoolDto::new(
dex_id,
pool_address.clone(),
crate::KbPoolKind::BondingCurve,
crate::KbPoolStatus::Pending,
);
let upsert_result = crate::upsert_pool(self.database.as_ref(), &pool_dto).await;
match upsert_result {
Ok(pool_id) => pool_id,
Err(error) => return Err(error),
}
}
};
let existing_pair_result =
crate::get_pair_by_pool_id(self.database.as_ref(), pool_id).await;
let existing_pair_option = match existing_pair_result {
Ok(existing_pair_option) => existing_pair_option,
Err(error) => return Err(error),
};
let created_pair = existing_pair_option.is_none();
let pair_symbol = kb_build_pair_symbol(base_mint.as_str(), quote_mint_ordered.as_str());
let pair_id = match existing_pair_option {
Some(pair) => {
let pair_id_option = pair.id;
match pair_id_option {
Some(pair_id) => pair_id,
None => {
return Err(crate::KbError::InvalidState(format!(
"pair for pool '{}' has no internal id",
pool_id
)));
}
}
}
None => {
let pair_dto = crate::KbPairDto::new(
dex_id,
pool_id,
base_token_id,
quote_token_id,
pair_symbol,
);
let upsert_result = crate::upsert_pair(self.database.as_ref(), &pair_dto).await;
match upsert_result {
Ok(pair_id) => pair_id,
Err(error) => return Err(error),
}
}
};
let upsert_base_pool_token_result = crate::upsert_pool_token(
self.database.as_ref(),
&crate::KbPoolTokenDto::new(
pool_id,
base_token_id,
crate::KbPoolTokenRole::Base,
None,
Some(0),
),
)
.await;
if let Err(error) = upsert_base_pool_token_result {
return Err(error);
}
let upsert_quote_pool_token_result = crate::upsert_pool_token(
self.database.as_ref(),
&crate::KbPoolTokenDto::new(
pool_id,
quote_token_id,
crate::KbPoolTokenRole::Quote,
None,
Some(1),
),
)
.await;
if let Err(error) = upsert_quote_pool_token_result {
return Err(error);
}
let existing_listing_result =
crate::get_pool_listing_by_pool_id(self.database.as_ref(), pool_id).await;
let existing_listing_option = match existing_listing_result {
Ok(existing_listing_option) => existing_listing_option,
Err(error) => return Err(error),
};
let created_listing = existing_listing_option.is_none();
let pool_listing_id = match existing_listing_option {
Some(pool_listing) => pool_listing.id,
None => {
let listing_id_result = self
.upsert_pool_listing_from_decoded_event(dex_id, pool_id, pair_id, transaction)
.await;
match listing_id_result {
Ok(listing_id) => Some(listing_id),
Err(error) => return Err(error),
}
}
};
let payload_value_result = kb_parse_payload_json(decoded_event.payload_json.as_str());
let payload_value = match payload_value_result {
Ok(payload_value) => payload_value,
Err(error) => return Err(error),
};
if created_pool {
let signal_result = self
.record_detection_signal(
transaction,
"signal.dex.pump_fun.new_pool",
crate::KbAnalysisSignalSeverity::Low,
payload_value.clone(),
)
.await;
if let Err(error) = signal_result {
return Err(error);
}
}
if created_pair {
let signal_result = self
.record_detection_signal(
transaction,
"signal.dex.pump_fun.new_pair",
crate::KbAnalysisSignalSeverity::Low,
payload_value.clone(),
)
.await;
if let Err(error) = signal_result {
return Err(error);
}
}
if created_listing {
let signal_result = self
.record_detection_signal(
transaction,
"signal.dex.pump_fun.first_listing_seen",
crate::KbAnalysisSignalSeverity::Low,
payload_value,
)
.await;
if let Err(error) = signal_result {
return Err(error);
}
}
Ok(crate::KbDexPoolDetectionResult {
decoded_event_id,
dex_id,
pool_id,
pair_id,
pool_listing_id,
created_pool,
created_pair,
created_listing,
})
}
async fn ensure_pump_fun_dex(&self) -> Result<i64, crate::KbError> {
let dex_result = crate::get_dex_by_code(self.database.as_ref(), "pump_fun").await;
let dex_option = match dex_result {
Ok(dex_option) => dex_option,
Err(error) => return Err(error),
};
match dex_option {
Some(dex) => match dex.id {
Some(dex_id) => Ok(dex_id),
None => Err(crate::KbError::InvalidState(
"pump_fun dex has no internal id".to_string(),
)),
},
None => {
let dex_dto = crate::KbDexDto::new(
"pump_fun".to_string(),
"Pump.fun".to_string(),
Some(crate::KB_PUMP_FUN_PROGRAM_ID.to_string()),
None,
true,
);
crate::upsert_dex(self.database.as_ref(), &dex_dto).await
}
}
}
async fn ensure_raydium_dex(&self) -> Result<i64, crate::KbError> {
let dex_result = crate::get_dex_by_code(self.database.as_ref(), "raydium").await;
let dex_option = match dex_result {
@@ -720,4 +991,118 @@ mod tests {
};
assert_eq!(listings.len(), 1);
}
async fn seed_decoded_pump_fun_event(
database: std::sync::Arc<crate::KbDatabase>,
signature: &str,
) {
let transaction_model = crate::KbTransactionModelService::new(database.clone());
let dex_decode = crate::KbDexDecodeService::new(database);
let resolved_transaction = serde_json::json!({
"slot": 910002,
"blockTime": 1779100002,
"version": 0,
"transaction": {
"message": {
"instructions": [
{
"programId": crate::KB_PUMP_FUN_PROGRAM_ID,
"program": "pump",
"stackHeight": 1,
"accounts": [
"MintPumpDetect111",
"MintAuthority111",
"BondingCurveDetect111",
"AssociatedBondingCurveDetect111",
"Global111",
"CreatorDetect111",
"System111",
"Token2022Program111",
"AtaProgram111"
],
"data": "opaque"
}
]
}
},
"meta": {
"err": null,
"logMessages": [
"Program log: Instruction: CreateV2"
]
}
});
let project_result = transaction_model
.persist_resolved_transaction(
signature,
Some("helius_primary_http".to_string()),
&resolved_transaction,
)
.await;
if let Err(error) = project_result {
panic!("projection must succeed: {}", error);
}
let decode_result = dex_decode.decode_transaction_by_signature(signature).await;
if let Err(error) = decode_result {
panic!("dex decode must succeed: {}", error);
}
}
#[tokio::test]
async fn detect_transaction_by_signature_creates_pump_fun_pool_pair_and_listing() {
let database = make_database().await;
seed_decoded_pump_fun_event(database.clone(), "sig-dex-detect-pump-1").await;
let detect_service = crate::KbDexDetectService::new(database.clone());
let detect_result = detect_service
.detect_transaction_by_signature("sig-dex-detect-pump-1")
.await;
let results = match detect_result {
Ok(results) => results,
Err(error) => panic!("dex detect must succeed: {}", error),
};
assert_eq!(results.len(), 1);
assert!(results[0].created_pool);
assert!(results[0].created_pair);
assert!(results[0].created_listing);
let pool_result =
crate::get_pool_by_address(database.as_ref(), "BondingCurveDetect111").await;
let pool_option = match pool_result {
Ok(pool_option) => pool_option,
Err(error) => panic!("pool fetch must succeed: {}", error),
};
let pool = match pool_option {
Some(pool) => pool,
None => panic!("pool must exist"),
};
assert_eq!(pool.id, Some(results[0].pool_id));
assert_eq!(pool.pool_kind, crate::KbPoolKind::BondingCurve);
let pair_result = crate::get_pair_by_pool_id(database.as_ref(), results[0].pool_id).await;
let pair_option = match pair_result {
Ok(pair_option) => pair_option,
Err(error) => panic!("pair fetch must succeed: {}", error),
};
let pair = match pair_option {
Some(pair) => pair,
None => panic!("pair must exist"),
};
assert_eq!(pair.id, Some(results[0].pair_id));
let listing_result =
crate::get_pool_listing_by_pool_id(database.as_ref(), results[0].pool_id).await;
let listing_option = match listing_result {
Ok(listing_option) => listing_option,
Err(error) => panic!("listing fetch must succeed: {}", error),
};
let listing = match listing_option {
Some(listing) => listing,
None => panic!("listing must exist"),
};
assert_eq!(listing.id, results[0].pool_listing_id);
let pool_tokens_result =
crate::list_pool_tokens_by_pool_id(database.as_ref(), results[0].pool_id).await;
let pool_tokens = match pool_tokens_result {
Ok(pool_tokens) => pool_tokens,
Err(error) => panic!("pool tokens list must succeed: {}", error),
};
assert_eq!(pool_tokens.len(), 2);
}
}

10
kb_lib/src/lib.rs
View File

@@ -200,10 +200,18 @@ pub use tx_resolution::KbWsTransactionResolutionEnvelope;
pub use tx_resolution::KbWsTransactionResolutionRelay;
pub use tx_resolution::KbWsTransactionResolutionRelayStats;
pub use tx_model::KbTransactionModelService;
pub use dex::KB_RAYDIUM_AMM_V4_PROGRAM_ID;
pub use dex::KbRaydiumAmmV4DecodedEvent;
pub use dex::KbRaydiumAmmV4Decoder;
pub use dex::KbRaydiumAmmV4Initialize2PoolDecoded;
pub use dex::KB_RAYDIUM_AMM_V4_PROGRAM_ID;
pub use dex::KB_PUMP_FUN_PROGRAM_ID;
pub use dex::KbPumpFunCreateV2TokenDecoded;
pub use dex::KbPumpFunDecodedEvent;
pub use dex::KbPumpFunDecoder;
pub use dex::KB_PUMP_SWAP_PROGRAM_ID;
pub use dex::KbPumpSwapDecodedEvent;
pub use dex::KbPumpSwapDecoder;
pub use dex::KbPumpSwapTradeDecoded;
pub use dex_decode::KbDexDecodeService;
pub use dex_detect::KbDexDetectService;
pub use dex_detect::KbDexPoolDetectionResult;