chore: code formatting

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

@@ -13,7 +13,6 @@ pub fn json_serialize_value<T: serde::Serialize + Debug>(v: T) -> Vec<u8> {
         .to_vec()
 }
 
-
 /// Encodes a list of addresses (in byte representation) into json.
 ///
 /// Converts each address to a 0x prefixed hex string and then serializes
@@ -30,7 +29,6 @@ pub fn json_serialize_address_list(addresses: &[Vec<u8>]) -> Vec<u8> {
     )
 }
 
-
 /// Encodes a list of BigInt values into json.
 ///
 /// Converts each integer to a 0x prefixed hex string and then serializes

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

@@ -10,13 +10,14 @@
 /// more [here](https://streamingfastio.medium.com/new-block-model-to-accelerate-chain-integration-9f65126e5425)
 use std::collections::HashMap;
 
+use crate::{
+    models::{InterimContractChange, TransactionChanges},
+    prelude::TransactionChangesBuilder,
+};
 use substreams_ethereum::pb::{
     eth,
-    eth::v2::block::DetailLevel, eth::v2::CallType
+    eth::v2::{block::DetailLevel, TransactionTrace, CallType},
 };
-use substreams_ethereum::pb::eth::v2::TransactionTrace;
-use crate::models::{InterimContractChange, TransactionChanges};
-use crate::prelude::TransactionChangesBuilder;
 
 /// Extracts and aggregates contract changes from a block.
 ///
@@ -49,10 +50,7 @@ pub fn extract_contract_changes<F: Fn(&[u8]) -> bool>(
     inclusion_predicate: F,
     transaction_changes: &mut HashMap<u64, TransactionChanges>,
 ) {
-    extract_contract_changes_generic(
+    extract_contract_changes_generic(block, inclusion_predicate, |tx, changed_contracts| {
-        block,
-        inclusion_predicate,
-        |tx, changed_contracts| {
         transaction_changes
             .entry(tx.index.into())
             .or_insert_with(|| TransactionChanges::new(&(tx.into())))
@@ -63,20 +61,15 @@ pub fn extract_contract_changes<F: Fn(&[u8]) -> bool>(
                     .into_values()
                     .map(|change| change.into()),
             );
-        },
+    })
-    )
 }
 
