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

0.7.7

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-04-28 20:38:44 +02:00
parent b6aa14b62d
commit 7cf4f3a0f0
8 changed files with 1469 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -37,3 +37,4 @@
0.7.4 - Ajout du premier lot multi-DEX v1 avec décodeurs Pump.fun (create_v2) et PumpSwap (buy/sell), plus détection métier Pump.fun vers token/pool/pair/listing
0.7.5 - Enrichissement de PumpSwap avec extraction des mints et du pool_v2, persistance des événements décodés enrichis et détection métier automatique pool/pair/listing
0.7.6 - Ajout du premier support Meteora DBC avec décodage create_pool/swap, persistance des événements décodés et détection métier automatique pool/pair/listing
0.7.7 - Ajout du premier support Meteora DAMM v2 avec décodage create_pool/swap, persistance des événements décodés et détection métier automatique pool/pair/listing

2
Cargo.toml
View File

@@ -8,7 +8,7 @@ members = [
]
[workspace.package]
version = "0.7.6"
version = "0.7.7"
edition = "2024"
license = "MIT"
repository = "https://git.sasedev.com/Sasedev/khadhroony-bobobot"

71
ROADMAP.md
View File

@@ -515,9 +515,42 @@ Réalisé :
- persistance des événements `Meteora DBC` dans `kb_dex_decoded_events`,
- ajout de la détection métier `Meteora DBC` vers `pool / pair / listing`,
- émission des signaux dédiés `new_pool`, `new_pair` et `first_listing_seen`,
- préparation du lot suivant pour `DAMM v1`, `DAMM v2` et `LaunchLab`.
- préparation du lot suivant pour `Meteora DAMM v2`, `Meteora DAMM v1` et `LaunchLab / Fun Launch`.
### 6.039. Version `0.7.7` — Consolidation multi-DEX
### 6.039. Version `0.7.7` — Meteora DAMM v2
Réalisé :
- ajout du premier décodeur `Meteora DAMM v2`,
- prise en charge initiale des événements de création de pool via `initialize_pool`, `initialize_pool_with_dynamic_config` et `initialize_customizable_pool`,
- prise en charge initiale des swaps via `swap` et `swap2`,
- persistance des événements `Meteora DAMM v2` dans `kb_dex_decoded_events`,
- ajout de la détection métier `Meteora DAMM v2` vers `pool / pair / listing`,
- préparation du rattachement futur entre `Meteora DBC` et `Meteora DAMM v2`.
### 6.040. Version `0.7.8` — Meteora DAMM v1
Objectif : couvrir l’autre destination de migration possible après `Meteora DBC`.
À faire :
- ajouter un décodeur dédié `Meteora DAMM v1`,
- identifier les événements utiles de création de pool et de swaps,
- persister les événements décodés dans `kb_dex_decoded_events`,
- ajouter la détection métier `pool / pair / listing` pour `Meteora DAMM v1`,
- préparer le rattachement futur entre un pool `Meteora DBC` et son pool migré `Meteora DAMM v1`,
- conserver une séparation stricte entre `DBC`, `DAMM v1`, `DAMM v2` et les autres produits Meteora.
### 6.041. Version `0.7.9` — LaunchLab / Fun Launch
Objectif : prendre en charge la couche de lancement construite autour de `Meteora DBC` et `Meteora DAMM v2`.
À faire :
- ajouter un décodeur dédié `LaunchLab / Fun Launch` lorsque des signatures ou programmes propres sont observables,
- distinguer les événements de scaffolding / launchpad des événements `DBC` et `DAMM v2`,
- relier les lancements `LaunchLab / Fun Launch` aux pools et paires finalement créés,
- préparer le suivi des transitions entre lancement, bonding curve et AMM,
- éviter de confondre la couche launchpad avec les programmes DEX sous-jacents.
### 6.042. Version `0.7.10` — Consolidation multi-DEX
Objectif : unifier le comportement des connecteurs DEX v1 avant l’ouverture des couches analytiques plus riches.
À faire :
@@ -528,7 +561,7 @@ Objectif : unifier le comportement des connecteurs DEX v1 avant l’ouverture de
- améliorer l’idempotence et la traçabilité inter-protocoles,
- préparer la base des futurs événements enrichis de liquidité, swaps et activité.
### 6.040. Version `0.7.8` — Wallets, holdings et participants observés
### 6.043. Version `0.7.11` — Wallets, holdings et participants observés
Objectif : préparer le suivi des acteurs on-chain autour des pools et tokens détectés.
À faire :
@@ -538,7 +571,7 @@ Objectif : préparer le suivi des acteurs on-chain autour des pools et tokens d
- préparer l’identification des créateurs, mint authorities, wallets d’activité et contreparties,
- éviter de limiter l’analyse future au seul niveau token/pool sans vision des participants.
### 6.041. Version `0.7.9` — Séries de prix, volumes et agrégats DEX
### 6.044. Version `0.7.12` — Séries de prix, volumes et agrégats DEX
Objectif : préparer la couche analytique fine à partir des événements métier normalisés.
À faire :
@@ -548,20 +581,21 @@ Objectif : préparer la couche analytique fine à partir des événements métie
- permettre plus tard le calcul d’OHLCV, volume, nombre de trades et liquidité par fenêtre,
- préparer le terrain pour la couche analytique `0.8.x`.
### 6.042. Version `0.7.x` — Couverture DEX v1
### 6.045. Version `0.7.x` — Couverture DEX v1
Objectif : structurer les connecteurs DEX autour d’un pipeline complet de résolution, décodage et normalisation métier.
Protocoles cibles :
- Meteora
- Meteora DBC
- Meteora DAMM v2
- Meteora DAMM v1
- LaunchLab / Fun Launch
- Pump.fun
- PumpSwap
- Raydium
- Orca
- Pump.fun
- Meteora DBC
- Bags
- FluxBeam
- LaunchLab
- Heaven
- DexLab
- Moonit
@@ -575,7 +609,7 @@ Résultat attendu :
- création d’objets métier riches pour tokens, pools, paires, listings et participants,
- remplacement progressif des scripts heuristiques externes par des composants Rust intégrés.
### 6.043. Version `0.8.x` — Analyse et filtrage
### 6.046. Version `0.8.x` — Analyse et filtrage
Objectif : transformer les événements bruts en signaux exploitables.
À faire :
@@ -586,7 +620,7 @@ Objectif : transformer les événements bruts en signaux exploitables.
- statistiques de comportement,
- premiers patterns.
### 6.044. Version `1.x.y` — Wallets et swap préparatoire
### 6.047. Version `1.x.y` — Wallets et swap préparatoire
Objectif : préparer la couche d’action.
À faire :
@@ -597,7 +631,7 @@ Objectif : préparer la couche d’action.
- préparation d’ordres et de swaps,
- simulation et garde-fous.
### 6.045. Version `2.x.y` — Trading semi-automatisé
### 6.048. Version `2.x.y` — Trading semi-automatisé
Objectif : brancher l’analyse à l’action tout en gardant des garde-fous explicites.
À faire :
@@ -608,7 +642,7 @@ Objectif : brancher l’analyse à l’action tout en gardant des garde-fous exp
- confirmations explicites ou semi-automatiques,
- journaux d’exécution.
### 6.046. Version `3.x.y` — Yellowstone gRPC
### 6.049. Version `3.x.y` — Yellowstone gRPC
Objectif : ajouter le connecteur gRPC dédié.
À faire :
@@ -696,10 +730,9 @@ Le projet doit maintenir au minimum :
La priorité immédiate est désormais la suivante :
1. démarrer la version `0.7.4` avec la première vague de connecteurs DEX supplémentaires,
2. ajouter `Pump.fun`, `PumpSwap`, `Meteora`, `Meteora DBC` et `LaunchLab` au pipeline complet,
1. démarrer la version `0.7.7` avec le premier support `Meteora DAMM v2`,
2. préparer le rattachement futur entre `Meteora DBC` et `Meteora DAMM v2`,
3. conserver un décodeur séparé par protocole et par version,
4. unifier les sorties métier `pool / pair / listing` entre tous les DEX supportés,
5. préparer ensuite la version `0.7.5` pour `Orca`, les compléments `Raydium`, `Bags`, `FluxBeam` et `DexLab`,
6. préparer enfin la version `0.7.6` pour `Heaven`, `Moonit`, `Zora` et la consolidation multi-DEX.
4. préparer ensuite la version `0.7.8` pour `Meteora DAMM v1`,
5. préparer ensuite la version `0.7.9` pour `LaunchLab / Fun Launch`,
6. garder l’unification multi-DEX et la consolidation métier pour `0.7.10`.

