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

0.7.28 - final

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-05-12 15:04:04 +02:00
parent 7f130dba6b
commit 4f6a4806e2
34 changed files with 4020 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@@ -58,4 +58,4 @@
0.7.25 - Enrichissement metadata des tokens, avec résolution locale limitée à SOL / WSOL, résolution des autres mints via comptes on-chain, Token-2022, Metaplex ou payloads DEX, et conservation explicite des cas non résolus
0.7.26 - Diagnostics locaux du pipeline persisté, correction de l’agrégation instruction-scoped des swaps Raydium, clarification des compteurs de replay/upsert, et validation qu’aucun trade candidate issu d’une transaction OK n’est perdu
0.7.27 - Validation multi-DEX et non-régression du pipeline sur Pump.fun, PumpSwap, Raydium CPMM et Raydium CLMM, avec corpus de tests, diagnostics de référence et garanties sur les événements non pricés
0.7.28 - nettoyer la couche DEX avant d’ajouter de nouveaux protocoles, sans modifier le transport HTTP/WS déjà stabilisé.
0.7.28 - Refactor DEX commun et verrouillage des invariants de normalisation : séparation des événements décodés, actionnables, trade candidates et candle candidates ; conservation des transactions failed comme traçables mais non actionnables ; ajout de la règle bloquante empêchant tout trade/candle candidate sans payload de montants exploitable, notamment pour le cas partiel `meteora_damm_v1.swap` sans base/quote amount.

11
Cargo.toml
View File