-
 pub fn extract_contract_changes_builder<F: Fn(&[u8]) -> bool>(
     block: &eth::v2::Block,
     inclusion_predicate: F,
     transaction_changes: &mut HashMap<u64, TransactionChangesBuilder>,
 ) {
-    extract_contract_changes_generic(
+    extract_contract_changes_generic(block, inclusion_predicate, |tx, changed_contracts| {
-        block,
-        inclusion_predicate,
-        |tx, changed_contracts| {
         let builder = transaction_changes
             .entry(tx.index.into())
             .or_insert_with(|| TransactionChangesBuilder::new(&(tx.into())));
@@ -84,11 +77,13 @@ pub fn extract_contract_changes_builder<F: Fn(&[u8]) -> bool>(
             .clone()
             .into_iter()
             .for_each(|(_, change)| builder.add_contract_changes(&change));
-        },
+    })
-    )
 }
 
-fn extract_contract_changes_generic<F: Fn(&[u8]) -> bool, G: FnMut(&TransactionTrace, &HashMap<Vec<u8>, InterimContractChange>)>(
+fn extract_contract_changes_generic<
+    F: Fn(&[u8]) -> bool,
+    G: FnMut(&TransactionTrace, &HashMap<Vec<u8>, InterimContractChange>),
+>(
     block: &eth::v2::Block,
     inclusion_predicate: F,
     mut store_changes: G,

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

@@ -1,10 +1,10 @@
 mod abi;
+pub mod attributes;
 pub mod balances;
 pub mod contract;
 mod mock_store;
 pub mod models;
 mod pb;
-pub mod attributes;
 
 pub mod prelude {
     pub use super::models::*;

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

@@ -66,7 +66,7 @@ impl TransactionChangesBuilder {
     }
 
     /// Unique contract/account addresses that have been changed so far.
-    pub fn changed_contracts(&self) -> impl Iterator<Item=&[u8]> {
+    pub fn changed_contracts(&self) -> impl Iterator<Item = &[u8]> {
         self.contract_changes
             .keys()
             .map(|k| k.as_slice())

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

@@ -1,11 +1,11 @@
 #![allow(clippy::all)]
-pub mod yearn_linear_pool_factory;
 pub mod composable_stable_pool_factory;
-pub mod vault;
-pub mod weighted_pool_tokens_factory;
-pub mod silo_linear_pool_factory;
-pub mod euler_linear_pool_factory;
-pub mod weighted_pool_factory;
-pub mod managed_pool_factory;
 pub mod erc_linear_pool_factory;
+pub mod euler_linear_pool_factory;
 pub mod gearbox_linear_pool_factory;
+pub mod managed_pool_factory;
+pub mod silo_linear_pool_factory;
+pub mod vault;
+pub mod weighted_pool_factory;
+pub mod weighted_pool_tokens_factory;
+pub mod yearn_linear_pool_factory;

substreams/ethereum-balancer/src/abi/weighted_pool_tokens_factory.rs

@@ -1,7 +1,8 @@
-    const INTERNAL_ERR: &'static str = "`ethabi_derive` internal error";
+
-    /// Contract's functions.
+const INTERNAL_ERR: &'static str = "`ethabi_derive` internal error";
-    #[allow(dead_code, unused_imports, unused_variables)]
+/// Contract's functions.
-    pub mod functions {
+#[allow(dead_code, unused_imports, unused_variables)]
+pub mod functions {
     use super::INTERNAL_ERR;
     #[derive(Debug, Clone, PartialEq)]
     pub struct Create {
@@ -15,9 +16,7 @@
     }
     impl Create {
         const METHOD_ID: [u8; 4] = [21u8, 150u8, 1u8, 155u8];
-            pub fn decode(
+        pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<Self, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<Self, String> {
             let maybe_data = call.input.get(4..);
             if maybe_data.is_none() {
                 return Err("no data to decode".to_string());
@@ -26,12 +25,8 @@
                 &[
                     ethabi::ParamType::String,
                     ethabi::ParamType::String,
-                            ethabi::ParamType::Array(
+                    ethabi::ParamType::Array(Box::new(ethabi::ParamType::Address)),
-                                Box::new(ethabi::ParamType::Address),
+                    ethabi::ParamType::Array(Box::new(ethabi::ParamType::Uint(256usize))),
-                            ),
-                            ethabi::ParamType::Array(
-                                Box::new(ethabi::ParamType::Uint(256usize)),
-                            ),
                     ethabi::ParamType::Uint(256usize),
                     ethabi::ParamType::Bool,
                     ethabi::ParamType::Address,
@@ -58,7 +53,11 @@
                     .expect(INTERNAL_ERR)
                     .into_iter()
                     .map(|inner| {
-                            inner.into_address().expect(INTERNAL_ERR).as_bytes().to_vec()
+                        inner
+                            .into_address()
+                            .expect(INTERNAL_ERR)
+                            .as_bytes()
+                            .to_vec()
                     })
                     .collect(),
                 weights: values
@@ -101,17 +100,14 @@
             })
         }
         pub fn encode(&self) -> Vec<u8> {
-                let data = ethabi::encode(
+            let data = ethabi::encode(&[
-                    &[
                 ethabi::Token::String(self.name.clone()),
                 ethabi::Token::String(self.symbol.clone()),
                 {
                     let v = self
                         .tokens
                         .iter()
-                                .map(|inner| ethabi::Token::Address(
+                        .map(|inner| ethabi::Token::Address(ethabi::Address::from_slice(&inner)))
-                                    ethabi::Address::from_slice(&inner),
-                                ))
                         .collect();
                     ethabi::Token::Array(v)
                 },
@@ -119,8 +115,8 @@
                     let v = self
                         .weights
                         .iter()
-                                .map(|inner| ethabi::Token::Uint(
+                        .map(|inner| {
-                                    ethabi::Uint::from_big_endian(
+                            ethabi::Token::Uint(ethabi::Uint::from_big_endian(
                                 match inner.clone().to_bytes_be() {
                                     (num_bigint::Sign::Plus, bytes) => bytes,
                                     (num_bigint::Sign::NoSign, bytes) => bytes,
@@ -129,14 +125,17 @@
                                     }
                                 }
                                 .as_slice(),
-                                    ),
                             ))
+                        })
                         .collect();
                     ethabi::Token::Array(v)
                 },
-                        ethabi::Token::Uint(
+                ethabi::Token::Uint(ethabi::Uint::from_big_endian(
-                            ethabi::Uint::from_big_endian(
+                    match self
-                                match self.swap_fee_percentage.clone().to_bytes_be() {
+                        .swap_fee_percentage
+                        .clone()
+                        .to_bytes_be()
+                    {
                         (num_bigint::Sign::Plus, bytes) => bytes,
                         (num_bigint::Sign::NoSign, bytes) => bytes,
                         (num_bigint::Sign::Minus, _) => {
@@ -144,12 +143,10 @@
                         }
                     }
                     .as_slice(),
-                            ),
+                )),
-                        ),
                 ethabi::Token::Bool(self.oracle_enabled.clone()),
                 ethabi::Token::Address(ethabi::Address::from_slice(&self.owner)),
-                    ],
+            ]);
-                );
             let mut encoded = Vec::with_capacity(4 + data.len());
             encoded.extend(Self::METHOD_ID);
             encoded.extend(data);
@@ -161,20 +158,15 @@
             Self::output(call.return_data.as_ref())
         }
         pub fn output(data: &[u8]) -> Result<Vec<u8>, String> {
-                let mut values = ethabi::decode(
+            let mut values = ethabi::decode(&[ethabi::ParamType::Address], data.as_ref())
-                        &[ethabi::ParamType::Address],
-                        data.as_ref(),
-                    )
                 .map_err(|e| format!("unable to decode output data: {:?}", e))?;
-                Ok(
+            Ok(values
-                    values
                 .pop()
                 .expect("one output data should have existed")
                 .into_address()
                 .expect(INTERNAL_ERR)
                 .as_bytes()
-                        .to_vec(),
+                .to_vec())
-                )
         }
         pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool {
             match call.input.get(0..4) {
@@ -185,9 +177,7 @@
         pub fn call(&self, address: Vec<u8>) -> Option<Vec<u8>> {
             use substreams_ethereum::pb::eth::rpc;
             let rpc_calls = rpc::RpcCalls {
-                    calls: vec![
+                calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }],
-                        rpc::RpcCall { to_addr : address, data : self.encode(), }
-                    ],
             };
             let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses;
             let response = responses
@@ -202,7 +192,8 @@
                     use substreams_ethereum::Function;
                     substreams::log::info!(
                         "Call output for function `{}` failed to decode with error: {}",
-                            Self::NAME, err
+                        Self::NAME,
+                        err
                     );
                     None
                 }
@@ -214,9 +205,7 @@
         fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool {
             Self::match_call(call)
         }
-            fn decode(
+        fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<Self, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<Self, String> {
             Self::decode(call)
         }
         fn encode(&self) -> Vec<u8> {
@@ -232,9 +221,7 @@
     pub struct GetPauseConfiguration {}
     impl GetPauseConfiguration {
         const METHOD_ID: [u8; 4] = [45u8, 164u8, 124u8, 64u8];
-            pub fn decode(
+        pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<Self, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<Self, String> {
             Ok(Self {})
         }
         pub fn encode(&self) -> Vec<u8> {
@@ -246,23 +233,14 @@
         }
         pub fn output_call(
             call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<
+        ) -> Result<(substreams::scalar::BigInt, substreams::scalar::BigInt), String> {
-                (substreams::scalar::BigInt, substreams::scalar::BigInt),
-                String,
-            > {
             Self::output(call.return_data.as_ref())
         }
         pub fn output(
             data: &[u8],
-            ) -> Result<
+        ) -> Result<(substreams::scalar::BigInt, substreams::scalar::BigInt), String> {
-                (substreams::scalar::BigInt, substreams::scalar::BigInt),
-                String,
-            > {
             let mut values = ethabi::decode(
-                        &[
+                &[ethabi::ParamType::Uint(256usize), ethabi::ParamType::Uint(256usize)],
-                            ethabi::ParamType::Uint(256usize),
-                            ethabi::ParamType::Uint(256usize),
-                        ],
                 data.as_ref(),
             )
             .map_err(|e| format!("unable to decode output data: {:?}", e))?;
@@ -302,9 +280,7 @@
         ) -> Option<(substreams::scalar::BigInt, substreams::scalar::BigInt)> {
             use substreams_ethereum::pb::eth::rpc;
             let rpc_calls = rpc::RpcCalls {
-                    calls: vec![
+                calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }],
-                        rpc::RpcCall { to_addr : address, data : self.encode(), }
-                    ],
             };
             let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses;
             let response = responses
@@ -319,7 +295,8 @@
                     use substreams_ethereum::Function;
                     substreams::log::info!(
                         "Call output for function `{}` failed to decode with error: {}",
-                            Self::NAME, err
+                        Self::NAME,
+                        err
                     );
                     None
                 }
@@ -331,24 +308,22 @@
         fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool {
             Self::match_call(call)
         }
-            fn decode(
+        fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<Self, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<Self, String> {
             Self::decode(call)
         }
         fn encode(&self) -> Vec<u8> {
             self.encode()
         }
     }
-        impl substreams_ethereum::rpc::RPCDecodable<
+    impl
-            (substreams::scalar::BigInt, substreams::scalar::BigInt),
+        substreams_ethereum::rpc::RPCDecodable<(
-        > for GetPauseConfiguration {
+            substreams::scalar::BigInt,
+            substreams::scalar::BigInt,
+        )> for GetPauseConfiguration
+    {
         fn output(
             data: &[u8],
-            ) -> Result<
+        ) -> Result<(substreams::scalar::BigInt, substreams::scalar::BigInt), String> {
-                (substreams::scalar::BigInt, substreams::scalar::BigInt),
-                String,
-            > {
             Self::output(data)
         }
     }
@@ -356,9 +331,7 @@
     pub struct GetVault {}
     impl GetVault {
         const METHOD_ID: [u8; 4] = [141u8, 146u8, 138u8, 248u8];
-            pub fn decode(
+        pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<Self, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<Self, String> {
             Ok(Self {})
         }
         pub fn encode(&self) -> Vec<u8> {
@@ -374,20 +347,15 @@
             Self::output(call.return_data.as_ref())
         }
         pub fn output(data: &[u8]) -> Result<Vec<u8>, String> {
-                let mut values = ethabi::decode(
+            let mut values = ethabi::decode(&[ethabi::ParamType::Address], data.as_ref())
-                        &[ethabi::ParamType::Address],
-                        data.as_ref(),
-                    )
                 .map_err(|e| format!("unable to decode output data: {:?}", e))?;
-                Ok(
+            Ok(values
-                    values
                 .pop()
                 .expect("one output data should have existed")
                 .into_address()
                 .expect(INTERNAL_ERR)
                 .as_bytes()
-                        .to_vec(),
+                .to_vec())
-                )
         }
         pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool {
             match call.input.get(0..4) {
@@ -398,9 +366,7 @@
         pub fn call(&self, address: Vec<u8>) -> Option<Vec<u8>> {
             use substreams_ethereum::pb::eth::rpc;
             let rpc_calls = rpc::RpcCalls {
-                    calls: vec![
+                calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }],
-                        rpc::RpcCall { to_addr : address, data : self.encode(), }
-                    ],
             };
             let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses;
             let response = responses
@@ -415,7 +381,8 @@
                     use substreams_ethereum::Function;
                     substreams::log::info!(
                         "Call output for function `{}` failed to decode with error: {}",
-                            Self::NAME, err
+                        Self::NAME,
+                        err
                     );
                     None
                 }
@@ -427,9 +394,7 @@
         fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool {
             Self::match_call(call)
         }
-            fn decode(
+        fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<Self, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<Self, String> {
             Self::decode(call)
         }
         fn encode(&self) -> Vec<u8> {
@@ -447,17 +412,12 @@
     }
     impl IsPoolFromFactory {
         const METHOD_ID: [u8; 4] = [102u8, 52u8, 183u8, 83u8];
-            pub fn decode(
+        pub fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<Self, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<Self, String> {
             let maybe_data = call.input.get(4..);
             if maybe_data.is_none() {
                 return Err("no data to decode".to_string());
             }
-                let mut values = ethabi::decode(
+            let mut values = ethabi::decode(&[ethabi::ParamType::Address], maybe_data.unwrap())
-                        &[ethabi::ParamType::Address],
-                        maybe_data.unwrap(),
-                    )
                 .map_err(|e| format!("unable to decode call.input: {:?}", e))?;
             values.reverse();
             Ok(Self {
@@ -471,32 +431,24 @@
             })
         }
         pub fn encode(&self) -> Vec<u8> {
-                let data = ethabi::encode(
+            let data =
-                    &[ethabi::Token::Address(ethabi::Address::from_slice(&self.pool))],
+                ethabi::encode(&[ethabi::Token::Address(ethabi::Address::from_slice(&self.pool))]);
-                );
             let mut encoded = Vec::with_capacity(4 + data.len());
             encoded.extend(Self::METHOD_ID);
             encoded.extend(data);
             encoded
         }
-            pub fn output_call(
+        pub fn output_call(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<bool, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<bool, String> {
             Self::output(call.return_data.as_ref())
         }
         pub fn output(data: &[u8]) -> Result<bool, String> {
-                let mut values = ethabi::decode(
+            let mut values = ethabi::decode(&[ethabi::ParamType::Bool], data.as_ref())
-                        &[ethabi::ParamType::Bool],
-                        data.as_ref(),
-                    )
                 .map_err(|e| format!("unable to decode output data: {:?}", e))?;
-                Ok(
+            Ok(values
-                    values
                 .pop()
                 .expect("one output data should have existed")
                 .into_bool()
-                        .expect(INTERNAL_ERR),
+                .expect(INTERNAL_ERR))
-                )
         }
         pub fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool {
             match call.input.get(0..4) {
@@ -507,9 +459,7 @@
         pub fn call(&self, address: Vec<u8>) -> Option<bool> {
             use substreams_ethereum::pb::eth::rpc;
             let rpc_calls = rpc::RpcCalls {
-                    calls: vec![
+                calls: vec![rpc::RpcCall { to_addr: address, data: self.encode() }],
-                        rpc::RpcCall { to_addr : address, data : self.encode(), }
-                    ],
             };
             let responses = substreams_ethereum::rpc::eth_call(&rpc_calls).responses;
             let response = responses
@@ -524,7 +474,8 @@
                     use substreams_ethereum::Function;
                     substreams::log::info!(
                         "Call output for function `{}` failed to decode with error: {}",
-                            Self::NAME, err
+                        Self::NAME,
+                        err
                     );
                     None
                 }
@@ -536,9 +487,7 @@
         fn match_call(call: &substreams_ethereum::pb::eth::v2::Call) -> bool {
             Self::match_call(call)
         }
-            fn decode(
+        fn decode(call: &substreams_ethereum::pb::eth::v2::Call) -> Result<Self, String> {
-                call: &substreams_ethereum::pb::eth::v2::Call,
-            ) -> Result<Self, String> {
             Self::decode(call)
         }
         fn encode(&self) -> Vec<u8> {
@@ -550,10 +499,10 @@
             Self::output(data)
         }
     }
-    }
+}
-    /// Contract's events.
+/// Contract's events.
-    #[allow(dead_code, unused_imports, unused_variables)]
+#[allow(dead_code, unused_imports, unused_variables)]
-    pub mod events {
+pub mod events {
     use super::INTERNAL_ERR;
     #[derive(Debug, Clone, PartialEq)]
     pub struct PoolCreated {
@@ -561,38 +510,9 @@
     }
     impl PoolCreated {
         const TOPIC_ID: [u8; 32] = [
-                131u8,
+            131u8, 164u8, 143u8, 188u8, 252u8, 153u8, 19u8, 53u8, 49u8, 78u8, 116u8, 208u8, 73u8,
-                164u8,
+            106u8, 171u8, 106u8, 25u8, 135u8, 233u8, 146u8, 221u8, 200u8, 93u8, 221u8, 188u8,
-                143u8,
+            196u8, 214u8, 221u8, 110u8, 242u8, 233u8, 252u8,
-                188u8,
-                252u8,
-                153u8,
-                19u8,
-                53u8,
-                49u8,
-                78u8,
-                116u8,
-                208u8,
-                73u8,
-                106u8,
-                171u8,
-                106u8,
-                25u8,
-                135u8,
-                233u8,
-                146u8,
-                221u8,
-                200u8,
-                93u8,
-                221u8,
-                188u8,
-                196u8,
-                214u8,
-                221u8,
-                110u8,
-                242u8,
-                233u8,
-                252u8,
         ];
         pub fn match_log(log: &substreams_ethereum::pb::eth::v2::Log) -> bool {
             if log.topics.len() != 2usize {
@@ -601,17 +521,16 @@
             if log.data.len() != 0usize {
                 return false;
             }
-                return log.topics.get(0).expect("bounds already checked").as_ref()
+            return log
+                .topics
+                .get(0)
+                .expect("bounds already checked")
+                .as_ref()
                 == Self::TOPIC_ID;
         }
-            pub fn decode(
+        pub fn decode(log: &substreams_ethereum::pb::eth::v2::Log) -> Result<Self, String> {
-                log: &substreams_ethereum::pb::eth::v2::Log,
-            ) -> Result<Self, String> {
             Ok(Self {
-                    pool: ethabi::decode(
+                pool: ethabi::decode(&[ethabi::ParamType::Address], log.topics[1usize].as_ref())
-                            &[ethabi::ParamType::Address],
-                            log.topics[1usize].as_ref(),
-                        )
                     .map_err(|e| {
                         format!(
                             "unable to decode param 'pool' from topic of type 'address': {:?}",
@@ -632,10 +551,8 @@
         fn match_log(log: &substreams_ethereum::pb::eth::v2::Log) -> bool {
             Self::match_log(log)
         }
-            fn decode(
+        fn decode(log: &substreams_ethereum::pb::eth::v2::Log) -> Result<Self, String> {
-                log: &substreams_ethereum::pb::eth::v2::Log,
-            ) -> Result<Self, String> {
             Self::decode(log)
         }
     }
-    }
+}

substreams/ethereum-balancer/src/modules.rs

@@ -176,7 +176,10 @@ pub fn map_protocol_changes(
                 .or_insert_with(|| TransactionChangesBuilder::new(tx));
 
             // iterate over individual components created within this tx
-            tx_component.components.iter().for_each(|component| {
+            tx_component
+                .components
+                .iter()
+                .for_each(|component| {
                     builder.add_protocol_component(component);
                     let entity_change = EntityChanges {
                         component_id: component.id.clone(),
@@ -196,7 +199,9 @@ pub fn map_protocol_changes(
             let builder = transaction_changes
                 .entry(tx.index)
                 .or_insert_with(|| TransactionChangesBuilder::new(&tx));
-            balances.values().for_each(|bc| builder.add_balance_change(bc));
+            balances
+                .values()
+                .for_each(|bc| builder.add_balance_change(bc));
         });
 
     // Extract and insert any storage changes that happened for any of the components.
@@ -211,10 +216,17 @@ pub fn map_protocol_changes(
         &mut transaction_changes,
     );
 
-    transaction_changes.iter_mut().for_each(|(_, change)| {
+    transaction_changes
+        .iter_mut()
+        .for_each(|(_, change)| {
             // this indirection is necessary due to borrowing rules.
-        let addresses = change.changed_contracts().map(|e| e.to_vec()).collect::<Vec<_>>();
+            let addresses = change
-        addresses.into_iter().for_each(|address| {
+                .changed_contracts()
+                .map(|e| e.to_vec())
+                .collect::<Vec<_>>();
+            addresses
+                .into_iter()
+                .for_each(|address| {
                     if address != VAULT_ADDRESS {
                         // We reconstruct the component_id from the address here
                         change.mark_component_as_updated(&format!("0x{}", hex::encode(address)))
@@ -229,9 +241,7 @@ pub fn map_protocol_changes(
         changes: transaction_changes
             .drain()
             .sorted_unstable_by_key(|(index, _)| *index)
-            .filter_map(|(_, builder)| {
+            .filter_map(|(_, builder)| builder.build())
-                builder.build()
-            })
             .collect::<Vec<_>>(),
     })
 }

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

@@ -929,8 +929,8 @@ pub mod events {
                 .topics
                 .get(0)
                 .expect("bounds already checked")
-                .as_ref() ==
+                .as_ref()
-                Self::TOPIC_ID;
+                == Self::TOPIC_ID;
         }
         pub fn decode(log: &substreams_ethereum::pb::eth::v2::Log) -> Result<Self, String> {
             let mut values =
@@ -1009,8 +1009,8 @@ pub mod events {
                 .topics
                 .get(0)
                 .expect("bounds already checked")
-                .as_ref() ==
+                .as_ref()
-                Self::TOPIC_ID;
+                == Self::TOPIC_ID;
         }
         pub fn decode(log: &substreams_ethereum::pb::eth::v2::Log) -> Result<Self, String> {
             let mut values =

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 crypto_swap_ng_factory;
-pub mod meta_registry;
-pub mod tricrypto_factory;
-pub mod twocrypto_factory;
 pub mod erc20;
 pub mod meta_pool_factory;
+pub mod meta_registry;
+pub mod stableswap_factory;
+pub mod tricrypto_factory;
+pub mod twocrypto_factory;