Added logic to indexing to get relayer transactions that are missing …

…from db

src/db.rs

@@ -3,7 +3,7 @@
 use std::time::Duration;
 
 use chrono::{DateTime, Utc};
-use ethers::types::{Address, H256, U256};
+use ethers::types::{Address, Transaction, H256, U256};
 use sqlx::migrate::{MigrateDatabase, Migrator};
 use sqlx::types::{BigDecimal, Json};
 use sqlx::{Pool, Postgres, Row};
@@ -548,6 +548,7 @@ impl Database {
     pub async fn mine_txs(
         &self,
         chain_id: u64,
+        relayed_transactions: Vec<Transaction>,
     ) -> eyre::Result<Vec<(String, H256)>> {
         let updated_txs: Vec<(String, H256Wrapper)> = sqlx::query_as(
             r#"
@@ -578,6 +579,24 @@ impl Database {
         .fetch_all(&self.pool)
         .await?;
 
+        let missing_transactions = relayed_transactions
+            .into_iter()
+            .filter(|tx| {
+                !updated_txs.iter().any(|(_, tx_hash)| tx_hash.0 == tx.hash)
+            })
+            .collect::<Vec<Transaction>>();
+
+        if !missing_transactions.is_empty() {
+
+            //TODO: get the tx id for each transaction
+
+            //TODO: insert into tx_hashes
+
+            //TODO: insert into sent_transactions
+
+            //TODO: update updated txs
+        }
+
         Ok(updated_txs
             .into_iter()
             .map(|(id, hash)| (id, hash.0))
@@ -1087,7 +1106,7 @@ mod tests {
         chain_id: u64,
         finalization_timestamp: DateTime<Utc>,
     ) -> eyre::Result<()> {
-        db.mine_txs(chain_id).await?;
+        db.mine_txs(chain_id, vec![]).await?;
 
         db.handle_soft_reorgs().await?;
         db.handle_hard_reorgs().await?;

src/tasks/index.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 
 use chrono::{DateTime, Utc};
 use ethers::providers::{Http, Middleware, Provider};
-use ethers::types::BlockNumber;
+use ethers::types::{Block, BlockNumber, Transaction, H256};
 use eyre::{Context, ContextCompat};
 use futures::stream::FuturesUnordered;
 use futures::StreamExt;
@@ -53,7 +53,16 @@ pub async fn index_chain(app: Arc<App>, chain_id: u64) -> eyre::Result<()> {
                 )
                 .await?;
 
-            let mined_txs = app.db.mine_txs(chain_id).await?;
+            // Get transactions from the block where the from address is a relayer
+            let txs = get_transactions_from_block(&block, &rpc).await?;
+            let relayers = app.db.get_relayer_addresses(chain_id).await?;
+            let relayed_transactions = txs
+                .into_iter()
+                .filter(|tx| relayers.contains(&tx.from))
+                .collect::<Vec<Transaction>>();
+
+            let mined_txs =
+                app.db.mine_txs(chain_id, relayed_transactions).await?;
 
             let metric_labels = [("chain_id", chain_id.to_string())];
             for tx in mined_txs {
@@ -196,3 +205,23 @@ pub async fn get_block_fee_estimates(
 
     Ok(fee_estimates)
 }
+
+pub async fn get_transactions_from_block(
+    block: &Block<H256>,
+    rpc: &Provider<Http>,
+) -> eyre::Result<Vec<Transaction>> {
+    let mut futures = FuturesUnordered::new();
+
+    block
+        .transactions
+        .iter()
+        .map(|tx_hash| futures.push(rpc.get_transaction(*tx_hash)));
+
+    let mut transactions = vec![];
+    while let Some(tx) = futures.next().await {
+        let tx = tx?.context("Tx not found")?;
+        transactions.push(tx);
+    }
+
+    Ok(transactions)
+}