0.7.24-pre.3

kb_lib/src/dex_decode.rs

@@ -86,7 +86,7 @@ impl KbDexDecodeService {
         };
         for decoded_event in &raydium_decoded {
             let persist_result = self
-                .persist_raydium_event(&transaction, decoded_event)
+                .persist_raydium_amm_v4_event(&transaction, decoded_event)
                 .await;
             let persisted_event = match persist_result {
                 Ok(persisted_event) => persisted_event,
@@ -94,6 +94,16 @@ impl KbDexDecodeService {
             };
             persisted.push(persisted_event);
         }
+        let raydium_cpmm_persisted_result = self
+            .decode_and_persist_raydium_cpmm_events(&transaction, &instructions)
+            .await;
+        let raydium_cpmm_persisted = match raydium_cpmm_persisted_result {
+            Ok(raydium_cpmm_persisted) => raydium_cpmm_persisted,
+            Err(error) => return Err(error),
+        };
+        for persisted_event in raydium_cpmm_persisted {
+            persisted.push(persisted_event);
+        }
         let pump_fun_decoded_result = self
             .pump_fun_decoder
             .decode_transaction(&transaction, &instructions);
@@ -1414,7 +1424,7 @@ impl KbDexDecodeService {
         }
     }
 
-    async fn persist_raydium_event(
+    async fn persist_raydium_amm_v4_event(
         &self,
         transaction: &crate::KbChainTransactionDto,
         decoded_event: &crate::KbRaydiumAmmV4DecodedEvent,
@@ -1517,6 +1527,168 @@ impl KbDexDecodeService {
         }
     }
 
+    async fn decode_and_persist_raydium_cpmm_events(
+        &self,
+        transaction: &crate::KbChainTransactionDto,
+        instructions: &[crate::KbChainInstructionDto],
+    ) -> Result<std::vec::Vec<crate::KbDexDecodedEventDto>, crate::KbError> {
+        let mut persisted = std::vec::Vec::new();
+        for instruction in instructions {
+            let program_id = match instruction.program_id.as_ref() {
+                Some(program_id) => program_id,
+                None => continue,
+            };
+            if program_id.as_str() != crate::KB_RAYDIUM_CPMM_PROGRAM_ID {
+                continue;
+            }
+            let data_json = match instruction.data_json.as_ref() {
+                Some(data_json) => data_json,
+                None => continue,
+            };
+            let decoded_events = crate::kb_decode_raydium_cpmm_instruction(
+                instruction.accounts_json.as_str(),
+                data_json.as_str(),
+            );
+            for decoded_event in &decoded_events {
+                let persist_result = self
+                    .persist_raydium_cpmm_event(transaction, instruction, decoded_event)
+                    .await;
+                let persisted_event = match persist_result {
+                    Ok(persisted_event) => persisted_event,
+                    Err(error) => return Err(error),
+                };
+                persisted.push(persisted_event);
+            }
+        }
+        Ok(persisted)
+    }
+
+    async fn persist_raydium_cpmm_event(
+        &self,
+        transaction: &crate::KbChainTransactionDto,
+        instruction: &crate::KbChainInstructionDto,
+        decoded_event: &crate::KbRaydiumCpmmDecodedEvent,
+    ) -> Result<crate::KbDexDecodedEventDto, crate::KbError> {
+        let transaction_id = match transaction.id {
+            Some(transaction_id) => transaction_id,
+            None => {
+                return Err(crate::KbError::InvalidState(format!(
+                    "transaction '{}' has no internal id",
+                    transaction.signature
+                )));
+            }
+        };
+        let instruction_id = match instruction.id {
+            Some(instruction_id) => instruction_id,
+            None => {
+                return Err(crate::KbError::InvalidState(format!(
+                    "raydium cpmm instruction for transaction '{}' has no internal id",
+                    transaction.signature
+                )));
+            }
+        };
+        let payload_json = match decoded_event.to_payload_json() {
+            Some(payload_json) => payload_json,
+            None => {
+                return Err(crate::KbError::Json(
+                    "cannot serialize decoded raydium cpmm payload".to_string(),
+                ));
+            }
+        };
+        let event_kind = decoded_event.event_kind().to_string();
+        let existing_result = crate::get_dex_decoded_event_by_key(
+            self.database.as_ref(),
+            transaction_id,
+            Some(instruction_id),
+            event_kind.as_str(),
+        )
+        .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(
+            transaction_id,
+            Some(instruction_id),
+            "raydium_cpmm".to_string(),
+            crate::KB_RAYDIUM_CPMM_PROGRAM_ID.to_string(),
+            event_kind.clone(),
+            Some(decoded_event.pool_account().to_string()),
+            None,
+            Some(decoded_event.base_mint().to_string()),
+            Some(decoded_event.quote_mint().to_string()),
+            None,
+            payload_json.clone(),
+        );
+        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(),
+            transaction_id,
+            Some(instruction_id),
+            event_kind.as_str(),
+        )
+        .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 raydium cpmm event disappeared after upsert".to_string(),
+                ));
+            }
+        };
+        if !already_present {
+            let payload_value_result =
+                serde_json::from_str::<serde_json::Value>(payload_json.as_str());
+            let payload_value = match payload_value_result {
+                Ok(payload_value) => payload_value,
+                Err(error) => {
+                    return Err(crate::KbError::Json(format!(
+                        "cannot parse raydium cpmm payload after serialization: {}",
+                        error
+                    )));
+                }
+            };
+            let observation_result = self
+                .persistence
+                .record_observation(&crate::KbDetectionObservationInput::new(
+                    format!("dex.{}", event_kind),
+                    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(
+                    format!("signal.dex.{}", event_kind),
+                    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_fun_event(
         &self,
         transaction: &crate::KbChainTransactionDto,