khadhroony-bot3/kb-app-demo-desktop/src/demo_decode_replay.rs

// file: kb-app-demo-desktop/src/demo_decode_replay.rs
// version: 30

//! Tauri commands and UI payloads for contextual instruction decode replay.

use tauri::Emitter; // rust-rules: trait-import
use ts_rs::TS; // rust-rules: derive-import

/// One decoder selectable by the decode replay demo.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayDecoderOption.ts"
)]
pub(crate) struct DemoDecodeReplayDecoderOption {
    /// Stable decoder name.
    pub(crate) name: std::string::String,
    /// Stable decoder version.
    pub(crate) version: std::string::String,
    /// Exact supported program identifiers.
    pub(crate) program_ids: std::vec::Vec<std::string::String>,
}

/// One known program suggested by the free-form Program ID field.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayProgramOption.ts"
)]
pub(crate) struct DemoDecodeReplayProgramOption {
    /// Stable program code.
    pub(crate) code: std::string::String,
    /// Canonical program identifier.
    pub(crate) program_id: std::string::String,
}

/// Initial options shown by the contextual decode replay demo.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayOptionsPayload.ts"
)]
pub(crate) struct DemoDecodeReplayOptionsPayload {
    /// Common orchestration version.
    pub(crate) pipeline_version: std::string::String,
    /// Available contextual decoders.
    pub(crate) decoders: std::vec::Vec<crate::DemoDecodeReplayDecoderOption>,
    /// Stable names of materializers currently registered by the demo.
    pub(crate) materializer_names: std::vec::Vec<std::string::String>,
    /// Known programs offered as non-blocking autocomplete suggestions.
    pub(crate) programs: std::vec::Vec<crate::DemoDecodeReplayProgramOption>,
    /// Default bounded selection limit.
    pub(crate) default_limit: u32,
    /// Default maximum concurrent contextual inputs.
    pub(crate) default_max_concurrent_inputs: u32,
    /// Whether one replay campaign is running.
    pub(crate) running: bool,
}

/// UI request for one bounded contextual decode replay campaign.
#[derive(Clone, Debug, serde::Deserialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayRequest.ts"
)]
pub(crate) struct DemoDecodeReplayRequest {
    /// Newline-separated exact transaction signatures.
    pub(crate) signatures_text: std::option::Option<std::string::String>,
    /// Optional exact program identifier.
    pub(crate) program_id: std::option::Option<std::string::String>,
    /// Instruction processing state code, actionable or incomplete_signatures.
    pub(crate) instruction_state: std::string::String,
    /// Newline-separated exact stable instruction paths.
    pub(crate) instruction_paths_text: std::option::Option<std::string::String>,
    /// Explicit selected decoder names.
    pub(crate) decoder_names: std::vec::Vec<std::string::String>,
    /// Maximum selected contextual inputs.
    pub(crate) limit: u32,
    /// Maximum concurrent contextual inputs.
    pub(crate) max_concurrent_inputs: u32,
    /// Whether every compatible decoder must run.
    pub(crate) all_compatible: bool,
    /// Replaces only processor-owned outputs for the same version and input.
    pub(crate) force_replay: bool,
    /// Explicitly authorizes a bounded force replay without exact signatures.
    pub(crate) force_replay_all_matching: bool,
    /// Runs compatible materializers after decoded event persistence.
    pub(crate) materialize_after_decode: bool,
}

/// One progress event emitted to the decode replay window.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayProgressPayload.ts"
)]
pub(crate) struct DemoDecodeReplayProgressPayload {
    /// Stable process-local campaign identifier.
    pub(crate) campaign_id: std::string::String,
    /// RFC 3339 timestamp.
    pub(crate) timestamp: std::string::String,
    /// Stable severity code.
    pub(crate) level: std::string::String,
    /// Human-readable message.
    pub(crate) message: std::string::String,
    /// Number of terminal contextual inputs.
    #[ts(type = "number")]
    pub(crate) completed: u64,
    /// Total selected contextual inputs.
    #[ts(type = "number")]
    pub(crate) total: u64,
}

