0.7.25

kb_lib/src/local_pipeline_replay.rs

@@ -0,0 +1,244 @@
+// file: kb_lib/src/local_pipeline_replay.rs
+
+//! Local pipeline replay from already persisted raw transaction data.
+//!
+//! This service does not fetch historical transactions from Solana RPC. It
+//! reuses rows already present in `kb_chain_transactions` and replays the
+//! deterministic local pipeline over their signatures.
+
+/// Configuration for a local pipeline replay pass.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct KbLocalPipelineReplayConfig {
+    /// Maximum number of persisted transactions to replay.
+    pub limit: std::option::Option<i64>,
+    /// Whether token metadata backfill should run after local replay.
+    pub refresh_missing_token_metadata: bool,
+    /// Maximum number of missing token metadata rows to resolve.
+    pub token_metadata_limit: std::option::Option<i64>,
+}
+
+impl Default for KbLocalPipelineReplayConfig {
+    fn default() -> Self {
+        return Self {
+            limit: Some(10_000),
+            refresh_missing_token_metadata: false,
+            token_metadata_limit: Some(250),
+        };
+    }
+}
+
+/// Summary of a local pipeline replay pass.
+#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct KbLocalPipelineReplayResult {
+    /// Number of transaction signatures selected for replay.
+    pub selected_transaction_count: usize,
+    /// Number of transactions replayed without a fatal per-signature error.
+    pub replayed_transaction_count: usize,
+    /// Number of transactions that produced a decode error.
+    pub decode_error_count: usize,
+    /// Number of transactions that produced a detect error.
+    pub detect_error_count: usize,
+    /// Number of transactions that produced a trade aggregation error.
+    pub trade_aggregation_error_count: usize,
+    /// Number of transactions that produced a candle aggregation error.
+    pub pair_candle_error_count: usize,
+    /// Number of transactions that produced an analytic signal error.
+    pub analytic_signal_error_count: usize,
+    /// Total decoded events returned by replayed decode calls.
+    pub decoded_event_count: usize,
+    /// Total detection results returned by replayed detect calls.
+    pub detection_count: usize,
+    /// Total trade aggregation results returned by replayed aggregation calls.
+    pub trade_event_count: usize,
+    /// Total candle aggregation results returned by replayed candle calls.
+    pub pair_candle_count: usize,
+    /// Total analytic signal results returned by replayed analytic calls.
+    pub analytic_signal_count: usize,
+    /// Number of token metadata rows updated after replay.
+    pub token_metadata_updated_count: usize,
+    /// Number of pair symbols updated after replay.
+    pub pair_symbol_updated_count: usize,
+    /// Number of errors outside per-signature replay.
+    pub global_error_count: usize,
+}
+
+/// Local pipeline replay service.
+#[derive(Debug, Clone)]
+pub struct KbLocalPipelineReplayService {
+    database: std::sync::Arc<crate::KbDatabase>,
+    http_pool: std::option::Option<std::sync::Arc<crate::HttpEndpointPool>>,
+    http_role: std::string::String,
+}
+
+impl KbLocalPipelineReplayService {
+    /// Creates a new local-only pipeline replay service.
+    pub fn new(database: std::sync::Arc<crate::KbDatabase>) -> Self {
+        return Self {
+            database,
+            http_pool: None,
+            http_role: "local_metadata".to_string(),
+        };
+    }
+
+    /// Creates a new pipeline replay service able to refresh token metadata over Solana HTTP RPC.
+    pub fn new_with_http_pool(
+        http_pool: std::sync::Arc<crate::HttpEndpointPool>,
+        database: std::sync::Arc<crate::KbDatabase>,
+        http_role: std::string::String,
+    ) -> Self {
+        return Self {
+            database,
+            http_pool: Some(http_pool),
+            http_role,
+        };
+    }
+
+    /// Replays the local pipeline from persisted raw chain transaction rows.
+    pub async fn replay_local_pipeline(
+        &self,
+        config: &crate::KbLocalPipelineReplayConfig,
+    ) -> Result<crate::KbLocalPipelineReplayResult, crate::KbError> {
+        let signatures_result = crate::list_chain_transaction_signatures_for_replay(
+            self.database.as_ref(),
+            config.limit,
+        )
+        .await;
+        let signatures = match signatures_result {
+            Ok(signatures) => signatures,
+            Err(error) => return Err(error),
+        };
+        let dex_decode = crate::KbDexDecodeService::new(self.database.clone());
+        let dex_detect = crate::KbDexDetectService::new(self.database.clone());
+        let trade_aggregation = crate::KbTradeAggregationService::new(self.database.clone());
+        let pair_candle_aggregation =
+            crate::KbPairCandleAggregationService::new(self.database.clone());
+        let pair_analytic_signal = crate::KbPairAnalyticSignalService::new(self.database.clone());
+        let mut result = KbLocalPipelineReplayResult {
+            selected_transaction_count: signatures.len(),
+            ..Default::default()
+        };
+        for signature in signatures {
+            tracing::debug!(
+                signature = %signature,
+                "replaying local pipeline for persisted transaction"
+            );
+            let decode_result =
+                dex_decode.decode_transaction_by_signature(signature.as_str()).await;
+            match decode_result {
+                Ok(decoded_events) => {
+                    result.decoded_event_count += decoded_events.len();
+                },
+                Err(error) => {
+                    result.decode_error_count += 1;
+                    tracing::warn!(
+                        signature = %signature,
+                        error = %error,
+                        "local pipeline replay decode step failed"
+                    );
+                    continue;
+                },
+            }
+            let detect_result =
+                dex_detect.detect_transaction_by_signature(signature.as_str()).await;
+            match detect_result {
+                Ok(detections) => {
+                    result.detection_count += detections.len();
+                },
+                Err(error) => {
+                    result.detect_error_count += 1;
+                    tracing::warn!(
+                        signature = %signature,
+                        error = %error,
+                        "local pipeline replay detect step failed"
+                    );
+                    continue;
+                },
+            }
+            let trade_result =
+                trade_aggregation.record_transaction_by_signature(signature.as_str()).await;
+            match trade_result {
+                Ok(trade_results) => {
+                    result.trade_event_count += trade_results.len();
+                },
+                Err(error) => {
+                    result.trade_aggregation_error_count += 1;
+                    tracing::warn!(
+                        signature = %signature,
+                        error = %error,
+                        "local pipeline replay trade aggregation step failed"
+                    );
+                    continue;
+                },
+            }
+            let candle_result = pair_candle_aggregation
+                .record_transaction_by_signature(signature.as_str())
+                .await;
+            match candle_result {
+                Ok(candle_results) => {
+                    result.pair_candle_count += candle_results.len();
+                },
+                Err(error) => {
+                    result.pair_candle_error_count += 1;
+                    tracing::warn!(
+                        signature = %signature,
+                        error = %error,
+                        "local pipeline replay candle aggregation step failed"
+                    );
+                },
+            }
+            let analytic_result =
+                pair_analytic_signal.record_transaction_by_signature(signature.as_str()).await;
+            match analytic_result {
+                Ok(analytic_results) => {
+                    result.analytic_signal_count += analytic_results.len();
+                },
+                Err(error) => {
+                    result.analytic_signal_error_count += 1;
+                    tracing::warn!(
+                        signature = %signature,
+                        error = %error,
+                        "local pipeline replay analytic signal step failed"
+                    );
+                },
+            }
+            result.replayed_transaction_count += 1;
+        }
+        if config.refresh_missing_token_metadata {
+            let metadata_service = match &self.http_pool {
+                Some(http_pool) => crate::KbTokenMetadataBackfillService::new(
+                    http_pool.clone(),
+                    self.database.clone(),
+                    self.http_role.clone(),
+                ),
+                None => crate::KbTokenMetadataBackfillService::new_local(self.database.clone()),
+            };
+            let metadata_result = metadata_service
+                .backfill_missing_token_metadata(config.token_metadata_limit)
+                .await;
+            match metadata_result {
+                Ok(metadata_result) => {
+                    result.token_metadata_updated_count += metadata_result.updated_token_count;
+                },
+                Err(error) => {
+                    result.global_error_count += 1;
+                    tracing::warn!(
+                        error = %error,
+                        "token metadata refresh failed after local pipeline replay"
+                    );
+                },
+            }
+        }
+        return Ok(result);
+    }
+}
+
+/// Replays the local pipeline from persisted raw chain transaction rows.
+pub async fn kb_replay_local_pipeline(
+    database: std::sync::Arc<crate::KbDatabase>,
+    config: &crate::KbLocalPipelineReplayConfig,
+) -> Result<crate::KbLocalPipelineReplayResult, crate::KbError> {
+    let service = crate::KbLocalPipelineReplayService::new(database);
+    return service.replay_local_pipeline(config).await;
+}