7
kb_lib/src/dex.rs
View File

@@ -2,11 +2,18 @@
//! DEX-specific transaction decoders.
mod meteora_damm_v2;
mod meteora_dbc;
mod pump_fun;
mod pump_swap;
mod raydium_amm_v4;
pub use meteora_damm_v2::KB_METEORA_DAMM_V2_PROGRAM_ID;
pub use meteora_damm_v2::KbMeteoraDammV2CreatePoolDecoded;
pub use meteora_damm_v2::KbMeteoraDammV2DecodedEvent;
pub use meteora_damm_v2::KbMeteoraDammV2Decoder;
pub use meteora_damm_v2::KbMeteoraDammV2SwapDecoded;
pub use meteora_dbc::KB_METEORA_DBC_PROGRAM_ID;
pub use meteora_dbc::KbMeteoraDbcCreatePoolDecoded;
pub use meteora_dbc::KbMeteoraDbcDecodedEvent;

686
kb_lib/src/dex/meteora_damm_v2.rs Normal file
View File

@@ -0,0 +1,686 @@
// file: kb_lib/src/dex/meteora_damm_v2.rs
//! Meteora DAMM v2 transaction decoder.
/// Meteora DAMM v2 program id.
pub const KB_METEORA_DAMM_V2_PROGRAM_ID: &str = "cpamdpZCGKUy5JxQXB4dcpGPiikHawvSWAd6mEn1sGG";
/// Decoded Meteora DAMM v2 create-pool event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct KbMeteoraDammV2CreatePoolDecoded {
/// 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 pool account.
pub pool_account: std::option::Option<std::string::String>,
/// Optional token A mint.
pub token_a_mint: std::option::Option<std::string::String>,
/// Optional token B mint.
pub token_b_mint: std::option::Option<std::string::String>,
/// Optional static or dynamic config account.
pub config_account: std::option::Option<std::string::String>,
/// Optional creator / payer.
pub creator: std::option::Option<std::string::String>,
/// Instruction flavor: static / dynamic / customizable.
pub create_kind: std::string::String,
/// Decoded payload.
pub payload_json: serde_json::Value,
}
/// Decoded Meteora DAMM v2 swap event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct KbMeteoraDammV2SwapDecoded {
/// 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 relative to normalized base.
pub trade_side: crate::KbSwapTradeSide,
/// Optional pool account.
pub pool_account: std::option::Option<std::string::String>,
/// Optional token A mint.
pub token_a_mint: std::option::Option<std::string::String>,
/// Optional token B mint.
pub token_b_mint: std::option::Option<std::string::String>,
/// Whether the source instruction looked like swap2.
pub used_swap2: bool,
/// Decoded payload.
pub payload_json: serde_json::Value,
}
/// Decoded Meteora DAMM v2 event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum KbMeteoraDammV2DecodedEvent {
/// Pool creation.
CreatePool(KbMeteoraDammV2CreatePoolDecoded),
/// Swap / swap2.
Swap(KbMeteoraDammV2SwapDecoded),
}
/// Meteora DAMM v2 decoder.
#[derive(Debug, Clone, Default)]
pub struct KbMeteoraDammV2Decoder;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum KbMeteoraDammV2InstructionKind {
CreatePoolStatic,
CreatePoolDynamic,
CreatePoolCustomizable,
Swap,
Unknown,
}
impl KbMeteoraDammV2Decoder {
/// Creates a new decoder.
pub fn new() -> Self {
Self
}
/// Decodes one projected transaction into zero or more Meteora DAMM v2 events.
pub fn decode_transaction(
&self,
transaction: &crate::KbChainTransactionDto,
instructions: &[crate::KbChainInstructionDto],
) -> Result<std::vec::Vec<crate::KbMeteoraDammV2DecodedEvent>, 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_METEORA_DAMM_V2_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 parsed_json_result =
kb_parse_optional_parsed_json(instruction.parsed_json.as_ref());
let parsed_json = match parsed_json_result {
Ok(parsed_json) => parsed_json,
Err(error) => return Err(error),
};
let instruction_kind =
kb_classify_instruction_kind(parsed_json.as_ref(), &log_messages);
let pool_account = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&["pool", "poolAddress", "poolAccount", "poolState", "cpAmm"],
)
.or_else(|| kb_extract_account(&accounts, 0));
let token_a_mint = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&["tokenAMint", "mintA", "baseMint", "token0Mint", "mint0"],
)
.or_else(|| kb_extract_account(&accounts, 1));
let token_b_mint = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&["tokenBMint", "mintB", "quoteMint", "token1Mint", "mint1"],
)
.or_else(|| kb_extract_account(&accounts, 2));
let config_account = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&["staticConfig", "dynamicConfig", "config", "poolConfig"],
)
.or_else(|| kb_extract_account(&accounts, 3));
let creator = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&["creator", "payer", "user", "owner"],
)
.or_else(|| kb_extract_account(&accounts, 4));
if instruction_kind == KbMeteoraDammV2InstructionKind::CreatePoolStatic
|| instruction_kind == KbMeteoraDammV2InstructionKind::CreatePoolDynamic
|| instruction_kind == KbMeteoraDammV2InstructionKind::CreatePoolCustomizable
{
let create_kind = match instruction_kind {
KbMeteoraDammV2InstructionKind::CreatePoolStatic => "static".to_string(),
KbMeteoraDammV2InstructionKind::CreatePoolDynamic => "dynamic".to_string(),
KbMeteoraDammV2InstructionKind::CreatePoolCustomizable => {
"customizable".to_string()
}
_ => "unknown".to_string(),
};
let payload_json = serde_json::json!({
"decoder": "meteora_damm_v2",
"eventKind": "create_pool",
"classifiedInstructionKind": create_kind,
"signature": transaction.signature,
"instructionId": instruction_id,
"instructionIndex": instruction.instruction_index,
"accounts": accounts,
"parsed": parsed_json,
"logMessages": log_messages,
"poolAccount": pool_account,
"tokenAMint": token_a_mint,
"tokenBMint": token_b_mint,
"configAccount": config_account,
"creator": creator
});
decoded_events.push(crate::KbMeteoraDammV2DecodedEvent::CreatePool(
crate::KbMeteoraDammV2CreatePoolDecoded {
transaction_id,
instruction_id,
signature: transaction.signature.clone(),
program_id: program_id.clone(),
pool_account,
token_a_mint,
token_b_mint,
config_account,
creator,
create_kind,
payload_json,
},
));
continue;
}
if instruction_kind == KbMeteoraDammV2InstructionKind::Swap {
let used_swap2 = kb_log_messages_contain_keyword(&log_messages, "swap2")
|| kb_value_contains_any_key(parsed_json.as_ref(), &["swap2", "isSwap2"]);
let trade_side = kb_infer_trade_side(&log_messages);
let payload_json = serde_json::json!({
"decoder": "meteora_damm_v2",
"eventKind": "swap",
"classifiedInstructionKind": if used_swap2 { "swap2" } else { "swap" },
"signature": transaction.signature,
"instructionId": instruction_id,
"instructionIndex": instruction.instruction_index,
"accounts": accounts,
"parsed": parsed_json,
"logMessages": log_messages,
"poolAccount": pool_account,
"tokenAMint": token_a_mint,
"tokenBMint": token_b_mint,
"tradeSide": format!("{:?}", trade_side)
});
decoded_events.push(crate::KbMeteoraDammV2DecodedEvent::Swap(
crate::KbMeteoraDammV2SwapDecoded {
transaction_id,
instruction_id,
signature: transaction.signature.clone(),
program_id: program_id.clone(),
trade_side,
pool_account,
token_a_mint,
token_b_mint,
used_swap2,
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_classify_instruction_kind(
parsed_json: std::option::Option<&serde_json::Value>,
log_messages: &[std::string::String],
) -> KbMeteoraDammV2InstructionKind {
let parsed_instruction_name = kb_extract_string_by_candidate_keys(
parsed_json,
&["instruction", "instructionName", "type", "name"],
);
if let Some(parsed_instruction_name) = parsed_instruction_name {
let normalized = kb_normalize_text(parsed_instruction_name.as_str());
if normalized.contains("initializepoolwithdynamicconfig") {
return KbMeteoraDammV2InstructionKind::CreatePoolDynamic;
}
if normalized.contains("initializecustomizablepool") {
return KbMeteoraDammV2InstructionKind::CreatePoolCustomizable;
}
if normalized.contains("initializepool") {
return KbMeteoraDammV2InstructionKind::CreatePoolStatic;
}
if normalized == "swap" || normalized == "swap2" {
return KbMeteoraDammV2InstructionKind::Swap;
}
}
if kb_value_contains_any_key(parsed_json, &["dynamicConfig"]) {
return KbMeteoraDammV2InstructionKind::CreatePoolDynamic;
}
if kb_value_contains_any_key(parsed_json, &["staticConfig"]) {
return KbMeteoraDammV2InstructionKind::CreatePoolStatic;
}
if kb_value_contains_any_key(parsed_json, &["poolActivationPoint", "isCustomizablePool"]) {
return KbMeteoraDammV2InstructionKind::CreatePoolCustomizable;
}
if kb_log_messages_contain_keyword(log_messages, "initialize_pool_with_dynamic_config")
|| kb_log_messages_contain_keyword(log_messages, "initializepoolwithdynamicconfig")
{
return KbMeteoraDammV2InstructionKind::CreatePoolDynamic;
}
if kb_log_messages_contain_keyword(log_messages, "initialize_customizable_pool")
|| kb_log_messages_contain_keyword(log_messages, "initializecustomizablepool")
{
return KbMeteoraDammV2InstructionKind::CreatePoolCustomizable;
}
if kb_log_messages_contain_keyword(log_messages, "initialize_pool")
|| kb_log_messages_contain_keyword(log_messages, "initializepool")
{
return KbMeteoraDammV2InstructionKind::CreatePoolStatic;
}
if kb_log_messages_contain_keyword(log_messages, "swap2")
|| kb_log_messages_contain_keyword(log_messages, "swap")
{
return KbMeteoraDammV2InstructionKind::Swap;
}
KbMeteoraDammV2InstructionKind::Unknown
}
fn kb_log_messages_contain_keyword(log_messages: &[std::string::String], keyword: &str) -> bool {
let keyword_normalized = kb_normalize_text(keyword);
for log_message in log_messages {
let log_normalized = kb_normalize_text(log_message.as_str());
if log_normalized.contains(keyword_normalized.as_str()) {
return true;
}
}
false
}
fn kb_normalize_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
}
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_parse_optional_parsed_json(
parsed_json: std::option::Option<&std::string::String>,
) -> Result<std::option::Option<serde_json::Value>, crate::KbError> {
let parsed_json = match parsed_json {
Some(parsed_json) => parsed_json,
None => return Ok(None),
};
let value_result = serde_json::from_str::<serde_json::Value>(parsed_json.as_str());
match value_result {
Ok(value) => Ok(Some(value)),
Err(error) => Err(crate::KbError::Json(format!(
"cannot parse instruction parsed_json '{}': {}",
parsed_json, error
))),
}
}
fn kb_extract_string_by_candidate_keys(
value: std::option::Option<&serde_json::Value>,
candidate_keys: &[&str],
) -> std::option::Option<std::string::String> {
let value = match value {
Some(value) => value,
None => return None,
};
kb_extract_string_by_candidate_keys_inner(value, candidate_keys)
}
fn kb_extract_string_by_candidate_keys_inner(
value: &serde_json::Value,
candidate_keys: &[&str],
) -> std::option::Option<std::string::String> {
if let Some(object) = value.as_object() {
for candidate_key in candidate_keys {
let direct_option = object.get(*candidate_key);
if let Some(direct) = direct_option {
let direct_text_option = direct.as_str();
if let Some(direct_text) = direct_text_option {
return Some(direct_text.to_string());
}
}
}
for nested_value in object.values() {
let nested_result =
kb_extract_string_by_candidate_keys_inner(nested_value, candidate_keys);
if nested_result.is_some() {
return nested_result;
}
}
return None;
}
if let Some(array) = value.as_array() {
for nested_value in array {
let nested_result =
kb_extract_string_by_candidate_keys_inner(nested_value, candidate_keys);
if nested_result.is_some() {
return nested_result;
}
}
}
None
}
fn kb_value_contains_any_key(
value: std::option::Option<&serde_json::Value>,
candidate_keys: &[&str],
) -> bool {
let value = match value {
Some(value) => value,
None => return false,
};
kb_value_contains_any_key_inner(value, candidate_keys)
}
fn kb_value_contains_any_key_inner(value: &serde_json::Value, candidate_keys: &[&str]) -> bool {
if let Some(object) = value.as_object() {
for candidate_key in candidate_keys {
if object.contains_key(*candidate_key) {
return true;
}
}
for nested_value in object.values() {
if kb_value_contains_any_key_inner(nested_value, candidate_keys) {
return true;
}
}
return false;
}
if let Some(array) = value.as_array() {
for nested_value in array {
if kb_value_contains_any_key_inner(nested_value, candidate_keys) {
return true;
}
}
}
false
}
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_infer_trade_side(log_messages: &[std::string::String]) -> crate::KbSwapTradeSide {
if kb_log_messages_contain_keyword(log_messages, "buy") {
return crate::KbSwapTradeSide::BuyBase;
}
if kb_log_messages_contain_keyword(log_messages, "sell") {
return crate::KbSwapTradeSide::SellBase;
}
crate::KbSwapTradeSide::Unknown
}
#[cfg(test)]
mod tests {
fn make_create_transaction() -> crate::KbChainTransactionDto {
let mut dto = crate::KbChainTransactionDto::new(
"sig-meteora-damm-v2-create-1".to_string(),
Some(889001),
Some(1779400001),
Some("helius_primary_http".to_string()),
Some("0".to_string()),
None,
None,
serde_json::json!({
"slot": 889001,
"meta": {
"logMessages": [
"Program log: Instruction: InitializeCustomizablePool"
]
},
"transaction": {
"message": {
"instructions": []
}
}
})
.to_string(),
);
dto.id = Some(401);
dto
}
fn make_create_instruction() -> crate::KbChainInstructionDto {
let mut dto = crate::KbChainInstructionDto::new(
401,
None,
0,
None,
Some(crate::KB_METEORA_DAMM_V2_PROGRAM_ID.to_string()),
Some("meteora-damm-v2".to_string()),
Some(1),
serde_json::json!([
"DammV2Pool111",
"DammV2TokenA111",
"So11111111111111111111111111111111111111112",
"DammV2Config111",
"DammV2Creator111"
])
.to_string(),
None,
None,
Some(
serde_json::json!({
"info": {
"instruction": "initialize_customizable_pool",
"pool": "DammV2Pool111",
"tokenAMint": "DammV2TokenA111",
"tokenBMint": "So11111111111111111111111111111111111111112",
"creator": "DammV2Creator111",
"isCustomizablePool": true
}
})
.to_string(),
),
);
dto.id = Some(402);
dto
}
fn make_swap_transaction() -> crate::KbChainTransactionDto {
let mut dto = crate::KbChainTransactionDto::new(
"sig-meteora-damm-v2-swap-1".to_string(),
Some(889002),
Some(1779400002),
Some("helius_primary_http".to_string()),
Some("0".to_string()),
None,
None,
serde_json::json!({
"slot": 889002,
"meta": {
"logMessages": [
"Program log: Instruction: Swap2"
]
},
"transaction": {
"message": {
"instructions": []
}
}
})
.to_string(),
);
dto.id = Some(403);
dto
}
fn make_swap_instruction() -> crate::KbChainInstructionDto {
let mut dto = crate::KbChainInstructionDto::new(
403,
None,
0,
None,
Some(crate::KB_METEORA_DAMM_V2_PROGRAM_ID.to_string()),
Some("meteora-damm-v2".to_string()),
Some(1),
serde_json::json!([
"DammV2SwapPool111",
"DammV2SwapTokenA111",
"So11111111111111111111111111111111111111112"
])
.to_string(),
None,
None,
Some(
serde_json::json!({
"info": {
"instruction": "swap2",
"pool": "DammV2SwapPool111",
"tokenAMint": "DammV2SwapTokenA111",
"tokenBMint": "So11111111111111111111111111111111111111112"
}
})
.to_string(),
),
);
dto.id = Some(404);
dto
}
#[test]
fn meteora_damm_v2_create_pool_is_detected() {
let decoder = crate::KbMeteoraDammV2Decoder::new();
let transaction = make_create_transaction();
let instructions = vec![make_create_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::KbMeteoraDammV2DecodedEvent::CreatePool(event) => {
assert_eq!(event.transaction_id, 401);
assert_eq!(event.instruction_id, 402);
assert_eq!(event.pool_account, Some("DammV2Pool111".to_string()));
assert_eq!(event.token_a_mint, Some("DammV2TokenA111".to_string()));
assert_eq!(
event.token_b_mint,
Some("So11111111111111111111111111111111111111112".to_string())
);
assert_eq!(event.create_kind, "customizable".to_string());
}
crate::KbMeteoraDammV2DecodedEvent::Swap(_) => {
panic!("unexpected swap event")
}
}
}
#[test]
fn meteora_damm_v2_swap_is_detected() {
let decoder = crate::KbMeteoraDammV2Decoder::new();
let transaction = make_swap_transaction();
let instructions = vec![make_swap_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::KbMeteoraDammV2DecodedEvent::Swap(event) => {
assert_eq!(event.transaction_id, 403);
assert_eq!(event.instruction_id, 404);
assert_eq!(event.pool_account, Some("DammV2SwapPool111".to_string()));
assert_eq!(event.token_a_mint, Some("DammV2SwapTokenA111".to_string()));
assert_eq!(
event.token_b_mint,
Some("So11111111111111111111111111111111111111112".to_string())
);
assert!(event.used_swap2);
}
crate::KbMeteoraDammV2DecodedEvent::CreatePool(_) => {
panic!("unexpected create event")
}
}
}
}

303
kb_lib/src/dex_decode.rs
View File

@@ -11,6 +11,7 @@ pub struct KbDexDecodeService {
pump_fun_decoder: crate::KbPumpFunDecoder,
pump_swap_decoder: crate::KbPumpSwapDecoder,
meteora_dbc_decoder: crate::KbMeteoraDbcDecoder,
meteora_damm_v2_decoder: crate::KbMeteoraDammV2Decoder,
}
impl KbDexDecodeService {
@@ -24,6 +25,7 @@ impl KbDexDecodeService {
pump_fun_decoder: crate::KbPumpFunDecoder::new(),
pump_swap_decoder: crate::KbPumpSwapDecoder::new(),
meteora_dbc_decoder: crate::KbMeteoraDbcDecoder::new(),
meteora_damm_v2_decoder: crate::KbMeteoraDammV2Decoder::new(),
}
}
@@ -135,9 +137,223 @@ impl KbDexDecodeService {
};
persisted.push(persisted_event);
}
let meteora_damm_v2_decoded_result = self
.meteora_damm_v2_decoder
.decode_transaction(&transaction, &instructions);
let meteora_damm_v2_decoded = match meteora_damm_v2_decoded_result {
Ok(meteora_damm_v2_decoded) => meteora_damm_v2_decoded,
Err(error) => return Err(error),
};
for decoded_event in &meteora_damm_v2_decoded {
let persist_result = self
.persist_meteora_damm_v2_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_damm_v2_event(
&self,
transaction: &crate::KbChainTransactionDto,
decoded_event: &crate::KbMeteoraDammV2DecodedEvent,
) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
match decoded_event {
crate::KbMeteoraDammV2DecodedEvent::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 damm v2 payload: {}",
error
)));
}
};
let existing_result = crate::get_dex_decoded_event_by_key(
self.database.as_ref(),
event.transaction_id,
Some(event.instruction_id),
"meteora_damm_v2.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_damm_v2".to_string(),
event.program_id.clone(),
"meteora_damm_v2.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_damm_v2.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_damm_v2.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_damm_v2.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::KbMeteoraDammV2DecodedEvent::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 damm v2 payload: {}",
error
)));
}
};
let existing_result = crate::get_dex_decoded_event_by_key(
self.database.as_ref(),
event.transaction_id,
Some(event.instruction_id),
"meteora_damm_v2.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_damm_v2".to_string(),
event.program_id.clone(),
"meteora_damm_v2.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_damm_v2.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_damm_v2.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_damm_v2.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_dbc_event(
&self,
transaction: &crate::KbChainTransactionDto,
@@ -1023,4 +1239,91 @@ mod tests {
Some("DbcConfigDecode111".to_string())
);
}
async fn seed_projected_meteora_damm_v2_transaction(
database: std::sync::Arc<crate::KbDatabase>,
signature: &str,
) {
let service = crate::KbTransactionModelService::new(database);
let resolved_transaction = serde_json::json!({
"slot": 999005,
"blockTime": 1779000005,
"version": 0,
"transaction": {
"message": {
"instructions": [
{
"programId": crate::KB_METEORA_DAMM_V2_PROGRAM_ID,
"program": "meteora-damm-v2",
"stackHeight": 1,
"accounts": [
"DammV2DecodePool111",
"DammV2DecodeTokenA111",
"So11111111111111111111111111111111111111112",
"DammV2DecodeConfig111",
"DammV2DecodeCreator111"
],
"parsed": {
"info": {
"instruction": "initialize_customizable_pool",
"pool": "DammV2DecodePool111",
"tokenAMint": "DammV2DecodeTokenA111",
"tokenBMint": "So11111111111111111111111111111111111111112",
"creator": "DammV2DecodeCreator111",
"isCustomizablePool": true
}
},
"data": "opaque"
}
]
}
},
"meta": {
"err": null,
"logMessages": [
"Program log: Instruction: InitializeCustomizablePool"
]
}
});
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_damm_v2_event() {
let database = make_database().await;
seed_projected_meteora_damm_v2_transaction(database.clone(), "sig-dex-decode-dammv2-1")
.await;
let service = crate::KbDexDecodeService::new(database.clone());
let decoded_result = service
.decode_transaction_by_signature("sig-dex-decode-dammv2-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_damm_v2");
assert_eq!(decoded[0].event_kind, "meteora_damm_v2.create_pool");
assert_eq!(
decoded[0].pool_account,
Some("DammV2DecodePool111".to_string())
);
assert_eq!(
decoded[0].token_a_mint,
Some("DammV2DecodeTokenA111".to_string())
);
assert_eq!(
decoded[0].token_b_mint,
Some("So11111111111111111111111111111111111111112".to_string())
);
}
}