/// One processor counter row returned to the UI.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeProcessorSummaryPayload.ts"
)]
pub(crate) struct DemoDecodeProcessorSummaryPayload {
    /// Stable processor name.
    pub(crate) processor_name: std::string::String,
    /// Stable processor version.
    pub(crate) processor_version: std::string::String,
    /// Compatible dispatch count.
    #[ts(type = "number")]
    pub(crate) dispatched: u64,
    /// Version and hash skip count.
    #[ts(type = "number")]
    pub(crate) skipped: u64,
    /// Decoded input count.
    #[ts(type = "number")]
    pub(crate) decoded: u64,
    /// Ignored input count.
    #[ts(type = "number")]
    pub(crate) ignored: u64,
    /// Unsupported input count.
    #[ts(type = "number")]
    pub(crate) unsupported: u64,
    /// Failed input count.
    #[ts(type = "number")]
    pub(crate) failed: u64,
    /// Materialized output count.
    #[ts(type = "number")]
    pub(crate) materialized_outputs: u64,
    /// Materialization refusal count.
    #[ts(type = "number")]
    pub(crate) materialization_refused: u64,
}

/// Final UI-safe summary for one contextual decode replay campaign.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplaySummaryPayload.ts"
)]
pub(crate) struct DemoDecodeReplaySummaryPayload {
    /// Stable process-local campaign identifier.
    pub(crate) campaign_id: std::string::String,
    /// Common pipeline version.
    pub(crate) pipeline_version: std::string::String,
    /// Number of selected contextual inputs.
    #[ts(type = "number")]
    pub(crate) selected: u64,
    /// Number admitted to execution.
    #[ts(type = "number")]
    pub(crate) started: u64,
    /// Number reaching a terminal state.
    #[ts(type = "number")]
    pub(crate) completed: u64,
    /// Number with no compatible enabled decoder.
    #[ts(type = "number")]
    pub(crate) unmatched: u64,
    /// Number never started after cancellation.
    #[ts(type = "number")]
    pub(crate) not_started: u64,
    /// Number of failed contextual inputs.
    #[ts(type = "number")]
    pub(crate) failed_inputs: u64,
    /// Whether cancellation was observed.
    pub(crate) cancelled: bool,
    /// Per-processor counters.
    pub(crate) processors: std::vec::Vec<crate::DemoDecodeProcessorSummaryPayload>,
    /// Campaign start timestamp.
    pub(crate) started_at: std::string::String,
    /// Campaign finish timestamp.
    pub(crate) finished_at: std::string::String,
}

/// One aggregated coverage row returned to the UI.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeCoverageSummaryPayload.ts"
)]
pub(crate) struct DemoDecodeCoverageSummaryPayload {
    /// Stable processor name.
    pub(crate) processor_name: std::string::String,
    /// Stable processor version.
    pub(crate) processor_version: std::string::String,
    /// Exact program identifier.
    pub(crate) program_id: std::string::String,
    /// Optional stable surface code.
    pub(crate) surface_code: std::option::Option<std::string::String>,
    /// Stable entry classifier.
    pub(crate) entry_code: std::string::String,
    /// Declared count.
    #[ts(type = "number")]
    pub(crate) declared_count: i64,
    /// Observed count.
    #[ts(type = "number")]
    pub(crate) observed_count: i64,
    /// Recognized count.
    #[ts(type = "number")]
    pub(crate) recognized_count: i64,
    /// Decoded observation count.
    #[ts(type = "number")]
    pub(crate) decoded_count: i64,
    /// Materialized output count.
    #[ts(type = "number")]
    pub(crate) materialized_count: i64,
    /// Error count.
    #[ts(type = "number")]
    pub(crate) error_count: i64,
    /// Unknown or unsupported observation count.
    #[ts(type = "number")]
    pub(crate) unknown_count: i64,
    /// Successful source transaction count.
    #[ts(type = "number")]
    pub(crate) successful_transaction_count: i64,
    /// Failed source transaction count.
    #[ts(type = "number")]
    pub(crate) failed_transaction_count: i64,
}

/// Read-only decode, ledger and coverage diagnostics.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeDiagnosticsPayload.ts"
)]
pub(crate) struct DemoDecodeDiagnosticsPayload {
    /// Decode, materialization and ledger table diagnostics.
    pub(crate) tables: std::vec::Vec<crate::DemoSqlTableSnapshot>,
    /// Aggregated declared and observed coverage.
    pub(crate) coverage: std::vec::Vec<crate::DemoDecodeCoverageSummaryPayload>,
}