@@ -19,6 +19,7 @@ publish = false
argon2 = { version = "^0.5", features = ["std", "zeroize"] }
async-trait = { version = "^0.1", features = [] }
base64 = { version = "^0.22", features = [] }
borsh = { version = "^1.6", features = ["ascii", "bson", "bytes", "default", "derive", "de_strict_order", "borsh-derive", "indexmap", "std", "rc"] }
bs58 = {version = "^0.5", features = ["default", "cb58", "check"] }
chacha20poly1305 = { version = "^0.10", features = ["std", "stream"] }
chrono = { version = "^0.4", features = ["serde"] }
@@ -30,23 +31,23 @@ reqwest = { version = "^0.13", default-features = false, features = ["charset",
rustls = { version = "^0.23", features = ["aws-lc-rs"] }
serde = { version = "^1.0", features = ["derive"] }
serde_json = { version = "^1.0", features = [] }
solana-account-decoder-client-types = { version = ">=4.0.0-rc.0", features = ["zstd"] }
solana-address-lookup-table-interface = { version = "^3.1", features = ["bincode", "serde"] }
solana-account-decoder-client-types = { version = ">=4.0.0-rc.1", features = ["zstd"] }
solana-address-lookup-table-interface = { version = "^3.1", features = ["serde"] }
solana-client = { version = ">=4.0.0-rc.0", features = [] }
solana-compute-budget-interface = { version = "^3.0", features = ["borsh", "serde"] }
solana-rpc-client-api = { version = ">=4.0.0-rc.0", features = [] }
solana-rpc-client-types = { version = ">=4.0.0-rc.0", features = [] }
solana-sdk = { version = "^4.0", features = ["full"] }
solana-sdk-ids = { version = "^3.1", features = [] }
solana-system-interface = { version = "^3.2", features = ["alloc", "bincode", "serde", "std"] }
solana-transaction-status-client-types = { version = ">=4.0.0-rc.0", features = [] }
solana-system-interface = { version = "^3.2", features = ["alloc", "serde", "std"] }
solana-transaction-status-client-types = { version = ">=4.0.0-rc.1", features = [] }
spl-associated-token-account-interface = { version = "^2.0", features = ["borsh"] }
spl-memo-interface = { version = "^2.0", features = [] }
spl-token-interface = { version = "^2.0", features = [] }
spl-token-2022-interface = { version = "^2.1", features = [] }
sqlx = { version = "^0.8", features = ["chrono", "uuid", "bigdecimal", "json", "sqlite", "runtime-tokio-rustls"] }
tauri = { version = "^2.11", features = ["default", "tray-icon"] }
tauri-build = { version = "2", features = [] }
tauri-build = { version = "^2.6", features = [] }
tauri-plugin-tracing = { version = "^0.3", default-features = false, features = [] }
tempfile = { version = "^3", features = [] }
tokio = { version = "^1.52", features = ["full"] }

12
README.md
View File

@@ -246,15 +246,3 @@ Pour reprendre rapidement le codage dans une nouvelle session, fournir au minimu
- `kb_lib/src/db/queries.rs` et `kb_lib/src/db/queries/*`.
Ajouter `kb_demo_app/src/demo_pipeline*.rs` seulement si la tâche concerne l’UI ou les diagnostics affichés.
## 10. Prompt court de reprise
```text
Je reprends le workspace Rust khadhroony-bobobot autour de la version 0.7.27.
Objectif actuel : finaliser le pipeline DEX Solana avant trading.
Ne pas toucher pour le moment à ws_client/ws_manager/http_client/http_pool : ils fonctionnent et sont non bloquants.
Priorité : refactor DEX commun à partir de 0.7.28, matrice DEX, transactions inconnues/protocol candidates, événements non-trade, puis ajout/consolidation des DEX et launch surfaces.
Respecter les contraintes : Rust 2024, pas de mod.rs, pas de anyhow/thiserror, pas de ?/unwrap/expect dans le code applicatif, rustdoc utile sur l’API publique.
Les connecteurs à verrouiller avant extension sont pump_fun, pump_swap, raydium_cpmm et raydium_clmm.
Les launch surfaces sont importantes comme première source de mint, même si le token migre ensuite vers Raydium/Meteora/autre.
```

10
ROADMAP.md
View File

@@ -749,7 +749,9 @@ Réalisé :
- seuls les trade candidates issus de transactions échouées restent ignorés.
### 6.059. Version `0.7.27` — Validation multi-DEX des connecteurs déjà branchés
Réalisé :
Objectif : verrouiller la non-régression du pipeline actuel avant d’ajouter de nouveaux DEX ou d’ouvrir la phase d’analyse `0.8.x`.
À faire :
- rejouer des bases neuves de test pour `pump_fun`, `pump_swap`, `raydium_cpmm` et `raydium_clmm`,
- ne pas ajouter de nouveau DEX dans cette version ; cette version sert uniquement à valider les connecteurs déjà branchés,
@@ -764,7 +766,9 @@ Réalisé :
- valider que les transactions échouées restent traçables dans les événements décodés sans produire de `k_sol_trade_events`.
### 6.060. Version `0.7.28` — Refactor DEX commun et préparation extension
Réalisé :
Objectif : nettoyer la couche DEX avant d’ajouter de nouveaux protocoles, sans modifier le transport HTTP/WS déjà stabilisé.
À faire :
- ne pas toucher à `ws_client.rs`, `ws_manager.rs`, `http_client.rs`, `http_pool.rs` ni aux couches JSON-RPC déjà stabilisées,
- extraire depuis `dex_decode.rs` les catégories communes d’événements : trade, candle candidate, liquidity candidate, fee candidate, reward candidate, admin candidate, pool lifecycle candidate,
@@ -1163,7 +1167,7 @@ Le projet doit maintenir au minimum :
- un `README.md` global,
- un `ROADMAP.md` global,
- un `CHANGELOG.md` global,
- des `README.md` et `TODO.md` par crate à mesure de l’évolution (surtout en version 1.0),
- des `README.md` et `TODO.md` par crate à mesure de l’évolution,
- des tests unitaires robustes,
- les bindings TS générés via `cargo test export_bindings` lorsque les types partagés évoluent.

64
kb_demo_app/frontend/demo_pipeline2.html
View File

@@ -168,7 +168,50 @@
Diagnose local pipeline
</button>
<button id="demoPipeline2ValidateLocalPipelineButton" type="button" class="btn btn-outline-success">
Validate 0.7.27
Validate
</button>
</div>
</div>
</div>
</div>
<div class="accordion-item border-0 shadow-sm mb-3">
<h2 class="accordion-header" id="demoPipeline2DiscriminatorDiagnosticsHeading">
<button class="accordion-button collapsed" type="button" data-bs-toggle="collapse" data-bs-target="#demoPipeline2DiscriminatorDiagnosticsCollapse" aria-expanded="false" aria-controls="demoPipeline2DiscriminatorDiagnosticsCollapse">
Instruction discriminators
</button>
</h2>
<div id="demoPipeline2DiscriminatorDiagnosticsCollapse" class="accordion-collapse collapse" aria-labelledby="demoPipeline2DiscriminatorDiagnosticsHeading" data-bs-parent="#demoPipeline2LeftAccordion">
<div class="accordion-body">
<p class="small text-body-secondary mb-3">
Groupe les instructions d’un programme par discriminator décodé, nombre de comptes, stack height et statut decoded/non-decoded.
</p>
<div class="mb-3">
<label for="demoPipeline2DiscriminatorProgramIdInput" class="form-label">Program id</label>
<input
id="demoPipeline2DiscriminatorProgramIdInput"
type="text"
class="form-control font-monospace"
value="LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo"
/>
</div>
<div class="mb-3">
<label for="demoPipeline2DiscriminatorLimitInput" class="form-label">Instruction row limit</label>
<input
id="demoPipeline2DiscriminatorLimitInput"
type="number"
min="1"
step="1"
class="form-control"
value="200"
/>
</div>
<div class="d-flex gap-2">
<button id="demoPipeline2LoadDiscriminatorSummariesButton" type="button" class="btn btn-outline-warning">
Load discriminator summaries
</button>
</div>
</div>
@@ -290,6 +333,25 @@
</div>
</div>
</div>
<div class="accordion-item border-0 shadow-sm mb-3">
<h2 class="accordion-header" id="demoPipeline2DiscriminatorSummaryHeading">
<button class="accordion-button collapsed" type="button" data-bs-toggle="collapse" data-bs-target="#demoPipeline2DiscriminatorSummaryCollapse" aria-expanded="false" aria-controls="demoPipeline2DiscriminatorSummaryCollapse">
Instruction discriminator summaries
</button>
</h2>
<div id="demoPipeline2DiscriminatorSummaryCollapse" class="accordion-collapse collapse" aria-labelledby="demoPipeline2DiscriminatorSummaryHeading" data-bs-parent="#demoPipeline2ContentAccordion">
<div class="accordion-body">
<textarea
id="demoPipeline2DiscriminatorSummariesTextarea"
class="form-control font-monospace"
rows="18"
readonly
spellcheck="false"
></textarea>
</div>
</div>
</div>
<div class="accordion-item border-0 shadow-sm mb-3">
<h2 class="accordion-header" id="demoPipeline2ProtocolCandidateSummaryHeading">

10
kb_demo_app/frontend/ts/bindings/DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload.ts Normal file
View File

@@ -0,0 +1,10 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
/**
* Response payload for program instruction discriminator summaries.
*/
export type DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload = {
/**
* Pretty JSON summary rows.
*/
summariesJson: string, };

14
kb_demo_app/frontend/ts/bindings/DemoPipeline2ProgramInstructionDiscriminatorSummaryRequest.ts Normal file
View File

@@ -0,0 +1,14 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
/**
* Request payload for program instruction discriminator summaries.
*/
export type DemoPipeline2ProgramInstructionDiscriminatorSummaryRequest = {
/**
* Program id to inspect.
*/
programId: string,
/**
* Maximum number of instruction rows to inspect before grouping.
*/
limit: number, };

60
kb_demo_app/frontend/ts/demo_pipeline2.ts
View File

@@ -15,6 +15,8 @@ import type { DemoPipeline2PairCandlesRequest } from "./bindings/DemoPipeline2Pa
import type { DemoPipeline2PairCandlesPayload } from "./bindings/DemoPipeline2PairCandlesPayload.ts";
import type { DemoPipeline2LocalDiagnosticsPayload } from "./bindings/DemoPipeline2LocalDiagnosticsPayload.ts";
import type { DemoPipeline2LocalValidationPayload } from "./bindings/DemoPipeline2LocalValidationPayload.ts";
import type { DemoPipeline2ProgramInstructionDiscriminatorSummaryRequest } from "./bindings/DemoPipeline2ProgramInstructionDiscriminatorSummaryRequest.ts";
import type { DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload } from "./bindings/DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload.ts";
import { DemoPipeline2ProtocolCandidateSummaryRequest } from './bindings/DemoPipeline2ProtocolCandidateSummaryRequest.ts';
import { DemoPipeline2ProtocolCandidateSummaryPayload } from './bindings/DemoPipeline2ProtocolCandidateSummaryPayload.ts';
@@ -356,6 +358,10 @@ document.addEventListener("DOMContentLoaded", async () => {
const diagnoseLocalPipelineButton = document.querySelector<HTMLButtonElement>("#demoPipeline2DiagnoseLocalPipelineButton");
const validateLocalPipelineButton = document.querySelector<HTMLButtonElement>("#demoPipeline2ValidateLocalPipelineButton");
const discriminatorProgramIdInput = document.querySelector<HTMLInputElement>("#demoPipeline2DiscriminatorProgramIdInput");
const discriminatorLimitInput = document.querySelector<HTMLInputElement>("#demoPipeline2DiscriminatorLimitInput");
const loadDiscriminatorSummariesButton = document.querySelector<HTMLButtonElement>("#demoPipeline2LoadDiscriminatorSummariesButton");
const protocolCandidateLimitInput = document.querySelector<HTMLInputElement>("#demoPipeline2ProtocolCandidateLimitInput");
const refreshProtocolCandidatesButton = document.querySelector<HTMLButtonElement>("#demoPipeline2RefreshProtocolCandidatesButton");
@@ -371,6 +377,8 @@ document.addEventListener("DOMContentLoaded", async () => {
const localDiagnosticsTextarea = document.querySelector<HTMLTextAreaElement>("#demoPipeline2LocalDiagnosticsTextarea");
const localValidationTextarea = document.querySelector<HTMLTextAreaElement>("#demoPipeline2LocalValidationTextarea");
const discriminatorSummariesTextarea = document.querySelector<HTMLTextAreaElement>("#demoPipeline2DiscriminatorSummariesTextarea");
const protocolCandidateSummariesTextarea = document.querySelector<HTMLTextAreaElement>("#demoPipeline2ProtocolCandidateSummariesTextarea");
const clearLogButton = document.querySelector<HTMLButtonElement>("#demoPipeline2ClearLogButton");
@@ -395,6 +403,9 @@ document.addEventListener("DOMContentLoaded", async () => {
!replayLocalPipelineButton ||
!diagnoseLocalPipelineButton ||
!validateLocalPipelineButton ||
!discriminatorProgramIdInput ||
!discriminatorLimitInput ||
!loadDiscriminatorSummariesButton ||
!protocolCandidateLimitInput ||
!refreshProtocolCandidatesButton ||
!pairSelect ||
@@ -405,6 +416,7 @@ document.addEventListener("DOMContentLoaded", async () => {
!backfillSummaryTextarea ||
!localDiagnosticsTextarea ||
!localValidationTextarea ||
!discriminatorSummariesTextarea ||
!protocolCandidateSummariesTextarea ||
!chartElement ||
!chartMeta ||
@@ -635,7 +647,7 @@ document.addEventListener("DOMContentLoaded", async () => {
});
validateLocalPipelineButton.addEventListener("click", async () => {
appendLogLine(logTextarea, "[ui] validating local pipeline with 0.7.27 profile");
appendLogLine(logTextarea, "[ui] validating local pipeline with 0.7.28 profile");
try {
const payload = await invoke<DemoPipeline2LocalValidationPayload>(
@@ -693,6 +705,52 @@ document.addEventListener("DOMContentLoaded", async () => {
}
});
loadDiscriminatorSummariesButton.addEventListener("click", async () => {
const programId = discriminatorProgramIdInput.value.trim();
if (programId === "") {
appendLogLine(logTextarea, "[ui] discriminator program id is required");
return;
}
const limit = readPositiveIntegerInput(
discriminatorLimitInput,
logTextarea,
"discriminatorSummaryLimit",
);
if (limit === undefined) {
return;
}
appendLogLine(
logTextarea,
`[ui] loading instruction discriminator summaries for program '${programId}' with limit '${limit.toString()}'`,
);
const request: DemoPipeline2ProgramInstructionDiscriminatorSummaryRequest = {
programId,
limit,
};
try {
const payload = await invoke<DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload>(
"demo_pipeline2_get_program_instruction_discriminator_summaries",
{ request },
);
discriminatorSummariesTextarea.value = payload.summariesJson;
appendLogLine(
logTextarea,
"[ui] instruction discriminator summaries loaded",
);
} catch (error) {
appendLogLine(
logTextarea,
`[ui] instruction discriminator summary error: ${String(error)}`,
);
}
});
loadCandlesButton.addEventListener("click", async () => {
const pairIdText = pairSelect.value.trim();
if (pairIdText === "") {

74
kb_demo_app/src/demo_pipeline2.rs
View File

@@ -10,6 +10,31 @@
use tauri::Manager;
use ts_rs::TS;
/// Request payload for program instruction discriminator summaries.
#[derive(Clone, Debug, serde::Deserialize, ts_rs::TS)]
#[ts(
export,
export_to = "../frontend/ts/bindings/DemoPipeline2ProgramInstructionDiscriminatorSummaryRequest.ts"
)]
#[serde(rename_all = "camelCase")]
pub(crate) struct DemoPipeline2ProgramInstructionDiscriminatorSummaryRequest {
/// Program id to inspect.
pub program_id: std::string::String,
/// Maximum number of instruction rows to inspect before grouping.
pub limit: u32,
}
/// Response payload for program instruction discriminator summaries.
#[derive(Clone, Debug, serde::Serialize, ts_rs::TS)]
#[ts(
export,
export_to = "../frontend/ts/bindings/DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload.ts"
)]
#[serde(rename_all = "camelCase")]
pub(crate) struct DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload {
/// Pretty JSON summary rows.
pub summaries_json: std::string::String,
}
/// Request payload for protocol candidate summary diagnostics.
#[derive(Clone, Debug, serde::Deserialize, ts_rs::TS)]
@@ -35,9 +60,6 @@ pub(crate) struct DemoPipeline2ProtocolCandidateSummaryPayload {
pub summaries_json: std::string::String,
}
/// Local diagnostics payload returned to the UI.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[ts(
@@ -718,6 +740,43 @@ pub(crate) struct DemoPipeline2PairCandlesPayload {
pub candles_json: std::string::String,
}
/// Lists program instruction discriminator summaries for one program id.
#[tauri::command]
pub(crate) async fn demo_pipeline2_get_program_instruction_discriminator_summaries(
state: tauri::State<'_, crate::AppState>,
request: DemoPipeline2ProgramInstructionDiscriminatorSummaryRequest,
) -> Result<DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload, std::string::String> {
if request.program_id.trim().is_empty() {
return Err("program id must not be empty".to_string());
}
if request.limit == 0 {
return Err("instruction discriminator summary limit must be > 0".to_string());
}
let rows_result = kb_lib::query_program_instruction_discriminator_summaries_list_by_program_id(
state.database.as_ref(),
request.program_id.as_str(),
request.limit,
)
.await;
let rows = match rows_result {
Ok(rows) => rows,
Err(error) => {
return Err(format!(
"cannot list instruction discriminator summaries for program_id '{}': {}",
request.program_id, error
));
},
};
let summaries_json_result = serde_json::to_string_pretty(&rows);
let summaries_json = match summaries_json_result {
Ok(summaries_json) => summaries_json,
Err(error) => {
return Err(format!("cannot serialize instruction discriminator summaries: {}", error));
},
};
return Ok(DemoPipeline2ProgramInstructionDiscriminatorSummaryPayload { summaries_json });
}
/// Lists protocol candidate summaries ordered by investigation priority.
#[tauri::command]
pub(crate) async fn demo_pipeline2_get_protocol_candidate_summaries(
@@ -745,10 +804,7 @@ pub(crate) async fn demo_pipeline2_get_protocol_candidate_summaries(
let summaries_json = match summaries_json_result {
Ok(summaries_json) => summaries_json,
Err(error) => {
return Err(format!(
"cannot serialize protocol candidate summaries: {}",
error
));
return Err(format!("cannot serialize protocol candidate summaries: {}", error));
},
};
return Ok(DemoPipeline2ProtocolCandidateSummaryPayload { summaries_json });
@@ -783,14 +839,14 @@ pub(crate) async fn demo_pipeline2_diagnose_local_pipeline(
})
}
/// Validates the local pipeline with the strict `0.7.27` non-regression profile.
/// Validates the local pipeline with the `0.7.28` multi-DEX non-regression profile.
#[tauri::command]
pub(crate) async fn demo_pipeline2_validate_local_pipeline(
state: tauri::State<'_, crate::AppState>,
) -> Result<DemoPipeline2LocalValidationPayload, std::string::String> {
let database = state.database.clone();
let service = kb_lib::LocalPipelineValidationService::new(database.clone());
let run_result = service.validate_v0_7_27_current_database().await;
let run_result = service.validate_v0_7_28_current_database().await;
let run = match run_result {
Ok(run) => run,
Err(error) => {

1
kb_demo_app/src/lib.rs
View File

@@ -154,6 +154,7 @@ pub async fn run() -> Result<(), kb_lib::Error> {
crate::demo_pipeline2::demo_pipeline2_diagnose_local_pipeline,
crate::demo_pipeline2::demo_pipeline2_validate_local_pipeline,
crate::demo_pipeline2::demo_pipeline2_get_protocol_candidate_summaries,
crate::demo_pipeline2::demo_pipeline2_get_program_instruction_discriminator_summaries,
]);
tauri_builder = tauri_builder.plugin(tracing_builder.build::<tauri::Wry>());
tauri_builder = tauri_builder.setup(|app| {

2
kb_lib/Cargo.toml
View File

@@ -10,6 +10,8 @@ publish.workspace = true
[dependencies]
chrono.workspace = true
base64.workspace = true
borsh.workspace = true
bs58.workspace = true
futures-util.workspace = true
reqwest.workspace = true

6
kb_lib/src/db.rs
View File

@@ -50,6 +50,8 @@ pub use dtos::PoolDto;
pub use dtos::PoolListingDto;
pub use dtos::PoolOriginDto;
pub use dtos::PoolTokenDto;
pub use dtos::ProgramInstructionDiagnosticDto;
pub use dtos::ProgramInstructionDiscriminatorSummaryDto;
pub use dtos::ProtocolCandidateDto;
pub use dtos::ProtocolCandidateSummaryDto;
pub use dtos::SwapDto;
@@ -85,6 +87,8 @@ pub use entities::PoolEntity;
pub use entities::PoolListingEntity;
pub use entities::PoolOriginEntity;
pub use entities::PoolTokenEntity;
pub use entities::ProgramInstructionDiagnosticEntity;
pub use entities::ProgramInstructionDiscriminatorRowEntity;
pub use entities::ProtocolCandidateEntity;
pub use entities::ProtocolCandidateSummaryEntity;
pub use entities::SwapEntity;
@@ -178,6 +182,8 @@ pub use queries::query_pool_tokens_upsert;
pub use queries::query_pools_get_by_address;
pub use queries::query_pools_list;
pub use queries::query_pools_upsert;
pub use queries::query_program_instruction_diagnostics_list_by_program_id;
pub use queries::query_program_instruction_discriminator_summaries_list_by_program_id;
pub use queries::query_protocol_candidate_summaries_list_by_priority;
pub use queries::query_protocol_candidates_delete_by_transaction_id;
pub use queries::query_protocol_candidates_insert;

4
kb_lib/src/db/dtos.rs
View File

@@ -27,6 +27,7 @@ mod pool;
mod pool_listing;
mod pool_origin;
mod pool_token;
mod program_instruction_diagnostic;
mod protocol_candidate;
mod protocol_candidate_summary;
mod swap;
@@ -37,6 +38,7 @@ mod trade_event;
mod transaction_classification;
mod wallet;
mod wallet_holding;
mod program_instruction_discriminator_summary;
mod wallet_participation;
pub(crate) use local_pipeline_diagnostics::LocalDecodedEventDiagnosticSummaryRow;
@@ -50,6 +52,7 @@ pub(crate) use local_pipeline_diagnostics::LocalPairDiagnosticSummaryRow;
pub(crate) use local_pipeline_diagnostics::LocalPairGapDiagnosticSampleRow;
pub(crate) use local_pipeline_diagnostics::LocalPipelineDiagnosticCountersRow;
pub use program_instruction_discriminator_summary::ProgramInstructionDiscriminatorSummaryDto;
pub use analysis_signal::AnalysisSignalDto;
pub use chain_instruction::ChainInstructionDto;
pub use chain_slot::ChainSlotDto;
@@ -85,6 +88,7 @@ pub use pool::PoolDto;
pub use pool_listing::PoolListingDto;
pub use pool_origin::PoolOriginDto;
pub use pool_token::PoolTokenDto;
pub use program_instruction_diagnostic::ProgramInstructionDiagnosticDto;
pub use protocol_candidate::ProtocolCandidateDto;
pub use protocol_candidate_summary::ProtocolCandidateSummaryDto;
pub use swap::SwapDto;

298
kb_lib/src/db/dtos/program_instruction_diagnostic.rs Normal file
View File

@@ -0,0 +1,298 @@
// file: kb_lib/src/db/dtos/program_instruction_diagnostic.rs
//! Program instruction diagnostic DTO.
/// Diagnostic row for instructions of one Solana program.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ProgramInstructionDiagnosticDto {
/// Parent transaction id.
pub transaction_id: i64,
/// Transaction signature.
pub signature: std::string::String,
/// Optional Solana slot.
pub slot: std::option::Option<u64>,
/// Internal instruction id.
pub instruction_id: i64,
/// Optional parent instruction id.
pub parent_instruction_id: std::option::Option<i64>,
/// Outer instruction index.
pub instruction_index: u32,
/// Optional inner instruction index.
pub inner_instruction_index: std::option::Option<u32>,
/// Program id.
pub program_id: std::option::Option<std::string::String>,
/// Optional program name.
pub program_name: std::option::Option<std::string::String>,
/// Optional stack height.
pub stack_height: std::option::Option<u32>,
/// Number of accounts in `accounts_json`.
pub accounts_count: u64,
/// First account, when present.
pub account_0: std::option::Option<std::string::String>,
/// Second account, when present.
pub account_1: std::option::Option<std::string::String>,
/// Third account, when present.
pub account_2: std::option::Option<std::string::String>,
/// Fourth account, when present.
pub account_3: std::option::Option<std::string::String>,
/// Last account, when present.
pub last_account: std::option::Option<std::string::String>,
/// Optional parsed instruction type.
pub parsed_type: std::option::Option<std::string::String>,
/// True when `data_json` exists.
pub has_data_json: bool,
/// True when `parsed_json` exists.
pub has_parsed_json: bool,
/// Short data JSON preview.
pub data_json_preview: std::option::Option<std::string::String>,
/// Short parsed JSON preview.
pub parsed_json_preview: std::option::Option<std::string::String>,
/// JSON array of useful log hints.
pub log_hints_json: std::string::String,
}
impl TryFrom<crate::ProgramInstructionDiagnosticEntity> for ProgramInstructionDiagnosticDto {
type Error = crate::Error;
fn try_from(entity: crate::ProgramInstructionDiagnosticEntity) -> Result<Self, Self::Error> {
let slot = match entity.slot {
Some(slot) => match u64::try_from(slot) {
Ok(slot) => Some(slot),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert program instruction diagnostic slot '{}' to u64: {}",
slot, error
)));
},
},
None => None,
};
let instruction_index = match u32::try_from(entity.instruction_index) {
Ok(instruction_index) => instruction_index,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert program instruction diagnostic instruction_index '{}' to u32: {}",
entity.instruction_index, error
)));
},
};
let inner_instruction_index = match entity.inner_instruction_index {
Some(inner_instruction_index) => match u32::try_from(inner_instruction_index) {
Ok(inner_instruction_index) => Some(inner_instruction_index),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert program instruction diagnostic inner_instruction_index '{}' to u32: {}",
inner_instruction_index, error
)));
},
},
None => None,
};
let stack_height = match entity.stack_height {
Some(stack_height) => match u32::try_from(stack_height) {
Ok(stack_height) => Some(stack_height),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert program instruction diagnostic stack_height '{}' to u32: {}",
stack_height, error
)));
},
},
None => None,
};
let accounts = parse_accounts_json(entity.accounts_json.as_str());
let accounts_count = match u64::try_from(accounts.len()) {
Ok(accounts_count) => accounts_count,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert accounts count to u64: {}",
error
)));
},
};
let account_0 = account_at(&accounts, 0);
let account_1 = account_at(&accounts, 1);
let account_2 = account_at(&accounts, 2);
let account_3 = account_at(&accounts, 3);
let last_account = match accounts.last() {
Some(last_account) => Some(last_account.clone()),
None => None,
};
let log_hints =
collect_log_hints(entity.meta_json.as_deref(), entity.transaction_json.as_str());
let log_hints_json_result = serde_json::to_string(&log_hints);
let log_hints_json = match log_hints_json_result {
Ok(log_hints_json) => log_hints_json,
Err(error) => {
return Err(crate::Error::Json(format!(
"cannot serialize program instruction log hints: {}",
error
)));
},
};
return Ok(Self {
transaction_id: entity.transaction_id,
signature: entity.signature,
slot,
instruction_id: entity.instruction_id,
parent_instruction_id: entity.parent_instruction_id,
instruction_index,
inner_instruction_index,
program_id: entity.program_id,
program_name: entity.program_name,
stack_height,
accounts_count,
account_0,
account_1,
account_2,
account_3,
last_account,
parsed_type: entity.parsed_type,
has_data_json: entity.data_json.is_some(),
has_parsed_json: entity.parsed_json.is_some(),
data_json_preview: preview_text(entity.data_json.as_deref(), 600),
parsed_json_preview: preview_text(entity.parsed_json.as_deref(), 1200),
log_hints_json,
});
}
}
fn account_at(
accounts: &[std::string::String],
index: usize,
) -> std::option::Option<std::string::String> {
match accounts.get(index) {
Some(account) => return Some(account.clone()),
None => return None,
}
}
fn preview_text(
text: std::option::Option<&str>,
max_len: usize,
) -> std::option::Option<std::string::String> {
let text = match text {
Some(text) => text,
None => return None,
};
if text.len() <= max_len {
return Some(text.to_string());
}
let mut preview = text.chars().take(max_len).collect::<std::string::String>();
preview.push_str("...");
return Some(preview);
}
fn parse_accounts_json(accounts_json: &str) -> std::vec::Vec<std::string::String> {
let parsed_result = serde_json::from_str::<serde_json::Value>(accounts_json);
let parsed = match parsed_result {
Ok(parsed) => parsed,
Err(_) => return std::vec::Vec::new(),
};
let array = match parsed.as_array() {
Some(array) => array,
None => return std::vec::Vec::new(),
};
let mut accounts = std::vec::Vec::new();
for item in array {
if let Some(text) = item.as_str() {
accounts.push(text.to_string());
continue;
}
if let Some(pubkey) = item.get("pubkey").and_then(|value| value.as_str()) {
accounts.push(pubkey.to_string());
}
}
return accounts;
}
fn collect_log_hints(
meta_json: std::option::Option<&str>,
transaction_json: &str,
) -> std::vec::Vec<std::string::String> {
let mut hints = std::vec::Vec::new();
if let Some(meta_json) = meta_json {
collect_log_hints_from_json_text(meta_json, &mut hints);
}
collect_log_hints_from_json_text(transaction_json, &mut hints);
hints.sort();
hints.dedup();
return hints;
}
fn collect_log_hints_from_json_text(
json_text: &str,
hints: &mut std::vec::Vec<std::string::String>,
) {
let value_result = serde_json::from_str::<serde_json::Value>(json_text);
let value = match value_result {
Ok(value) => value,
Err(_) => return,
};
collect_log_hints_from_value(&value, hints);
}
fn collect_log_hints_from_value(
value: &serde_json::Value,
hints: &mut std::vec::Vec<std::string::String>,
) {
match value {
serde_json::Value::String(text) => {
let normalized = text.to_ascii_lowercase();
if normalized.contains("instruction:")
|| normalized.contains("meteora")
|| normalized.contains("dlmm")
|| normalized.contains("lb")
|| normalized.contains("swap")
|| normalized.contains("bin")
{
hints.push(text.clone());
}
},
serde_json::Value::Array(values) => {
for nested in values {
collect_log_hints_from_value(nested, hints);
}
},
serde_json::Value::Object(object) => {
for nested in object.values() {
collect_log_hints_from_value(nested, hints);
}
},
_ => {},
}
}
#[cfg(test)]
mod tests {
#[test]
fn accounts_json_extracts_string_accounts() {
let accounts = super::parse_accounts_json(
serde_json::json!(["A111", "B222", "C333"]).to_string().as_str(),
);
assert_eq!(accounts.len(), 3);
assert_eq!(accounts[0], "A111");
assert_eq!(accounts[1], "B222");
assert_eq!(accounts[2], "C333");
}
#[test]
fn log_hints_are_extracted_from_nested_json() {
let mut hints = std::vec::Vec::new();
super::collect_log_hints_from_json_text(
serde_json::json!({
"meta": {
"logMessages": [
"Program log: Instruction: Swap",
"irrelevant"
]
}
})
.to_string()
.as_str(),
&mut hints,
);
assert_eq!(hints.len(), 1);
assert_eq!(hints[0], "Program log: Instruction: Swap");
}
}

