valdiate: 3pool and steth-eth

substreams/ethereum-curve/src/modules.rs

@@ -4,6 +4,7 @@ use anyhow::Result;
 use itertools::Itertools;
 use substreams::{
     pb::substreams::StoreDeltas,
+    scalar::BigInt,
     store::{
         StoreAdd, StoreAddBigInt, StoreAddInt64, StoreGet, StoreGetInt64, StoreGetString, StoreNew,
         StoreSet, StoreSetString,
@@ -179,7 +180,7 @@ pub fn map_protocol_changes(
                 .extend_from_slice(&tx_component.components);
         });
 
-    // Balance changes are gathered by the `StoreDelta` based on `PoolBalanceChanged` creating
+    // Balance changes are gathered by the `StoreDelta` based on `TokenExchange`, etc. creating
     //  `BalanceDeltas`. We essentially just process the changes that occured to the `store` this
     //  block. Then, these balance changes are merged onto the existing map of tx contract changes,
     //  inserting a new one if it doesn't exist.
@@ -188,14 +189,16 @@ pub fn map_protocol_changes(
         .into_iter()
         .zip(deltas.balance_deltas)
         .map(|(store_delta, balance_delta)| {
-            let pool_id = key::segment_at(&store_delta.key, 1);
-            let token_id = key::segment_at(&store_delta.key, 3);
+            let new_value_string = String::from_utf8_lossy(&store_delta.new_value).to_string();
             (
                 balance_delta.tx.unwrap(),
                 BalanceChange {
-                    token: hex::decode(token_id).expect("Token ID not valid hex"),
-                    balance: store_delta.new_value,
-                    component_id: hex::decode(pool_id).expect("Token ID not valid hex"),
+                    token: balance_delta.token,
+                    balance: BigInt::try_from(new_value_string)
+                        .unwrap()
+                        .to_signed_bytes_be(),
+                    component_id: hex::decode(balance_delta.component_id)
+                        .expect("Pool ID not valid hex"),
                 },
             )
         })

substreams/ethereum-curve/src/pool_changes.rs

@@ -1,4 +1,5 @@
 use substreams::{
+    hex,
     scalar::BigInt,
     store::{StoreGet, StoreGetInt64, StoreGetString},
 };
@@ -37,9 +38,10 @@ pub fn emit_deltas(
         .collect::<Vec<_>>();
 
     if let Some(event) = abi::pool::events::TokenExchange::match_and_decode(log) {
-        token_change_deltas(event, log)
+        token_change_deltas(tokens, event, log)
     } else if let Some(event) = abi::pool_3pool::events::TokenExchange::match_and_decode(log) {
         token_change_deltas(
+            tokens,
             abi::pool::events::TokenExchange {
                 sold_id: event.sold_id,
                 bought_id: event.bought_id,
@@ -51,6 +53,7 @@ pub fn emit_deltas(
         )
     } else if let Some(event) = abi::pool_steth::events::TokenExchange::match_and_decode(log) {
         token_change_deltas(
+            tokens,
             abi::pool::events::TokenExchange {
                 sold_id: event.sold_id,
                 bought_id: event.bought_id,
@@ -62,6 +65,7 @@ pub fn emit_deltas(
         )
     } else if let Some(event) = abi::pool_tricrypto::events::TokenExchange::match_and_decode(log) {
         token_change_deltas(
+            tokens,
             abi::pool::events::TokenExchange {
                 sold_id: event.sold_id,
                 bought_id: event.bought_id,
@@ -94,24 +98,27 @@ pub fn emit_deltas(
 }
 
 fn token_change_deltas(
+    tokens: Vec<String>,
     event: abi::pool::events::TokenExchange,
     log: LogView<'_>,
 ) -> Option<Vec<BalanceDelta>> {
     let tokens_bought_delta: BigInt = event.tokens_bought * -1;
+    let sold_token_id = &tokens[event.sold_id.to_u64() as usize];
+    let bought_token_id = &tokens[event.bought_id.to_u64() as usize];
     Some(vec![
         BalanceDelta {
             ord: log.log.ordinal,
             tx: Some(tx_from_log(&log)),
-            token: event.sold_id.to_signed_bytes_be(),
+            token: hex::decode(sold_token_id.clone()).unwrap(),
             delta: event.tokens_sold.to_signed_bytes_be(),
-            component_id: log.address().into(),
+            component_id: hex::encode(log.address()).into(),
         },
         BalanceDelta {
             ord: log.log.ordinal,
             tx: Some(tx_from_log(&log)),
-            token: event.bought_id.to_signed_bytes_be(),
+            token: hex::decode(bought_token_id.clone()).unwrap(),
             delta: tokens_bought_delta.to_signed_bytes_be(),
-            component_id: log.address().into(),
+            component_id: hex::encode(log.address()).into(),
         },
     ])
 }
@@ -128,9 +135,9 @@ fn add_liquidity_deltas(
             .map(move |(token_amount, token_id)| BalanceDelta {
                 ord: log.log.ordinal,
                 tx: Some(tx_from_log(&log)),
-                token: token_id.as_str().into(),
+                token: hex::decode(token_id).unwrap(),
                 delta: token_amount.to_signed_bytes_be(),
-                component_id: log.address().into(),
+                component_id: hex::encode(log.address()).into(),
             })
             .collect::<Vec<_>>(),
     )
@@ -148,9 +155,9 @@ fn remove_liquidity_deltas(
             .map(move |(token_amount, token_id)| BalanceDelta {
                 ord: log.log.ordinal,
                 tx: Some(tx_from_log(&log)),
-                token: token_id.as_str().into(),
+                token: hex::decode(token_id).unwrap(),
                 delta: token_amount.to_signed_bytes_be(),
-                component_id: log.address().into(),
+                component_id: hex::encode(log.address()).into(),
             })
             .collect::<Vec<_>>(),
     )

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.