/// Bounded read request for committed transaction annotations.
#[derive(Clone, Debug, serde::Deserialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoTransactionAnnotationRequest.ts"
)]
pub(crate) struct DemoTransactionAnnotationRequest {
    /// Optional partial transaction signature.
    pub(crate) signature_contains: std::option::Option<std::string::String>,
    /// Maximum returned rows.
    pub(crate) limit: u32,
}

/// UI-safe committed transaction annotation row.
#[derive(Clone, Debug, serde::Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(
    export,
    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoTransactionAnnotationRow.ts"
)]
pub(crate) struct DemoTransactionAnnotationRow {
    /// Materializer processor version.
    pub(crate) processor_version: std::string::String,
    /// Source decoder and version.
    pub(crate) decoder: std::string::String,
    /// Transaction signature.
    pub(crate) signature: std::string::String,
    /// Decimal slot rendered as text to preserve JSON precision.
    pub(crate) slot: std::string::String,
    /// Stable outer or inner instruction path.
    pub(crate) instruction_path: std::string::String,
    /// Exact Memo generation.
    pub(crate) generation: std::string::String,
    /// Exact Memo Program ID.
    pub(crate) program_id: std::string::String,
    /// Complete bounded UTF-8 Memo text.
    pub(crate) text: std::string::String,
    /// Exact payload byte length.
    pub(crate) payload_length_bytes: u32,
    /// Canonical payload SHA-256.
    pub(crate) payload_sha256: std::string::String,
    /// Runtime-verified ordered signer keys.
    pub(crate) verified_signers: std::vec::Vec<std::string::String>,
    /// Stable materializer idempotence key.
    pub(crate) idempotence_key: std::string::String,
    /// Database creation timestamp.
    pub(crate) created_at: std::string::String,
    /// Database replacement timestamp.
    pub(crate) updated_at: std::string::String,
}

pub(crate) struct DemoDecodeReplayObserver<'a> {
    pub(crate) app_handle: tauri::AppHandle,
    pub(crate) campaign_id: std::string::String,
    pub(crate) cancel_requested: &'a std::sync::atomic::AtomicBool,
}

impl kb_pipeline::DecodeReplayObserver for crate::DemoDecodeReplayObserver<'_> {
    fn on_progress(&self, event: &kb_pipeline::DecodeReplayProgressEvent) {
        let payload = crate::DemoDecodeReplayProgressPayload {
            campaign_id: self.campaign_id.clone(),
            timestamp: event.timestamp.clone(),
            level: event.level.code().to_string(),
            message: event.message.clone(),
            completed: event.completed,
            total: event.total,
        };
        let emit_result =
            self.app_handle
                .emit_to("demo_decode_replay", "demo-decode-replay-progress", payload);
        if let std::result::Result::Err(error) = emit_result {
            tracing::warn!(target: crate::TRACING_TARGET, action = "emit_progress", campaign_id = %self.campaign_id, error = %error, "cannot emit decode replay progress");
        }
    }

    fn is_cancelled(&self) -> bool {
        return self.cancel_requested.load(std::sync::atomic::Ordering::Relaxed);
    }
}

pub(crate) struct DemoDecodeReplayRunGuard<'a> {
    pub(crate) running: &'a std::sync::atomic::AtomicBool,
    pub(crate) campaign_id: &'a std::sync::Mutex<std::option::Option<std::string::String>>,
}

impl std::ops::Drop for crate::DemoDecodeReplayRunGuard<'_> {
    fn drop(&mut self) {
        self.running.store(false, std::sync::atomic::Ordering::Release);
        let lock_result = self.campaign_id.lock();
        if let std::result::Result::Ok(mut active_campaign_id) = lock_result {
            *active_campaign_id = std::option::Option::None;
        }
    }
}

pub(crate) fn register_active_campaign(
    campaign_slot: &std::sync::Mutex<std::option::Option<std::string::String>>,
    campaign_id: &str,
) -> std::result::Result<(), std::string::String> {
    let lock_result = campaign_slot.lock();
    let mut active_campaign_id = match lock_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => {
            return std::result::Result::Err(
                "cannot register active contextual decode campaign".to_string(),
            );
        },
    };
    *active_campaign_id = std::option::Option::Some(campaign_id.to_string());
    return std::result::Result::Ok(());
}

