fix(curve-substreams): miscellaneous fixes for balances extractions in Curve (#118)

* fix(curve-substreams): filter out reverted calls in `emit_eth_deltas`

* feat(substreams-sdk): extract balances from `Deposit` events in `extract_balance_deltas_from_tx`

* ci: ignore fmt for abi files

* feat(sdk): also account for `Withdrawal` event from WETH

* chore: reorder mod.rs, remove unused import

* chore: cargo fmt

---------

Co-authored-by: zizou <111426680+flopell@users.noreply.github.com>
Co-authored-by: Thales <thales@datarevenue.com>

substreams/crates/tycho-substreams/src/abi/mod.rs

@@ -1,2 +1,3 @@
 #![allow(clippy::all)]
 pub mod erc20;
+pub mod weth;

substreams/crates/tycho-substreams/src/balances.rs

@@ -172,10 +172,10 @@ pub fn aggregate_balances_changes(
 
 /// Extracts balance deltas from a transaction trace based on a given address predicate.
 ///
-/// This function processes the logs within a transaction trace to identify ERC-20 token transfer
+/// This function processes the logs within a transaction trace to identify ERC-20 token
-/// events. It applies the given predicate to determine which addresses are of interest and extracts
+/// transfer and weth specific events (Deposit and Withdrawal). It applies the given predicate to
-/// the balance changes (deltas) for those addresses. The balance deltas are then returned as a
+/// determine which addresses are of interest and extracts the balance changes (deltas) for those
-/// vector.
+/// addresses. The balance deltas are then returned as a vector.
 ///
 /// # Arguments
 ///
@@ -215,23 +215,30 @@ pub fn extract_balance_deltas_from_tx<F: Fn(&[u8], &[u8]) -> bool>(
 
     tx.logs_with_calls()
         .for_each(|(log, _)| {
+            let mut create_balance_delta = |transactor: &[u8], delta: BigInt| {
+                balance_deltas.push(BalanceDelta {
+                    ord: log.ordinal,
+                    tx: Some(tx.into()),
+                    token: log.address.clone(),
+                    delta: delta.to_signed_bytes_be(),
+                    component_id: hex::encode(transactor).into(),
+                });
+            };
             if let Some(transfer) = abi::erc20::events::Transfer::match_and_decode(log) {
-                let mut create_balance_delta = |transactor: &[u8], delta: BigInt| {
-                    balance_deltas.push(BalanceDelta {
-                        ord: log.ordinal,
-                        tx: Some(tx.into()),
-                        token: log.address.clone(),
-                        delta: delta.to_signed_bytes_be(),
-                        component_id: hex::encode(transactor).into(),
-                    });
-                };
-
                 if address_predicate(&log.address, &transfer.from) {
                     create_balance_delta(&transfer.from, transfer.value.neg());
                 }
                 if address_predicate(&log.address, &transfer.to) {
                     create_balance_delta(&transfer.to, transfer.value);
                 }
+            } else if let Some(deposit) = abi::weth::events::Deposit::match_and_decode(log) {
+                if address_predicate(&log.address, &deposit.dst) {
+                    create_balance_delta(&deposit.dst, deposit.wad);
+                }
+            } else if let Some(withdrawal) = abi::weth::events::Withdrawal::match_and_decode(log) {
+                if address_predicate(&log.address, &withdrawal.src) {
+                    create_balance_delta(&withdrawal.src, withdrawal.wad.neg());
+                }
             }
         });
 

substreams/ethereum-balancer-v2/src/abi/mod.rs

@@ -1,14 +1,14 @@
 #![allow(clippy::all)]
-pub mod yearn_linear_pool_factory;
-pub mod composable_stable_pool_factory;
-pub mod vault;
-pub mod weighted_pool_factory_v4;
-pub mod weighted_pool_tokens_factory;
-pub mod silo_linear_pool_factory;
-pub mod weighted_pool_factory_v3;
-pub mod weighted_pool_factory_v2;
-pub mod euler_linear_pool_factory;
-pub mod weighted_pool_factory_v1;
-pub mod managed_pool_factory;
 pub mod erc_linear_pool_factory;
 pub mod gearbox_linear_pool_factory;
+pub mod weighted_pool_tokens_factory;
+pub mod vault;
+pub mod yearn_linear_pool_factory;
+pub mod managed_pool_factory;
+pub mod weighted_pool_factory_v1;
+pub mod euler_linear_pool_factory;
+pub mod weighted_pool_factory_v2;
+pub mod weighted_pool_factory_v3;
+pub mod weighted_pool_factory_v4;
+pub mod silo_linear_pool_factory;
+pub mod composable_stable_pool_factory;

substreams/ethereum-curve/src/abi/mod.rs

@@ -1,9 +1,9 @@
 #![allow(clippy::all)]
 pub mod crypto_pool_factory;
-pub mod stableswap_factory;
+pub mod erc20;
 pub mod crypto_swap_ng_factory;
 pub mod meta_registry;
+pub mod stableswap_factory;
 pub mod tricrypto_factory;
 pub mod twocrypto_factory;
-pub mod erc20;
 pub mod meta_pool_factory;

substreams/ethereum-curve/src/pool_changes.rs

@@ -5,7 +5,7 @@ use substreams::{
 use substreams_ethereum::pb::eth::v2::TransactionTrace;
 use tycho_substreams::prelude::*;
 
-use crate::consts::{ETH_ADDRESS, WETH_ADDRESS};
+use crate::consts::ETH_ADDRESS;
 
 fn get_pool_tokens(pool_address: &Vec<u8>, tokens_store: &StoreGetString) -> Option<Vec<String>> {
     let pool_key = format!("pool:{}", hex::encode(pool_address));
@@ -27,6 +27,7 @@ fn get_pool_tokens(pool_address: &Vec<u8>, tokens_store: &StoreGetString) -> Opt
 ///  - If neither, it's likely an erroneous ETH transactions that many older pools don't reject.
 pub fn emit_eth_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) -> Vec<BalanceDelta> {
     tx.calls()
+        .filter(|call| !call.call.state_reverted)
         .flat_map(|call| {
             call.call
                 .balance_changes
@@ -35,13 +36,9 @@ pub fn emit_eth_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) ->
                     if let Some(pool_tokens) =
                         get_pool_tokens(&balance_change.address, tokens_store)
                     {
-                        let token = if pool_tokens.contains(&hex::encode(ETH_ADDRESS)) {
+                        if !pool_tokens.contains(&hex::encode(ETH_ADDRESS)) {
-                            ETH_ADDRESS.to_vec()
+                            // The pool that was matched to the call doesn't contain ETH so
-                        } else if pool_tokens.contains(&hex::encode(WETH_ADDRESS)) {
+                            // found eth balance changes are not relevant.
-                            WETH_ADDRESS.to_vec()
-                        } else {
-                            // The pool that was matched to the call doesn't contain either ETH
-                            //  or WETH so found eth balance changes are erroneous.
                             return None;
                         };
 
@@ -68,7 +65,7 @@ pub fn emit_eth_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) ->
                                 hash: tx.hash.clone(),
                                 index: tx.index.into(),
                             }),
-                            token,
+                            token: ETH_ADDRESS.to_vec(),
                             delta: delta.to_signed_bytes_be(),
                             component_id: hex::encode(balance_change.address.clone()).into(),
                         })

substreams/rustfmt.toml

@@ -10,6 +10,7 @@ use_field_init_shorthand = true
 chain_width = 40
 ignore = [
     "crates/tycho-substreams/src/pb",
+    "crates/tycho-substreams/src/abi",
     "ethereum-balancer-v2/src/abi",
     "ethereum-sfraxeth/src/abi",
     "ethereum-sfrax/src/abi",