46
kb_lib/src/db/dtos/program_instruction_discriminator_summary.rs Normal file
View File

@@ -0,0 +1,46 @@
// file: kb_lib/src/db/dtos/program_instruction_discriminator_summary.rs
//! Program instruction discriminator summary DTO.
/// Aggregated instruction discriminator diagnostic row.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ProgramInstructionDiscriminatorSummaryDto {
/// Program id.
pub program_id: std::string::String,
/// First eight decoded instruction-data bytes as hex.
pub discriminator_hex: std::option::Option<std::string::String>,
/// Known instruction name, when recognized by local diagnostic mapping.
pub known_instruction_name: std::option::Option<std::string::String>,
/// Number of accounts in the instruction.
pub accounts_count: u64,
/// Optional stack height.
pub stack_height: std::option::Option<u32>,
/// True when the grouped instructions are inner instructions.
pub is_inner_instruction: bool,
/// Number of instruction rows in this group.
pub occurrence_count: u64,
/// Number of distinct transactions in this group.
pub transaction_count: u64,
/// Number of instruction rows already linked to decoded events.
pub decoded_event_count: u64,
/// Number of instruction rows not linked to decoded events.
pub undecoded_occurrence_count: u64,
/// Latest observed slot.
pub latest_slot: std::option::Option<u64>,
/// Latest signature in this group.
pub latest_signature: std::string::String,
/// Latest instruction id in this group.
pub latest_instruction_id: i64,
/// Latest outer instruction index.
pub latest_instruction_index: u32,
/// Latest inner instruction index.
pub latest_inner_instruction_index: std::option::Option<u32>,
/// Latest parsed type.
pub latest_parsed_type: std::option::Option<std::string::String>,
/// Latest decoded event kind.
pub latest_decoded_event_kind: std::option::Option<std::string::String>,
/// Data JSON preview from the latest row.
pub latest_data_json_preview: std::option::Option<std::string::String>,
/// Accounts JSON preview from the latest row.
pub latest_accounts_json_preview: std::option::Option<std::string::String>,
}