pub(crate) fn available_materializers()
-> std::vec::Vec<std::sync::Arc<dyn kb_lib::MtApiEventMaterializer>> {
    return std::vec![
        std::sync::Arc::new(kb_lib::MtAdminMaterializer),
        std::sync::Arc::new(kb_lib::MtComplianceAuditMaterializer),
        std::sync::Arc::new(kb_lib::MtFeesMaterializer),
        std::sync::Arc::new(kb_lib::MtLifecycleMaterializer),
        std::sync::Arc::new(kb_lib::MtStakingMaterializer),
        std::sync::Arc::new(kb_lib::MtTokenAccountsMaterializer),
        std::sync::Arc::new(kb_lib::MtRiskMaterializer),
        std::sync::Arc::new(kb_lib::MtTransactionAnnotationMaterializer,),
    ];
}

pub(crate) fn available_decoders()
-> std::vec::Vec<std::sync::Arc<dyn kb_lib::DcApiInstructionDecoder>> {
    return std::vec![
        std::sync::Arc::new(kb_lib::DcSolanaCoreDecoder),
        std::sync::Arc::new(kb_lib::DcMetadataMetaplexTokenMetadataDecoder),
        std::sync::Arc::new(kb_lib::DcSplAssociatedTokenAccountDecoder,),
        std::sync::Arc::new(kb_lib::DcSplElgamalRegistryDecoder),
        std::sync::Arc::new(kb_lib::DcSplMemoDecoder),
        std::sync::Arc::new(kb_lib::DcSplTokenDecoder),
        std::sync::Arc::new(kb_lib::DcSplToken2022Decoder),
    ];
}

pub(crate) fn build_demo_decode_replay_pipeline_request(
    request: crate::DemoDecodeReplayRequest,
    campaign_id: std::string::String,
) -> std::result::Result<kb_pipeline::DecodeReplayRequest, std::string::String> {
    let states_result = processing_states(request.instruction_state.as_str());
    let states = match states_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let signatures = split_lines(request.signatures_text.as_deref());
    let instruction_paths = split_lines(request.instruction_paths_text.as_deref());
    let incomplete_signatures = request.instruction_state.trim() == "incomplete_signatures";
    let program_ids = match request.program_id {
        std::option::Option::Some(value) if !value.trim().is_empty() => {
            std::vec![value.trim().to_string()]
        },
        _ => std::vec::Vec::new(),
    };
    let selection_result = kb_store::DecodeSelectionFilter::new(
        signatures,
        states,
        std::option::Option::None,
        std::option::Option::None,
        program_ids,
        instruction_paths,
        incomplete_signatures,
        request.limit,
    );
    let selection = match selection_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
    };
    let dispatch_policy = if request.all_compatible {
        kb_pipeline::DecodeDispatchPolicy::AllCompatible
    } else {
        kb_pipeline::DecodeDispatchPolicy::HighestPriority
    };
    let pipeline_request = kb_pipeline::DecodeReplayRequest {
        campaign_id,
        selection,
        decoder_names: request.decoder_names,
        dispatch_policy,
        max_concurrent_inputs: request.max_concurrent_inputs,
        force_replay: request.force_replay,
        force_replay_all_matching: request.force_replay_all_matching,
        materialize_after_decode: request.materialize_after_decode,
    };
    let validation_result = pipeline_request.validate();
    if let std::result::Result::Err(error) = validation_result {
        return std::result::Result::Err(error.to_string());
    }
    tracing::debug!(
        target: crate::TRACING_TARGET,
        action = "build_request",
        campaign_id = %pipeline_request.campaign_id,
        signature_count = pipeline_request.selection.signatures.len(),
        signature_sample = ?text_sample(pipeline_request.selection.signatures.as_slice(), 5),
        processing_states = ?pipeline_request.selection.processing_states,
        program_ids = ?pipeline_request.selection.program_ids,
        instruction_paths = ?pipeline_request.selection.instruction_paths,
        incomplete_signatures = pipeline_request.selection.incomplete_signatures,
        limit = pipeline_request.selection.limit,
        decoder_names = ?pipeline_request.decoder_names,
        dispatch_policy = ?pipeline_request.dispatch_policy,
        max_concurrent_inputs = pipeline_request.max_concurrent_inputs,
        force_replay = pipeline_request.force_replay,
        force_replay_all_matching = pipeline_request.force_replay_all_matching,
        materialize_after_decode = pipeline_request.materialize_after_decode,
        "built normalized contextual decode replay request"
    );
    return std::result::Result::Ok(pipeline_request);
}

