fix(curve): some general fixes including

Bytes decoding from BalanceChange, sort  BlockBalanceDeltas, transform `0xeeee...` into `0x000...` for consistency, and add a placeholder for missing tokens in `get_token_from_pool`.

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

@@ -5,6 +5,6 @@ pub mod crypto_swap_ng_factory;
 pub mod meta_registry;
 pub mod tricrypto_factory;
 pub mod main_registry;
-pub mod ERC20;
+pub mod erc20;
 pub mod meta_pool_factory;
 pub mod crypto_swap_registry;

substreams/ethereum-curve/src/modules.rs

@@ -110,7 +110,7 @@ pub fn map_relative_balances(
 ) -> Result<BlockBalanceDeltas, anyhow::Error> {
     Ok(BlockBalanceDeltas {
         balance_deltas: {
-            block
+            let mut deltas: Vec<_> = block
                 .transactions()
                 .into_iter()
                 .flat_map(|tx| {
@@ -118,7 +118,13 @@ pub fn map_relative_balances(
                         .into_iter()
                         .chain(emit_deltas(&tx, &tokens_store))
                 })
-                .collect::<Vec<_>>()
+                .collect();
+
+            // Keep it consistent with how it's inserted in the store. This step is important
+            // because we use a zip on the store deltas and balance deltas later.
+            deltas.sort_unstable_by(|a, b| a.ord.cmp(&b.ord));
+
+            deltas
         },
     })
 }
@@ -234,6 +240,18 @@ pub fn map_protocol_changes(
         &mut transaction_contract_changes,
     );
 
+    for change in transaction_contract_changes.values_mut() {
+        for balance_change in change.balance_changes.iter_mut() {
+            replace_eth_address(&mut balance_change.token);
+        }
+
+        for component_change in change.component_changes.iter_mut() {
+            for token in component_change.tokens.iter_mut() {
+                replace_eth_address(token);
+            }
+        }
+    }
+
     // Process all `transaction_contract_changes` for final output in the `BlockContractChanges`,
     //  sorted by transaction index (the key).
     Ok(BlockContractChanges {
@@ -264,3 +282,10 @@ pub fn map_protocol_changes(
             .collect::<Vec<_>>(),
     })
 }
+
+fn replace_eth_address(token: &mut Vec<u8>) {
+    let eth_address = [238u8; 20]; // 0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee
+    if *token == eth_address {
+        *token = [0u8; 20].to_vec();
+    }
+}

substreams/ethereum-curve/src/pool_changes.rs

@@ -26,7 +26,7 @@ pub fn emit_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) -> Vec<
     tx.logs_with_calls()
         .into_iter()
         .filter_map(|(log, _)| {
-            let transfer = abi::ERC20::events::Transfer::match_and_decode(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) {
@@ -75,7 +75,9 @@ pub fn emit_eth_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) ->
                 .balance_changes
                 .iter()
                 .filter_map(|balance_change| {
-                    if let Some(pool_tokens) = get_pool_tokens(&call.call.address, tokens_store) {
+                    if let Some(pool_tokens) =
+                        get_pool_tokens(&balance_change.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)) {
@@ -88,11 +90,19 @@ pub fn emit_eth_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) ->
 
                         // 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 =
+                        let delta = BigInt::from_unsigned_bytes_be(
-                            BigInt::from_store_bytes(&balance_change.new_value.clone()?.bytes) -
+                            &balance_change
-                                BigInt::from_store_bytes(
+                                .new_value
-                                    &balance_change.old_value.clone()?.bytes,
+                                .clone()
-                                );
+                                .unwrap_or_default()
+                                .bytes,
+                        ) - BigInt::from_unsigned_bytes_be(
+                            &balance_change
+                                .old_value
+                                .clone()
+                                .unwrap_or_default()
+                                .bytes,
+                        );
                         Some(BalanceDelta {
                             ord: call.call.end_ordinal,
                             tx: Some(Transaction {
@@ -103,7 +113,7 @@ pub fn emit_eth_deltas(tx: &TransactionTrace, tokens_store: &StoreGetString) ->
                             }),
                             token,
                             delta: delta.to_signed_bytes_be(),
-                            component_id: call.call.address.clone(),
+                            component_id: hex::encode(balance_change.address.clone()).into(),
                         })
                     } else {
                         None

substreams/ethereum-curve/src/pool_factories.rs

@@ -693,7 +693,7 @@ pub fn address_map(
 /// If all else fails, we force an `unwrap` to trigger a `panic` so that we can resolve this by
 ///  adding onto our map of `pool` -> `token` addresses.
 fn get_token_from_pool(pool: &Vec<u8>) -> Vec<u8> {
-    abi::ERC20::functions::Name {}
+    abi::erc20::functions::Name {}
         .call(pool.clone())
         .and(Some(pool.clone()))
         .or_else(|| {
@@ -707,7 +707,9 @@ fn get_token_from_pool(pool: &Vec<u8>) -> Vec<u8> {
                 "bebc44782c7db0a1a60cb6fe97d0b483032ff1c7" => {
                     hex::decode("6c3F90f043a72FA612cbac8115EE7e52BDe6E490").ok()
                 }
-                _ => None,
+                // Placeholder if we can't find the token. It will help us to detect these missing
+                // token easily with a SQL query.
+                _ => hex::decode("1111111111111111111111111111111111111111").ok(),
             }
         })
         .unwrap()

substreams/ethereum-curve/src/pools.rs

@@ -24,7 +24,7 @@ struct PoolQueryParams {
 /// Static attributes are defined as a vector of tuples with the name and value of the attribute.
 ///  These contain things like the pool type, specific pool fees, etc. You can see
 ///  `pool_factories.rs` for an example of the modern curve pool attributes and also the ones chosen
-///  for 3pool, etc. 
+///  for 3pool, etc.
 ///
 /// This function can error based on some basic parsing errors and deeper down hex decoding errors
 ///  if various addresses are not formatted properly.

substreams/rustfmt.toml

@@ -11,4 +11,5 @@ chain_width = 40
 ignore = [
     "crates/tycho-substreams/src/pb",
     "ethereum-balancer/src/abi",
-]
+    "ethereum-curve/src/abi",
+]