4
kb_lib/src/db/entities.rs
View File

@@ -28,6 +28,8 @@ mod pool;
mod pool_listing;
mod pool_origin;
mod pool_token;
mod program_instruction_diagnostic;
mod program_instruction_discriminator_row;
mod protocol_candidate;
mod protocol_candidate_summary;
mod swap;
@@ -64,6 +66,8 @@ pub use pool::PoolEntity;
pub use pool_listing::PoolListingEntity;
pub use pool_origin::PoolOriginEntity;
pub use pool_token::PoolTokenEntity;
pub use program_instruction_diagnostic::ProgramInstructionDiagnosticEntity;
pub use program_instruction_discriminator_row::ProgramInstructionDiscriminatorRowEntity;
pub use protocol_candidate::ProtocolCandidateEntity;
pub use protocol_candidate_summary::ProtocolCandidateSummaryEntity;
pub use swap::SwapEntity;

40
kb_lib/src/db/entities/program_instruction_diagnostic.rs Normal file
View File

@@ -0,0 +1,40 @@
// file: kb_lib/src/db/entities/program_instruction_diagnostic.rs
//! Program instruction diagnostic entity.
/// Raw diagnostic row for instructions of one Solana program.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)]
pub struct ProgramInstructionDiagnosticEntity {
/// Parent transaction id.
pub transaction_id: i64,
/// Transaction signature.
pub signature: std::string::String,
/// Optional Solana slot.
pub slot: std::option::Option<i64>,
/// Internal instruction id.
pub instruction_id: i64,
/// Optional parent instruction id.
pub parent_instruction_id: std::option::Option<i64>,
/// Outer instruction index.
pub instruction_index: i64,
/// Optional inner instruction index.
pub inner_instruction_index: std::option::Option<i64>,
/// Optional program id.
pub program_id: std::option::Option<std::string::String>,
/// Optional program name.
pub program_name: std::option::Option<std::string::String>,
/// Optional stack height.
pub stack_height: std::option::Option<i64>,
/// Serialized accounts JSON.
pub accounts_json: std::string::String,
/// Optional serialized data JSON.
pub data_json: std::option::Option<std::string::String>,
/// Optional parsed instruction type.
pub parsed_type: std::option::Option<std::string::String>,
/// Optional serialized parsed JSON.
pub parsed_json: std::option::Option<std::string::String>,
/// Optional transaction meta JSON.
pub meta_json: std::option::Option<std::string::String>,
/// Full transaction JSON.
pub transaction_json: std::string::String,
}

37
kb_lib/src/db/entities/program_instruction_discriminator_row.rs Normal file
View File

@@ -0,0 +1,37 @@
// file: kb_lib/src/db/entities/program_instruction_discriminator_row.rs
//! Program instruction discriminator diagnostic row entity.
/// Raw row used to summarize instruction discriminators for one Solana program.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)]
pub struct ProgramInstructionDiscriminatorRowEntity {
/// Parent transaction id.
pub transaction_id: i64,
/// Transaction signature.
pub signature: std::string::String,
/// Optional Solana slot.
pub slot: std::option::Option<i64>,
/// Internal instruction id.
pub instruction_id: i64,
/// Optional parent instruction id.
pub parent_instruction_id: std::option::Option<i64>,
/// Outer instruction index.
pub instruction_index: i64,
/// Optional inner instruction index.
pub inner_instruction_index: std::option::Option<i64>,
/// Optional program id.
pub program_id: std::option::Option<std::string::String>,
/// Optional stack height.
pub stack_height: std::option::Option<i64>,
/// Serialized accounts JSON.
pub accounts_json: std::string::String,
/// Optional serialized data JSON.
pub data_json: std::option::Option<std::string::String>,
/// Optional parsed instruction type.
pub parsed_type: std::option::Option<std::string::String>,
/// Optional decoded event id for this instruction.
pub decoded_event_id: std::option::Option<i64>,
/// Optional decoded event kind for this instruction.
pub decoded_event_kind: std::option::Option<std::string::String>,
}

3
kb_lib/src/db/queries.rs
View File

@@ -27,6 +27,7 @@ mod pool;
mod pool_listing;
mod pool_origin;
mod pool_token;
mod program_instruction_diagnostic;
mod protocol_candidate;
mod swap;
mod token;
@@ -120,6 +121,8 @@ pub use pool_origin::query_pool_origins_list;
pub use pool_origin::query_pool_origins_upsert;
pub use pool_token::query_pool_tokens_list_by_pool_id;
pub use pool_token::query_pool_tokens_upsert;
pub use program_instruction_diagnostic::query_program_instruction_diagnostics_list_by_program_id;
pub use program_instruction_diagnostic::query_program_instruction_discriminator_summaries_list_by_program_id;
pub use protocol_candidate::query_protocol_candidate_summaries_list_by_priority;
pub use protocol_candidate::query_protocol_candidates_delete_by_transaction_id;
pub use protocol_candidate::query_protocol_candidates_insert;

451
kb_lib/src/db/queries/program_instruction_diagnostic.rs Normal file
View File