pub(crate) fn optional_line_count(value: &std::option::Option<std::string::String>) -> usize {
    return match value {
        std::option::Option::Some(text) => {
            text.lines().filter(|line| return !line.trim().is_empty()).count()
        },
        std::option::Option::None => 0,
    };
}

pub(crate) fn text_sample(values: &[std::string::String], limit: usize) -> std::vec::Vec<&str> {
    return values.iter().take(limit).map(std::string::String::as_str).collect();
}

fn processing_states(
    value: &str,
) -> std::result::Result<std::vec::Vec<kb_store::CoreInstructionProcessingState>, std::string::String>
{
    return match value.trim() {
        "incomplete_signatures" | "actionable" => std::result::Result::Ok(std::vec![
            kb_store::CoreInstructionProcessingState::Pending,
            kb_store::CoreInstructionProcessingState::Failed,
            kb_store::CoreInstructionProcessingState::ReplayRequested,
        ]),
        "pending" => {
            std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Pending])
        },
        "failed" => {
            std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Failed])
        },
        "replay_requested" => std::result::Result::Ok(std::vec![
            kb_store::CoreInstructionProcessingState::ReplayRequested
        ]),
        "decoded" => {
            std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Decoded])
        },
        "ignored" => {
            std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Ignored])
        },
        "materialized" => std::result::Result::Ok(std::vec![
            kb_store::CoreInstructionProcessingState::Materialized
        ]),
        _ => std::result::Result::Err("unsupported instruction processing state".to_string()),
    };
}

fn split_lines(text: std::option::Option<&str>) -> std::vec::Vec<std::string::String> {
    let source = match text {
        std::option::Option::Some(value) => value,
        std::option::Option::None => "",
    };
    let mut unique = std::collections::BTreeSet::<std::string::String>::new();
    let mut output = std::vec::Vec::new();
    for line in source.lines() {
        let value = line.trim();
        if value.is_empty() {
            continue;
        }
        if unique.insert(value.to_string()) {
            output.push(value.to_string());
        }
    }
    return output;
}

pub(crate) fn demo_decode_replay_summary_payload(
    summary: kb_pipeline::DecodeReplaySummary,
) -> crate::DemoDecodeReplaySummaryPayload {
    return crate::DemoDecodeReplaySummaryPayload {
        campaign_id: summary.campaign_id,
        pipeline_version: summary.pipeline_version,
        selected: summary.selected,
        started: summary.started,
        completed: summary.completed,
        unmatched: summary.unmatched,
        not_started: summary.not_started,
        failed_inputs: summary.failed_inputs,
        cancelled: summary.cancelled,
        processors: summary
            .processors
            .into_iter()
            .map(|processor| {
                return crate::DemoDecodeProcessorSummaryPayload {
                    processor_name: processor.processor_name,
                    processor_version: processor.processor_version,
                    dispatched: processor.dispatched,
                    skipped: processor.skipped,
                    decoded: processor.decoded,
                    ignored: processor.ignored,
                    unsupported: processor.unsupported,
                    failed: processor.failed,
                    materialized_outputs: processor.materialized_outputs,
                    materialization_refused: processor.materialization_refused,
                };
            })
            .collect(),
        started_at: summary.started_at,
        finished_at: summary.finished_at,
    };
}

pub(crate) fn coverage_payload(
    value: kb_store::DecodeCoverageSummaryRow,
) -> crate::DemoDecodeCoverageSummaryPayload {
    return crate::DemoDecodeCoverageSummaryPayload {
        processor_name: value.processor_name,
        processor_version: value.processor_version,
        program_id: value.program_id,
        surface_code: value.surface_code,
        entry_code: value.entry_code,
        declared_count: value.declared_count,
        observed_count: value.observed_count,
        recognized_count: value.recognized_count,
        decoded_count: value.decoded_count,
        materialized_count: value.materialized_count,
        error_count: value.error_count,
        unknown_count: value.unknown_count,
        successful_transaction_count: value.successful_transaction_count,
        failed_transaction_count: value.failed_transaction_count,
    };
}

