0.7.34

kb_lib/src/local_pipeline_replay.rs

@@ -45,6 +45,8 @@ pub struct LocalPipelineReplayResult {
     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 non-trade materialization error.
+    pub non_trade_materialization_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.
@@ -55,6 +57,10 @@ pub struct LocalPipelineReplayResult {
     pub detection_count: usize,
     /// Total trade aggregation results returned by replayed aggregation calls.
     pub trade_event_count: usize,
+    /// Total liquidity event materialization results returned by replayed non-trade calls.
+    pub liquidity_event_count: usize,
+    /// Total pool lifecycle event materialization results returned by replayed non-trade calls.
+    pub pool_lifecycle_event_count: usize,
     /// Total candle upsert results returned by replayed candle calls.
     ///
     /// This is a replay write/result counter, not the number of distinct rows
@@ -141,6 +147,8 @@ impl LocalPipelineReplayService {
         let dex_decode = crate::DexDecodeService::new(self.database.clone());
         let dex_detect = crate::DexDetectService::new(self.database.clone());
         let trade_aggregation = crate::TradeAggregationService::new(self.database.clone());
+        let non_trade_materialization =
+            crate::NonTradeEventMaterializationService::new(self.database.clone());
         let pair_candle_aggregation =
             crate::PairCandleAggregationService::new(self.database.clone());
         let pair_analytic_signal = crate::PairAnalyticSignalService::new(self.database.clone());
@@ -187,6 +195,25 @@ impl LocalPipelineReplayService {
                     );
                 },
             }
+            let non_trade_result = non_trade_materialization
+                .record_transaction_by_signature(signature.as_str())
+                .await;
+            match non_trade_result {
+                Ok(non_trade_result) => {
+                    result.liquidity_event_count += non_trade_result.liquidity_event_count;
+                    result.pool_lifecycle_event_count +=
+                        non_trade_result.pool_lifecycle_event_count;
+                },
+                Err(error) => {
+                    result.non_trade_materialization_error_count += 1;
+                    tracing::warn!(
+                        signature = %signature,
+                        error = %error,
+                        "local pipeline replay non-trade materialization step failed"
+                    );
+                    continue;
+                },
+            }
             let trade_result =
                 trade_aggregation.record_transaction_by_signature(signature.as_str()).await;
             match trade_result {