adjusts to return bytes

Author: unbound26z

Cargo.lock

@@ -28,6 +28,7 @@ solana-transaction-status = { workspace = true }
 async-trait = { workspace = true }
 borsh = { version = "0.10.4" }
 bs58 = { workspace = true }
+base64 = { workspace = true }
 log = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

crates/core/Cargo.toml

@@ -70,14 +70,14 @@ enum LogType {
 }
 
 impl InstructionMetadata {
-    /// Extracts the `data` log messages associated with this instruction.
+    /// Extracts the `data` from log messages associated with this instruction.
     ///
     /// This method filters the transaction's log messages to return only those
     /// that correspond to the current instruction, based on its stack height and
     /// absolute path within the instruction stack.
     ///
-    /// Returns `Vec<String>` containing the `data` log messages.
-    pub fn extract_logs(&self) -> Vec<String> {
+    /// Returns `Vec<Vec<u8>>` containing the `data` bytes (base64 encoded) from log messages.
+    pub fn extract_logs(&self) -> Vec<Vec<u8>> {
         let logs = match &self.transaction_metadata.meta.log_messages {
             Some(logs) => logs,
             None => return Vec::new(),
@@ -128,7 +128,13 @@ impl InstructionMetadata {
 
             // Extract if we're on target path and it's a data log
             if current_path == self.absolute_path && matches!(parsed_log, LogType::Data) {
-                extracted_logs.push(log.clone());
+                if let Some(data) = log.split_whitespace().last() {
+                    if let Ok(buf) =
+                        base64::Engine::decode(&base64::engine::general_purpose::STANDARD, data)
+                    {
+                        extracted_logs.push(buf);
+                    }
+                }
             }
         }
 

crates/core/src/instruction.rs

@@ -5,7 +5,7 @@ edition = { workspace = true }
 
 [dependencies]
 carbon-core = { workspace = true }
-carbon-jupiter-swap-decoder = { workspace = true }
+carbon-raydium-cpmm-decoder = { workspace = true }
 carbon-log-metrics = { workspace = true }
 carbon-yellowstone-grpc-datasource = { workspace = true }
 
@@ -15,6 +15,7 @@ env_logger = { workspace = true }
 solana-instruction = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 yellowstone-grpc-proto = { workspace = true }
+base64 = { workspace = true }
 
 [dependencies.rustls]
 default-features = false

examples/log-events-example/Cargo.toml

@@ -1,16 +1,17 @@
 use {
     async_trait::async_trait,
     carbon_core::{
+        borsh::BorshDeserialize,
         error::CarbonResult,
         instruction::{DecodedInstruction, InstructionMetadata, NestedInstructions},
         metrics::MetricsCollection,
         processor::Processor,
     },
-    carbon_jupiter_swap_decoder::{
-        instructions::JupiterSwapInstruction, JupiterSwapDecoder,
-        PROGRAM_ID as JUPITER_SWAP_PROGRAM_ID,
-    },
     carbon_log_metrics::LogMetrics,
+    carbon_raydium_cpmm_decoder::{
+        instructions::RaydiumCpmmInstruction, types::SwapEvent, RaydiumCpmmDecoder,
+        PROGRAM_ID as RAYDIUM_CPMM_PROGRAM_ID,
+    },
     carbon_yellowstone_grpc_datasource::{
         YellowstoneGrpcClientConfig, YellowstoneGrpcGeyserClient,
     },
@@ -39,7 +40,7 @@ pub async fn main() -> CarbonResult<()> {
         failed: Some(false),
         account_include: vec![],
         account_exclude: vec![],
-        account_required: vec![JUPITER_SWAP_PROGRAM_ID.to_string().clone()],
+        account_required: vec![RAYDIUM_CPMM_PROGRAM_ID.to_string().clone()],
         signature: None,
     };
 
@@ -75,7 +76,7 @@ pub async fn main() -> CarbonResult<()> {
         .datasource(yellowstone_grpc)
         .metrics(Arc::new(LogMetrics::new()))
         .metrics_flush_interval(3)
-        .instruction(JupiterSwapDecoder, JupiterSwapInstructionProcessor)
+        .instruction(RaydiumCpmmDecoder, RaydiumCpmmInstructionProcessor)
         .shutdown_strategy(carbon_core::pipeline::ShutdownStrategy::Immediate)
         .build()?
         .run()
@@ -84,13 +85,13 @@ pub async fn main() -> CarbonResult<()> {
     Ok(())
 }
 
-pub struct JupiterSwapInstructionProcessor;
+pub struct RaydiumCpmmInstructionProcessor;
 
 #[async_trait]
-impl Processor for JupiterSwapInstructionProcessor {
+impl Processor for RaydiumCpmmInstructionProcessor {
     type InputType = (
         InstructionMetadata,
-        DecodedInstruction<JupiterSwapInstruction>,
+        DecodedInstruction<RaydiumCpmmInstruction>,
         NestedInstructions,
         solana_instruction::Instruction,
     );
@@ -107,9 +108,12 @@ impl Processor for JupiterSwapInstructionProcessor {
                 "Processing instruction - Index: {}, Stack Height: {}, Path: {:?}",
                 metadata.index, metadata.stack_height, metadata.absolute_path
             );
-            println!("Extracted {} logs:", logs.len());
-            for (i, log) in logs.iter().enumerate() {
-                println!("  {}: {}", i, log);
+            for log in logs {
+                if log.len() > 8 {
+                    if let Ok(swap_event) = SwapEvent::deserialize(&mut &log[8..]) {
+                        println!("SwapEvent: {:?}", swap_event);
+                    }
+                }
             }
             println!("--------------------------------");
         }