pub(crate) fn annotation_payload(
    row: kb_store::MaterializedEventQueryRow,
) -> std::result::Result<crate::DemoTransactionAnnotationRow, std::string::String> {
    if row.processor_name != "transaction_annotations"
        || row.materialized_family != "transaction_annotation"
    {
        return std::result::Result::Err(
            "materialized row is not a transaction annotation".to_string(),
        );
    }
    let payload = &row.payload_json;
    let byte_length = match payload.get("payloadLengthBytes").and_then(serde_json::Value::as_u64) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => {
            return std::result::Result::Err(
                "transaction annotation payloadLengthBytes is absent".to_string(),
            );
        },
    };
    let byte_length_result = u32::try_from(byte_length);
    let payload_length_bytes = match byte_length_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            return std::result::Result::Err(format!(
                "transaction annotation payloadLengthBytes is not UI-safe: {error}"
            ));
        },
    };
    let signers = match payload.get("signers") {
        std::option::Option::Some(value) => value,
        std::option::Option::None => {
            return std::result::Result::Err(
                "transaction annotation signers are absent".to_string(),
            );
        },
    };
    let verified_values = match signers.get("verified").and_then(serde_json::Value::as_array) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => {
            return std::result::Result::Err(
                "transaction annotation verified signers are invalid".to_string(),
            );
        },
    };
    let mut verified_signers = std::vec::Vec::with_capacity(verified_values.len());
    for value in verified_values {
        match value.as_str() {
            std::option::Option::Some(signer) if !signer.trim().is_empty() => {
                verified_signers.push(signer.to_string());
            },
            _ => {
                return std::result::Result::Err(
                    "transaction annotation verified signer is invalid".to_string(),
                );
            },
        }
    }
    let instruction_path = match required_annotation_text(payload, "instructionPath") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let generation = match required_annotation_text(payload, "generation") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let program_id = match required_annotation_text(payload, "programId") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let text = match required_annotation_text(payload, "text") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let payload_sha256 = match required_annotation_text(payload, "payloadSha256") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let idempotence_key = match required_annotation_text(payload, "idempotenceKey") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return std::result::Result::Ok(crate::DemoTransactionAnnotationRow {
        processor_version: row.processor_version,
        decoder: format!("{}@{}", row.source_decoder_name, row.source_decoder_version),
        signature: row.signature,
        slot: row.slot.to_string(),
        instruction_path,
        generation,
        program_id,
        text,
        payload_length_bytes,
        payload_sha256,
        verified_signers,
        idempotence_key,
        created_at: row.created_at,
        updated_at: row.updated_at,
    });
}

fn required_annotation_text(
    payload: &serde_json::Value,
    field: &str,
) -> std::result::Result<std::string::String, std::string::String> {
    return match payload.get(field).and_then(serde_json::Value::as_str) {
        std::option::Option::Some(value) => std::result::Result::Ok(value.to_string()),
        std::option::Option::None => {
            std::result::Result::Err(format!("transaction annotation {field} is absent"))
        },
    };
}

#[cfg(test)]
mod tests {
    fn ata_observation(entry: &str) -> kb_lib::DcApiDecodedObservation {
        return kb_lib::DcApiDecodedObservation {
            event_key: format!("ata:{entry}:0"),
            event: kb_lib::MdDecodedProtocolEvent {
                signature: kb_lib::MdSignature("signature".to_string()),
                slot: kb_lib::MdSlot(1),
                instruction_path: kb_lib::MdInstructionPath("0".to_string()),
                program_id: kb_lib::MdProgramId(
                    kb_program_ids::ASSOCIATED_TOKEN_PROGRAM_ID.to_string(),
                ),
                protocol_code: kb_lib::MdProtocolCode("spl_associated_token_account".to_string()),
                surface_code: kb_lib::MdSurfaceCode("spl_associated_token_account".to_string()),
                event_code: kb_lib::MdEventCode(format!("spl_associated_token_account.{entry}")),
                event_name: kb_lib::MdEventName(entry.to_string()),
                event_family: kb_lib::MdEventFamily::Lifecycle,
                source_kind: kb_lib::MdEventSourceKind::Instruction,
                confidence: kb_lib::MdDecoderConfidence::ManualExact,
            },
            payload_json: serde_json::json!({}),
            transaction_failed: false,
            transaction_error: std::option::Option::None,
            observation_committed: true,
            proof: kb_lib::DcApiDecoderProof {
                kind: kb_lib::DcApiDecoderProofKind::Manual,
                confidence: kb_lib::MdDecoderConfidence::ManualExact,
                evidence: std::vec!["fixture".to_string()],
            },
        };
    }

