0.7.40

kb_lib/src/token_backfill.rs

@@ -84,6 +84,41 @@ pub struct PoolBackfillResult {
     pub pair_candle_count: usize,
 }
 
+/// One signature-backfill result summary.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct SignatureBackfillResult {
+    /// Input transaction signature.
+    pub signature: std::string::String,
+    /// Number of transactions resolved through HTTP during this run.
+    pub resolved_transaction_count: usize,
+    /// Number of signatures whose `getTransaction` lookup returned `null`.
+    pub missing_transaction_count: usize,
+    /// Total number of decoded DEX events replayed during this run.
+    pub decoded_event_count: usize,
+    /// Total number of DEX detection results produced during this run.
+    pub detection_count: usize,
+    /// Total number of launch-attribution results produced during this run.
+    pub launch_attribution_count: usize,
+    /// Total number of pool-origin results produced during this run.
+    pub pool_origin_count: usize,
+    /// Total number of wallet-participation observations produced during this run.
+    pub wallet_participation_count: usize,
+    /// Total number of trade-aggregation results produced during this run.
+    pub trade_event_count: usize,
+    /// Total number of liquidity event materialization results produced during this run.
+    pub liquidity_event_count: usize,
+    /// Total number of pool lifecycle event materialization results produced during this run.
+    pub pool_lifecycle_event_count: usize,
+    /// Total number of fee event materialization results produced during this run.
+    pub fee_event_count: usize,
+    /// Total number of reward event materialization results produced during this run.
+    pub reward_event_count: usize,
+    /// Total number of pool administration event materialization results produced during this run.
+    pub pool_admin_event_count: usize,
+    /// Total number of pair-candle aggregation results produced during this run.
+    pub pair_candle_count: usize,
+}
+
 /// Historical token backfill service.
 ///
 /// This service reuses the existing transaction projection and downstream
@@ -681,6 +716,87 @@ impl TokenBackfillService {
         return Ok(result);
     }
 
+    /// Replays one known transaction signature through the existing pipeline.
+    pub async fn backfill_signature(
+        &self,
+        signature: &str,
+    ) -> Result<crate::SignatureBackfillResult, crate::Error> {
+        let trimmed_signature = signature.trim().to_string();
+        if trimmed_signature.is_empty() {
+            return Err(crate::Error::Config("signature must not be empty".to_string()));
+        }
+        let replay_result = self.replay_signature(trimmed_signature.clone()).await;
+        let replay = match replay_result {
+            Ok(replay) => replay,
+            Err(error) => return Err(error),
+        };
+        self.backfill_missing_token_metadata_best_effort(100).await;
+        let result = crate::SignatureBackfillResult {
+            signature: trimmed_signature.clone(),
+            resolved_transaction_count: replay.resolved_transaction_count,
+            missing_transaction_count: replay.missing_transaction_count,
+            decoded_event_count: replay.decoded_event_count,
+            detection_count: replay.detection_count,
+            launch_attribution_count: replay.launch_attribution_count,
+            pool_origin_count: replay.pool_origin_count,
+            wallet_participation_count: replay.wallet_participation_count,
+            trade_event_count: replay.trade_event_count,
+            liquidity_event_count: replay.liquidity_event_count,
+            pool_lifecycle_event_count: replay.pool_lifecycle_event_count,
+            fee_event_count: replay.fee_event_count,
+            reward_event_count: replay.reward_event_count,
+            pool_admin_event_count: replay.pool_admin_event_count,
+            pair_candle_count: replay.pair_candle_count,
+        };
+        let summary_payload = serde_json::json!({
+            "signature": result.signature.clone(),
+            "resolvedTransactionCount": result.resolved_transaction_count,
+            "missingTransactionCount": result.missing_transaction_count,
+            "decodedEventCount": result.decoded_event_count,
+            "detectionCount": result.detection_count,
+            "launchAttributionCount": result.launch_attribution_count,
+            "poolOriginCount": result.pool_origin_count,
+            "walletParticipationCount": result.wallet_participation_count,
+            "tradeEventCount": result.trade_event_count,
+            "liquidityEventCount": result.liquidity_event_count,
+            "poolLifecycleEventCount": result.pool_lifecycle_event_count,
+            "feeEventCount": result.fee_event_count,
+            "rewardEventCount": result.reward_event_count,
+            "poolAdminEventCount": result.pool_admin_event_count,
+            "pairCandleCount": result.pair_candle_count
+        });
+        let observation_result = self
+            .persistence
+            .record_observation(&crate::DetectionObservationInput::new(
+                "signature.backfill.completed".to_string(),
+                crate::ObservationSourceKind::HttpRpc,
+                Some(format!("backfill:{}", self.http_role)),
+                trimmed_signature.clone(),
+                None,
+                summary_payload.clone(),
+            ))
+            .await;
+        let observation_id = match observation_result {
+            Ok(observation_id) => observation_id,
+            Err(error) => return Err(error),
+        };
+        let signal_result = self
+            .persistence
+            .record_signal(&crate::DetectionSignalInput::new(
+                "signal.signature.backfill.completed".to_string(),
+                crate::AnalysisSignalSeverity::Low,
+                trimmed_signature,
+                Some(observation_id),
+                None,
+                summary_payload,
+            ))
+            .await;
+        if let Err(error) = signal_result {
+            return Err(error);
+        }
+        return Ok(result);
+    }
+
     async fn backfill_missing_token_metadata_best_effort(&self, limit: i64) {
         let metadata_result =
             self.token_metadata_service.backfill_missing_token_metadata(Some(limit)).await;