@@ -0,0 +1,451 @@
// file: kb_lib/src/db/queries/program_instruction_diagnostic.rs
//! Queries for program instruction diagnostics.
/// Lists diagnostic instruction rows for one program id.
pub async fn query_program_instruction_diagnostics_list_by_program_id(
database: &crate::Database,
program_id: &str,
limit: u32,
) -> Result<std::vec::Vec<crate::ProgramInstructionDiagnosticDto>, crate::Error> {
if limit == 0 {
return Ok(std::vec::Vec::new());
}
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result =
sqlx::query_as::<sqlx::Sqlite, crate::ProgramInstructionDiagnosticEntity>(
r#"
SELECT
tx.id AS transaction_id,
tx.signature AS signature,
tx.slot AS slot,
ins.id AS instruction_id,
ins.parent_instruction_id AS parent_instruction_id,
ins.instruction_index AS instruction_index,
ins.inner_instruction_index AS inner_instruction_index,
ins.program_id AS program_id,
ins.program_name AS program_name,
ins.stack_height AS stack_height,
ins.accounts_json AS accounts_json,
ins.data_json AS data_json,
ins.parsed_type AS parsed_type,
ins.parsed_json AS parsed_json,
tx.meta_json AS meta_json,
tx.transaction_json AS transaction_json
FROM k_sol_chain_instructions ins
JOIN k_sol_chain_transactions tx
ON tx.id = ins.transaction_id
WHERE ins.program_id = ?
ORDER BY
tx.slot DESC,
tx.id DESC,
ins.instruction_index ASC,
ins.inner_instruction_index ASC,
ins.id ASC
LIMIT ?
"#,
)
.bind(program_id.to_string())
.bind(i64::from(limit))
.fetch_all(pool)
.await;
let entities = match query_result {
Ok(entities) => entities,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot list program instruction diagnostics for program_id '{}' on sqlite: {}",
program_id, error
)));
},
};
let mut dtos = std::vec::Vec::new();
for entity in entities {
let dto_result = crate::ProgramInstructionDiagnosticDto::try_from(entity);
let dto = match dto_result {
Ok(dto) => dto,
Err(error) => return Err(error),
};
dtos.push(dto);
}
return Ok(dtos);
},
}
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct ProgramInstructionDiscriminatorSummaryKey {
program_id: std::string::String,
discriminator_hex: std::option::Option<std::string::String>,
accounts_count: u64,
stack_height: std::option::Option<u32>,
is_inner_instruction: bool,
}
#[derive(Debug, Clone)]
struct ProgramInstructionDiscriminatorSummaryAccumulator {
key: ProgramInstructionDiscriminatorSummaryKey,
known_instruction_name: std::option::Option<std::string::String>,
occurrence_count: u64,
decoded_event_count: u64,
transaction_signatures: std::collections::BTreeSet<std::string::String>,
latest_slot: std::option::Option<u64>,
latest_signature: std::string::String,
latest_instruction_id: i64,
latest_instruction_index: u32,
latest_inner_instruction_index: std::option::Option<u32>,
latest_parsed_type: std::option::Option<std::string::String>,
latest_decoded_event_kind: std::option::Option<std::string::String>,
latest_data_json_preview: std::option::Option<std::string::String>,
latest_accounts_json_preview: std::option::Option<std::string::String>,
}
fn build_program_instruction_discriminator_summaries(
rows: std::vec::Vec<crate::ProgramInstructionDiscriminatorRowEntity>,
) -> Result<std::vec::Vec<crate::ProgramInstructionDiscriminatorSummaryDto>, crate::Error> {
let mut grouped = std::collections::BTreeMap::<
ProgramInstructionDiscriminatorSummaryKey,
ProgramInstructionDiscriminatorSummaryAccumulator,
>::new();
for row in rows {
let summary_row_result = build_summary_row_from_discriminator_entity(row);
let summary_row = match summary_row_result {
Ok(summary_row) => summary_row,
Err(error) => return Err(error),
};
let existing = grouped.get_mut(&summary_row.key);
match existing {
Some(existing) => {
existing.occurrence_count += 1;
if summary_row.decoded_event_count > 0 {
existing.decoded_event_count += 1;
}
existing.transaction_signatures.insert(summary_row.latest_signature.clone());
if summary_row.latest_instruction_id > existing.latest_instruction_id {
existing.latest_slot = summary_row.latest_slot;
existing.latest_signature = summary_row.latest_signature;
existing.latest_instruction_id = summary_row.latest_instruction_id;
existing.latest_instruction_index = summary_row.latest_instruction_index;
existing.latest_inner_instruction_index =
summary_row.latest_inner_instruction_index;
existing.latest_parsed_type = summary_row.latest_parsed_type;
existing.latest_decoded_event_kind = summary_row.latest_decoded_event_kind;
existing.latest_data_json_preview = summary_row.latest_data_json_preview;
existing.latest_accounts_json_preview =
summary_row.latest_accounts_json_preview;
}
},
None => {
grouped.insert(summary_row.key.clone(), summary_row);
},
}
}
let mut summaries = std::vec::Vec::new();
for (_, accumulator) in grouped {
let transaction_count_result = u64::try_from(accumulator.transaction_signatures.len());
let transaction_count = match transaction_count_result {
Ok(transaction_count) => transaction_count,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert discriminator summary transaction_count to u64: {}",
error
)));
},
};
let undecoded_occurrence_count =
accumulator.occurrence_count.saturating_sub(accumulator.decoded_event_count);
summaries.push(crate::ProgramInstructionDiscriminatorSummaryDto {
program_id: accumulator.key.program_id,
discriminator_hex: accumulator.key.discriminator_hex,
known_instruction_name: accumulator.known_instruction_name,
accounts_count: accumulator.key.accounts_count,
stack_height: accumulator.key.stack_height,
is_inner_instruction: accumulator.key.is_inner_instruction,
occurrence_count: accumulator.occurrence_count,
transaction_count,
decoded_event_count: accumulator.decoded_event_count,
undecoded_occurrence_count,
latest_slot: accumulator.latest_slot,
latest_signature: accumulator.latest_signature,
latest_instruction_id: accumulator.latest_instruction_id,
latest_instruction_index: accumulator.latest_instruction_index,
latest_inner_instruction_index: accumulator.latest_inner_instruction_index,
latest_parsed_type: accumulator.latest_parsed_type,
latest_decoded_event_kind: accumulator.latest_decoded_event_kind,
latest_data_json_preview: accumulator.latest_data_json_preview,
latest_accounts_json_preview: accumulator.latest_accounts_json_preview,
});
}
summaries.sort_by(|left, right| {
right
.undecoded_occurrence_count
.cmp(&left.undecoded_occurrence_count)
.then(right.transaction_count.cmp(&left.transaction_count))
.then(right.occurrence_count.cmp(&left.occurrence_count))
.then(right.latest_instruction_id.cmp(&left.latest_instruction_id))
});
return Ok(summaries);
}
fn build_summary_row_from_discriminator_entity(
row: crate::ProgramInstructionDiscriminatorRowEntity,
) -> Result<ProgramInstructionDiscriminatorSummaryAccumulator, crate::Error> {
let program_id = match row.program_id.clone() {
Some(program_id) => program_id,
None => "unknown".to_string(),
};
let accounts_count = accounts_count_from_json(row.accounts_json.as_str());
let stack_height = match row.stack_height {
Some(stack_height) => match u32::try_from(stack_height) {
Ok(stack_height) => Some(stack_height),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert discriminator summary stack_height '{}' to u32: {}",
stack_height, error
)));
},
},
None => None,
};
let instruction_index = match u32::try_from(row.instruction_index) {
Ok(instruction_index) => instruction_index,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert discriminator summary instruction_index '{}' to u32: {}",
row.instruction_index, error
)));
},
};
let inner_instruction_index = match row.inner_instruction_index {
Some(inner_instruction_index) => match u32::try_from(inner_instruction_index) {
Ok(inner_instruction_index) => Some(inner_instruction_index),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert discriminator summary inner_instruction_index '{}' to u32: {}",
inner_instruction_index, error
)));
},
},
None => None,
};
let latest_slot = match row.slot {
Some(slot) => match u64::try_from(slot) {
Ok(slot) => Some(slot),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert discriminator summary slot '{}' to u64: {}",
slot, error
)));
},
},
None => None,
};
let decoded_data = decode_instruction_data_json(row.data_json.as_ref());
let discriminator_hex = match decoded_data {
Some(decoded_data) => first_8_bytes_hex(decoded_data.as_slice()),
None => None,
};
let known_instruction_name = known_instruction_name_for_program_discriminator(
program_id.as_str(),
discriminator_hex.as_deref(),
);
let mut transaction_signatures = std::collections::BTreeSet::new();
transaction_signatures.insert(row.signature.clone());
let decoded_event_count = if row.decoded_event_id.is_some() { 1_u64 } else { 0_u64 };
let key = ProgramInstructionDiscriminatorSummaryKey {
program_id,
discriminator_hex,
accounts_count,
stack_height,
is_inner_instruction: row.parent_instruction_id.is_some(),
};
return Ok(ProgramInstructionDiscriminatorSummaryAccumulator {
key,
known_instruction_name,
occurrence_count: 1,
decoded_event_count,
transaction_signatures,
latest_slot,
latest_signature: row.signature,
latest_instruction_id: row.instruction_id,
latest_instruction_index: instruction_index,
latest_inner_instruction_index: inner_instruction_index,
latest_parsed_type: row.parsed_type,
latest_decoded_event_kind: row.decoded_event_kind,
latest_data_json_preview: preview_text(row.data_json.as_deref(), 300),
latest_accounts_json_preview: preview_text(Some(row.accounts_json.as_str()), 600),
});
}
fn accounts_count_from_json(accounts_json: &str) -> u64 {
let parsed_result = serde_json::from_str::<serde_json::Value>(accounts_json);
let parsed = match parsed_result {
Ok(parsed) => parsed,
Err(_) => return 0,
};
let array = match parsed.as_array() {
Some(array) => array,
None => return 0,
};
let count_result = u64::try_from(array.len());
match count_result {
Ok(count) => return count,
Err(_) => return 0,
}
}
fn decode_instruction_data_json(
data_json: std::option::Option<&std::string::String>,
) -> std::option::Option<std::vec::Vec<u8>> {
let data_json = match data_json {
Some(data_json) => data_json,
None => return None,
};
let parsed_result = serde_json::from_str::<serde_json::Value>(data_json.as_str());
let parsed = match parsed_result {
Ok(parsed) => parsed,
Err(_) => return None,
};
if let serde_json::Value::String(base58_text) = parsed {
let decoded_result = decode_base58(base58_text.as_str());
match decoded_result {
Ok(decoded) => return Some(decoded),
Err(_) => return None,
}
}
return None;
}
fn decode_base58(input: &str) -> Result<std::vec::Vec<u8>, crate::Error> {
let decoded_result = bs58::decode(input).into_vec();
match decoded_result {
Ok(decoded) => return Ok(decoded),
Err(error) => {
return Err(crate::Error::Json(format!(
"cannot decode instruction data from base58: {}",
error
)));
},
}
}
fn first_8_bytes_hex(bytes: &[u8]) -> std::option::Option<std::string::String> {
if bytes.len() < 8 {
return None;
}
return Some(format!(
"{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
));
}
fn known_instruction_name_for_program_discriminator(
program_id: &str,
discriminator_hex: std::option::Option<&str>,
) -> std::option::Option<std::string::String> {
if program_id != crate::METEORA_DLMM_PROGRAM_ID {
return None;
}
let discriminator_hex = match discriminator_hex {
Some(discriminator_hex) => discriminator_hex,
None => return None,
};
let name = match discriminator_hex {
"2d9aedd2dd0fa65c" => "initialize_lb_pair",
"493b2478ed536cc6" => "initialize_lb_pair2",
"2e2729876fb7c840" => "initialize_customizable_permissionless_lb_pair",
"f349817e3313f16b" => "initialize_customizable_permissionless_lb_pair2",
"f8c69e91e17587c8" => "swap",
"414b3f4ceb5b5b88" => "swap2",
"fa49652126cf4bb8" => "swap_exact_out",
"2bd7f784893cf351" => "swap_exact_out2",
"38ade6d0ade49ccd" => "swap_with_price_impact",
"235613b94ed44bd3" => "initialize_bin_array",
"dbc0ea47bebf6650" => "initialize_position",
"b59d59438fb63448" => "add_liquidity",
"5055d14818ceb16c" => "remove_liquidity",
"e445a52e51cb9a1d" => "anchor_self_cpi_log_event",
"70bf65ab1c907fbb" => "claim_fee2",
_ => return None,
};
return Some(name.to_string());
}
fn preview_text(
text: std::option::Option<&str>,
max_len: usize,
) -> std::option::Option<std::string::String> {
let text = match text {
Some(text) => text,
None => return None,
};
if text.len() <= max_len {
return Some(text.to_string());
}
let mut preview = text.chars().take(max_len).collect::<std::string::String>();
preview.push_str("...");
return Some(preview);
}
/// Lists instruction discriminator summaries for one program id.
pub async fn query_program_instruction_discriminator_summaries_list_by_program_id(
database: &crate::Database,
program_id: &str,
limit: u32,
) -> Result<std::vec::Vec<crate::ProgramInstructionDiscriminatorSummaryDto>, crate::Error> {
if limit == 0 {
return Ok(std::vec::Vec::new());
}
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result =
sqlx::query_as::<sqlx::Sqlite, crate::ProgramInstructionDiscriminatorRowEntity>(
r#"
SELECT
tx.id AS transaction_id,
tx.signature AS signature,
tx.slot AS slot,
ins.id AS instruction_id,
ins.parent_instruction_id AS parent_instruction_id,
ins.instruction_index AS instruction_index,
ins.inner_instruction_index AS inner_instruction_index,
ins.program_id AS program_id,
ins.stack_height AS stack_height,
ins.accounts_json AS accounts_json,
ins.data_json AS data_json,
ins.parsed_type AS parsed_type,
de.id AS decoded_event_id,
de.event_kind AS decoded_event_kind
FROM k_sol_chain_instructions ins
JOIN k_sol_chain_transactions tx
ON tx.id = ins.transaction_id
LEFT JOIN k_sol_dex_decoded_events de
ON de.transaction_id = tx.id
AND de.instruction_id = ins.id
WHERE ins.program_id = ?
ORDER BY
tx.slot DESC,
tx.id DESC,
ins.instruction_index ASC,
ins.inner_instruction_index ASC,
ins.id ASC
LIMIT ?
"#,
)
.bind(program_id.to_string())
.bind(i64::from(limit))
.fetch_all(pool)
.await;
let rows = match query_result {
Ok(rows) => rows,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot list program instruction discriminator diagnostics for program_id '{}' on sqlite: {}",
program_id, error
)));
},
};
return build_program_instruction_discriminator_summaries(rows);
},
}
}

5
kb_lib/src/dex.rs
View File

@@ -7,6 +7,7 @@ mod fluxbeam;
mod meteora_damm_v1;
mod meteora_damm_v2;
mod meteora_dbc;
mod meteora_dlmm;
mod orca_whirlpools;
mod pump_fun;
mod pump_swap;
@@ -34,6 +35,10 @@ pub use meteora_dbc::MeteoraDbcCreatePoolDecoded;
pub use meteora_dbc::MeteoraDbcDecodedEvent;
pub use meteora_dbc::MeteoraDbcDecoder;
pub use meteora_dbc::MeteoraDbcSwapDecoded;
pub use meteora_dlmm::MeteoraDlmmCreatePoolDecoded;
pub use meteora_dlmm::MeteoraDlmmDecodedEvent;
pub use meteora_dlmm::MeteoraDlmmDecoder;
pub use meteora_dlmm::MeteoraDlmmSwapDecoded;
pub use orca_whirlpools::OrcaWhirlpoolsCreatePoolDecoded;
pub use orca_whirlpools::OrcaWhirlpoolsDecodedEvent;
pub use orca_whirlpools::OrcaWhirlpoolsDecoder;