    #[test]
    fn native_memo_token2022_elgamal_classic_token_and_ata_decoders_are_registered() {
        let decoders = crate::available_decoders();
        assert_eq!(decoders.len(), 7);
        let mut names = decoders
            .iter()
            .map(|decoder| return decoder.identity().name)
            .collect::<std::vec::Vec<_>>();
        names.sort();
        let mut expected_names = std::vec![
            "metadata_metaplex_token_metadata".to_string(),
            "solana_native_classifier".to_string(),
            "spl_associated_token_account".to_string(),
            "spl_elgamal_registry".to_string(),
            "spl_memo".to_string(),
            "spl_token".to_string(),
            "spl_token2022".to_string(),
        ];
        expected_names.sort();
        assert_eq!(names, expected_names);
    }

    #[test]
    fn runtime_materializer_registry_is_complete_for_current_instructional_surfaces() {
        let materializers = crate::available_materializers();
        assert_eq!(materializers.len(), 8);
        let names = materializers
            .iter()
            .map(|materializer| return materializer.identity().name)
            .collect::<std::vec::Vec<_>>();
        assert_eq!(
            names,
            std::vec![
                "solana_native_admin".to_string(),
                "solana_native_compliance_audit".to_string(),
                "fees".to_string(),
                "solana_native_lifecycle".to_string(),
                "solana_native_staking".to_string(),
                "spl_token_accounts".to_string(),
                "spl_token_risk".to_string(),
                "transaction_annotations".to_string(),
            ]
        );
    }

    #[test]
    fn ata_materializer_ownership_is_exact_in_the_runtime_registry() {
        for (entry, expected) in [
            ("create", std::vec!["spl_token_accounts".to_string()]),
            (
                "recover_nested",
                std::vec!["spl_token_accounts".to_string(), "spl_token_risk".to_string(),],
            ),
        ] {
            let observation = ata_observation(entry);
            let owners = crate::available_materializers()
                .iter()
                .filter(|materializer| return materializer.accepts_observation(&observation))
                .map(|materializer| return materializer.identity().name)
                .collect::<std::vec::Vec<_>>();
            assert_eq!(owners, expected);
        }
    }

    #[test]
    fn line_selection_is_trimmed_and_deduplicated() {
        let values = super::split_lines(std::option::Option::Some(" a \n\na\nb\n"));
        assert_eq!(values, std::vec!["a".to_string(), "b".to_string()]);
    }

    fn request(
        signatures_text: std::option::Option<&str>,
        force_replay: bool,
        force_replay_all_matching: bool,
        materialize_after_decode: bool,
    ) -> crate::DemoDecodeReplayRequest {
        return crate::DemoDecodeReplayRequest {
            signatures_text: signatures_text.map(|value| return value.to_string()),
            program_id: std::option::Option::None,
            instruction_state: "actionable".to_string(),
            instruction_paths_text: std::option::Option::None,
            decoder_names: std::vec!["solana_native_classifier".to_string()],
            limit: 10,
            max_concurrent_inputs: 2,
            all_compatible: false,
            force_replay,
            force_replay_all_matching,
            materialize_after_decode,
        };
    }

    #[test]
    fn force_replay_requires_signatures_or_all_matching_authorization() {
        let result = crate::build_demo_decode_replay_pipeline_request(
            request(std::option::Option::None, true, false, false),
            "decode-test".to_string(),
        );
        assert!(result.is_err());
    }

