listening add meteora programs ids for a single bundler and saving logs

.env

@@ -1,5 +1,5 @@
 YELLOWSTONE_ENDPOINT=https://solana-yellowstone-grpc.publicnode.com:443
-YELLOWSTONE_COMMITMENT=confirmed
+YELLOWSTONE_COMMITMENT=processed
 
 BUNDLER=5jYaYv7HoiFVrY9bAcruj6dH8fCBseky4sBmnTFGSaeW
 

src/listener.rs

@@ -12,11 +12,13 @@ use yellowstone_grpc_proto::geyser::{
     SubscribeUpdate,
     SubscribeUpdateTransaction
 };
+use bs58;
+
+pub const METEORA_DBC:   &str = "dbcij3LWUppWqq96dh6gJWwBifmcGfLSB5D4DuSMaqN";
+pub const METEORA_DLMM:  &str = "LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo";
+pub const METEORA_POOLS: &str = "Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB";
+pub const METEORA_VAULT: &str = "24Uqj9JCLxUeoC3hGfh5W3s9FM9uCHDS2SG3LYwBpyTi";
 
-pub const METEORA_DBC_PROGRAM:   &str = "dbcij3LWUppWqq96dh6gJWwBifmcGfLSB5D4DuSMaqN";
-pub const METEORA_DLMM_PROGRAM:  &str = "LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo";
-pub const METEORA_POOLS_PROGRAM: &str = "Eo7WjKq67rjJQSZxS6z3YkapzY3eMj6Xy8X5EQVn5UaB";
-pub const METEORA_VAULT_PROGRAM: &str = "24Uqj9JCLxUeoC3hGfh5W3s9FM9uCHDS2SG3LYwBpyTi";
 
 pub struct YellowstoneSource {
     stream: Pin<Box<dyn Stream<Item = Result<SubscribeUpdate, yellowstone_grpc_proto::tonic::Status>> + Send>>,
@@ -56,20 +58,13 @@ impl YellowstoneSource {
         };
 
         tx_filter.account_required.push(bundler.clone());
-        for pid in [
-            METEORA_DBC_PROGRAM,
-            METEORA_DLMM_PROGRAM,
-            METEORA_POOLS_PROGRAM,
-            METEORA_VAULT_PROGRAM,
-        ] {
-            tx_filter.account_include.push(pid.to_string());
-        }
 
 
         
 
         // Determine commitment level for subscriptions
         let commitment = match commitment.as_deref() {
+            Some("processed") => CommitmentLevel::Processed,
             Some("confirmed") => CommitmentLevel::Confirmed,
             Some("finalized") => CommitmentLevel::Finalized,
             _ => CommitmentLevel::Processed,
@@ -106,18 +101,23 @@ impl YellowstoneSource {
 
     // Wait until the next transaction update is available
     async fn next(&mut self) -> Result<Option<SubscribeUpdateTransaction>> {
-        // Determine the next update from the stream
-        match self.stream.next().await {
-            None => Ok(None),
-            Some(Err(e)) => Err(e.into()),
-            Some(Ok(update)) => {
-                if let Some(UpdateOneof::Transaction(txu)) = update.update_oneof {
-                    tracing::info!("[INFO] received transaction from Yellowstone: \n{:#?}", txu);
-                    Ok(Some(txu))
+        loop {
+            let Some(update_res) = self.stream.as_mut().next().await else {
+                return Ok(None);
+            };
+
+            let update = update_res.map_err(anyhow::Error::from)?;
+
+            if let Some(UpdateOneof::Transaction(txu)) = update.update_oneof {
+                if program_key_hit(&txu) {
+                    return Ok(Some(txu));
                 } else {
-                    Ok(None)
+                    // no coincide → sigue leyendo el stream
+                    continue;
                 }
             }
+
+            // no era una transacción → sigue leyendo
         }
     }
 }
@@ -137,4 +137,41 @@ pub async fn yellowstone_forward_source_to_channel(
     }
 
     Ok(())
-}
+}
+
+// Ponlo en listener.rs (o en utils.rs si prefieres), en el mismo módulo.
+fn program_key_hit(txu: &SubscribeUpdateTransaction) -> bool {
+    let Some(msg) = txu
+        .transaction.as_ref()
+        .and_then(|t| t.transaction.as_ref())
+        .and_then(|uv| uv.message.as_ref())
+    else { return false; };
+
+    // account_keys: Vec<Vec<u8>> → Vec<String> (base58)
+    let keys_b58: Vec<String> = msg.account_keys
+        .iter()
+        .map(|k| bs58::encode(k).into_string())
+        .collect();
+
+    // 1) ¿Aparece el programa como key directa?
+    if keys_b58.iter().any(|k| matches!(k.as_str(),
+        METEORA_DBC | METEORA_DLMM | METEORA_POOLS | METEORA_VAULT))
+    {
+        return true;
+    }
+
+    // 2) ¿Aparece como program_id_index en alguna instrucción?
+    for ix in &msg.instructions {
+        let idx = ix.program_id_index as usize;
+        if let Some(pid_bytes) = msg.account_keys.get(idx) {
+            let pid_b58 = bs58::encode(pid_bytes).into_string();
+            if matches!(pid_b58.as_str(),
+                METEORA_DBC | METEORA_DLMM | METEORA_POOLS | METEORA_VAULT)
+            {
+                return true;
+            }
+        }
+    }
+
+    false
+}

src/main.rs

@@ -11,6 +11,8 @@ use sniper_bot::listener::YellowstoneSource;
 use sniper_bot::utils::save_tx_update;
 
 
+
+
 #[tokio::main]
 async fn main() -> Result<()> {
     init_tracing();
@@ -48,7 +50,7 @@ async fn main() -> Result<()> {
             tracing::warn!("No se pudo guardar el frame: {e:?}");
         }
         
-        tracing::info!("[INFO] received transaction update: \n{:#?}", txu);
+        //tracing::info!("[INFO] received transaction update: \n{:#?}", txu);
         let sig = txu
             .transaction
             .as_ref()

src/utils.rs

@@ -7,6 +7,7 @@ use tokio::fs;
 use yellowstone_grpc_proto::geyser::SubscribeUpdateTransaction;
 use yellowstone_grpc_proto::prost::Message;
 
+
 // Save file from tx received
 pub async fn save_tx_update(txu: &SubscribeUpdateTransaction) -> Result<(PathBuf, PathBuf)> {
     // 1) Carpetas