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

0.7.10

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-04-29 08:38:42 +02:00
parent d7b03c91b9
commit 3313f8245e
8 changed files with 1516 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

309
kb_lib/src/dex_decode.rs
View File

@@ -10,6 +10,7 @@ pub struct KbDexDecodeService {
raydium_amm_v4_decoder: crate::KbRaydiumAmmV4Decoder,
pump_fun_decoder: crate::KbPumpFunDecoder,
pump_swap_decoder: crate::KbPumpSwapDecoder,
orca_whirlpools_decoder: crate::KbOrcaWhirlpoolsDecoder,
meteora_dbc_decoder: crate::KbMeteoraDbcDecoder,
meteora_damm_v1_decoder: crate::KbMeteoraDammV1Decoder,
meteora_damm_v2_decoder: crate::KbMeteoraDammV2Decoder,
@@ -25,6 +26,7 @@ impl KbDexDecodeService {
raydium_amm_v4_decoder: crate::KbRaydiumAmmV4Decoder::new(),
pump_fun_decoder: crate::KbPumpFunDecoder::new(),
pump_swap_decoder: crate::KbPumpSwapDecoder::new(),
orca_whirlpools_decoder: crate::KbOrcaWhirlpoolsDecoder::new(),
meteora_dbc_decoder: crate::KbMeteoraDbcDecoder::new(),
meteora_damm_v1_decoder: crate::KbMeteoraDammV1Decoder::new(),
meteora_damm_v2_decoder: crate::KbMeteoraDammV2Decoder::new(),
@@ -173,9 +175,224 @@ impl KbDexDecodeService {
};
persisted.push(persisted_event);
}
let orca_whirlpools_decoded_result = self
.orca_whirlpools_decoder
.decode_transaction(&transaction, &instructions);
let orca_whirlpools_decoded = match orca_whirlpools_decoded_result {
Ok(orca_whirlpools_decoded) => orca_whirlpools_decoded,
Err(error) => return Err(error),
};
for decoded_event in &orca_whirlpools_decoded {
let persist_result = self
.persist_orca_whirlpools_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)
}
async fn persist_orca_whirlpools_event(
&self,
transaction: &crate::KbChainTransactionDto,
decoded_event: &crate::KbOrcaWhirlpoolsDecodedEvent,
) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
match decoded_event {
crate::KbOrcaWhirlpoolsDecodedEvent::CreatePool(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 orca whirlpools payload: {}",
error
)));
}
};
let existing_result = crate::get_dex_decoded_event_by_key(
self.database.as_ref(),
event.transaction_id,
Some(event.instruction_id),
"orca_whirlpools.create_pool",
)
.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),
"orca_whirlpools".to_string(),
event.program_id.clone(),
"orca_whirlpools.create_pool".to_string(),
event.pool_account.clone(),
None,
event.token_a_mint.clone(),
event.token_b_mint.clone(),
event.config_account.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),
"orca_whirlpools.create_pool",
)
.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.orca_whirlpools.create_pool".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.orca_whirlpools.create_pool".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)
}
crate::KbOrcaWhirlpoolsDecodedEvent::Swap(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 orca whirlpools payload: {}",
error
)));
}
};
let existing_result = crate::get_dex_decoded_event_by_key(
self.database.as_ref(),
event.transaction_id,
Some(event.instruction_id),
"orca_whirlpools.swap",
)
.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),
"orca_whirlpools".to_string(),
event.program_id.clone(),
"orca_whirlpools.swap".to_string(),
event.pool_account.clone(),
None,
event.token_a_mint.clone(),
event.token_b_mint.clone(),
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),
"orca_whirlpools.swap",
)
.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.orca_whirlpools.swap".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.orca_whirlpools.swap".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_meteora_damm_v1_event(
&self,
transaction: &crate::KbChainTransactionDto,
@@ -1629,4 +1846,96 @@ mod tests {
Some("So11111111111111111111111111111111111111112".to_string())
);
}
async fn seed_projected_orca_whirlpools_transaction(
database: std::sync::Arc<crate::KbDatabase>,
signature: &str,
) {
let service = crate::KbTransactionModelService::new(database);
let resolved_transaction = serde_json::json!({
"slot": 999007,
"blockTime": 1779000007,
"version": 0,
"transaction": {
"message": {
"instructions": [
{
"programId": crate::KB_ORCA_WHIRLPOOLS_PROGRAM_ID,
"program": "orca-whirlpools",
"stackHeight": 1,
"accounts": [
"OrcaDecodePool111",
"OrcaDecodeTokenA111",
"So11111111111111111111111111111111111111112",
"OrcaDecodeConfig111",
"OrcaDecodeCreator111"
],
"parsed": {
"info": {
"instruction": "initialize_pool_v2",
"whirlpool": "OrcaDecodePool111",
"tokenMintA": "OrcaDecodeTokenA111",
"tokenMintB": "So11111111111111111111111111111111111111112",
"whirlpoolsConfig": "OrcaDecodeConfig111",
"funder": "OrcaDecodeCreator111",
"tokenProgramA": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA",
"tokenProgramB": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"
}
},
"data": "opaque"
}
]
}
},
"meta": {
"err": null,
"logMessages": [
"Program log: Instruction: InitializePoolV2"
]
}
});
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_orca_whirlpools_event() {
let database = make_database().await;
seed_projected_orca_whirlpools_transaction(database.clone(), "sig-dex-decode-orca-1").await;
let service = crate::KbDexDecodeService::new(database.clone());
let decoded_result = service
.decode_transaction_by_signature("sig-dex-decode-orca-1")
.await;
let decoded = match decoded_result {
Ok(decoded) => decoded,
Err(error) => panic!("decode must succeed: {}", error),
};
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].protocol_name, "orca_whirlpools");
assert_eq!(decoded[0].event_kind, "orca_whirlpools.create_pool");
assert_eq!(
decoded[0].pool_account,
Some("OrcaDecodePool111".to_string())
);
assert_eq!(
decoded[0].token_a_mint,
Some("OrcaDecodeTokenA111".to_string())
);
assert_eq!(
decoded[0].token_b_mint,
Some("So11111111111111111111111111111111111111112".to_string())
);
assert_eq!(
decoded[0].market_account,
Some("OrcaDecodeConfig111".to_string())
);
}
}