feat: Add Raydium DEX integration support

Add Raydium logs processing, event parsing and GRPC stream handling functionality,
remove redundant PumpSwap subscription module, optimize project structure.

src/grpc/shred_stream.rs

@@ -8,6 +8,7 @@ use log::error;
 use solana_sdk::transaction::VersionedTransaction;
 
 use crate::common::pumpswap::PumpSwapInstruction;
+use crate::common::raydium::{RaydiumEvent, RaydiumInstruction};
 use crate::common::AnyResult;
 
 use solana_sdk::pubkey::Pubkey;
@@ -16,6 +17,7 @@ use crate::common::pumpfun::logs_events::PumpfunEvent;
 use crate::common::pumpswap::logs_events::PumpSwapEvent;
 use crate::common::pumpfun::logs_filters::LogFilter;
 use crate::common::pumpswap::logs_filters::LogFilter as PumpswapLogFilter;
+use crate::common::raydium::logs_filters::LogFilter as RaydiumLogFilter;
 use crate::swqos::jito_grpc::shredstream::shredstream_proxy_client::ShredstreamProxyClient;
 use crate::swqos::jito_grpc::shredstream::SubscribeEntriesRequest;
 
@@ -121,6 +123,47 @@ impl ShredStreamGrpc {
         Ok(())
     }
 
+    pub async fn shredstream_subscribe_raydium<F>(&self, callback: F) -> AnyResult<()> 
+    where
+        F: Fn(RaydiumEvent) + Send + Sync + 'static,
+    {
+        let request = tonic::Request::new(SubscribeEntriesRequest {});
+        let mut client = (*self.shredstream_client).clone();
+        let mut stream = client.subscribe_entries(request).await?.into_inner();
+        let (mut tx, mut rx) = mpsc::channel::<TransactionWithSlot>(CHANNEL_SIZE);
+        let callback = Box::new(callback);
+        tokio::spawn(async move {
+            while let Some(message) = stream.next().await {
+                match message {
+                    Ok(msg) => {
+                        if let Ok(entries) = bincode::deserialize::<Vec<Entry>>(&msg.entries) {
+                            for entry in entries {
+                                for transaction in entry.transactions {
+                                    let _ = tx.try_send(TransactionWithSlot {
+                                        transaction: transaction.clone(),
+                                        slot: msg.slot,
+                                    });
+                                }
+                            }
+                        }
+                    }
+                    Err(error) => {
+                        error!("Stream error: {error:?}");
+                        break;
+                    }
+                }
+            }
+        });
+
+        while let Some(transaction_with_slot) = rx.next().await {
+            if let Err(e) = Self::process_raydium_transaction(transaction_with_slot, &*callback).await {
+                error!("Error processing transaction: {:?}", e);
+            }
+        }
+    
+        Ok(())
+    }
+
     async fn process_pumpfun_transaction<F>(transaction_with_slot: TransactionWithSlot, callback: &F, bot_wallet: Option<Pubkey>) -> AnyResult<()> 
     where
         F: Fn(PumpfunEvent) + Send + Sync,
@@ -211,4 +254,35 @@ impl ShredStreamGrpc {
         }
         Ok(())
     }
+
+    async fn process_raydium_transaction<F>(transaction_with_slot: TransactionWithSlot, callback: &F) -> AnyResult<()> 
+    where
+        F: Fn(RaydiumEvent) + Send + Sync,
+    {
+        let slot = transaction_with_slot.slot;
+        let versioned_tx = transaction_with_slot.transaction;
+        let signature = versioned_tx.signatures[0].to_string();
+        let instructions: Vec<RaydiumInstruction> = RaydiumLogFilter::parse_raydium_compiled_instruction(versioned_tx).unwrap();
+        for instruction in instructions {
+            match instruction {
+                RaydiumInstruction::V4Swap(mut v4_swap_event) => {
+                    v4_swap_event.slot = slot;
+                    v4_swap_event.signature = signature.clone();
+                    callback(RaydiumEvent::V4Swap(v4_swap_event));
+                }
+                RaydiumInstruction::SwapBaseInput(mut swap_base_input_event) => {
+                    swap_base_input_event.slot = slot;
+                    swap_base_input_event.signature = signature.clone();
+                    callback(RaydiumEvent::SwapBaseInput(swap_base_input_event));
+                }
+                RaydiumInstruction::SwapBaseOutput(mut swap_base_output_event) => {
+                    swap_base_output_event.slot = slot;
+                    swap_base_output_event.signature = signature.clone();
+                    callback(RaydiumEvent::SwapBaseOutput(swap_base_output_event));
+                }
+                _ => {}
+            }
+        }
+        Ok(())
+    }
 }

