fix: Replace buggy balance extraction with SDK extract_balance_deltas_from_tx

The bug occurred when a Transfer event involved both the sender and recipient being pools. In such cases, the previous implementation only created a BalanceDelta for the "to" address and missed accounting for the "from" address.

substreams/ethereum-curve/src/modules.rs

@@ -10,13 +10,11 @@ use substreams::{
 
 use substreams_ethereum::pb::eth;
 
-use crate::{
+use crate::{pool_changes::emit_eth_deltas, pool_factories, pools::emit_specific_pools};
-    pool_changes::{emit_deltas, emit_eth_deltas},
-    pool_factories,
-    pools::emit_specific_pools,
-};
 use tycho_substreams::{
-    balances::store_balance_changes, contract::extract_contract_changes, prelude::*,
+    balances::{extract_balance_deltas_from_tx, store_balance_changes},
+    contract::extract_contract_changes,
+    prelude::*,
 };
 
 /// This struct purely exists to spoof the `PartialEq` trait for `Transaction` so we can use it in
@@ -115,7 +113,19 @@ pub fn map_relative_balances(
                 .flat_map(|tx| {
                     emit_eth_deltas(tx, &tokens_store)
                         .into_iter()
-                        .chain(emit_deltas(tx, &tokens_store))
+                        .chain(extract_balance_deltas_from_tx(tx, |token, transactor| {
+                            let pool_key = format!("pool:{}", hex::encode(transactor));
+                            if let Some(tokens) = tokens_store.get_last(pool_key) {
+                                let token_id = hex::encode(token);
+                                tokens
+                                    .split(":")
+                                    .map(|token| token.to_owned())
+                                    .collect::<Vec<_>>()
+                                    .contains(&token_id)
+                            } else {
+                                false
+                            }
+                        }))
                 })
                 .collect();
 

substreams/ethereum-curve/src/pool_changes.rs

@@ -2,13 +2,10 @@ use substreams::{
     scalar::BigInt,
     store::{StoreGet, StoreGetString},
 };
-use substreams_ethereum::{pb::eth::v2::TransactionTrace, Event};
+use substreams_ethereum::pb::eth::v2::TransactionTrace;
 use tycho_substreams::prelude::*;
 
-use crate::{
+use crate::consts::{ETH_ADDRESS, WETH_ADDRESS};
-    abi,
-    consts::{ETH_ADDRESS, WETH_ADDRESS},
-};
 
 fn get_pool_tokens(pool_address: &Vec<u8>, tokens_store: &StoreGetString) -> Option<Vec<String>> {
     let pool_key = format!("pool:{}", hex::encode(pool_address));
@@ -21,42 +18,6 @@ fn get_pool_tokens(pool_address: &Vec<u8>, tokens_store: &StoreGetString) -> Opt
     )
 }
 
-/// Tracks `Transfers` in and out of tracked pools if it matches the specific tokens.
-pub fn emit_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) -> Vec<BalanceDelta> {
-    tx.logs_with_calls()
-        .filter_map(|(log, _)| {
-            let transfer = abi::erc20::events::Transfer::match_and_decode(log)?;
-            let (component_id, pool_tokens, is_incoming) =
-                if let Some(pool_tokens) = get_pool_tokens(&transfer.to, tokens_store) {
-                    (hex::encode(&transfer.to), pool_tokens, true)
-                } else if let Some(pool_tokens) = get_pool_tokens(&transfer.from, tokens_store) {
-                    (hex::encode(&transfer.from), pool_tokens, false)
-                } else {
-                    return None;
-                };
-
-            let token_id = hex::encode(log.address.clone());
-            if pool_tokens.contains(&token_id) {
-                let delta = if is_incoming { transfer.value } else { transfer.value * -1 };
-                Some(BalanceDelta {
-                    ord: log.ordinal,
-                    tx: Some(Transaction {
-                        to: tx.to.clone(),
-                        from: tx.from.clone(),
-                        hash: tx.hash.clone(),
-                        index: tx.index.into(),
-                    }),
-                    token: hex::decode(token_id).unwrap(),
-                    delta: delta.to_signed_bytes_be(),
-                    component_id: component_id.into(),
-                })
-            } else {
-                None
-            }
-        })
-        .collect::<Vec<_>>()
-}
-
 /// Tracks ETH balance changes in and out of tracked pools if it matches the specific tokens.
 /// Note: Pools might report as WETH or ETH. Some pools might even accept either WETH or ETH and
 ///  convert them on the fly (checkout pools with `WETHOptimized` in the name). It's a bit tricky