0.7.6
This commit is contained in:
@@ -10,6 +10,7 @@ pub struct KbDexDecodeService {
|
||||
raydium_amm_v4_decoder: crate::KbRaydiumAmmV4Decoder,
|
||||
pump_fun_decoder: crate::KbPumpFunDecoder,
|
||||
pump_swap_decoder: crate::KbPumpSwapDecoder,
|
||||
meteora_dbc_decoder: crate::KbMeteoraDbcDecoder,
|
||||
}
|
||||
|
||||
impl KbDexDecodeService {
|
||||
@@ -22,6 +23,7 @@ impl KbDexDecodeService {
|
||||
raydium_amm_v4_decoder: crate::KbRaydiumAmmV4Decoder::new(),
|
||||
pump_fun_decoder: crate::KbPumpFunDecoder::new(),
|
||||
pump_swap_decoder: crate::KbPumpSwapDecoder::new(),
|
||||
meteora_dbc_decoder: crate::KbMeteoraDbcDecoder::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -116,9 +118,223 @@ impl KbDexDecodeService {
|
||||
};
|
||||
persisted.push(persisted_event);
|
||||
}
|
||||
let meteora_dbc_decoded_result = self
|
||||
.meteora_dbc_decoder
|
||||
.decode_transaction(&transaction, &instructions);
|
||||
let meteora_dbc_decoded = match meteora_dbc_decoded_result {
|
||||
Ok(meteora_dbc_decoded) => meteora_dbc_decoded,
|
||||
Err(error) => return Err(error),
|
||||
};
|
||||
for decoded_event in &meteora_dbc_decoded {
|
||||
let persist_result = self
|
||||
.persist_meteora_dbc_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_meteora_dbc_event(
|
||||
&self,
|
||||
transaction: &crate::KbChainTransactionDto,
|
||||
decoded_event: &crate::KbMeteoraDbcDecodedEvent,
|
||||
) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
|
||||
match decoded_event {
|
||||
crate::KbMeteoraDbcDecodedEvent::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 meteora dbc payload: {}",
|
||||
error
|
||||
)));
|
||||
}
|
||||
};
|
||||
let existing_result = crate::get_dex_decoded_event_by_key(
|
||||
self.database.as_ref(),
|
||||
event.transaction_id,
|
||||
Some(event.instruction_id),
|
||||
"meteora_dbc.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),
|
||||
"meteora_dbc".to_string(),
|
||||
event.program_id.clone(),
|
||||
"meteora_dbc.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),
|
||||
"meteora_dbc.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.meteora_dbc.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.meteora_dbc.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::KbMeteoraDbcDecodedEvent::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 meteora dbc payload: {}",
|
||||
error
|
||||
)));
|
||||
}
|
||||
};
|
||||
let existing_result = crate::get_dex_decoded_event_by_key(
|
||||
self.database.as_ref(),
|
||||
event.transaction_id,
|
||||
Some(event.instruction_id),
|
||||
"meteora_dbc.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),
|
||||
"meteora_dbc".to_string(),
|
||||
event.program_id.clone(),
|
||||
"meteora_dbc.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),
|
||||
"meteora_dbc.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.meteora_dbc.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.meteora_dbc.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_raydium_event(
|
||||
&self,
|
||||
transaction: &crate::KbChainTransactionDto,
|
||||
@@ -718,4 +934,93 @@ mod tests {
|
||||
Some("PumpSwapPoolV2_111".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
async fn seed_projected_meteora_dbc_transaction(
|
||||
database: std::sync::Arc<crate::KbDatabase>,
|
||||
signature: &str,
|
||||
) {
|
||||
let service = crate::KbTransactionModelService::new(database);
|
||||
let resolved_transaction = serde_json::json!({
|
||||
"slot": 999004,
|
||||
"blockTime": 1779000004,
|
||||
"version": 0,
|
||||
"transaction": {
|
||||
"message": {
|
||||
"instructions": [
|
||||
{
|
||||
"programId": crate::KB_METEORA_DBC_PROGRAM_ID,
|
||||
"program": "meteora-dbc",
|
||||
"stackHeight": 1,
|
||||
"accounts": [
|
||||
"DbcPoolDecode111",
|
||||
"DbcTokenDecode111",
|
||||
"So11111111111111111111111111111111111111112",
|
||||
"DbcConfigDecode111",
|
||||
"DbcCreatorDecode111"
|
||||
],
|
||||
"parsed": {
|
||||
"info": {
|
||||
"pool": "DbcPoolDecode111",
|
||||
"baseMint": "DbcTokenDecode111",
|
||||
"quoteMint": "So11111111111111111111111111111111111111112",
|
||||
"poolConfig": "DbcConfigDecode111",
|
||||
"creator": "DbcCreatorDecode111"
|
||||
}
|
||||
},
|
||||
"data": "opaque"
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"meta": {
|
||||
"err": null,
|
||||
"logMessages": [
|
||||
"Program log: Instruction: CreatePool"
|
||||
]
|
||||
}
|
||||
});
|
||||
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_meteora_dbc_event() {
|
||||
let database = make_database().await;
|
||||
seed_projected_meteora_dbc_transaction(database.clone(), "sig-dex-decode-dbc-1").await;
|
||||
let service = crate::KbDexDecodeService::new(database.clone());
|
||||
let decoded_result = service
|
||||
.decode_transaction_by_signature("sig-dex-decode-dbc-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, "meteora_dbc");
|
||||
assert_eq!(decoded[0].event_kind, "meteora_dbc.create_pool");
|
||||
assert_eq!(
|
||||
decoded[0].pool_account,
|
||||
Some("DbcPoolDecode111".to_string())
|
||||
);
|
||||
assert_eq!(
|
||||
decoded[0].token_a_mint,
|
||||
Some("DbcTokenDecode111".to_string())
|
||||
);
|
||||
assert_eq!(
|
||||
decoded[0].token_b_mint,
|
||||
Some("So11111111111111111111111111111111111111112".to_string())
|
||||
);
|
||||
assert_eq!(
|
||||
decoded[0].market_account,
|
||||
Some("DbcConfigDecode111".to_string())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user