SubscribeDeshred

use yellowstone_grpc_client::GeyserGrpcClient;
use yellowstone_grpc_proto::prelude::{
    SubscribeDeshredRequest,
    SubscribeRequestFilterDeshredTransactions,
};
use futures::stream::StreamExt;
use std::collections::HashMap;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connect to the gRPC endpoint
    let mut client = GeyserGrpcClient::build_from_shared("http://127.0.0.1:10000")?
        .x_token(Some("your-token-here"))?
        .connect()
        .await?;

    // Create deshred subscription request
    let mut deshred_transactions = HashMap::new();
    deshred_transactions.insert(
        "raydium_swaps".to_string(),
        SubscribeRequestFilterDeshredTransactions {
            vote: Some(false),
            account_include: vec![
                "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8".to_string(), // Raydium AMM program
            ],
            account_exclude: vec![],
            account_required: vec![],
        },
    );

    let request = SubscribeDeshredRequest {
        deshred_transactions,
        ping: None,
    };

    // Subscribe and receive deshred updates
    let (mut subscribe_tx, mut stream) = client
        .subscribe_deshred_with_request(Some(request))
        .await?;

    println!("Deshred subscription started, receiving pre-execution transactions...");
    
    while let Some(message) = stream.next().await {
        match message {
            Ok(msg) => {
                if let Some(update) = msg.update_oneof {
                    match update {
                        SubscribeUpdateDeshred::DeshredTransaction(tx_update) => {
                            let tx = tx_update.transaction.unwrap();
                            println!("Deshred transaction at slot {}:", tx_update.slot);
                            println!("  Signature: {:?}", tx.signature);
                            println!("  Is vote: {}", tx.is_vote);
                            println!("  Loaded writable addresses: {}", 
                                tx.loaded_writable_addresses.len());
                            println!("  Loaded readonly addresses: {}", 
                                tx.loaded_readonly_addresses.len());
                        }
                        SubscribeUpdateDeshred::Ping(_) => {
                            // Respond to ping to keep connection alive
                            subscribe_tx.send(SubscribeDeshredRequest {
                                ping: Some(SubscribeRequestPing { id: 1 }),
                                ..Default::default()
                            }).await?;
                        }
                        _ => {}
                    }
                }
            }
            Err(error) => {
                eprintln!("Error: {:?}", error);
                break;
            }
        }
    }

    Ok(())
}