diff --git a/CHANGELOG.md b/CHANGELOG.md index 5077c8b..1075af6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,13 @@ # CHANGELOG +## 0.1.0-pre.029 + +- Migration initiale de `kb-pipeline` : primitives de planification et extraction canonique Solana vers les tables core. +- Adaptation aux façades `kb-lib`, `kb-store` et `kb-onchain-transport`. +- Normalisation du target de tracing `kb-pipeline` et ajout d’un test de configuration. + + ## 0.1.0-pre.024 - Première tranche fonctionnelle de `kb-onchain-transport`. diff --git a/README.md b/README.md index c149564..64a3e3c 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ Khadhroony Bot3 is the consolidated successor workspace to `khadhroony-bot2`. - `kb-pipeline`: ingestion, decoding, materialization and execution orchestration. - `kb-onchain-transport`: HTTP and WebSocket Solana transports. - `kb-store`: store-neutral contracts and PostgreSQL implementation modules. -- `kb-wallet`: stockage local protégé de keypairs temporaires, résumés non secrets et signature de messages. +- `kb-wallet`: wallet and signer support. - `kb-app-demo`: the only binary retained during the initial migration. ## Migration principle @@ -89,3 +89,8 @@ Les modules de matérialisation sont privés. Tous les contrats, types concrets `0.1.0-pre.015` rend `ExSolanaCoreExecutor` fonctionnel. Il construit des plans déterministes pour 109 opérations natives réparties sur 18 surfaces, conserve les refus explicites des programmes historiques non exécutables et expose ses intents sous les préfixes bot3 `ExSolanaCore*` et `EX_SOLANA_CORE_*`. Les 103 autres exécuteurs restent réservés jusqu’à leur port fonctionnel. `0.1.0-pre.016` intègre le moteur transversal `ExSafetyChecker` et rend `ExSplMemoExecutor` fonctionnel. Memo v4 produit une instruction officielle bornée et soumise à simulation ; les générations historiques v1/v3 restent decode-only. Les 102 autres exécuteurs restent réservés jusqu’à leur port fonctionnel. + + +### Migration pipeline + +La tranche `0.1.0-pre.029` migre la planification et l’extraction core de `kb-pipeline`. diff --git a/ROADMAP.md b/ROADMAP.md index a41740f..80682de 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -276,3 +276,11 @@ - [x] Routes canoniques `kb-wallet`. - [x] Configuration d’exemple sans clé fournisseur en clair. - [ ] Migration fonctionnelle de `kb-pipeline` à poursuivre à partir de `pre.029`. + + +### 0.1.0-pre.029 — Pipeline, fondations + +- [x] Planification et scopes de replay. +- [x] Extraction canonique vers les tables core. +- [ ] Backfill et replay de décodage. +- [ ] Orchestrations Solana stateful et exécution. diff --git a/config/example.config.json b/config/example.config.json index d4ff4b6..4a87581 100644 --- a/config/example.config.json +++ b/config/example.config.json @@ -46,7 +46,7 @@ "kb-store", "kb-onchain-transport", "kb-wallet", - "kb_pipeline", + "kb-pipeline", "kb_app_demo", "kb_executor_metadata_metaplex_token_metadata", "kb_executor_metadata_spl_name_service", @@ -927,42 +927,42 @@ "ansi": false }, { - "name": "file_kb_pipeline_debug", + "name": "file_kb-pipeline_debug", "enabled": true, "sink": "file", "level": "debug", - "path": "logs/devnet/kb_pipeline/debug.log", + "path": "logs/devnet/kb-pipeline/debug.log", "rotation": "daily", "format": "human", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { - "name": "file_kb_pipeline_info", + "name": "file_kb-pipeline_info", "enabled": true, "sink": "file", "level": "info", - "path": "logs/devnet/kb_pipeline/info.log", + "path": "logs/devnet/kb-pipeline/info.log", "rotation": "daily", "format": "human", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { - "name": "file_kb_pipeline_error", + "name": "file_kb-pipeline_error", "enabled": true, "sink": "file", "level": "error", - "path": "logs/devnet/kb_pipeline/error.jsonl", + "path": "logs/devnet/kb-pipeline/error.jsonl", "rotation": "daily", "format": "json", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { @@ -1690,7 +1690,7 @@ "kb-lib.materializer.staking", "kb-lib.materializer.token", "kb-lib.materializer.transaction", - "kb_pipeline", + "kb-pipeline", "kb-onchain-transport", "kb-store", "kb-wallet" @@ -2568,42 +2568,42 @@ "ansi": false }, { - "name": "file_kb_pipeline_debug", + "name": "file_kb-pipeline_debug", "enabled": true, "sink": "file", "level": "debug", - "path": "logs/mainnet_research/kb_pipeline/debug.log", + "path": "logs/mainnet_research/kb-pipeline/debug.log", "rotation": "daily", "format": "human", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { - "name": "file_kb_pipeline_info", + "name": "file_kb-pipeline_info", "enabled": true, "sink": "file", "level": "info", - "path": "logs/mainnet_research/kb_pipeline/info.log", + "path": "logs/mainnet_research/kb-pipeline/info.log", "rotation": "daily", "format": "human", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { - "name": "file_kb_pipeline_error", + "name": "file_kb-pipeline_error", "enabled": true, "sink": "file", "level": "error", - "path": "logs/mainnet_research/kb_pipeline/error.jsonl", + "path": "logs/mainnet_research/kb-pipeline/error.jsonl", "rotation": "daily", "format": "json", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { @@ -3531,7 +3531,7 @@ "kb-lib.materializer.staking", "kb-lib.materializer.token", "kb-lib.materializer.transaction", - "kb_pipeline", + "kb-pipeline", "kb-onchain-transport", "kb-store", "kb-wallet" @@ -4409,42 +4409,42 @@ "ansi": false }, { - "name": "file_kb_pipeline_debug", + "name": "file_kb-pipeline_debug", "enabled": true, "sink": "file", "level": "debug", - "path": "logs/mainnet/kb_pipeline/debug.log", + "path": "logs/mainnet/kb-pipeline/debug.log", "rotation": "daily", "format": "human", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { - "name": "file_kb_pipeline_info", + "name": "file_kb-pipeline_info", "enabled": true, "sink": "file", "level": "info", - "path": "logs/mainnet/kb_pipeline/info.log", + "path": "logs/mainnet/kb-pipeline/info.log", "rotation": "daily", "format": "human", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { - "name": "file_kb_pipeline_error", + "name": "file_kb-pipeline_error", "enabled": true, "sink": "file", "level": "error", - "path": "logs/mainnet/kb_pipeline/error.jsonl", + "path": "logs/mainnet/kb-pipeline/error.jsonl", "rotation": "daily", "format": "json", "ansi": false, "targets": [ - "kb_pipeline" + "kb-pipeline" ] }, { diff --git a/kb-app-demo/src/main.rs b/kb-app-demo/src/main.rs index 0a569df..e16b3e8 100644 --- a/kb-app-demo/src/main.rs +++ b/kb-app-demo/src/main.rs @@ -1,5 +1,5 @@ // file: kb-app-demo/src/main.rs -// version: 1 +// version: 2 #![forbid(unsafe_code)] #![deny(unreachable_pub)] @@ -10,5 +10,9 @@ fn main() { println!("khadhroony-bot3 workspace scaffold"); println!("logging target: {}", kb_logging::tracing_target()); - println!("pipeline: {}", kb_pipeline::migration_status()); + println!( + "pipeline: {} v{}", + kb_pipeline::CORE_EXTRACTION_PROCESSOR_NAME, + kb_pipeline::CORE_EXTRACTION_PROCESSOR_VERSION + ); } diff --git a/kb-config/src/settings.rs b/kb-config/src/settings.rs index db7da9f..5152523 100644 --- a/kb-config/src/settings.rs +++ b/kb-config/src/settings.rs @@ -1361,6 +1361,16 @@ mod tests { } } + #[test] + fn example_config_uses_canonical_pipeline_tracing_routes() { + let source = include_str!("../../config/example.config.json"); + assert!(source.contains("\"kb-pipeline\"")); + assert!(source.contains("logs/devnet/kb-pipeline/debug.log")); + assert!(source.contains("logs/mainnet_research/kb-pipeline/info.log")); + assert!(source.contains("logs/mainnet/kb-pipeline/error.jsonl")); + assert!(!source.contains("kb_pipeline")); + } + #[test] fn example_config_parses_and_resolves_active_profile() { let config_result = super::parse_config_json(EXAMPLE_CONFIG); diff --git a/kb-pipeline/Cargo.toml b/kb-pipeline/Cargo.toml index 27a4995..f314820 100644 --- a/kb-pipeline/Cargo.toml +++ b/kb-pipeline/Cargo.toml @@ -1,5 +1,5 @@ # file: kb-pipeline/Cargo.toml -# version: 2 +# version: 4 [package] name = "kb-pipeline" @@ -9,11 +9,19 @@ license.workspace = true publish.workspace = true [dependencies] +async-trait.workspace = true +chrono.workspace = true +futures-util.workspace = true kb-core = { path = "../kb-core" } kb-lib = { path = "../kb-lib" } -kb-store = { path = "../kb-store" } kb-onchain-transport = { path = "../kb-onchain-transport" } +kb-store = { path = "../kb-store" } kb-wallet = { path = "../kb-wallet" } +serde_json.workspace = true +sha2.workspace = true +tokio.workspace = true +tracing.workspace = true +uuid.workspace = true [lints] workspace = true diff --git a/kb-pipeline/src/constants.rs b/kb-pipeline/src/constants.rs new file mode 100644 index 0000000..b6a4434 --- /dev/null +++ b/kb-pipeline/src/constants.rs @@ -0,0 +1,7 @@ +// file: kb-pipeline/src/constants.rs +// version: 2 + +//! Pipeline constants. + +/// Canonical tracing target for pipeline orchestration. +pub(crate) const TRACING_TARGET: &str = "kb-pipeline"; diff --git a/kb-pipeline/src/core_extraction.rs b/kb-pipeline/src/core_extraction.rs new file mode 100644 index 0000000..d0f4f57 --- /dev/null +++ b/kb-pipeline/src/core_extraction.rs @@ -0,0 +1,1845 @@ +// file: kb-pipeline/src/core_extraction.rs +// version: 10 + +//! Canonical Solana transaction to normalized core extraction pipeline. + +use futures_util::FutureExt; // rust-rules: trait-import +use futures_util::StreamExt; // rust-rules: trait-import +use sha2::Digest; // rust-rules: trait-import + +/// Stable processing stage used by the canonical to core extractor. +pub const CORE_EXTRACTION_STAGE: &str = "core_extraction"; +/// Stable processor name used by the canonical to core extractor. +pub const CORE_EXTRACTION_PROCESSOR_NAME: &str = "canonical_to_core"; +/// Current extractor implementation version. +pub const CORE_EXTRACTION_PROCESSOR_VERSION: &str = "1"; + +/// Source selection used by one core extraction campaign. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum CoreExtractionSource { + /// Exact canonical transaction signatures. + Signatures(std::vec::Vec), + /// Canonical transactions still in the received state. + Pending, + /// Canonical transactions inside an inclusive slot range. + SlotRange { + /// Inclusive minimum slot. + min_slot: u64, + /// Inclusive maximum slot. + max_slot: u64, + }, + /// Transactions already containing one resolved core instruction for a program id. + ProgramId { + /// Solana program id. + program_id: std::string::String, + }, +} + +/// Complete bounded core extraction request. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct CoreExtractionRequest { + /// Source selection. + pub source: crate::CoreExtractionSource, + /// Maximum number of canonical transactions selected. + pub limit: u32, + /// Maximum concurrent extraction operations. + pub max_concurrent_extractions: u32, + /// Forces replacement even when the same processor version and hash succeeded. + pub force_replay: bool, +} + +impl CoreExtractionRequest { + /// Validates campaign bounds and source values. + pub fn validate(&self) -> kb_core::Result<()> { + if self.limit == 0 { + return std::result::Result::Err(kb_core::Error::config( + "core extraction limit must be greater than zero", + )); + } + if self.max_concurrent_extractions == 0 { + return std::result::Result::Err(kb_core::Error::config( + "core extraction concurrency must be greater than zero", + )); + } + match &self.source { + crate::CoreExtractionSource::Signatures(signatures) => { + if signatures.is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "signature extraction requires at least one signature", + )); + } + for signature in signatures { + let validation_result = + kb_onchain_transport::validate_transaction_signature_text( + signature.as_str(), + "core extraction signature", + ); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + } + }, + crate::CoreExtractionSource::Pending => {}, + crate::CoreExtractionSource::SlotRange { min_slot, max_slot } => { + if min_slot > max_slot { + return std::result::Result::Err(kb_core::Error::config( + "core extraction minimum slot must not exceed maximum slot", + )); + } + }, + crate::CoreExtractionSource::ProgramId { program_id } => { + let validation_result = kb_onchain_transport::validate_solana_pubkey_text( + program_id.as_str(), + "core extraction program id", + ); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + }, + } + return std::result::Result::Ok(()); + } + + fn selection_filter(&self) -> kb_core::Result { + return match &self.source { + crate::CoreExtractionSource::Signatures(signatures) => { + kb_store::CoreExtractionSelectionFilter::new( + signatures.clone(), + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + self.limit, + ) + }, + crate::CoreExtractionSource::Pending => { + kb_store::CoreExtractionSelectionFilter::pending(self.limit) + }, + crate::CoreExtractionSource::SlotRange { min_slot, max_slot } => { + kb_store::CoreExtractionSelectionFilter::new( + std::vec::Vec::new(), + std::option::Option::Some(*min_slot), + std::option::Option::Some(*max_slot), + std::option::Option::None, + std::option::Option::None, + self.limit, + ) + }, + crate::CoreExtractionSource::ProgramId { program_id } => { + kb_store::CoreExtractionSelectionFilter::new( + std::vec::Vec::new(), + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + std::option::Option::Some(program_id.clone()), + self.limit, + ) + }, + }; + } +} + +/// Core extraction progress severity. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum CoreExtractionProgressLevel { + /// Diagnostic detail. + Debug, + /// Normal campaign information. + Info, + /// Recoverable issue or cancellation. + Warning, + /// Extraction failure. + Error, +} + +impl CoreExtractionProgressLevel { + /// Returns the stable lowercase level code. + pub fn code(&self) -> &'static str { + return match self { + crate::CoreExtractionProgressLevel::Debug => "debug", + crate::CoreExtractionProgressLevel::Info => "info", + crate::CoreExtractionProgressLevel::Warning => "warn", + crate::CoreExtractionProgressLevel::Error => "error", + }; + } +} + +/// One operator-visible extraction progress event. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct CoreExtractionProgressEvent { + /// UTC timestamp rendered in RFC 3339. + pub timestamp: std::string::String, + /// Severity. + pub level: crate::CoreExtractionProgressLevel, + /// Human-readable message. + pub message: std::string::String, + /// Number of terminal candidates. + pub completed: u64, + /// Total selected candidates. + pub total: u64, +} + +impl CoreExtractionProgressEvent { + fn new( + level: crate::CoreExtractionProgressLevel, + message: impl std::convert::Into, + completed: u64, + total: u64, + ) -> Self { + return Self { + timestamp: chrono::Utc::now().to_rfc3339(), + level, + message: message.into(), + completed, + total, + }; + } +} + +/// Progress and cancellation contract implemented by applications. +pub trait CoreExtractionObserver: Sync { + /// Receives one progress event. + fn on_progress(&self, event: &crate::CoreExtractionProgressEvent); + /// Returns true when cooperative cancellation was requested. + fn is_cancelled(&self) -> bool; +} + +/// Final counters for one core extraction campaign. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct CoreExtractionSummary { + /// Current extractor implementation version. + pub processor_version: std::string::String, + /// Number of selected canonical transactions. + pub selected: u64, + /// Number admitted to the bounded execution queue. + pub started: u64, + /// Number reaching a terminal result. + pub completed: u64, + /// Number skipped because the version and hash already succeeded. + pub skipped: u64, + /// Number extracted and committed successfully. + pub extracted: u64, + /// Number failing extraction or persistence. + pub failed: u64, + /// Number admitted but cancelled before a terminal result. + pub cancelled_candidates: u64, + /// Number never admitted after cancellation. + pub not_started: u64, + /// Whether the campaign was cancelled. + pub cancelled: bool, + /// Campaign start timestamp. + pub started_at: std::string::String, + /// Campaign finish timestamp. + pub finished_at: std::string::String, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum CoreExtractionItemStatus { + Skipped, + Extracted, + Failed, + Cancelled, +} + +struct CoreExtractionItemOutcome { + index: usize, + status: CoreExtractionItemStatus, + message: std::string::String, +} + +/// Executes one bounded canonical transaction to core extraction campaign. +pub async fn execute_core_extraction( + store: &S, + request: &crate::CoreExtractionRequest, + observer: &O, +) -> kb_core::Result +where + S: kb_store::CoreExtractionStore + Sync, + O: crate::CoreExtractionObserver, +{ + let validation_result = request.validate(); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + let filter_result = request.selection_filter(); + let filter = match filter_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let rows_result = store.list_raw_transactions_for_core_extraction(&filter).await; + let rows = match rows_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let started_at = chrono::Utc::now().to_rfc3339(); + let total = match u64::try_from(rows.len()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "core extraction selected count overflow: {error}" + ))); + }, + }; + tracing::info!( + target: crate::TRACING_TARGET, + selected = total, + processor_version = crate::CORE_EXTRACTION_PROCESSOR_VERSION, + force_replay = request.force_replay, + "core extraction campaign started" + ); + observer.on_progress(&crate::CoreExtractionProgressEvent::new( + crate::CoreExtractionProgressLevel::Info, + format!( + "core extraction started with processor version {} and {total} selected transactions", + crate::CORE_EXTRACTION_PROCESSOR_VERSION + ), + 0, + total, + )); + let concurrency = match usize::try_from(request.max_concurrent_extractions) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::config(format!( + "core extraction concurrency conversion failed: {error}" + ))); + }, + }; + let mut next_index = 0_usize; + let mut in_flight = futures_util::stream::FuturesUnordered::new(); + let mut summary = crate::CoreExtractionSummary { + processor_version: crate::CORE_EXTRACTION_PROCESSOR_VERSION.to_string(), + selected: total, + started: 0, + completed: 0, + skipped: 0, + extracted: 0, + failed: 0, + cancelled_candidates: 0, + not_started: 0, + cancelled: false, + started_at, + finished_at: std::string::String::new(), + }; + loop { + while in_flight.len() < concurrency && next_index < rows.len() && !observer.is_cancelled() { + let row = rows[next_index].clone(); + let item_index = next_index; + let force_replay = request.force_replay; + in_flight.push( + async move { + return crate::core_extraction::process_one_core_extraction( + store, + row, + item_index, + force_replay, + ) + .await; + } + .boxed(), + ); + next_index += 1; + summary.started += 1; + } + if observer.is_cancelled() { + summary.cancelled = true; + } + let next_result = in_flight.next().await; + let outcome = match next_result { + std::option::Option::Some(value) => value, + std::option::Option::None => break, + }; + match outcome.status { + CoreExtractionItemStatus::Skipped => summary.skipped += 1, + CoreExtractionItemStatus::Extracted => summary.extracted += 1, + CoreExtractionItemStatus::Failed => summary.failed += 1, + CoreExtractionItemStatus::Cancelled => summary.cancelled_candidates += 1, + } + if outcome.status != CoreExtractionItemStatus::Cancelled { + summary.completed += 1; + } + let level = match outcome.status { + CoreExtractionItemStatus::Skipped => crate::CoreExtractionProgressLevel::Debug, + CoreExtractionItemStatus::Extracted => crate::CoreExtractionProgressLevel::Info, + CoreExtractionItemStatus::Failed => crate::CoreExtractionProgressLevel::Error, + CoreExtractionItemStatus::Cancelled => crate::CoreExtractionProgressLevel::Warning, + }; + observer.on_progress(&crate::CoreExtractionProgressEvent::new( + level, + format!("candidate {}: {}", outcome.index + 1, outcome.message), + summary.completed, + total, + )); + } + let started_and_completed = summary.completed + summary.cancelled_candidates; + if summary.started > started_and_completed { + summary.cancelled_candidates += summary.started - started_and_completed; + } + if summary.selected > summary.started { + summary.not_started = summary.selected - summary.started; + } + summary.finished_at = chrono::Utc::now().to_rfc3339(); + let final_level = if summary.failed > 0 || summary.cancelled { + crate::CoreExtractionProgressLevel::Warning + } else { + crate::CoreExtractionProgressLevel::Info + }; + tracing::info!( + target: crate::TRACING_TARGET, + extracted = summary.extracted, + skipped = summary.skipped, + failed = summary.failed, + cancelled = summary.cancelled_candidates, + not_started = summary.not_started, + "core extraction campaign completed" + ); + observer.on_progress(&crate::CoreExtractionProgressEvent::new( + final_level, + format!( + "core extraction completed: extracted={}, skipped={}, failed={}, cancelled={}, not_started={}", + summary.extracted, + summary.skipped, + summary.failed, + summary.cancelled_candidates, + summary.not_started + ), + summary.completed, + total, + )); + return std::result::Result::Ok(summary); +} + +async fn process_one_core_extraction( + store: &S, + row: kb_store::RawTransactionRow, + index: usize, + force_replay: bool, +) -> crate::core_extraction::CoreExtractionItemOutcome +where + S: kb_store::CoreExtractionStore + Sync, +{ + let identity_result = crate::core_extraction::ledger_identity_from_raw(&row); + let identity = match identity_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return crate::core_extraction::CoreExtractionItemOutcome { + index, + status: CoreExtractionItemStatus::Failed, + message: error.to_string(), + }; + }, + }; + if !force_replay { + let current_result = store.is_core_extraction_current(&identity).await; + match current_result { + std::result::Result::Ok(true) => { + return crate::core_extraction::CoreExtractionItemOutcome { + index, + status: CoreExtractionItemStatus::Skipped, + message: "already current for the same version and hash".to_string(), + }; + }, + std::result::Result::Ok(false) => {}, + std::result::Result::Err(error) => { + return crate::core_extraction::CoreExtractionItemOutcome { + index, + status: CoreExtractionItemStatus::Failed, + message: error.to_string(), + }; + }, + } + } + tracing::debug!( + target: crate::TRACING_TARGET, + signature = row.signature.as_str(), + index, + force_replay, + "extracting canonical transaction to core" + ); + let extraction_result = crate::extract_raw_transaction_to_core(&row); + let bundle = match extraction_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + let failure_result = kb_store::CoreExtractionFailure::new( + row.id, + identity, + "canonical_to_core_extraction_failed", + error.to_string(), + ); + if let std::result::Result::Ok(failure) = failure_result { + let _mark_result = store.mark_core_extraction_failed(&failure).await; + } + return crate::core_extraction::CoreExtractionItemOutcome { + index, + status: CoreExtractionItemStatus::Failed, + message: error.to_string(), + }; + }, + }; + let persist_result = store.persist_core_extraction(&bundle, force_replay).await; + return match persist_result { + std::result::Result::Ok(_) => crate::core_extraction::CoreExtractionItemOutcome { + index, + status: CoreExtractionItemStatus::Extracted, + message: "extracted and committed".to_string(), + }, + std::result::Result::Err(error) => { + let failure_result = kb_store::CoreExtractionFailure::new( + row.id, + bundle.ledger_identity, + "canonical_to_core_persist_failed", + error.to_string(), + ); + if let std::result::Result::Ok(failure) = failure_result { + let _mark_result = store.mark_core_extraction_failed(&failure).await; + } + crate::core_extraction::CoreExtractionItemOutcome { + index, + status: CoreExtractionItemStatus::Failed, + message: error.to_string(), + } + }, + }; +} + +/// Extracts one canonical raw transaction into a complete normalized core bundle. +pub fn extract_raw_transaction_to_core( + row: &kb_store::RawTransactionRow, +) -> kb_core::Result { + let canonical_json = match &row.canonical_json { + std::option::Option::Some(value) => value.clone(), + std::option::Option::None => { + return std::result::Result::Err(kb_core::Error::invalid_state( + "canonical JSON is unavailable for core extraction", + )); + }, + }; + let input_hash = match &row.canonical_json_hash { + std::option::Option::Some(value) if !value.trim().is_empty() => value.clone(), + _ => { + return std::result::Result::Err(kb_core::Error::invalid_state( + "canonical JSON hash is unavailable for core extraction", + )); + }, + }; + let expected_format_version = + match i32::try_from(kb_lib::MD_CANONICAL_TRANSACTION_FORMAT_VERSION) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "canonical format version conversion failed: {error}" + ))); + }, + }; + if row.canonical_format_version != expected_format_version { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "unsupported canonical format version in raw store: {}", + row.canonical_format_version + ))); + } + let parse_result = serde_json::from_value::(canonical_json); + let transaction = match parse_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::json(format!( + "cannot parse canonical transaction for core extraction: {error}" + ))); + }, + }; + let validation_result = transaction.validate(); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + if transaction.primary_signature != row.signature { + return std::result::Result::Err(kb_core::Error::invalid_state( + "canonical transaction signature does not match raw row signature", + )); + } + let row_slot_result = u64::try_from(row.slot); + let row_slot = match row_slot_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "raw transaction slot is negative: {error}" + ))); + }, + }; + if transaction.slot != row_slot { + return std::result::Result::Err(kb_core::Error::invalid_state( + "canonical transaction slot does not match raw row slot", + )); + } + let computed_hash_result = transaction.canonical_json_hash(); + let computed_hash = match computed_hash_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if computed_hash != input_hash { + return std::result::Result::Err(kb_core::Error::invalid_state( + "canonical transaction hash does not match the stored hash", + )); + } + let ledger_identity_result = crate::core_extraction::ledger_identity_from_raw(row); + let ledger_identity = match ledger_identity_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if let std::option::Option::Some(metadata) = &transaction.metadata { + return crate::core_extraction::build_bundle_with_metadata( + row, + &transaction, + metadata, + ledger_identity, + ); + } + let core_transaction_result = kb_store::CoreTransactionInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + false, + std::option::Option::None, + ); + let core_transaction = match core_transaction_result { + std::result::Result::Ok(value) => value.with_raw_transaction_id(row.id), + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let account_keys_result = crate::core_extraction::extract_account_keys(&transaction); + let account_keys = match account_keys_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let instruction_result = crate::core_extraction::extract_instructions(&transaction); + let (instructions, inner_instructions) = match instruction_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(kb_store::CoreExtractionBundle { + raw_transaction_id: row.id, + ledger_identity, + transaction: core_transaction, + account_keys, + instructions, + inner_instructions, + logs: std::vec::Vec::new(), + balance_changes: std::vec::Vec::new(), + }); +} + +fn build_bundle_with_metadata( + row: &kb_store::RawTransactionRow, + transaction: &kb_lib::MdCanonicalTransaction, + metadata: &kb_lib::MdCanonicalTransactionMetadata, + ledger_identity: kb_store::ProcessingLedgerIdentity, +) -> kb_core::Result { + let failed = metadata.status == kb_lib::MdCanonicalTransactionStatus::Failed; + let core_transaction_result = kb_store::CoreTransactionInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + failed, + metadata.error.clone(), + ); + let core_transaction = match core_transaction_result { + std::result::Result::Ok(value) => value.with_raw_transaction_id(row.id), + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let account_keys_result = crate::core_extraction::extract_account_keys(transaction); + let account_keys = match account_keys_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let instruction_result = crate::core_extraction::extract_instructions(transaction); + let (instructions, inner_instructions) = match instruction_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let logs_result = crate::core_extraction::extract_logs(transaction, metadata); + let logs = match logs_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let balances_result = crate::core_extraction::extract_balance_changes(transaction, metadata); + let balance_changes = match balances_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(kb_store::CoreExtractionBundle { + raw_transaction_id: row.id, + ledger_identity, + transaction: core_transaction, + account_keys, + instructions, + inner_instructions, + logs, + balance_changes, + }); +} + +fn ledger_identity_from_raw( + row: &kb_store::RawTransactionRow, +) -> kb_core::Result { + let input_hash = match &row.canonical_json_hash { + std::option::Option::Some(value) => value.clone(), + std::option::Option::None => { + return std::result::Result::Err(kb_core::Error::invalid_state( + "canonical JSON hash is required for the processing ledger", + )); + }, + }; + return kb_store::ProcessingLedgerIdentity::new( + crate::CORE_EXTRACTION_STAGE, + crate::CORE_EXTRACTION_PROCESSOR_NAME, + crate::CORE_EXTRACTION_PROCESSOR_VERSION, + row.signature.clone(), + input_hash, + ); +} + +fn resolved_account_keys( + transaction: &kb_lib::MdCanonicalTransaction, +) -> std::vec::Vec { + let mut keys = std::vec::Vec::with_capacity(transaction.message.resolved_account_count()); + keys.extend(transaction.message.static_account_keys.iter().cloned()); + keys.extend(transaction.message.loaded_addresses.writable.iter().cloned()); + keys.extend(transaction.message.loaded_addresses.readonly.iter().cloned()); + return keys; +} + +fn extract_account_keys( + transaction: &kb_lib::MdCanonicalTransaction, +) -> kb_core::Result> { + let static_count = transaction.message.static_account_keys.len(); + let required = usize::from(transaction.message.header.num_required_signatures); + let readonly_signed = usize::from(transaction.message.header.num_readonly_signed_accounts); + let readonly_unsigned = usize::from(transaction.message.header.num_readonly_unsigned_accounts); + let writable_signed_count = required - readonly_signed; + let writable_unsigned_end = static_count - readonly_unsigned; + let mut output = std::vec::Vec::with_capacity(transaction.message.resolved_account_count()); + for (index, key) in transaction.message.static_account_keys.iter().enumerate() { + let account_index_result = u32::try_from(index); + let account_index = match account_index_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "static account index overflow: {error}" + ))); + }, + }; + let signer = index < required; + let writable = if signer { + index < writable_signed_count + } else { + index < writable_unsigned_end + }; + let insert_result = kb_store::CoreAccountKeyInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + account_index, + key.clone(), + kb_store::CoreAccountKeySource::Static, + writable, + signer, + std::option::Option::None, + ); + let insert = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + output.push(insert); + } + for key in &transaction.message.loaded_addresses.writable { + let account_index_result = u32::try_from(output.len()); + let account_index = match account_index_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "loaded writable account index overflow: {error}" + ))); + }, + }; + let insert_result = kb_store::CoreAccountKeyInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + account_index, + key.clone(), + kb_store::CoreAccountKeySource::LoadedWritable, + true, + false, + std::option::Option::None, + ); + let insert = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + output.push(insert); + } + for key in &transaction.message.loaded_addresses.readonly { + let account_index_result = u32::try_from(output.len()); + let account_index = match account_index_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "loaded readonly account index overflow: {error}" + ))); + }, + }; + let insert_result = kb_store::CoreAccountKeyInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + account_index, + key.clone(), + kb_store::CoreAccountKeySource::LoadedReadonly, + false, + false, + std::option::Option::None, + ); + let insert = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + output.push(insert); + } + return std::result::Result::Ok(output); +} + +fn extract_instructions( + transaction: &kb_lib::MdCanonicalTransaction, +) -> kb_core::Result<( + std::vec::Vec, + std::vec::Vec, +)> { + let keys = crate::core_extraction::resolved_account_keys(transaction); + let mut instructions = std::vec::Vec::new(); + for (index, instruction) in transaction.message.instructions.iter().enumerate() { + let path = index.to_string(); + let resolved_result = crate::core_extraction::resolve_instruction(&keys, instruction); + let (program_id, accounts_json, payload_json, payload_hash) = match resolved_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let insert_result = kb_store::CoreInstructionInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + path, + program_id, + accounts_json, + payload_json, + ); + let insert = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let hash_result = insert.with_payload_json_hash(payload_hash); + let hashed = match hash_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + instructions.push(hashed); + } + let mut inner_instructions = std::vec::Vec::new(); + if let std::option::Option::Some(metadata) = &transaction.metadata { + for group in &metadata.inner_instructions { + let parent_path = group.parent_instruction_index.to_string(); + for (inner_index, instruction) in group.instructions.iter().enumerate() { + let path = format!("{parent_path}/{inner_index}"); + let resolved_result = + crate::core_extraction::resolve_instruction(&keys, instruction); + let (program_id, accounts_json, payload_json, payload_hash) = match resolved_result + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let insert_result = kb_store::CoreInnerInstructionInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + parent_path.clone(), + path, + program_id, + accounts_json, + payload_json, + ); + let insert = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let hash_result = insert.with_payload_json_hash(payload_hash); + let hashed = match hash_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + inner_instructions.push(hashed); + } + } + } + return std::result::Result::Ok((instructions, inner_instructions)); +} + +fn resolve_instruction( + keys: &[std::string::String], + instruction: &kb_lib::MdCanonicalCompiledInstruction, +) -> kb_core::Result<(std::string::String, serde_json::Value, serde_json::Value, std::string::String)> +{ + let program_id = match keys.get(usize::from(instruction.program_id_index)) { + std::option::Option::Some(value) => value.clone(), + std::option::Option::None => { + return std::result::Result::Err(kb_core::Error::invalid_state( + "instruction program id index is outside resolved accounts", + )); + }, + }; + let mut accounts = std::vec::Vec::with_capacity(instruction.account_indexes.len()); + for account_index in &instruction.account_indexes { + let account_key = match keys.get(usize::from(*account_index)) { + std::option::Option::Some(value) => value.clone(), + std::option::Option::None => { + return std::result::Result::Err(kb_core::Error::invalid_state( + "instruction account index is outside resolved accounts", + )); + }, + }; + accounts.push(serde_json::json!({ + "accountIndex": account_index, + "accountKey": account_key, + })); + } + let accounts_json = serde_json::Value::Array(accounts); + let payload_json = serde_json::json!({ + "programIdIndex": instruction.program_id_index, + "dataBase64": instruction.data_base64, + "stackHeight": instruction.stack_height, + }); + let payload_hash_result = crate::core_extraction::hash_json(&payload_json); + let payload_hash = match payload_hash_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok((program_id, accounts_json, payload_json, payload_hash)); +} + +fn extract_logs( + transaction: &kb_lib::MdCanonicalTransaction, + metadata: &kb_lib::MdCanonicalTransactionMetadata, +) -> kb_core::Result> { + let mut output = std::vec::Vec::with_capacity(metadata.log_messages.len()); + let links = crate::core_extraction::resolve_log_links(transaction, metadata); + for (index, text) in metadata.log_messages.iter().enumerate() { + let log_index_result = u32::try_from(index); + let log_index = match log_index_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "log index overflow: {error}" + ))); + }, + }; + let link = match links.get(index) { + std::option::Option::Some(value) => value.clone(), + std::option::Option::None => (std::option::Option::None, std::option::Option::None), + }; + let insert_result = kb_store::CoreLogInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + log_index, + link.1, + link.0, + text.clone(), + ); + let insert = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let hash = crate::core_extraction::hash_bytes(text.as_bytes()); + let hash_result = insert.with_log_text_hash(hash); + let hashed = match hash_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + output.push(hashed); + } + return std::result::Result::Ok(output); +} + +fn resolve_log_links( + transaction: &kb_lib::MdCanonicalTransaction, + metadata: &kb_lib::MdCanonicalTransactionMetadata, +) -> std::vec::Vec<( + std::option::Option, + std::option::Option, +)> { + let keys = crate::core_extraction::resolved_account_keys(transaction); + let mut top_level = std::vec::Vec::<(std::string::String, std::string::String)>::new(); + for (index, instruction) in transaction.message.instructions.iter().enumerate() { + if let std::option::Option::Some(program_id) = + keys.get(usize::from(instruction.program_id_index)) + { + top_level.push((program_id.clone(), index.to_string())); + } + } + let mut inner_by_parent = std::collections::BTreeMap::< + std::string::String, + std::vec::Vec<(std::string::String, std::string::String)>, + >::new(); + for group in &metadata.inner_instructions { + let parent = group.parent_instruction_index.to_string(); + let entries = inner_by_parent.entry(parent.clone()).or_default(); + for (index, instruction) in group.instructions.iter().enumerate() { + if let std::option::Option::Some(program_id) = + keys.get(usize::from(instruction.program_id_index)) + { + entries.push((program_id.clone(), format!("{parent}/{index}"))); + } + } + } + let mut top_cursor = 0_usize; + let mut inner_cursors = std::collections::BTreeMap::::new(); + let mut stack = + std::vec::Vec::<(std::string::String, std::option::Option)>::new(); + let mut output = std::vec::Vec::with_capacity(metadata.log_messages.len()); + for text in &metadata.log_messages { + let program_line = crate::core_extraction::parse_program_log(text.as_str()); + match program_line { + std::option::Option::Some(( + program_id, + crate::core_extraction::ProgramLogKind::Invoke(depth), + )) => { + let path = if depth == 1 { + let mut resolved = std::option::Option::None; + while top_cursor < top_level.len() { + let candidate = &top_level[top_cursor]; + top_cursor += 1; + if candidate.0 == program_id { + resolved = std::option::Option::Some(candidate.1.clone()); + break; + } + } + resolved + } else { + let parent_path = stack.first().and_then(|entry| return entry.1.clone()); + match parent_path { + std::option::Option::Some(parent) => { + let cursor = inner_cursors.entry(parent.clone()).or_insert(0); + let mut resolved = std::option::Option::None; + if let std::option::Option::Some(entries) = + inner_by_parent.get(parent.as_str()) + { + while *cursor < entries.len() { + let candidate = &entries[*cursor]; + *cursor += 1; + if candidate.0 == program_id { + resolved = std::option::Option::Some(candidate.1.clone()); + break; + } + } + } + resolved + }, + std::option::Option::None => std::option::Option::None, + } + }; + stack.push((program_id.clone(), path.clone())); + output.push((std::option::Option::Some(program_id), path)); + }, + std::option::Option::Some(( + program_id, + crate::core_extraction::ProgramLogKind::Terminal, + )) => { + let path = stack.last().and_then(|entry| { + return if entry.0 == program_id { + entry.1.clone() + } else { + std::option::Option::None + }; + }); + output.push((std::option::Option::Some(program_id.clone()), path)); + if stack.last().map(|entry| return entry.0.as_str()) + == std::option::Option::Some(program_id.as_str()) + { + stack.pop(); + } + }, + std::option::Option::None => { + let current = stack.last().cloned(); + match current { + std::option::Option::Some((program_id, path)) => { + output.push((std::option::Option::Some(program_id), path)) + }, + std::option::Option::None => { + output.push((std::option::Option::None, std::option::Option::None)) + }, + } + }, + } + } + return output; +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum ProgramLogKind { + Invoke(u32), + Terminal, +} + +fn parse_program_log(text: &str) -> std::option::Option<(std::string::String, ProgramLogKind)> { + let remainder = match text.strip_prefix("Program ") { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let split = remainder.split_once(' '); + let (program_id, suffix) = match split { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + if let std::option::Option::Some(depth_text) = + suffix.strip_prefix("invoke [").and_then(|value| return value.strip_suffix(']')) + { + let parse_result = depth_text.parse::(); + if let std::result::Result::Ok(depth) = parse_result { + return std::option::Option::Some(( + program_id.to_string(), + ProgramLogKind::Invoke(depth), + )); + } + } + if suffix == "success" || suffix.starts_with("failed:") { + return std::option::Option::Some((program_id.to_string(), ProgramLogKind::Terminal)); + } + return std::option::Option::None; +} + +fn extract_balance_changes( + transaction: &kb_lib::MdCanonicalTransaction, + metadata: &kb_lib::MdCanonicalTransactionMetadata, +) -> kb_core::Result> { + let keys = crate::core_extraction::resolved_account_keys(transaction); + let mut output = std::vec::Vec::new(); + for index in 0..metadata.pre_balances.len() { + let pre = metadata.pre_balances[index]; + let post = metadata.post_balances[index]; + let delta = i128::from(post) - i128::from(pre); + let account_index_result = u32::try_from(index); + let account_index = match account_index_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "native balance account index overflow: {error}" + ))); + }, + }; + let balance_index_result = u32::try_from(output.len()); + let balance_index = match balance_index_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "balance change index overflow: {error}" + ))); + }, + }; + let account_key = keys.get(index).cloned(); + let insert_result = kb_store::CoreBalanceChangeInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + balance_index, + kb_store::CoreBalanceChangeKind::NativeLamports, + std::option::Option::Some(account_index), + account_key, + std::option::Option::None, + std::option::Option::None, + std::option::Option::Some(serde_json::Value::String(pre.to_string())), + std::option::Option::Some(serde_json::Value::String(post.to_string())), + std::option::Option::Some(serde_json::Value::String(delta.to_string())), + ); + let insert = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + output.push(insert); + } + let mut token_keys = std::collections::BTreeSet::<( + u16, + std::string::String, + std::option::Option, + )>::new(); + for balance in &metadata.pre_token_balances { + token_keys.insert(( + balance.account_index, + balance.mint.clone(), + balance.program_id.clone(), + )); + } + for balance in &metadata.post_token_balances { + token_keys.insert(( + balance.account_index, + balance.mint.clone(), + balance.program_id.clone(), + )); + } + for key in token_keys { + let pre = metadata.pre_token_balances.iter().find(|balance| { + return balance.account_index == key.0 + && balance.mint == key.1 + && balance.program_id == key.2; + }); + let post = metadata.post_token_balances.iter().find(|balance| { + return balance.account_index == key.0 + && balance.mint == key.1 + && balance.program_id == key.2; + }); + let decimals = match (pre, post) { + (std::option::Option::Some(left), std::option::Option::Some(right)) => { + if left.decimals != right.decimals { + return std::result::Result::Err(kb_core::Error::invalid_state( + "token balance decimals changed between pre and post metadata", + )); + } + left.decimals + }, + (std::option::Option::Some(value), std::option::Option::None) => value.decimals, + (std::option::Option::None, std::option::Option::Some(value)) => value.decimals, + (std::option::Option::None, std::option::Option::None) => continue, + }; + let pre_amount = match pre { + std::option::Option::Some(balance) => balance.amount.as_str(), + std::option::Option::None => "0", + }; + let post_amount = match post { + std::option::Option::Some(balance) => balance.amount.as_str(), + std::option::Option::None => "0", + }; + let delta_amount = + crate::core_extraction::subtract_unsigned_decimal(post_amount, pre_amount); + let delta_decimal = + crate::core_extraction::signed_fixed_decimal(delta_amount.as_str(), decimals); + let balance_index_result = u32::try_from(output.len()); + let balance_index = match balance_index_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "token balance change index overflow: {error}" + ))); + }, + }; + let account_key = keys.get(usize::from(key.0)).cloned(); + let owner = post + .and_then(|balance| return balance.owner.clone()) + .or_else(|| return pre.and_then(|balance| return balance.owner.clone())); + let pre_json = match pre { + std::option::Option::Some(balance) => serde_json::json!({ + "amount": balance.amount, + "decimals": balance.decimals, + "decimalAmount": balance.decimal_amount, + "programId": balance.program_id, + }), + std::option::Option::None => serde_json::json!({ + "amount": "0", + "decimals": decimals, + "decimalAmount": crate::core_extraction::signed_fixed_decimal("0", decimals), + "programId": key.2.clone(), + }), + }; + let post_json = match post { + std::option::Option::Some(balance) => serde_json::json!({ + "amount": balance.amount, + "decimals": balance.decimals, + "decimalAmount": balance.decimal_amount, + "programId": balance.program_id, + }), + std::option::Option::None => serde_json::json!({ + "amount": "0", + "decimals": decimals, + "decimalAmount": crate::core_extraction::signed_fixed_decimal("0", decimals), + "programId": key.2.clone(), + }), + }; + let delta_json = serde_json::json!({ + "amount": delta_amount, + "decimals": decimals, + "decimalAmount": delta_decimal, + "programId": key.2.clone(), + }); + let insert_result = kb_store::CoreBalanceChangeInsert::new( + transaction.primary_signature.clone(), + transaction.slot, + balance_index, + kb_store::CoreBalanceChangeKind::TokenAmount, + std::option::Option::Some(u32::from(key.0)), + account_key, + std::option::Option::Some(key.1), + owner, + std::option::Option::Some(pre_json), + std::option::Option::Some(post_json), + std::option::Option::Some(delta_json), + ); + let insert = match insert_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + output.push(insert); + } + return std::result::Result::Ok(output); +} + +fn subtract_unsigned_decimal(left: &str, right: &str) -> std::string::String { + let ordering = crate::core_extraction::compare_unsigned_decimal(left, right); + return match ordering { + std::cmp::Ordering::Equal => "0".to_string(), + std::cmp::Ordering::Greater => crate::core_extraction::subtract_magnitude(left, right), + std::cmp::Ordering::Less => { + format!("-{}", crate::core_extraction::subtract_magnitude(right, left)) + }, + }; +} + +fn compare_unsigned_decimal(left: &str, right: &str) -> std::cmp::Ordering { + let left_trimmed = left.trim_start_matches('0'); + let right_trimmed = right.trim_start_matches('0'); + let left_value = if left_trimmed.is_empty() { "0" } else { left_trimmed }; + let right_value = if right_trimmed.is_empty() { "0" } else { right_trimmed }; + return match left_value.len().cmp(&right_value.len()) { + std::cmp::Ordering::Equal => left_value.cmp(right_value), + ordering => ordering, + }; +} + +fn subtract_magnitude(larger: &str, smaller: &str) -> std::string::String { + let mut result = std::vec::Vec::::new(); + let larger_bytes = larger.as_bytes(); + let smaller_bytes = smaller.as_bytes(); + let mut borrow = 0_i16; + for offset in 0..larger_bytes.len() { + let larger_index = larger_bytes.len() - 1 - offset; + let mut digit = i16::from(larger_bytes[larger_index] - b'0') - borrow; + let smaller_digit = if offset < smaller_bytes.len() { + i16::from(smaller_bytes[smaller_bytes.len() - 1 - offset] - b'0') + } else { + 0 + }; + if digit < smaller_digit { + digit += 10; + borrow = 1; + } else { + borrow = 0; + } + let conversion_result = u8::try_from(digit - smaller_digit); + let value = match conversion_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => 0, + }; + result.push(value + b'0'); + } + while result.len() > 1 && result.last() == std::option::Option::Some(&b'0') { + result.pop(); + } + result.reverse(); + let text_result = std::string::String::from_utf8(result); + return match text_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => "0".to_string(), + }; +} + +fn signed_fixed_decimal(amount: &str, decimals: u8) -> std::string::String { + let negative = amount.starts_with('-'); + let magnitude = match amount.strip_prefix('-') { + std::option::Option::Some(value) => value, + std::option::Option::None => amount, + }; + let scale = usize::from(decimals); + let unsigned = if scale == 0 { + magnitude.to_string() + } else if magnitude.len() > scale { + let split = magnitude.len() - scale; + format!("{}.{}", &magnitude[..split], &magnitude[split..]) + } else { + format!("0.{}{}", "0".repeat(scale - magnitude.len()), magnitude) + }; + if negative && magnitude != "0" { + return format!("-{unsigned}"); + } + return unsigned; +} + +fn hash_json(value: &serde_json::Value) -> kb_core::Result { + let serialization_result = serde_json::to_vec(value); + let bytes = match serialization_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::json(format!( + "cannot serialize core payload for hashing: {error}" + ))); + }, + }; + return std::result::Result::Ok(crate::core_extraction::hash_bytes(bytes.as_slice())); +} + +fn hash_bytes(bytes: &[u8]) -> std::string::String { + let digest = sha2::Sha256::digest(bytes); + let mut output = std::string::String::with_capacity(64); + for byte in digest { + output.push_str(format!("{byte:02x}").as_str()); + } + return output; +} + +#[cfg(test)] +mod tests { + struct TestCoreExtractionStore { + rows: std::vec::Vec, + current: std::sync::atomic::AtomicBool, + persist_calls: std::sync::atomic::AtomicU64, + failure_calls: std::sync::atomic::AtomicU64, + } + + impl TestCoreExtractionStore { + fn new(rows: std::vec::Vec, current: bool) -> Self { + return Self { + rows, + current: std::sync::atomic::AtomicBool::new(current), + persist_calls: std::sync::atomic::AtomicU64::new(0), + failure_calls: std::sync::atomic::AtomicU64::new(0), + }; + } + } + + #[async_trait::async_trait] + impl kb_store::CoreExtractionStore for TestCoreExtractionStore { + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn list_raw_transactions_for_core_extraction( + &self, + _filter: &kb_store::CoreExtractionSelectionFilter, + ) -> kb_core::Result> { + return std::result::Result::Ok(self.rows.clone()); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn is_core_extraction_current( + &self, + _identity: &kb_store::ProcessingLedgerIdentity, + ) -> kb_core::Result { + return std::result::Result::Ok( + self.current.load(std::sync::atomic::Ordering::Acquire), + ); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn persist_core_extraction( + &self, + _bundle: &kb_store::CoreExtractionBundle, + _force_replay: bool, + ) -> kb_core::Result { + self.persist_calls.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + return std::result::Result::Ok(kb_store::InsertOutcome::new(1, 0, 0)); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn mark_core_extraction_failed( + &self, + _failure: &kb_store::CoreExtractionFailure, + ) -> kb_core::Result { + self.failure_calls.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + return std::result::Result::Ok(kb_store::InsertOutcome::new(0, 1, 0)); + } + } + + struct TestCoreExtractionObserver { + cancelled: bool, + events: std::sync::Mutex>, + } + + impl TestCoreExtractionObserver { + fn new(cancelled: bool) -> Self { + return Self { + cancelled, + events: std::sync::Mutex::new(std::vec::Vec::new()), + }; + } + } + + impl crate::CoreExtractionObserver for TestCoreExtractionObserver { + fn on_progress(&self, event: &crate::CoreExtractionProgressEvent) { + let lock_result = self.events.lock(); + if let std::result::Result::Ok(mut events) = lock_result { + events.push(event.clone()); + } + } + + fn is_cancelled(&self) -> bool { + return self.cancelled; + } + } + + fn sample_transaction() -> kb_lib::MdCanonicalTransaction { + let token_pre = match kb_lib::MdCanonicalTokenBalance::new( + 1, + "11111111111111111111111111111111", + std::option::Option::Some("SysvarC1ock11111111111111111111111111111111".to_string()), + std::option::Option::Some("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA".to_string()), + "100", + 2, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("sample token pre balance failed: {error}"), + }; + let token_post = match kb_lib::MdCanonicalTokenBalance::new( + 1, + "11111111111111111111111111111111", + std::option::Option::Some("SysvarC1ock11111111111111111111111111111111".to_string()), + std::option::Option::Some("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA".to_string()), + "125", + 2, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("sample token post balance failed: {error}"), + }; + return kb_lib::MdCanonicalTransaction { + format_version: kb_lib::MD_CANONICAL_TRANSACTION_FORMAT_VERSION, + primary_signature: "2Ana1pUpv2ZbMVkwF5FXapYeBEjdxDatLn7nvJkhgTSXbs59SyZSx866bXirPgj8QQVB57uxHJBG1YFvkRbFj4T".to_string(), + slot: 42, + block_time: std::option::Option::Some(1_700_000_000), + version: kb_lib::MdCanonicalTransactionVersion::Legacy, + signatures: std::vec!["2Ana1pUpv2ZbMVkwF5FXapYeBEjdxDatLn7nvJkhgTSXbs59SyZSx866bXirPgj8QQVB57uxHJBG1YFvkRbFj4T".to_string()], + message: kb_lib::MdCanonicalTransactionMessage { + header: kb_lib::MdCanonicalMessageHeader { + num_required_signatures: 1, + num_readonly_signed_accounts: 0, + num_readonly_unsigned_accounts: 1, + }, + static_account_keys: std::vec![ + "11111111111111111111111111111111".to_string(), + "SysvarC1ock11111111111111111111111111111111".to_string(), + ], + recent_blockhash: "11111111111111111111111111111111".to_string(), + instructions: std::vec![kb_lib::MdCanonicalCompiledInstruction { + program_id_index: 0, + account_indexes: std::vec![1], + data_base64: "AQ==".to_string(), + stack_height: std::option::Option::Some(1), + }], + address_table_lookups: std::vec::Vec::new(), + loaded_addresses: kb_lib::MdCanonicalLoadedAddresses::default(), + }, + metadata: std::option::Option::Some(kb_lib::MdCanonicalTransactionMetadata { + status: kb_lib::MdCanonicalTransactionStatus::Success, + error: std::option::Option::None, + fee: 5000, + pre_balances: std::vec![10_000, 1], + post_balances: std::vec![5_000, 1], + inner_instructions: std::vec![kb_lib::MdCanonicalInnerInstructionGroup { + parent_instruction_index: 0, + instructions: std::vec![kb_lib::MdCanonicalCompiledInstruction { + program_id_index: 0, + account_indexes: std::vec![1], + data_base64: "Ag==".to_string(), + stack_height: std::option::Option::Some(2), + }], + }], + log_messages: std::vec![ + "Program 11111111111111111111111111111111 invoke [1]".to_string(), + "Program 11111111111111111111111111111111 invoke [2]".to_string(), + "Program 11111111111111111111111111111111 success".to_string(), + "Program 11111111111111111111111111111111 success".to_string(), + ], + pre_token_balances: std::vec![token_pre], + post_token_balances: std::vec![token_post], + rewards: std::vec::Vec::new(), + return_data: std::option::Option::None, + compute_units_consumed: std::option::Option::Some(100), + cost_units: std::option::Option::Some(120), + }), + }; + } + + fn raw_row(transaction: &kb_lib::MdCanonicalTransaction) -> kb_store::RawTransactionRow { + let json = match transaction.to_canonical_json() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("sample canonical JSON failed: {error}"), + }; + let hash = match transaction.canonical_json_hash() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("sample canonical hash failed: {error}"), + }; + return kb_store::RawTransactionRow { + id: 1, + signature: transaction.primary_signature.clone(), + slot: 42, + canonical_json: std::option::Option::Some(json), + canonical_json_hash: std::option::Option::Some(hash), + canonical_format_version: 1, + retention_state: kb_store::RawPayloadRetentionState::Full, + processing_state: kb_store::RawPayloadProcessingState::Received, + created_at: chrono::Utc::now(), + updated_at: chrono::Utc::now(), + }; + } + + #[test] + fn extraction_resolves_accounts_instructions_logs_and_balances() { + let transaction = crate::core_extraction::tests::sample_transaction(); + let row = crate::core_extraction::tests::raw_row(&transaction); + let result = crate::extract_raw_transaction_to_core(&row); + let bundle = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("core extraction failed: {error}"), + }; + assert_eq!(bundle.account_keys.len(), 2); + assert_eq!(bundle.instructions.len(), 1); + assert_eq!(bundle.inner_instructions.len(), 1); + assert_eq!(bundle.logs.len(), 4); + assert_eq!(bundle.balance_changes.len(), 3); + assert_eq!(bundle.inner_instructions[0].instruction_path, "0/0"); + assert_eq!(bundle.logs[1].instruction_path.as_deref(), std::option::Option::Some("0/0")); + } + + #[test] + fn extraction_rejects_hash_mismatch() { + let transaction = crate::core_extraction::tests::sample_transaction(); + let mut row = crate::core_extraction::tests::raw_row(&transaction); + row.canonical_json_hash = std::option::Option::Some("bad".to_string()); + let result = crate::extract_raw_transaction_to_core(&row); + assert!(result.is_err()); + } + + #[test] + fn decimal_subtraction_handles_positive_negative_and_zero() { + assert_eq!(crate::core_extraction::subtract_unsigned_decimal("125", "100"), "25"); + assert_eq!(crate::core_extraction::subtract_unsigned_decimal("100", "125"), "-25"); + assert_eq!(crate::core_extraction::subtract_unsigned_decimal("000", "0"), "0"); + assert_eq!(crate::core_extraction::signed_fixed_decimal("-25", 2), "-0.25"); + } + + #[test] + fn static_account_flags_follow_message_header() { + let transaction = crate::core_extraction::tests::sample_transaction(); + let result = crate::core_extraction::extract_account_keys(&transaction); + let keys = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("account extraction failed: {error}"), + }; + assert!(keys[0].signer); + assert!(keys[0].writable); + assert!(!keys[1].signer); + assert!(!keys[1].writable); + } + + #[test] + fn version_zero_loaded_addresses_extend_resolved_account_space() { + let mut transaction = crate::core_extraction::tests::sample_transaction(); + transaction.version = kb_lib::MdCanonicalTransactionVersion::Number(0); + transaction.message.loaded_addresses.writable = + std::vec!["Vote111111111111111111111111111111111111111".to_string(),]; + transaction.message.loaded_addresses.readonly = + std::vec!["Stake11111111111111111111111111111111111111".to_string(),]; + transaction.message.instructions[0].program_id_index = 2; + transaction.message.instructions[0].account_indexes = std::vec![3]; + let result = crate::core_extraction::extract_account_keys(&transaction); + let keys = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("loaded account extraction failed: {error}"), + }; + assert_eq!(keys.len(), 4); + assert_eq!(keys[2].source, kb_store::CoreAccountKeySource::LoadedWritable); + assert!(keys[2].writable); + assert!(!keys[2].signer); + assert_eq!(keys[3].source, kb_store::CoreAccountKeySource::LoadedReadonly); + assert!(!keys[3].writable); + let instruction_result = crate::core_extraction::extract_instructions(&transaction); + let (instructions, _) = match instruction_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + panic!("loaded instruction extraction failed: {error}") + }, + }; + assert_eq!(instructions[0].program_id, keys[2].account_key); + } + + #[test] + fn failed_transaction_preserves_error_json() { + let mut transaction = crate::core_extraction::tests::sample_transaction(); + let metadata = match transaction.metadata.as_mut() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("sample metadata is required"), + }; + metadata.status = kb_lib::MdCanonicalTransactionStatus::Failed; + metadata.error = + std::option::Option::Some(serde_json::json!({"InstructionError": [0, "Custom"]})); + let expected_error = metadata.error.clone(); + let row = crate::core_extraction::tests::raw_row(&transaction); + let result = crate::extract_raw_transaction_to_core(&row); + let bundle = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + panic!("failed transaction extraction failed: {error}") + }, + }; + assert!(bundle.transaction.failed); + assert_eq!(bundle.transaction.err_json, expected_error); + } + + #[test] + fn transaction_without_metadata_keeps_structural_core_rows() { + let mut transaction = crate::core_extraction::tests::sample_transaction(); + transaction.metadata = std::option::Option::None; + let row = crate::core_extraction::tests::raw_row(&transaction); + let result = crate::extract_raw_transaction_to_core(&row); + let bundle = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("metadata-free extraction failed: {error}"), + }; + assert!(!bundle.transaction.failed); + assert_eq!(bundle.account_keys.len(), 2); + assert_eq!(bundle.instructions.len(), 1); + assert!(bundle.logs.is_empty()); + assert!(bundle.balance_changes.is_empty()); + } + + #[test] + fn invalid_instruction_indexes_are_rejected() { + let transaction = crate::core_extraction::tests::sample_transaction(); + let mut row = crate::core_extraction::tests::raw_row(&transaction); + let canonical_json = match row.canonical_json.as_mut() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("sample canonical JSON is required"), + }; + canonical_json["message"]["instructions"][0]["programIdIndex"] = serde_json::json!(99); + let result = crate::extract_raw_transaction_to_core(&row); + assert!(result.is_err()); + let mut account_row = crate::core_extraction::tests::raw_row(&transaction); + let account_json = match account_row.canonical_json.as_mut() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("sample canonical JSON is required"), + }; + account_json["message"]["instructions"][0]["accountIndexes"] = serde_json::json!([99]); + let account_result = crate::extract_raw_transaction_to_core(&account_row); + assert!(account_result.is_err()); + } + + #[test] + fn token_account_created_or_closed_produces_deterministic_changes() { + let mut created = crate::core_extraction::tests::sample_transaction(); + let created_metadata = match created.metadata.as_mut() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("sample metadata is required"), + }; + created_metadata.pre_token_balances.clear(); + let created_row = crate::core_extraction::tests::raw_row(&created); + let created_bundle = match crate::extract_raw_transaction_to_core(&created_row) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("created token extraction failed: {error}"), + }; + let created_token = match created_bundle.balance_changes.last() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("created token balance is required"), + }; + assert_eq!( + created_token.pre_balance_json, + std::option::Option::Some(serde_json::json!({ + "amount": "0", + "decimals": 2, + "decimalAmount": "0.00", + "programId": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA", + })) + ); + let mut closed = crate::core_extraction::tests::sample_transaction(); + let closed_metadata = match closed.metadata.as_mut() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("sample metadata is required"), + }; + closed_metadata.post_token_balances.clear(); + let closed_row = crate::core_extraction::tests::raw_row(&closed); + let closed_bundle = match crate::extract_raw_transaction_to_core(&closed_row) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("closed token extraction failed: {error}"), + }; + let closed_token = match closed_bundle.balance_changes.last() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("closed token balance is required"), + }; + assert_eq!( + closed_token.post_balance_json, + std::option::Option::Some(serde_json::json!({ + "amount": "0", + "decimals": 2, + "decimalAmount": "0.00", + "programId": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA", + })) + ); + } + + #[tokio::test] + async fn current_version_and_hash_are_skipped_without_persistence() { + let transaction = crate::core_extraction::tests::sample_transaction(); + let row = crate::core_extraction::tests::raw_row(&transaction); + let store = + crate::core_extraction::tests::TestCoreExtractionStore::new(std::vec![row], true); + let observer = crate::core_extraction::tests::TestCoreExtractionObserver::new(false); + let request = crate::CoreExtractionRequest { + source: crate::CoreExtractionSource::Pending, + limit: 1, + max_concurrent_extractions: 1, + force_replay: false, + }; + let result = crate::execute_core_extraction(&store, &request, &observer).await; + let summary = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("skip campaign failed: {error}"), + }; + assert_eq!(summary.skipped, 1); + assert_eq!(summary.extracted, 0); + assert_eq!(store.persist_calls.load(std::sync::atomic::Ordering::Acquire), 0); + } + + #[tokio::test] + async fn force_replay_persists_even_when_ledger_is_current() { + let transaction = crate::core_extraction::tests::sample_transaction(); + let row = crate::core_extraction::tests::raw_row(&transaction); + let store = + crate::core_extraction::tests::TestCoreExtractionStore::new(std::vec![row], true); + let observer = crate::core_extraction::tests::TestCoreExtractionObserver::new(false); + let request = crate::CoreExtractionRequest { + source: crate::CoreExtractionSource::Pending, + limit: 1, + max_concurrent_extractions: 1, + force_replay: true, + }; + let result = crate::execute_core_extraction(&store, &request, &observer).await; + let summary = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("force replay campaign failed: {error}"), + }; + assert_eq!(summary.extracted, 1); + assert_eq!(summary.skipped, 0); + assert_eq!(store.persist_calls.load(std::sync::atomic::Ordering::Acquire), 1); + } + + #[tokio::test] + async fn cancellation_before_admission_keeps_every_candidate_not_started() { + let transaction = crate::core_extraction::tests::sample_transaction(); + let row = crate::core_extraction::tests::raw_row(&transaction); + let store = + crate::core_extraction::tests::TestCoreExtractionStore::new(std::vec![row], false); + let observer = crate::core_extraction::tests::TestCoreExtractionObserver::new(true); + let request = crate::CoreExtractionRequest { + source: crate::CoreExtractionSource::Pending, + limit: 1, + max_concurrent_extractions: 1, + force_replay: false, + }; + let result = crate::execute_core_extraction(&store, &request, &observer).await; + let summary = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("cancelled campaign failed: {error}"), + }; + assert!(summary.cancelled); + assert_eq!(summary.started, 0); + assert_eq!(summary.completed, 0); + assert_eq!(summary.not_started, 1); + } +} diff --git a/kb-pipeline/src/lib.rs b/kb-pipeline/src/lib.rs index 7873bfb..9a367fb 100644 --- a/kb-pipeline/src/lib.rs +++ b/kb-pipeline/src/lib.rs @@ -1,5 +1,5 @@ // file: kb-pipeline/src/lib.rs -// version: 1 +// version: 3 #![forbid(unsafe_code)] #![deny(unreachable_pub)] @@ -7,7 +7,35 @@ //! Pipeline orchestration boundary. -/// Returns the pipeline migration status. -pub fn migration_status() -> &'static str { - return "scaffolded"; -} +mod constants; +mod core_extraction; +mod plan; + +/// Canonical tracing target for pipeline orchestration. +pub(crate) use self::constants::TRACING_TARGET; +/// Stable core extraction processor name. +pub use self::core_extraction::CORE_EXTRACTION_PROCESSOR_NAME; +/// Stable core extraction processor version. +pub use self::core_extraction::CORE_EXTRACTION_PROCESSOR_VERSION; +/// Stable core extraction stage code. +pub use self::core_extraction::CORE_EXTRACTION_STAGE; +/// Observer notified during core extraction campaigns. +pub use self::core_extraction::CoreExtractionObserver; +/// Progress event emitted by core extraction campaigns. +pub use self::core_extraction::CoreExtractionProgressEvent; +/// Severity of one core extraction progress event. +pub use self::core_extraction::CoreExtractionProgressLevel; +/// Bounded core extraction request. +pub use self::core_extraction::CoreExtractionRequest; +/// Source selection for a core extraction campaign. +pub use self::core_extraction::CoreExtractionSource; +/// Summary returned by a core extraction campaign. +pub use self::core_extraction::CoreExtractionSummary; +/// Executes one bounded core extraction campaign. +pub use self::core_extraction::execute_core_extraction; +/// Extracts one canonical raw transaction into normalized core rows. +pub use self::core_extraction::extract_raw_transaction_to_core; +/// Pipeline stage identifier. +pub use self::plan::PipelineStage; +/// Replay selection scope. +pub use self::plan::ReplayScope; diff --git a/kb-pipeline/src/plan.rs b/kb-pipeline/src/plan.rs new file mode 100644 index 0000000..df3d052 --- /dev/null +++ b/kb-pipeline/src/plan.rs @@ -0,0 +1,44 @@ +// file: kb-pipeline/src/plan.rs +// version: 2 + +//! Pipeline planning primitives. + +/// Pipeline stage identifier. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum PipelineStage { + /// Ingest raw transactions. + Ingest, + /// Extract generic Solana structures. + Extract, + /// Build program observations. + Observe, + /// Decode protocol events. + Decode, + /// Materialize business events. + Materialize, + /// Aggregate materialized events. + Aggregate, + /// Validate outputs. + Validate, +} + +/// Replay selection scope. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ReplayScope { + /// Optional module name. + pub module_name: std::option::Option, + /// Optional module version. + pub module_version: std::option::Option, + /// Optional program id. + pub program_id: std::option::Option, + /// Optional surface code. + pub surface_code: std::option::Option, + /// Optional 8-byte discriminator in hexadecimal. + pub discriminator_8: std::option::Option, + /// Optional inclusive start slot. + pub slot_start: std::option::Option, + /// Optional inclusive end slot. + pub slot_end: std::option::Option, + /// Whether to ignore existing ledger rows. + pub force: bool, +}