    #[test]
    fn force_replay_all_matching_is_forwarded() {
        let result = crate::build_demo_decode_replay_pipeline_request(
            request(std::option::Option::None, true, true, false),
            "decode-test".to_string(),
        );
        let request = match result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("request mapping failed: {error}"),
        };
        assert!(request.force_replay_all_matching);
        assert!(request.selection.signatures.is_empty());
    }

    #[test]
    fn run_guard_clears_active_campaign_state() {
        let running = std::sync::atomic::AtomicBool::new(true);
        let campaign_id =
            std::sync::Mutex::new(std::option::Option::Some("decode-test".to_string()));
        {
            let _guard = crate::DemoDecodeReplayRunGuard {
                running: &running,
                campaign_id: &campaign_id,
            };
        }
        assert!(!running.load(std::sync::atomic::Ordering::Acquire));
        let lock_result = campaign_id.lock();
        let active_campaign_id = match lock_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => panic!("campaign id mutex must not be poisoned"),
        };
        assert!(active_campaign_id.is_none());
    }

    #[test]
    fn register_active_campaign_sets_slot_without_exposing_guard() {
        let campaign_id = std::sync::Mutex::new(std::option::Option::None);
        let result = crate::register_active_campaign(&campaign_id, "decode-test");
        assert!(result.is_ok());
        let lock_result = campaign_id.lock();
        let active_campaign_id = match lock_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => panic!("campaign id mutex must not be poisoned"),
        };
        assert_eq!(active_campaign_id.as_deref(), std::option::Option::Some("decode-test"));
    }

    #[test]
    fn incomplete_signature_state_uses_actionable_states_and_expansion() {
        let mut request = request(std::option::Option::None, false, false, true);
        request.instruction_state = "incomplete_signatures".to_string();
        let result = crate::build_demo_decode_replay_pipeline_request(
            request,
            "decode-incomplete".to_string(),
        );
        let pipeline_request = match result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("request mapping failed: {error}"),
        };
        assert!(pipeline_request.selection.incomplete_signatures);
        assert_eq!(pipeline_request.selection.processing_states.len(), 3);
        assert!(!pipeline_request.force_replay);
        assert!(pipeline_request.materialize_after_decode);
    }

    #[test]
    fn actionable_state_contains_pending_failed_and_replay_requested() {
        let result = super::processing_states("actionable");
        let states = match result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("state mapping failed: {error}"),
        };
        assert_eq!(states.len(), 3);
    }

    #[test]
    fn committed_annotation_row_is_mapped_to_ui_safe_contract() {
        let row = kb_store::MaterializedEventQueryRow {
            processor_name: "transaction_annotations".to_string(),
            processor_version: "0.4.3".to_string(),
            input_key: "input".to_string(),
            output_key: "output".to_string(),
            source_event_key: "memo:0".to_string(),
            source_decoder_name: "spl_memo".to_string(),
            source_decoder_version: "0.4.3".to_string(),
            signature: "signature".to_string(),
            slot: u64::MAX,
            materialized_family: "transaction_annotation".to_string(),
            payload_json: serde_json::json!({
                "instructionPath": "1/0",
                "generation": "v4",
                "programId": kb_program_ids::SPL_MEMO_V4_PROGRAM_ID,
                "text": "annotation",
                "payloadLengthBytes": 10,
                "payloadSha256": "11",
                "signers": {"verified": ["signer"]},
                "idempotenceKey": "stable"
            }),
            created_at: "created".to_string(),
            updated_at: "updated".to_string(),
        };
        let result = crate::annotation_payload(row);
        let mapped = match result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("annotation mapping failed: {error}"),
        };
        assert_eq!(mapped.slot, u64::MAX.to_string());
        assert_eq!(mapped.text, "annotation");
        assert_eq!(mapped.verified_signers, std::vec!["signer".to_string()]);
    }

    #[test]
    fn malformed_annotation_payload_fails_closed() {
        let row = kb_store::MaterializedEventQueryRow {
            processor_name: "transaction_annotations".to_string(),
            processor_version: "0.4.3".to_string(),
            input_key: "input".to_string(),
            output_key: "output".to_string(),
            source_event_key: "memo:0".to_string(),
            source_decoder_name: "spl_memo".to_string(),
            source_decoder_version: "0.4.3".to_string(),
            signature: "signature".to_string(),
            slot: 1,
            materialized_family: "transaction_annotation".to_string(),
            payload_json: serde_json::json!({}),
            created_at: "created".to_string(),
            updated_at: "updated".to_string(),
        };
        assert!(crate::annotation_payload(row).is_err());
    }
}