0.7.24-pre.2

kb_lib/src/dex_decode.rs

@@ -1617,9 +1617,130 @@ impl KbDexDecodeService {
                 }
                 Ok(fetched)
             }
+            crate::KbPumpFunDecodedEvent::BuyTrade(event) => {
+                self.persist_pump_fun_trade_event(
+                    transaction,
+                    event,
+                    "pump_fun.buy",
+                    "signal.dex.pump_fun.buy",
+                    "dex.pump_fun.buy",
+                )
+                .await
+            }
+            crate::KbPumpFunDecodedEvent::SellTrade(event) => {
+                self.persist_pump_fun_trade_event(
+                    transaction,
+                    event,
+                    "pump_fun.sell",
+                    "signal.dex.pump_fun.sell",
+                    "dex.pump_fun.sell",
+                )
+                .await
+            }
         }
     }
 
+    async fn persist_pump_fun_trade_event(
+        &self,
+        transaction: &crate::KbChainTransactionDto,
+        event: &crate::KbPumpFunTradeDecoded,
+        event_kind: &str,
+        signal_kind: &str,
+        observation_kind: &str,
+    ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
+        let payload_json_result = serde_json::to_string(&event.payload_json);
+        let payload_json = match payload_json_result {
+            Ok(payload_json) => payload_json,
+            Err(error) => {
+                return Err(crate::KbError::Json(format!(
+                    "cannot serialize decoded pump.fun trade payload: {}",
+                    error
+                )));
+            }
+        };
+        let existing_result = crate::get_dex_decoded_event_by_key(
+            self.database.as_ref(),
+            event.transaction_id,
+            Some(event.instruction_id),
+            event_kind,
+        )
+        .await;
+        let existing_option = match existing_result {
+            Ok(existing_option) => existing_option,
+            Err(error) => return Err(error),
+        };
+        let already_present = existing_option.is_some();
+        let dto = crate::KbDexDecodedEventDto::new(
+            event.transaction_id,
+            Some(event.instruction_id),
+            "pump_fun".to_string(),
+            event.program_id.clone(),
+            event_kind.to_string(),
+            event.bonding_curve.clone(),
+            None,
+            event.mint.clone(),
+            Some(crate::WSOL_MINT_ID.to_string()),
+            event.associated_bonding_curve.clone(),
+            payload_json,
+        );
+        let upsert_result = crate::upsert_dex_decoded_event(self.database.as_ref(), &dto).await;
+        if let Err(error) = upsert_result {
+            return Err(error);
+        }
+        let fetched_result = crate::get_dex_decoded_event_by_key(
+            self.database.as_ref(),
+            event.transaction_id,
+            Some(event.instruction_id),
+            event_kind,
+        )
+        .await;
+        let fetched_option = match fetched_result {
+            Ok(fetched_option) => fetched_option,
+            Err(error) => return Err(error),
+        };
+        let fetched = match fetched_option {
+            Some(fetched) => fetched,
+            None => {
+                return Err(crate::KbError::InvalidState(
+                    "decoded pump.fun trade event disappeared after upsert".to_string(),
+                ));
+            }
+        };
+        if !already_present {
+            let payload_value = event.payload_json.clone();
+            let observation_result = self
+                .persistence
+                .record_observation(&crate::KbDetectionObservationInput::new(
+                    observation_kind.to_string(),
+                    crate::KbObservationSourceKind::HttpRpc,
+                    transaction.source_endpoint_name.clone(),
+                    transaction.signature.clone(),
+                    transaction.slot,
+                    payload_value.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::KbDetectionSignalInput::new(
+                    signal_kind.to_string(),
+                    crate::KbAnalysisSignalSeverity::Low,
+                    transaction.signature.clone(),
+                    Some(observation_id),
+                    None,
+                    payload_value,
+                ))
+                .await;
+            if let Err(error) = signal_result {
+                return Err(error);
+            }
+        }
+        Ok(fetched)
+    }
+
     async fn persist_pump_swap_event(
         &self,
         transaction: &crate::KbChainTransactionDto,