1318
kb_lib/src/dex/meteora_dlmm.rs Normal file
View File

File diff suppressed because it is too large Load Diff

48
kb_lib/src/dex/raydium_clmm.rs
View File

@@ -301,52 +301,10 @@ fn read_bool(data: &[u8], offset: usize) -> std::option::Option<bool> {
}
fn decode_base58(input: &str) -> std::option::Option<std::vec::Vec<u8>> {
let alphabet = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz".as_bytes();
let mut bytes: std::vec::Vec<u8> = std::vec::Vec::new();
for input_byte in input.bytes() {
let mut value_option = None;
let mut alphabet_index = 0_usize;
while alphabet_index < alphabet.len() {
if alphabet[alphabet_index] == input_byte {
value_option = Some(alphabet_index as u32);
break;
}
alphabet_index += 1;
}
let mut carry = match value_option {
Some(value) => value,
None => return None,
};
let mut byte_index = bytes.len();
while byte_index > 0 {
byte_index -= 1;
let value = (bytes[byte_index] as u32) * 58 + carry;
bytes[byte_index] = (value & 0xff) as u8;
carry = value >> 8;
}
while carry > 0 {
bytes.insert(0, (carry & 0xff) as u8);
carry >>= 8;
}
match bs58::decode(input).into_vec() {
Ok(decoded) => return Some(decoded),
Err(_) => return None,
}
let mut leading_zero_count = 0_usize;
for input_byte in input.bytes() {
if input_byte == b'1' {
leading_zero_count += 1;
} else {
break;
}
}
let mut result = std::vec::Vec::new();
let mut index = 0_usize;
while index < leading_zero_count {
result.push(0_u8);
index += 1;
}
for byte in bytes {
result.push(byte);
}
return Some(result);
}
#[cfg(test)]

10
kb_lib/src/dex_catalog.rs
View File

@@ -81,6 +81,15 @@ pub(crate) fn dex_catalog_item_by_code(
is_enabled: true,
});
},
"meteora_dlmm" => {
return Some(crate::dex_catalog::DexCatalogItem {
code: "meteora_dlmm",
name: "Meteora DLMM",
program_id: Some(crate::METEORA_DLMM_PROGRAM_ID),
router_program_id: None,
is_enabled: true,
});
},
"meteora_damm_v1" => {
return Some(crate::dex_catalog::DexCatalogItem {
code: "meteora_damm_v1",
@@ -250,6 +259,7 @@ mod tests {
"pump_fun",
"pump_swap",
"meteora_dbc",
"meteora_dlmm",
"meteora_damm_v1",
"meteora_damm_v2",
"orca_whirlpools",

274
kb_lib/src/dex_decode.rs
View File

@@ -12,6 +12,7 @@ pub struct DexDecodeService {
pump_swap_decoder: crate::PumpSwapDecoder,
orca_whirlpools_decoder: crate::OrcaWhirlpoolsDecoder,
meteora_dbc_decoder: crate::MeteoraDbcDecoder,
meteora_dlmm_decoder: crate::MeteoraDlmmDecoder,
meteora_damm_v1_decoder: crate::MeteoraDammV1Decoder,
meteora_damm_v2_decoder: crate::MeteoraDammV2Decoder,
fluxbeam_decoder: crate::FluxbeamDecoder,
@@ -30,6 +31,7 @@ impl DexDecodeService {
pump_swap_decoder: crate::PumpSwapDecoder::new(),
orca_whirlpools_decoder: crate::OrcaWhirlpoolsDecoder::new(),
meteora_dbc_decoder: crate::MeteoraDbcDecoder::new(),
meteora_dlmm_decoder: crate::MeteoraDlmmDecoder::new(),
meteora_damm_v1_decoder: crate::MeteoraDammV1Decoder::new(),
meteora_damm_v2_decoder: crate::MeteoraDammV2Decoder::new(),
fluxbeam_decoder: crate::FluxbeamDecoder::new(),
@@ -98,8 +100,7 @@ impl DexDecodeService {
}
let append_result = append_persisted_events_result(
&mut persisted,
self.decode_and_persist_meteora_damm_v2_events(&transaction, &instructions)
.await,
self.decode_and_persist_meteora_dlmm_events(&transaction, &instructions).await,
);
if let Err(error) = append_result {
return Err(error);
@@ -112,6 +113,14 @@ impl DexDecodeService {
if let Err(error) = append_result {
return Err(error);
}
let append_result = append_persisted_events_result(
&mut persisted,
self.decode_and_persist_meteora_damm_v2_events(&transaction, &instructions)
.await,
);
if let Err(error) = append_result {
return Err(error);
}
let append_result = append_persisted_events_result(
&mut persisted,
self.decode_and_persist_orca_whirlpools_events(&transaction, &instructions)
@@ -311,6 +320,96 @@ impl DexDecodeService {
}
}
async fn persist_meteora_dbc_event(
&self,
transaction: &crate::ChainTransactionDto,
decoded_event: &crate::MeteoraDbcDecodedEvent,
) -> Result<crate::DexDecodedEventDto, crate::Error> {
match decoded_event {
crate::MeteoraDbcDecodedEvent::CreatePool(event) => {
return self
.materialize_named_dex_event(
transaction,
event.transaction_id,
event.instruction_id,
"meteora_dbc",
event.program_id.clone(),
"meteora_dbc.create_pool",
event.pool_account.clone(),
None,
event.token_a_mint.clone(),
event.token_b_mint.clone(),
event.config_account.clone(),
event.payload_json.clone(),
)
.await;
},
crate::MeteoraDbcDecodedEvent::Swap(event) => {
return self
.materialize_named_dex_event(
transaction,
event.transaction_id,
event.instruction_id,
"meteora_dbc",
event.program_id.clone(),
"meteora_dbc.swap",
event.pool_account.clone(),
None,
event.token_a_mint.clone(),
event.token_b_mint.clone(),
None,
event.payload_json.clone(),
)
.await;
},
}
}
async fn persist_meteora_dlmm_event(
&self,
transaction: &crate::ChainTransactionDto,
decoded_event: &crate::MeteoraDlmmDecodedEvent,
) -> Result<crate::DexDecodedEventDto, crate::Error> {
match decoded_event {
crate::MeteoraDlmmDecodedEvent::CreatePool(event) => {
return self
.materialize_named_dex_event(
transaction,
event.transaction_id,
event.instruction_id,
"meteora_dlmm",
event.program_id.clone(),
"meteora_dlmm.create_pool",
event.pool_account.clone(),
None,
event.token_a_mint.clone(),
event.token_b_mint.clone(),
event.config_account.clone(),
event.payload_json.clone(),
)
.await;
},
crate::MeteoraDlmmDecodedEvent::Swap(event) => {
return self
.materialize_named_dex_event(
transaction,
event.transaction_id,
event.instruction_id,
"meteora_dlmm",
event.program_id.clone(),
"meteora_dlmm.swap",
event.pool_account.clone(),
None,
event.token_a_mint.clone(),
event.token_b_mint.clone(),
None,
event.payload_json.clone(),
)
.await;
},
}
}
async fn persist_meteora_damm_v1_event(
&self,
transaction: &crate::ChainTransactionDto,
@@ -336,6 +435,8 @@ impl DexDecodeService {
.await;
},
crate::MeteoraDammV1DecodedEvent::Swap(event) => {
let enrichment_payload_json =
prepare_meteora_damm_v1_swap_payload_for_classification(event);
return self
.materialize_named_dex_event(
transaction,
@@ -349,7 +450,7 @@ impl DexDecodeService {
event.token_a_mint.clone(),
event.token_b_mint.clone(),
None,
event.payload_json.clone(),
enrichment_payload_json,
)
.await;
},
@@ -401,51 +502,6 @@ impl DexDecodeService {
}
}
async fn persist_meteora_dbc_event(
&self,
transaction: &crate::ChainTransactionDto,
decoded_event: &crate::MeteoraDbcDecodedEvent,
) -> Result<crate::DexDecodedEventDto, crate::Error> {
match decoded_event {
crate::MeteoraDbcDecodedEvent::CreatePool(event) => {
return self
.materialize_named_dex_event(
transaction,
event.transaction_id,
event.instruction_id,
"meteora_dbc",
event.program_id.clone(),
"meteora_dbc.create_pool",
event.pool_account.clone(),
None,
event.token_a_mint.clone(),
event.token_b_mint.clone(),
event.config_account.clone(),
event.payload_json.clone(),
)
.await;
},
crate::MeteoraDbcDecodedEvent::Swap(event) => {
return self
.materialize_named_dex_event(
transaction,
event.transaction_id,
event.instruction_id,
"meteora_dbc",
event.program_id.clone(),
"meteora_dbc.swap",
event.pool_account.clone(),
None,
event.token_a_mint.clone(),
event.token_b_mint.clone(),
None,
event.payload_json.clone(),
)
.await;
},
}
}
async fn persist_raydium_amm_v4_event(
&self,
transaction: &crate::ChainTransactionDto,
@@ -900,6 +956,29 @@ impl DexDecodeService {
return Ok(persisted);
}
async fn decode_and_persist_meteora_dlmm_events(
&self,
transaction: &crate::ChainTransactionDto,
instructions: &[crate::ChainInstructionDto],
) -> Result<std::vec::Vec<crate::DexDecodedEventDto>, crate::Error> {
let decoded_result =
self.meteora_dlmm_decoder.decode_transaction(transaction, instructions);
let decoded_events = match decoded_result {
Ok(decoded_events) => decoded_events,
Err(error) => return Err(error),
};
let mut persisted = std::vec::Vec::new();
for decoded_event in &decoded_events {
let persist_result = self.persist_meteora_dlmm_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);
}
return Ok(persisted);
}
async fn decode_and_persist_meteora_damm_v1_events(
&self,
transaction: &crate::ChainTransactionDto,
@@ -1056,6 +1135,38 @@ fn enriched_raydium_payload_value(
return Ok(crate::enrich_dex_decoded_payload(protocol_name, event_kind, payload_value));
}
// Marks Meteora DAMM v1 swaps without direct amounts as non-materializable candidates
// before generic classification metadata is inserted.
fn prepare_meteora_damm_v1_swap_payload_for_classification(
event: &crate::MeteoraDammV1SwapDecoded,
) -> serde_json::Value {
let mut object = match event.payload_json.clone() {
serde_json::Value::Object(object) => object,
other => {
let mut object = serde_json::Map::new();
object.insert("rawPayload".to_string(), other);
object
},
};
let payload_json = serde_json::Value::Object(object.clone());
if crate::dex_event_classification::decoded_payload_has_trade_amount_or_price_payload(
&payload_json,
) {
return serde_json::Value::Object(object);
}
object.insert("tradeCandidate".to_string(), serde_json::Value::Bool(false));
object.insert("candleCandidate".to_string(), serde_json::Value::Bool(false));
object.insert(
"skipTradeReason".to_string(),
serde_json::Value::String("meteora_damm_v1_swap_without_amount_payload".to_string()),
);
object.insert(
"skipCandleReason".to_string(),
serde_json::Value::String("meteora_damm_v1_swap_without_amount_payload".to_string()),
);
return serde_json::Value::Object(object);
}
// Marks incomplete PumpSwap decoded trades as non-materializable candidates before generic
// classification metadata is inserted.
fn prepare_pump_swap_trade_payload_for_classification(
@@ -1313,6 +1424,73 @@ mod tests {
assert_eq!(decoded[0].token_a_mint, Some("MintPF111".to_string()));
}
#[test]
fn prepare_meteora_damm_v1_swap_without_amounts_marks_event_non_actionable() {
let event = crate::MeteoraDammV1SwapDecoded {
transaction_id: 1,
instruction_id: 2,
signature: "sig-damm-v1-no-amounts".to_string(),
program_id: crate::METEORA_DAMM_V1_PROGRAM_ID.to_string(),
trade_side: crate::SwapTradeSide::Unknown,
pool_account: Some("PoolDammV1".to_string()),
token_a_mint: Some("TokenA".to_string()),
token_b_mint: Some("TokenB".to_string()),
payload_json: serde_json::json!({
"decoder": "meteora_damm_v1",
"eventKind": "swap"
}),
};
let prepared_payload =
super::prepare_meteora_damm_v1_swap_payload_for_classification(&event);
let object_option = prepared_payload.as_object();
let object = match object_option {
Some(object) => object,
None => {
panic!("expected prepared payload object");
},
};
assert_eq!(object.get("tradeCandidate"), Some(&serde_json::Value::Bool(false)));
assert_eq!(object.get("candleCandidate"), Some(&serde_json::Value::Bool(false)));
assert_eq!(
object.get("skipTradeReason"),
Some(&serde_json::Value::String(
"meteora_damm_v1_swap_without_amount_payload".to_string()
))
);
}
#[test]
fn prepare_meteora_damm_v1_swap_with_amounts_keeps_event_actionable() {
let event = crate::MeteoraDammV1SwapDecoded {
transaction_id: 1,
instruction_id: 2,
signature: "sig-damm-v1-with-amounts".to_string(),
program_id: crate::METEORA_DAMM_V1_PROGRAM_ID.to_string(),
trade_side: crate::SwapTradeSide::Unknown,
pool_account: Some("PoolDammV1".to_string()),
token_a_mint: Some("TokenA".to_string()),
token_b_mint: Some("TokenB".to_string()),
payload_json: serde_json::json!({
"decoder": "meteora_damm_v1",
"eventKind": "swap",
"baseAmountRaw": "100",
"quoteAmountRaw": "250"
}),
};
let prepared_payload =
super::prepare_meteora_damm_v1_swap_payload_for_classification(&event);
let object_option = prepared_payload.as_object();
let object = match object_option {
Some(object) => object,
None => {
panic!("expected prepared payload object");
},
};
assert_eq!(object.get("tradeCandidate"), None);
assert_eq!(object.get("candleCandidate"), None);
assert_eq!(object.get("skipTradeReason"), None);
}
#[tokio::test]
async fn decode_transaction_by_signature_persists_decoded_pump_swap_event() {
let database = make_database().await;

36
kb_lib/src/dex_decoded_event_materialization.rs
View File

@@ -50,10 +50,15 @@ pub(crate) struct DexDecodedEventMaterializationInput<'a> {
pub(crate) async fn materialize_dex_decoded_event(
input: crate::dex_decoded_event_materialization::DexDecodedEventMaterializationInput<'_>,
) -> Result<crate::DexDecodedEventDto, crate::Error> {
let enrichment_payload_json =
crate::dex_decoded_event_materialization::prepare_payload_for_transaction_status(
input.transaction,
input.enrichment_payload_json,
);
let payload_json_result = crate::enrich_and_serialize_dex_decoded_payload(
input.protocol_name.as_str(),
input.event_kind.as_str(),
input.enrichment_payload_json,
enrichment_payload_json,
);
let payload_json = match payload_json_result {
Ok(payload_json) => payload_json,
@@ -138,3 +143,32 @@ pub(crate) async fn materialize_dex_decoded_event(
}
return Ok(fetched);
}
fn prepare_payload_for_transaction_status(
transaction: &crate::ChainTransactionDto,
payload_json: serde_json::Value,
) -> serde_json::Value {
if transaction.err_json.is_none() {
return payload_json;
}
let mut object = match payload_json {
serde_json::Value::Object(object) => object,
other => {
let mut object = serde_json::Map::new();
object.insert("rawPayload".to_string(), other);
object
},
};
object.insert("transactionFailed".to_string(), serde_json::Value::Bool(true));
object.insert("tradeCandidate".to_string(), serde_json::Value::Bool(false));
object.insert("candleCandidate".to_string(), serde_json::Value::Bool(false));
object.insert(
"skipTradeReason".to_string(),
serde_json::Value::String("failed_transaction".to_string()),
);
object.insert(
"skipCandleReason".to_string(),
serde_json::Value::String("failed_transaction".to_string()),
);
return serde_json::Value::Object(object);
}

75
kb_lib/src/dex_detect.rs
View File

@@ -115,12 +115,15 @@ impl DexDetectService {
crate::dex_detection_route::DexDetectionRoute::MeteoraDbcPool => {
self.detect_meteora_dbc_pool(&transaction, decoded_event).await
},
crate::dex_detection_route::DexDetectionRoute::MeteoraDammV2Pool => {
self.detect_meteora_damm_v2_pool(&transaction, decoded_event).await
crate::dex_detection_route::DexDetectionRoute::MeteoraDlmmPool => {
self.detect_meteora_dlmm_pool(&transaction, decoded_event).await
},
crate::dex_detection_route::DexDetectionRoute::MeteoraDammV1Pool => {
self.detect_meteora_damm_v1_pool(&transaction, decoded_event).await
},
crate::dex_detection_route::DexDetectionRoute::MeteoraDammV2Pool => {
self.detect_meteora_damm_v2_pool(&transaction, decoded_event).await
},
crate::dex_detection_route::DexDetectionRoute::OrcaWhirlpoolsPool => {
self.detect_orca_whirlpools_pool(&transaction, decoded_event).await
},
@@ -380,21 +383,58 @@ impl DexDetectService {
.await;
}
async fn detect_meteora_damm_v2_pool(
async fn detect_meteora_dlmm_pool(
&self,
transaction: &crate::ChainTransactionDto,
decoded_event: &crate::DexDecodedEventDto,
) -> Result<crate::DexPoolDetectionResult, crate::Error> {
return self
.detect_materialized_pool_from_decoded_event(
transaction,
let dex_id_result =
crate::dex_catalog::ensure_known_dex(self.database.as_ref(), "meteora_dlmm").await;
let dex_id = match dex_id_result {
Ok(dex_id) => dex_id,
Err(error) => return Err(error),
};
let payload_value_result = 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),
};
let reserve_x_account = extract_payload_string_field(&payload_value, "reserveXAccount");
let reserve_y_account = extract_payload_string_field(&payload_value, "reserveYAccount");
let input_result =
crate::dex_pool_materialization::DexPoolMaterializationInput::from_decoded_event(
decoded_event,
"meteora_damm_v2",
crate::PoolKind::Amm,
dex_id,
crate::PoolKind::Clmm,
crate::PoolStatus::Active,
"signal.dex.meteora_damm_v2",
crate::dex_pool_materialization::DexPoolTokenOrder::ChooseBaseQuoteFromTokenAB,
reserve_x_account,
reserve_y_account,
transaction.source_endpoint_name.clone(),
);
let input = match input_result {
Ok(input) => input,
Err(error) => return Err(error),
};
let detection_result =
crate::dex_pool_materialization::materialize_dex_pool(self.database.as_ref(), &input)
.await;
let detection_result = match detection_result {
Ok(detection_result) => detection_result,
Err(error) => return Err(error),
};
let signal_result = self
.record_pool_detection_signals(
transaction,
"signal.dex.meteora_dlmm",
&detection_result,
payload_value,
)
.await;
if let Err(error) = signal_result {
return Err(error);
}
return Ok(detection_result);
}
async fn detect_meteora_damm_v1_pool(
@@ -414,6 +454,23 @@ impl DexDetectService {
.await;
}
async fn detect_meteora_damm_v2_pool(
&self,
transaction: &crate::ChainTransactionDto,
decoded_event: &crate::DexDecodedEventDto,
) -> Result<crate::DexPoolDetectionResult, crate::Error> {
return self
.detect_materialized_pool_from_decoded_event(
transaction,
decoded_event,
"meteora_damm_v2",
crate::PoolKind::Amm,
crate::PoolStatus::Active,
"signal.dex.meteora_damm_v2",
)
.await;
}
async fn detect_orca_whirlpools_pool(
&self,
transaction: &crate::ChainTransactionDto,

20
kb_lib/src/dex_detection_route.rs
View File

@@ -21,10 +21,12 @@ pub(crate) enum DexDetectionRoute {
SkipIncompletePumpSwapTrade,
/// Meteora DBC pool route.
MeteoraDbcPool,
/// Meteora DAMM v2 pool route.
MeteoraDammV2Pool,
/// Meteora DLMM pool route.
MeteoraDlmmPool,
/// Meteora DAMM v1 pool route.
MeteoraDammV1Pool,
/// Meteora DAMM v2 pool route.
MeteoraDammV2Pool,
/// Orca Whirlpools pool route.
OrcaWhirlpoolsPool,
/// FluxBeam pool route.
@@ -83,11 +85,11 @@ pub(crate) fn dex_detection_route(
("meteora_dbc", "meteora_dbc.swap") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDbcPool);
},
("meteora_damm_v2", "meteora_damm_v2.create_pool") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDammV2Pool);
("meteora_dlmm", "meteora_dlmm.create_pool") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDlmmPool);
},
("meteora_damm_v2", "meteora_damm_v2.swap") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDammV2Pool);
("meteora_dlmm", "meteora_dlmm.swap") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDlmmPool);
},
("meteora_damm_v1", "meteora_damm_v1.create_pool") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDammV1Pool);
@@ -95,6 +97,12 @@ pub(crate) fn dex_detection_route(
("meteora_damm_v1", "meteora_damm_v1.swap") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDammV1Pool);
},
("meteora_damm_v2", "meteora_damm_v2.create_pool") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDammV2Pool);
},
("meteora_damm_v2", "meteora_damm_v2.swap") => {
return Some(crate::dex_detection_route::DexDetectionRoute::MeteoraDammV2Pool);
},
("orca_whirlpools", "orca_whirlpools.create_pool") => {
return Some(crate::dex_detection_route::DexDetectionRoute::OrcaWhirlpoolsPool);
},

79
kb_lib/src/dex_event_classification.rs
View File

@@ -195,6 +195,37 @@ pub fn is_dex_admin_event_kind(event_kind: &str) -> bool {
return false;
}
/// Returns true when a decoded payload contains at least one direct amount or price field.
///
/// This is a conservative payload-level check. It does not inspect transaction
/// token balance deltas and is intended for protocol decoders that cannot yet
/// produce a deterministic materializable swap payload.
pub(crate) fn decoded_payload_has_trade_amount_or_price_payload(
payload: &serde_json::Value,
) -> bool {
return value_contains_any_non_null_key(
payload,
&[
"baseAmountRaw",
"base_amount_raw",
"baseAmount",
"amountBase",
"amountInBase",
"quoteAmountRaw",
"quote_amount_raw",
"quoteAmount",
"amountQuote",
"amountOutQuote",
"amountIn",
"amountOut",
"priceQuotePerBase",
"price_quote_per_base",
"quotePerBase",
"lastPriceQuotePerBase",
],
);
}
/// Returns true when a decoded payload is marked as a trade candidate.
///
/// Explicit payload metadata wins over event-kind inference. This allows
@@ -332,6 +363,38 @@ fn json_insert_i64_if_missing(
object.insert(key.to_owned(), serde_json::Value::Number(serde_json::Number::from(value)));
}
fn value_contains_any_non_null_key(value: &serde_json::Value, candidate_keys: &[&str]) -> bool {
if let Some(object) = value.as_object() {
for candidate_key in candidate_keys {
let candidate_value_option = object.get(*candidate_key);
if let Some(candidate_value) = candidate_value_option {
if !candidate_value.is_null() {
if let Some(text) = candidate_value.as_str() {
if text.trim().is_empty() {
continue;
}
}
return true;
}
}
}
for nested_value in object.values() {
if value_contains_any_non_null_key(nested_value, candidate_keys) {
return true;
}
}
return false;
}
if let Some(array) = value.as_array() {
for nested_value in array {
if value_contains_any_non_null_key(nested_value, candidate_keys) {
return true;
}
}
}
return false;
}
fn extract_top_level_bool_by_candidate_keys(
payload: &serde_json::Value,
candidate_keys: &[&str],
@@ -506,4 +569,20 @@ mod tests {
assert!(!super::is_decoded_event_trade_candidate("pump_swap.buy", &payload_json));
assert!(!super::is_decoded_event_candle_candidate("pump_swap.buy", &payload_json));
}
#[test]
fn detects_direct_amount_or_price_payload_recursively() {
let payload_json = serde_json::json!({
"nested": {
"quoteAmountRaw": "2500"
}
});
assert!(super::decoded_payload_has_trade_amount_or_price_payload(&payload_json));
let empty_payload_json = serde_json::json!({
"nested": {
"quoteAmountRaw": ""
}
});
assert!(!super::decoded_payload_has_trade_amount_or_price_payload(&empty_payload_json));
}
}

20
kb_lib/src/lib.rs
View File

@@ -411,6 +411,14 @@ pub use db::PoolTokenDto;
pub use db::PoolTokenEntity;
/// Role of one token inside a normalized pool.
pub use db::PoolTokenRole;
/// Diagnostic row for instructions of one Solana program.
pub use db::ProgramInstructionDiagnosticDto;
/// Raw diagnostic row for instructions of one Solana program.
pub use db::ProgramInstructionDiagnosticEntity;
/// Raw row used to summarize instruction discriminators for one Solana program.
pub use db::ProgramInstructionDiscriminatorRowEntity;
/// Aggregated instruction discriminator diagnostic row.
pub use db::ProgramInstructionDiscriminatorSummaryDto;
/// Application-facing protocol candidate DTO.
///
/// A protocol candidate records a program/instruction that should be inspected
@@ -625,6 +633,10 @@ pub use db::query_pools_get_by_address;
pub use db::query_pools_list;
/// Inserts or updates one normalized pool row by address.
pub use db::query_pools_upsert;
/// Lists diagnostic instruction rows for one program id.
pub use db::query_program_instruction_diagnostics_list_by_program_id;
/// Lists instruction discriminator summaries for one program id.
pub use db::query_program_instruction_discriminator_summaries_list_by_program_id;
/// Lists protocol candidate summaries ordered by investigation priority.
pub use db::query_protocol_candidate_summaries_list_by_priority;
/// Deletes protocol candidates for one transaction.
@@ -761,6 +773,14 @@ pub use dex::MeteoraDbcDecodedEvent;
pub use dex::MeteoraDbcDecoder;
/// Decoded Meteora DBC swap event.
pub use dex::MeteoraDbcSwapDecoded;
/// Decoded Meteora DLMM create-pool event.
pub use dex::MeteoraDlmmCreatePoolDecoded;
/// Decoded Meteora DLMM event.
pub use dex::MeteoraDlmmDecodedEvent;
/// Meteora DLMM decoder.
pub use dex::MeteoraDlmmDecoder;
/// Decoded Meteora DLMM swap event.
pub use dex::MeteoraDlmmSwapDecoded;
/// Decoded Orca Whirlpools create-pool event.
pub use dex::OrcaWhirlpoolsCreatePoolDecoded;
/// Decoded Orca Whirlpools event.

81
kb_lib/src/local_pipeline_validation.rs
View File

@@ -84,6 +84,37 @@ impl LocalPipelineValidationConfig {
require_candles_per_dex: true,
};
}
/// Builds the `0.7.28` multi-DEX validation config.
///
/// This profile treats currently materialized DEX decoders as required:
/// PumpSwap, Raydium CPMM, Raydium CLMM, and Meteora DLMM.
///
/// It intentionally accepts additional observed DEX codes because some DEXes
/// can be detected before their trade/candle materialization is complete
/// (`meteora_damm_v1`, future launchpads, and other roadmap DEXes).
pub fn v0_7_28_multi_dex_non_regression() -> Self {
return Self {
profile_code: "0.7.28_multi_dex_non_regression".to_string(),
expected_dex_codes: vec![
"pump_swap".to_string(),
"raydium_cpmm".to_string(),
"raydium_clmm".to_string(),
"meteora_dlmm".to_string(),
],
require_all_expected_dexes: true,
allow_unexpected_dexes: true,
require_clean_diagnostics: false,
require_ok_trade_candidates_fully_materialized: true,
require_no_invalid_trade_events: true,
require_no_duplicate_decoded_event_trades: true,
require_no_duplicate_candle_buckets: true,
require_no_pair_gaps: false,
require_decoded_events_per_dex: true,
require_trade_events_per_dex: false,
require_candles_per_dex: false,
};
}
}
/// A single local pipeline validation issue.
@@ -185,6 +216,14 @@ impl LocalPipelineValidationService {
let config = crate::LocalPipelineValidationConfig::v0_7_27_multi_dex_non_regression();
return self.validate_current_database(&config).await;
}
/// Diagnoses the current database with the `0.7.28` multi-DEX non-regression profile.
pub async fn validate_v0_7_28_current_database(
&self,
) -> Result<crate::LocalPipelineValidationRunDto, crate::Error> {
let config = crate::LocalPipelineValidationConfig::v0_7_28_multi_dex_non_regression();
return self.validate_current_database(&config).await;
}
}
/// Validates a diagnostics summary without performing database access.
@@ -472,6 +511,48 @@ mod tests {
assert_eq!(report.warning_count, 0);
}
fn make_0_7_28_summary_with_meteora() -> crate::LocalPipelineDiagnosticSummaryDto {
let mut summary = make_clean_summary();
summary.dex_summaries.retain(|dex_summary| {
return dex_summary.dex_code != "pump_fun";
});
summary.dex_summaries.push(crate::LocalDexDiagnosticSummaryDto {
dex_code: "meteora_dlmm".to_string(),
pool_count: 67,
pair_count: 67,
decoded_event_count: 795,
decoded_trade_candidate_count: 795,
decoded_candle_candidate_count: 795,
trade_event_count: 404,
pair_candle_count: 448,
});
summary.dex_summaries.push(crate::LocalDexDiagnosticSummaryDto {
dex_code: "meteora_damm_v1".to_string(),
pool_count: 1,
pair_count: 1,
decoded_event_count: 1,
decoded_trade_candidate_count: 1,
decoded_candle_candidate_count: 1,
trade_event_count: 0,
pair_candle_count: 0,
});
summary.pool_count = 95;
summary.pair_count = 95;
return summary;
}
#[test]
fn validation_accepts_meteora_dlmm_and_partial_future_dex_in_0_7_28_profile() {
let summary = make_0_7_28_summary_with_meteora();
let config = crate::LocalPipelineValidationConfig::v0_7_28_multi_dex_non_regression();
let report = crate::validate_local_pipeline_diagnostics_summary(&summary, &config);
assert!(report.validation_passed);
assert_eq!(report.validation_profile_code, "0.7.28_multi_dex_non_regression");
assert_eq!(report.blocking_issue_count, 0);
assert!(report.observed_dex_codes.contains(&"meteora_dlmm".to_string()));
assert!(report.observed_dex_codes.contains(&"meteora_damm_v1".to_string()));
}
#[test]
fn validation_rejects_missing_expected_dex() {
let mut summary = make_clean_summary();

63
kb_lib/src/token_metadata.rs
View File

@@ -677,59 +677,16 @@ fn clean_metaplex_string(value: &str) -> std::option::Option<std::string::String
}
fn decode_base64_standard(text: &str) -> Result<std::vec::Vec<u8>, crate::Error> {
let mut output = std::vec::Vec::new();
let mut group = [0u8; 4];
let mut group_len = 0usize;
let mut padding_count = 0usize;
for byte in text.bytes() {
if byte == b'\r' || byte == b'\n' || byte == b'\t' || byte == b' ' {
continue;
}
let value_option = base64_value(byte);
let value = match value_option {
Some(value) => value,
None => {
return Err(crate::Error::Json(format!(
"invalid base64 character '{}'",
byte as char
)));
},
};
if byte == b'=' {
padding_count += 1;
}
group[group_len] = value;
group_len += 1;
if group_len == 4 {
output.push((group[0] << 2) | (group[1] >> 4));
if padding_count < 2 {
output.push((group[1] << 4) | (group[2] >> 2));
}
if padding_count == 0 {
output.push((group[2] << 6) | group[3]);
}
group = [0u8; 4];
group_len = 0;
padding_count = 0;
}
}
if group_len != 0 {
return Err(crate::Error::Json(
"invalid base64 length: trailing partial group".to_string(),
));
}
return Ok(output);
}
fn base64_value(byte: u8) -> std::option::Option<u8> {
match byte {
b'A'..=b'Z' => return Some(byte - b'A'),
b'a'..=b'z' => return Some(byte - b'a' + 26),
b'0'..=b'9' => return Some(byte - b'0' + 52),
b'+' => return Some(62),
b'/' => return Some(63),
b'=' => return Some(0),
_ => return None,
use base64::Engine;
let decoded_result = base64::engine::general_purpose::STANDARD.decode(text);
match decoded_result {
Ok(decoded) => return Ok(decoded),
Err(error) => {
return Err(crate::Error::Json(format!(
"cannot decode standard base64 payload: {}",
error
)));
},
}
}

7
kb_lib/src/trade_aggregation.rs
View File

@@ -118,7 +118,8 @@ impl TradeAggregationService {
}
let trade_side = crate::trade_side_resolution::extract_trade_side(
decoded_event.event_kind.as_str(),
&payload, );
&payload,
);
let amount_input = crate::trade_amount_resolution::TradeAmountResolutionInput {
database: self.database.as_ref(),
transaction: &transaction,
@@ -138,6 +139,10 @@ impl TradeAggregationService {
Ok(amount_resolution) => amount_resolution,
Err(error) => return Err(error),
};
let trade_side = match amount_resolution.resolved_trade_side.clone() {
Some(resolved_trade_side) => resolved_trade_side,
None => trade_side,
};
let base_amount_raw = amount_resolution.base_amount_raw.clone();
let quote_amount_raw = amount_resolution.quote_amount_raw.clone();
let price_quote_per_base = amount_resolution.price_quote_per_base;

1034
kb_lib/src/trade_amount_resolution.rs
View File

File diff suppressed because it is too large Load Diff