414
kb_lib/src/dex_detect.rs
View File

@@ -153,6 +153,30 @@ impl KbDexDetectService {
};
detection_results.push(detect_result);
}
if decoded_event.protocol_name == "meteora_damm_v2"
&& decoded_event.event_kind == "meteora_damm_v2.create_pool"
{
let detect_result = self
.detect_meteora_damm_v2_pool(&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);
}
if decoded_event.protocol_name == "meteora_damm_v2"
&& decoded_event.event_kind == "meteora_damm_v2.swap"
{
let detect_result = self
.detect_meteora_damm_v2_pool(&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)
}
@@ -1150,6 +1174,276 @@ impl KbDexDetectService {
})
}
async fn detect_meteora_damm_v2_pool(
&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_meteora_damm_v2_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_a_mint_option = decoded_event.token_a_mint.clone();
let token_a_mint = match token_a_mint_option {
Some(token_a_mint) => token_a_mint,
None => {
return Err(crate::KbError::InvalidState(format!(
"decoded event '{}' has no token_a_mint",
decoded_event_id
)));
}
};
let token_b_mint_option = decoded_event.token_b_mint.clone();
let token_b_mint = match token_b_mint_option {
Some(token_b_mint) => token_b_mint,
None => {
return Err(crate::KbError::InvalidState(format!(
"decoded event '{}' has no token_b_mint",
decoded_event_id
)));
}
};
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 {
token_a_mint.clone()
} else {
token_b_mint.clone()
};
let quote_mint = if base_is_token_a {
token_b_mint.clone()
} else {
token_a_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.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::Amm,
crate::KbPoolStatus::Active,
);
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.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.meteora_damm_v2.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.meteora_damm_v2.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.meteora_damm_v2.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_meteora_damm_v2_dex(&self) -> Result<i64, crate::KbError> {
let dex_result = crate::get_dex_by_code(self.database.as_ref(), "meteora_damm_v2").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(
"meteora_damm_v2 dex has no internal id".to_string(),
)),
},
None => {
let dex_dto = crate::KbDexDto::new(
"meteora_damm_v2".to_string(),
"Meteora DAMM v2".to_string(),
Some(crate::KB_METEORA_DAMM_V2_PROGRAM_ID.to_string()),
None,
true,
);
crate::upsert_dex(self.database.as_ref(), &dex_dto).await
}
}
}
async fn ensure_meteora_dbc_dex(&self) -> Result<i64, crate::KbError> {
let dex_result = crate::get_dex_by_code(self.database.as_ref(), "meteora_dbc").await;
let dex_option = match dex_result {
@@ -1957,4 +2251,124 @@ mod tests {
};
assert_eq!(pool_tokens.len(), 2);
}
async fn seed_decoded_meteora_damm_v2_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": 910005,
"blockTime": 1779100005,
"version": 0,
"transaction": {
"message": {
"instructions": [
{
"programId": crate::KB_METEORA_DAMM_V2_PROGRAM_ID,
"program": "meteora-damm-v2",
"stackHeight": 1,
"accounts": [
"DammV2DetectPool111",
"DammV2DetectTokenA111",
"So11111111111111111111111111111111111111112",
"DammV2DetectConfig111",
"DammV2DetectCreator111"
],
"parsed": {
"info": {
"instruction": "initialize_customizable_pool",
"pool": "DammV2DetectPool111",
"tokenAMint": "DammV2DetectTokenA111",
"tokenBMint": "So11111111111111111111111111111111111111112",
"creator": "DammV2DetectCreator111",
"isCustomizablePool": true
}
},
"data": "opaque"
}
]
}
},
"meta": {
"err": null,
"logMessages": [
"Program log: Instruction: InitializeCustomizablePool"
]
}
});
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_meteora_damm_v2_pool_pair_and_listing() {
let database = make_database().await;
seed_decoded_meteora_damm_v2_event(database.clone(), "sig-dex-detect-dammv2-1").await;
let detect_service = crate::KbDexDetectService::new(database.clone());
let detect_result = detect_service
.detect_transaction_by_signature("sig-dex-detect-dammv2-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(), "DammV2DetectPool111").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::Amm);
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);
}
}

5
kb_lib/src/lib.rs
View File

@@ -217,6 +217,11 @@ pub use dex::KbMeteoraDbcCreatePoolDecoded;
pub use dex::KbMeteoraDbcDecodedEvent;
pub use dex::KbMeteoraDbcDecoder;
pub use dex::KbMeteoraDbcSwapDecoded;
pub use dex::KB_METEORA_DAMM_V2_PROGRAM_ID;
pub use dex::KbMeteoraDammV2CreatePoolDecoded;
pub use dex::KbMeteoraDammV2DecodedEvent;
pub use dex::KbMeteoraDammV2Decoder;
pub use dex::KbMeteoraDammV2SwapDecoded;
pub use dex_decode::KbDexDecodeService;
pub use dex_detect::KbDexDetectService;
pub use dex_detect::KbDexPoolDetectionResult;