src/grpc/yellow_stone.rs

@@ -535,4 +535,113 @@ impl YellowstoneGrpc {
 
         Ok(())
     }
+
+    // ------------------------------------------------------------
+    // Raydium
+    // ------------------------------------------------------------
+
+    /// 订阅Raydium事件
+    pub async fn subscribe_raydium<F>(&self, callback: F) -> AnyResult<()>
+    where
+        F: Fn(crate::common::raydium::logs_events::RaydiumEvent) + Send + Sync + 'static,
+    {
+        // 使用constants中定义的AMM_PROGRAM
+        let raydium_v4_program_id = crate::constants::raydium::accounts::AMMV4_PROGRAM;
+        let raydium_cpmm_program_id = crate::constants::raydium::accounts::CPMM_PROGRAM;
+        let addrs = vec![raydium_v4_program_id.to_string(), raydium_cpmm_program_id.to_string()];
+
+        // 创建过滤器
+        let transactions = self.get_subscribe_request_filter(addrs, vec![], vec![]);
+
+        // 订阅事件
+        let (mut subscribe_tx, mut stream) = self.subscribe_with_request(transactions).await?;
+
+        // 创建通道
+        let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(1000);
+
+        // 创建回调函数
+        let callback = Box::new(callback);
+
+        // 启动处理流的任务
+        tokio::spawn(async move {
+            while let Some(message) = stream.next().await {
+                match message {
+                    Ok(msg) => {
+                        if let Err(e) =
+                            Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await
+                        {
+                            error!("Error handling message: {:?}", e);
+                            break;
+                        }
+                    }
+                    Err(error) => {
+                        error!("Stream error: {error:?}");
+                        break;
+                    }
+                }
+            }
+        });
+
+        // 处理交易
+        while let Some(transaction_pretty) = rx.next().await {
+            if let Err(e) = Self::process_raydium_transaction(transaction_pretty, &*callback).await
+            {
+                error!("Error processing transaction: {:?}", e);
+            }
+        }
+
+        Ok(())
+    }
+
+    /// 处理Raydium交易
+    async fn process_raydium_transaction<F>(
+        transaction_pretty: TransactionPretty,
+        callback: &F,
+    ) -> AnyResult<()>
+    where
+        F: Fn(crate::common::raydium::logs_events::RaydiumEvent) + Send + Sync,
+    {
+        let slot = transaction_pretty.slot;
+        let trade_raw: solana_transaction_status::EncodedTransactionWithStatusMeta =
+            transaction_pretty.tx;
+
+        // 检查交易元数据
+        let meta = trade_raw
+            .meta
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?;
+
+        // 检查交易是否成功
+        if meta.err.is_some() {
+            return Ok(());
+        }
+
+        if let Some(versioned_tx) = trade_raw.transaction.decode() {
+            let signature = versioned_tx.signatures[0].to_string();
+            let instructions: Vec<crate::common::raydium::logs_data::RaydiumInstruction> =
+                crate::common::raydium::logs_filters::LogFilter::parse_raydium_compiled_instruction(versioned_tx).unwrap();
+            for instruction in instructions {
+                match instruction {
+                    crate::common::raydium::logs_data::RaydiumInstruction::V4Swap(mut v4_swap_event) => {
+                        v4_swap_event.slot = slot;
+                        v4_swap_event.signature = signature.clone();
+                        callback(crate::common::raydium::logs_events::RaydiumEvent::V4Swap(v4_swap_event));
+                    }
+                    crate::common::raydium::logs_data::RaydiumInstruction::SwapBaseInput(mut swap_base_input_event) => {
+                        swap_base_input_event.slot = slot;
+                        swap_base_input_event.signature = signature.clone();
+                        callback(crate::common::raydium::logs_events::RaydiumEvent::SwapBaseInput(swap_base_input_event));
+                    }
+                    crate::common::raydium::logs_data::RaydiumInstruction::SwapBaseOutput(mut swap_base_output_event) => {
+                        swap_base_output_event.slot = slot;
+                        swap_base_output_event.signature = signature.clone();
+                        callback(crate::common::raydium::logs_events::RaydiumEvent::SwapBaseOutput(swap_base_output_event));
+                    }
+                    _ => {}
+                }
+            }
+        }
+
+        Ok(())
+    }
 }