From c0e02381f87351811d57b01f48712c0e3c80821a Mon Sep 17 00:00:00 2001
From: ysq <sgxiang1992@icloud.com>
Date: Wed, 9 Jul 2025 23:58:18 +0800
Subject: [PATCH] feat: add system transaction subscription module

Add yellow_stone_sub_system module for real-time system program transaction subscription and event handling
---
 src/grpc/mod.rs                     |  2 +
 src/grpc/yellow_stone.rs            |  2 +-
 src/grpc/yellow_stone_sub_system.rs | 96 +++++++++++++++++++++++++++++
 3 files changed, 99 insertions(+), 1 deletion(-)
 create mode 100644 src/grpc/yellow_stone_sub_system.rs

diff --git a/src/grpc/mod.rs b/src/grpc/mod.rs
index 217dac6..a98071c 100755
--- a/src/grpc/mod.rs
+++ b/src/grpc/mod.rs
@@ -1,5 +1,7 @@
 pub mod yellow_stone;
+pub mod yellow_stone_sub_system;    
 pub mod shred_stream;
 
 pub use yellow_stone::YellowstoneGrpc;
+pub use yellow_stone_sub_system::{SystemEvent, TransferInfo};
 pub use shred_stream::ShredStreamGrpc;
\ No newline at end of file
diff --git a/src/grpc/yellow_stone.rs b/src/grpc/yellow_stone.rs
index e049b89..a642f88 100755
--- a/src/grpc/yellow_stone.rs
+++ b/src/grpc/yellow_stone.rs
@@ -146,7 +146,7 @@ impl YellowstoneGrpc {
         transactions
     }
 
-    async fn handle_stream_message(
+    pub async fn handle_stream_message(
         msg: SubscribeUpdate,
         tx: &mut mpsc::Sender<TransactionPretty>,
         subscribe_tx: &mut (impl Sink<SubscribeRequest, Error = mpsc::SendError> + Unpin),
diff --git a/src/grpc/yellow_stone_sub_system.rs b/src/grpc/yellow_stone_sub_system.rs
new file mode 100644
index 0000000..fecf604
--- /dev/null
+++ b/src/grpc/yellow_stone_sub_system.rs
@@ -0,0 +1,96 @@
+use crate::{common::AnyResult, grpc::yellow_stone::{TransactionPretty, YellowstoneGrpc}};
+use solana_program::pubkey;
+use solana_sdk::{pubkey::Pubkey, transaction::VersionedTransaction};
+use futures::{channel::mpsc, sink::Sink, Stream, StreamExt, SinkExt};
+use log::{error, info};
+use solana_transaction_status::EncodedTransactionWithStatusMeta;
+
+const SYSTEM_PROGRAM_ID: Pubkey = pubkey!("11111111111111111111111111111111");
+const CHANNEL_SIZE: usize = 1000;
+
+#[derive(Debug)]
+pub enum SystemEvent {
+    NewTransfer(TransferInfo),
+    Error(String),
+}
+
+#[derive(Clone, Debug, Default, PartialEq)]
+pub struct TransferInfo {
+    pub slot: u64,
+    pub signature: String,
+    pub tx: Option<VersionedTransaction>,
+}
+
+impl YellowstoneGrpc {
+    pub async fn subscribe_system<F>(
+        &self,
+        callback: F,
+        account_include: Option<Vec<String>>,
+        account_exclude: Option<Vec<String>>,
+    ) -> AnyResult<()>
+    where
+        F: Fn(SystemEvent) + Send + Sync + 'static,
+    {
+        let addrs = vec![SYSTEM_PROGRAM_ID.to_string()];
+        let account_include = account_include.unwrap_or_default();
+        let account_exclude = account_exclude.unwrap_or_default();
+        let transactions =
+            self.get_subscribe_request_filter(account_include, account_exclude, addrs);
+        let (mut subscribe_tx, mut stream) = self.subscribe_with_request(transactions).await?;
+        let (mut tx, mut rx) = mpsc::channel::<TransactionPretty>(CHANNEL_SIZE);
+
+        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_system_transaction(transaction_pretty, &*callback).await {
+                error!("Error processing transaction: {:?}", e);
+            }
+        }
+        Ok(())
+    }
+
+    async fn process_system_transaction<F>(
+        transaction_pretty: TransactionPretty,
+        callback: &F,
+    ) -> AnyResult<()>
+    where
+        F: Fn(SystemEvent) + Send + Sync,
+    {
+        let trade_raw: 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(());
+        }
+
+        callback(SystemEvent::NewTransfer(TransferInfo {
+            slot: transaction_pretty.slot,
+            signature: transaction_pretty.signature.to_string(),
+            tx: trade_raw.transaction.decode(),
+        }));
+
+        Ok(())
+    }
+}