feat: add native eth transfers

substreams/Cargo.lock

@@ -237,7 +237,7 @@ dependencies = [
  "getrandom",
  "hex",
  "hex-literal 0.4.1",
- "itertools 0.12.1",
+ "itertools 0.13.0",
  "num-bigint",
  "prost 0.11.9",
  "prost-types 0.12.3",
@@ -453,6 +453,15 @@ dependencies = [
  "either",
 ]
 
+[[package]]
+name = "itertools"
+version = "0.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186"
+dependencies = [
+ "either",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.10"

substreams/ethereum-curve/Cargo.toml

@@ -18,10 +18,10 @@ hex.workspace = true
 bytes = "1.5.0"
 anyhow = "1.0.75"
 num-bigint = "0.4.4"
-itertools = "0.12.0"
 tycho-substreams.workspace = true
 serde = { version = "1.0", features = ["derive"] }
 serde_qs = "0.13.0"
+itertools = "0.13.0"
 
 [build-dependencies]
 anyhow = "1"

substreams/ethereum-curve/src/consts.rs

@@ -0,0 +1,16 @@
+use substreams::hex;
+
+// Registries
+pub const META_REGISTRY: [u8; 20] = hex!("F98B45FA17DE75FB1aD0e7aFD971b0ca00e379fC");
+
+// Factories
+pub const CRYPTO_POOL_FACTORY: [u8; 20] = hex!("F18056Bbd320E96A48e3Fbf8bC061322531aac99");
+pub const META_POOL_FACTORY: [u8; 20] = hex!("B9fC157394Af804a3578134A6585C0dc9cc990d4");
+pub const META_POOL_FACTORY_OLD: [u8; 20] = hex!("0959158b6040D32d04c301A72CBFD6b39E21c9AE");
+pub const CRYPTO_SWAP_NG_FACTORY: [u8; 20] = hex!("6A8cbed756804B16E05E741eDaBd5cB544AE21bf");
+pub const TRICRYPTO_FACTORY: [u8; 20] = hex!("0c0e5f2fF0ff18a3be9b835635039256dC4B4963");
+pub const STABLESWAP_FACTORY: [u8; 20] = hex!("4F8846Ae9380B90d2E71D5e3D042dff3E7ebb40d");
+
+// Important addresses
+pub const WETH_ADDRESS: [u8; 20] = hex!("C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2");
+pub const ETH_ADDRESS: [u8; 20] = hex!("EeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE");

substreams/ethereum-curve/src/lib.rs

@@ -1,4 +1,5 @@
 mod abi;
+mod consts;
 pub mod modules;
 mod pool_changes;
 mod pool_factories;

substreams/ethereum-curve/src/modules.rs

@@ -10,7 +10,11 @@ use substreams::{
 
 use substreams_ethereum::pb::eth;
 
-use crate::{pool_changes::emit_deltas, pool_factories, pools::emit_specific_pools};
+use crate::{
+    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::*,
 };
@@ -107,8 +111,13 @@ pub fn map_relative_balances(
     Ok(BlockBalanceDeltas {
         balance_deltas: {
             block
-                .logs()
+                .transactions()
-                .filter_map(|log| emit_deltas(log, &tokens_store))
+                .into_iter()
+                .flat_map(|tx| {
+                    emit_eth_deltas(&tx, &tokens_store)
+                        .into_iter()
+                        .chain(emit_deltas(&tx, &tokens_store))
+                })
                 .collect::<Vec<_>>()
         },
     })
@@ -195,7 +204,7 @@ pub fn map_protocol_changes(
             )
         })
         // We need to group the balance changes by tx hash for the `TransactionContractChanges` agg
-        .group_by(|(tx, _)| TransactionWrapper(tx.clone()))
+        .chunk_by(|(tx, _)| TransactionWrapper(tx.clone()))
         .into_iter()
         .for_each(|(tx_wrapped, group)| {
             let tx = tx_wrapped.0;

substreams/ethereum-curve/src/pool_changes.rs

@@ -1,17 +1,14 @@
-use substreams::store::{StoreGet, StoreGetString};
+use substreams::{
-use substreams_ethereum::{block_view::LogView, Event};
+    scalar::BigInt,
+    store::{StoreGet, StoreGetString},
+};
+use substreams_ethereum::{pb::eth::v2::TransactionTrace, Event};
 use tycho_substreams::prelude::*;
 
-use crate::abi;
+use crate::{
-
+    abi,
-fn tx_from_log(log: &LogView) -> Transaction {
+    consts::{ETH_ADDRESS, WETH_ADDRESS},
-    Transaction {
+};
-        hash: log.receipt.transaction.hash.clone(),
-        from: log.receipt.transaction.from.clone(),
-        to: log.receipt.transaction.to.clone(),
-        index: Into::<u64>::into(log.receipt.transaction.index),
-    }
-}
 
 fn get_pool_tokens(pool_address: &Vec<u8>, tokens_store: &StoreGetString) -> Option<Vec<String>> {
     let pool_key = format!("pool:{}", hex::encode(&pool_address));
@@ -25,30 +22,94 @@ 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(log: LogView, tokens_store: &StoreGetString) -> Option<BalanceDelta> {
+pub fn emit_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) -> Vec<BalanceDelta> {
-    let transfer = abi::ERC20::events::Transfer::match_and_decode(log)?;
+    tx.logs_with_calls()
+        .into_iter()
+        .filter_map(|(log, _)| {
+            let transfer = abi::ERC20::events::Transfer::match_and_decode(log)?;
 
-    let (component_id, pool_tokens, is_incoming) =
+            let (component_id, pool_tokens, is_incoming) =
-        if let Some(pool_tokens) = get_pool_tokens(&transfer.to, tokens_store) {
+                if let Some(pool_tokens) = get_pool_tokens(&transfer.to, tokens_store) {
-            (hex::encode(&transfer.to), pool_tokens, true)
+                    (hex::encode(&transfer.to), pool_tokens, true)
-        } else if let Some(pool_tokens) = get_pool_tokens(&transfer.from, tokens_store) {
+                } else if let Some(pool_tokens) = get_pool_tokens(&transfer.from, tokens_store) {
-            (hex::encode(&transfer.from), pool_tokens, false)
+                    (hex::encode(&transfer.from), pool_tokens, false)
-        } else {
+                } else {
-            return None;
+                    return None;
-        };
+                };
 
-    let token_id = hex::encode(log.address());
+            let token_id = hex::encode(log.address.clone());
-    if pool_tokens.contains(&token_id) {
+            if pool_tokens.contains(&token_id) {
-        let delta = if is_incoming { transfer.value } else { transfer.value * -1 };
+                let delta = if is_incoming { transfer.value } else { transfer.value * -1 };
-        Some(BalanceDelta {
+                Some(BalanceDelta {
-            ord: log.log.ordinal,
+                    ord: log.ordinal,
-            tx: Some(tx_from_log(&log)),
+                    tx: Some(Transaction {
-            token: hex::decode(token_id).unwrap(),
+                        to: tx.to.clone(),
-            delta: delta.to_signed_bytes_be(),
+                        from: tx.from.clone(),
-            component_id: component_id.into(),
+                        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 {
+                substreams::log::info!("Token {:?} not in pool: {:?}", token_id, &component_id);
+                None
+            }
         })
-    } else {
+        .collect::<Vec<_>>()
-        substreams::log::info!("Token {:?} not in pool: {:?}", token_id, &component_id);
+}
-        None
+
-    }
+/// 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
+///  to figure this stuff out on the fly, but our rule of thumb is as follows:
+///  - If a pool reports ETH in the `pool_tokens`, we use the fake ETH erc20 address.
+///  - If a pool reports WETH, we report with the WETH erc20 address.
+///  - 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()
+        .into_iter()
+        .flat_map(|call| {
+            call.call
+                .balance_changes
+                .iter()
+                .filter_map(|balance_change| {
+                    if let Some(pool_tokens) = get_pool_tokens(&call.call.address, tokens_store) {
+                        let token = if pool_tokens.contains(&hex::encode(ETH_ADDRESS)) {
+                            ETH_ADDRESS.to_vec()
+                        } else if pool_tokens.contains(&hex::encode(WETH_ADDRESS)) {
+                            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;
+                        };
+
+                        // We need to convert to the usable `BigInt` type to be able to calculate
+                        //  subtraction. This is seemingly the easiest way to do this.
+                        let delta =
+                            BigInt::from_store_bytes(&balance_change.new_value.clone()?.bytes) -
+                                BigInt::from_store_bytes(
+                                    &balance_change.old_value.clone()?.bytes,
+                                );
+                        Some(BalanceDelta {
+                            ord: call.call.end_ordinal,
+                            tx: Some(Transaction {
+                                to: tx.to.clone(),
+                                from: tx.from.clone(),
+                                hash: tx.hash.clone(),
+                                index: tx.index.into(),
+                            }),
+                            token,
+                            delta: delta.to_signed_bytes_be(),
+                            component_id: call.call.address.clone(),
+                        })
+                    } else {
+                        None
+                    }
+                })
+                .collect::<Vec<_>>()
+        })
+        .collect::<Vec<_>>()
 }

substreams/ethereum-curve/src/pool_factories.rs

@@ -4,22 +4,11 @@ use substreams_ethereum::{
 };
 
 use crate::abi;
-use substreams::hex;
 use tycho_substreams::prelude::*;
 
 use substreams::scalar::BigInt;
 
-const META_REGISTRY: [u8; 20] = hex!("F98B45FA17DE75FB1aD0e7aFD971b0ca00e379fC");
+use crate::consts::*;
-
-const CRYPTO_POOL_FACTORY: [u8; 20] = hex!("F18056Bbd320E96A48e3Fbf8bC061322531aac99");
-const META_POOL_FACTORY: [u8; 20] = hex!("B9fC157394Af804a3578134A6585C0dc9cc990d4");
-const META_POOL_FACTORY_OLD: [u8; 20] = hex!("0959158b6040D32d04c301A72CBFD6b39E21c9AE");
-const CRYPTO_SWAP_NG_FACTORY: [u8; 20] = hex!("6A8cbed756804B16E05E741eDaBd5cB544AE21bf");
-const TRICRYPTO_FACTORY: [u8; 20] = hex!("0c0e5f2fF0ff18a3be9b835635039256dC4B4963");
-const STABLESWAP_FACTORY: [u8; 20] = hex!("4F8846Ae9380B90d2E71D5e3D042dff3E7ebb40d");
-
-const WETH_ADDRESS: [u8; 20] = hex!("C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2");
-const ETH_ADDRESS: [u8; 20] = hex!("EeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE");
 
 /// This trait defines some helpers for serializing and deserializing `Vec<BigInt>` which is needed
 ///  to be able to encode some of the `Attribute`s. This should also be handled by any downstream