0.1.0

migration/khadhroony-bot2-reference/kb_store_pg/src/replay_candidates.rs

@@ -0,0 +1,380 @@
+// file: kb_store_pg/src/replay_candidates.rs
+// version: 2
+
+//! Read-only replay candidate filters and PostgreSQL result rows.
+
+/// Maximum number of rows returned by one replay candidate query.
+pub const MAX_REPLAY_CANDIDATE_ROWS: u32 = 5_000;
+
+/// Program occurrence scope used while filtering replay candidates.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum PostgresReplayProgramScope {
+    /// Match outer instructions, inner instructions or reliably linked logs.
+    Any,
+    /// Match only top-level instructions.
+    Outer,
+    /// Match only inner instructions.
+    Inner,
+    /// Match only logs with a reliably linked program id.
+    Logs,
+}
+
+impl PostgresReplayProgramScope {
+    /// Returns the stable SQL code for this scope.
+    pub fn as_sql(self) -> &'static str {
+        return match self {
+            Self::Any => "any",
+            Self::Outer => "outer",
+            Self::Inner => "inner",
+            Self::Logs => "logs",
+        };
+    }
+}
+
+/// Core entity kind used while filtering replay candidates.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum PostgresReplayEntityKind {
+    /// SPL or Token-2022 mint address.
+    Mint,
+    /// Token account owner address.
+    Owner,
+    /// Native or token account address.
+    AccountKey,
+}
+
+impl PostgresReplayEntityKind {
+    /// Returns the stable SQL code for this entity kind.
+    pub fn as_sql(self) -> &'static str {
+        return match self {
+            Self::Mint => "mint",
+            Self::Owner => "owner",
+            Self::AccountKey => "account_key",
+        };
+    }
+}
+
+/// Bounded read-only filter for transaction replay candidates.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PostgresReplayTransactionFilter {
+    /// Optional partial signature search.
+    pub signature_contains: std::option::Option<std::string::String>,
+    /// Optional inclusive minimum slot.
+    pub min_slot: std::option::Option<u64>,
+    /// Optional inclusive maximum slot.
+    pub max_slot: std::option::Option<u64>,
+    /// Optional raw processing state.
+    pub raw_processing_state: std::option::Option<std::string::String>,
+    /// Optional latest core extraction ledger status, including `not_started`.
+    pub ledger_status: std::option::Option<std::string::String>,
+    /// Optional exact program id.
+    pub program_id: std::option::Option<std::string::String>,
+    /// Program occurrence scope.
+    pub program_scope: crate::PostgresReplayProgramScope,
+    /// Optional core entity kind.
+    pub entity_kind: std::option::Option<crate::PostgresReplayEntityKind>,
+    /// Optional exact entity value.
+    pub entity_value: std::option::Option<std::string::String>,
+    /// Maximum returned rows.
+    pub limit: u32,
+    /// Orders newest slots first when true.
+    pub newest_first: bool,
+}
+
+impl PostgresReplayTransactionFilter {
+    /// Creates and validates a bounded transaction candidate filter.
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        signature_contains: std::option::Option<std::string::String>,
+        min_slot: std::option::Option<u64>,
+        max_slot: std::option::Option<u64>,
+        raw_processing_state: std::option::Option<std::string::String>,
+        ledger_status: std::option::Option<std::string::String>,
+        program_id: std::option::Option<std::string::String>,
+        program_scope: crate::PostgresReplayProgramScope,
+        entity_kind: std::option::Option<crate::PostgresReplayEntityKind>,
+        entity_value: std::option::Option<std::string::String>,
+        limit: u32,
+        newest_first: bool,
+    ) -> kb_core::Result<Self> {
+        let signature_contains_value = trim_optional_text(signature_contains);
+        let raw_processing_state_value = trim_optional_text(raw_processing_state);
+        let ledger_status_value = trim_optional_text(ledger_status);
+        let program_id_value = trim_optional_text(program_id);
+        let entity_value_value = trim_optional_text(entity_value);
+        let slot_result = validate_slot_range(min_slot, max_slot);
+        if let std::result::Result::Err(error) = slot_result {
+            return std::result::Result::Err(error);
+        }
+        let limit_result = validate_limit(limit);
+        if let std::result::Result::Err(error) = limit_result {
+            return std::result::Result::Err(error);
+        }
+        let raw_state_result = validate_optional_code(
+            raw_processing_state_value.as_deref(),
+            &["received", "core_extracted", "decoded", "materialized", "failed"],
+            "raw processing state",
+        );
+        if let std::result::Result::Err(error) = raw_state_result {
+            return std::result::Result::Err(error);
+        }
+        let ledger_status_result = validate_optional_code(
+            ledger_status_value.as_deref(),
+            &["not_started", "running", "succeeded", "failed"],
+            "ledger status",
+        );
+        if let std::result::Result::Err(error) = ledger_status_result {
+            return std::result::Result::Err(error);
+        }
+        if entity_kind.is_some() != entity_value_value.is_some() {
+            return std::result::Result::Err(kb_core::Error::db(
+                "replay entity kind and value must either both be present or both be absent",
+            ));
+        }
+        return std::result::Result::Ok(Self {
+            signature_contains: signature_contains_value,
+            min_slot,
+            max_slot,
+            raw_processing_state: raw_processing_state_value,
+            ledger_status: ledger_status_value,
+            program_id: program_id_value,
+            program_scope,
+            entity_kind,
+            entity_value: entity_value_value,
+            limit,
+            newest_first,
+        });
+    }
+}
+
+/// One raw transaction candidate enriched with core and ledger diagnostics.
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+pub struct PostgresReplayTransactionCandidate {
+    /// Canonical transaction signature.
+    pub signature: std::string::String,
+    /// Transaction slot.
+    pub slot: i64,
+    /// Current raw processing state.
+    pub raw_processing_state: std::string::String,
+    /// Current raw retention state.
+    pub retention_state: std::string::String,
+    /// Whether a core transaction row exists.
+    pub has_core_transaction: bool,
+    /// Core transaction failure flag when a core row exists.
+    pub transaction_failed: std::option::Option<bool>,
+    /// Latest core extraction ledger status or `not_started`.
+    pub ledger_status: std::string::String,
+    /// Latest core extraction processor version when available.
+    pub processor_version: std::option::Option<std::string::String>,
+    /// Latest core extraction attempt count.
+    pub attempt_count: i32,
+    /// Number of top-level instructions.
+    pub outer_instruction_count: i64,
+    /// Number of inner instructions.
+    pub inner_instruction_count: i64,
+    /// Number of distinct top-level programs.
+    pub outer_program_count: i64,
+    /// Number of distinct inner programs.
+    pub inner_program_count: i64,
+    /// Raw row update timestamp rendered by PostgreSQL.
+    pub updated_at: std::string::String,
+}
+
+/// Bounded read-only filter for program summaries.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PostgresReplayProgramFilter {
+    /// Optional partial program id search.
+    pub program_id_contains: std::option::Option<std::string::String>,
+    /// Maximum returned rows.
+    pub limit: u32,
+}
+
+impl PostgresReplayProgramFilter {
+    /// Creates and validates a bounded program summary filter.
+    pub fn new(
+        program_id_contains: std::option::Option<std::string::String>,
+        limit: u32,
+    ) -> kb_core::Result<Self> {
+        let limit_result = validate_limit(limit);
+        if let std::result::Result::Err(error) = limit_result {
+            return std::result::Result::Err(error);
+        }
+        return std::result::Result::Ok(Self {
+            program_id_contains: trim_optional_text(program_id_contains),
+            limit,
+        });
+    }
+}
+
+/// Aggregated program occurrences across outer, inner and linked logs.
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+pub struct PostgresReplayProgramSummary {
+    /// Program id.
+    pub program_id: std::string::String,
+    /// Number of distinct transactions containing the program.
+    pub transaction_count: i64,
+    /// Number of top-level instruction occurrences.
+    pub outer_instruction_count: i64,
+    /// Number of inner instruction occurrences.
+    pub inner_instruction_count: i64,
+    /// Number of reliably linked log occurrences.
+    pub log_count: i64,
+    /// Lowest observed slot.
+    pub min_slot: i64,
+    /// Highest observed slot.
+    pub max_slot: i64,
+}
+
+/// Bounded read-only filter for core entity summaries.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PostgresReplayEntityFilter {
+    /// Entity kind to aggregate.
+    pub entity_kind: crate::PostgresReplayEntityKind,
+    /// Optional partial entity value search.
+    pub entity_value_contains: std::option::Option<std::string::String>,
+    /// Maximum returned rows.
+    pub limit: u32,
+}
+
+impl PostgresReplayEntityFilter {
+    /// Creates and validates a bounded entity summary filter.
+    pub fn new(
+        entity_kind: crate::PostgresReplayEntityKind,
+        entity_value_contains: std::option::Option<std::string::String>,
+        limit: u32,
+    ) -> kb_core::Result<Self> {
+        let limit_result = validate_limit(limit);
+        if let std::result::Result::Err(error) = limit_result {
+            return std::result::Result::Err(error);
+        }
+        return std::result::Result::Ok(Self {
+            entity_kind,
+            entity_value_contains: trim_optional_text(entity_value_contains),
+            limit,
+        });
+    }
+}
+
+/// Aggregated mint, owner or account-key occurrences from core tables.
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+pub struct PostgresReplayEntitySummary {
+    /// Stable entity kind code.
+    pub entity_kind: std::string::String,
+    /// Mint, owner or account-key address.
+    pub entity_value: std::string::String,
+    /// Number of distinct transactions containing the entity.
+    pub transaction_count: i64,
+    /// Total number of core-table occurrences.
+    pub occurrence_count: i64,
+    /// Lowest observed slot.
+    pub min_slot: i64,
+    /// Highest observed slot.
+    pub max_slot: i64,
+}
+
+fn trim_optional_text(
+    value: std::option::Option<std::string::String>,
+) -> std::option::Option<std::string::String> {
+    return match value {
+        std::option::Option::Some(text) => {
+            let trimmed = text.trim();
+            if trimmed.is_empty() {
+                return std::option::Option::None;
+            }
+            std::option::Option::Some(trimmed.to_string())
+        },
+        std::option::Option::None => std::option::Option::None,
+    };
+}
+
+fn validate_slot_range(
+    min_slot: std::option::Option<u64>,
+    max_slot: std::option::Option<u64>,
+) -> kb_core::Result<()> {
+    if let (std::option::Option::Some(minimum), std::option::Option::Some(maximum)) =
+        (min_slot, max_slot)
+    {
+        if minimum > maximum {
+            return std::result::Result::Err(kb_core::Error::db(
+                "replay candidate minimum slot must not exceed maximum slot",
+            ));
+        }
+    }
+    return std::result::Result::Ok(());
+}
+
+fn validate_limit(limit: u32) -> kb_core::Result<()> {
+    if limit == 0 || limit > crate::MAX_REPLAY_CANDIDATE_ROWS {
+        return std::result::Result::Err(kb_core::Error::db(format!(
+            "replay candidate limit must be between 1 and {}",
+            crate::MAX_REPLAY_CANDIDATE_ROWS
+        )));
+    }
+    return std::result::Result::Ok(());
+}
+
+fn validate_optional_code(
+    value: std::option::Option<&str>,
+    allowed: &[&str],
+    label: &str,
+) -> kb_core::Result<()> {
+    let selected = match value {
+        std::option::Option::Some(code) => code,
+        std::option::Option::None => return std::result::Result::Ok(()),
+    };
+    for allowed_code in allowed {
+        if selected == *allowed_code {
+            return std::result::Result::Ok(());
+        }
+    }
+    return std::result::Result::Err(kb_core::Error::db(format!(
+        "unsupported {label}: {selected}"
+    )));
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn transaction_filter_rejects_inverted_slots() {
+        let result = crate::PostgresReplayTransactionFilter::new(
+            std::option::Option::None,
+            std::option::Option::Some(20),
+            std::option::Option::Some(10),
+            std::option::Option::None,
+            std::option::Option::None,
+            std::option::Option::None,
+            crate::PostgresReplayProgramScope::Any,
+            std::option::Option::None,
+            std::option::Option::None,
+            100,
+            true,
+        );
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn transaction_filter_requires_complete_entity_pair() {
+        let result = crate::PostgresReplayTransactionFilter::new(
+            std::option::Option::None,
+            std::option::Option::None,
+            std::option::Option::None,
+            std::option::Option::None,
+            std::option::Option::None,
+            std::option::Option::None,
+            crate::PostgresReplayProgramScope::Any,
+            std::option::Option::Some(crate::PostgresReplayEntityKind::Mint),
+            std::option::Option::None,
+            100,
+            true,
+        );
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn program_filter_rejects_limit_above_maximum() {
+        let result = crate::PostgresReplayProgramFilter::new(
+            std::option::Option::None,
+            crate::MAX_REPLAY_CANDIDATE_ROWS + 1,
+        );
+        assert!(result.is_err());
+    }
+}