fix: pool emitting

substreams/ethereum-curve/README.md

@@ -0,0 +1,5 @@
+# Instructions
+
+```bash
+$ substreams run -e mainnet.eth.streamingfast.io:443 substreams.yaml map_protocol_changes --start-block 11507454 --stop-block +100 -p map_components=`python params.py`
+```

substreams/ethereum-curve/params.json

@@ -1,11 +1,12 @@
 [
     {
-        "address": "0x5F890841f657d90E081bAbdB532A05996Af79Fe6",
+        "name": "3pool",
-        "tx_hash": "0xb71a66c1d93c525a2dd19a8db0da19e65be04f36e733af7f03e3c9dff41aa16a",
+        "address": "bEbc44782C7dB0a1A60Cb6fe97d0b483032FF1C7",
+        "tx_hash": "20793bbf260912aae189d5d261ff003c9b9166da8191d8f9d63ff1c7722f3ac6",
         "tokens": [
-            "0x6b175474e89094c44da98b954eedeac495271d0f",
+            "6b175474e89094c44da98b954eedeac495271d0f",
-            "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
+            "a0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
-            "0xdac17f958d2ee523a2206206994597c13d831ec7"
+            "dac17f958d2ee523a2206206994597c13d831ec7"
         ],
         "attributes": {}
     }

substreams/ethereum-curve/params.py

@@ -47,7 +47,7 @@ def encode_json_to_query_params(params: list[dict[str, Any]]):
 def main():
     with open(PARAMETERS, "r") as f:
         params = json.load(f)
-    print('"', encode_json_to_query_params(params), '"', sep="")
+    print(encode_json_to_query_params(params))
 
 
 if __name__ == "__main__":

substreams/ethereum-curve/src/modules.rs

@@ -58,9 +58,11 @@ pub fn map_components(
                     })
                     .collect::<Vec<_>>();
 
-                components.extend(emit_specific_pools(&params, &block).expect(
+                if let Some(component) = emit_specific_pools(&params, &tx).expect(
                     "An unexpected error occured when parsing params for emitting specific pools",
-                ));
+                ) {
+                    components.push(component)
+                }
 
                 if !components.is_empty() {
                     Some(TransactionProtocolComponents {

substreams/ethereum-curve/src/pools.rs

@@ -1,7 +1,7 @@
 use anyhow::{Context, Result};
 use serde::Deserialize;
 use std::{collections::HashMap, iter::zip};
-use substreams_ethereum::pb::eth;
+use substreams_ethereum::pb::eth::v2::TransactionTrace;
 use tycho_substreams::prelude::*;
 
 const PARAMS_SEPERATOR: &str = ",";
@@ -11,8 +11,8 @@ struct PoolQueryParams {
     address: String,
     tx_hash: String,
     tokens: Vec<String>,
-    attribute_keys: Vec<String>,
+    attribute_keys: Option<Vec<String>>,
-    attribute_vals: Vec<String>,
+    attribute_vals: Option<Vec<String>>,
 }
 
 /// This function parses the `params` string and extracts the pool query parameters. `params` are
@@ -30,60 +30,63 @@ struct PoolQueryParams {
 ///  if various addresses are not formatted properly.
 pub fn emit_specific_pools(
     params: &String,
-    block: &eth::v2::Block,
+    tx: &TransactionTrace,
-) -> Result<Vec<ProtocolComponent>> {
+) -> Result<Option<ProtocolComponent>> {
     let pools = parse_params(params)?;
-    create_components(block, pools)
+    create_component(tx, pools)
 }
 
-fn create_components(
+fn create_component(
-    block: &eth::v2::Block,
+    tx: &TransactionTrace,
     pools: HashMap<String, PoolQueryParams>,
-) -> Result<Vec<ProtocolComponent>, anyhow::Error> {
+) -> Result<Option<ProtocolComponent>> {
-    let mut components: Vec<ProtocolComponent> = vec![];
+    let encoded_hash = hex::encode(tx.hash.clone());
-    for tx in block.transactions() {
+    if let Some(pool) = pools.get(&encoded_hash) {
-        let encoded_hash = hex::encode(tx.hash.clone());
+        Ok(Some(ProtocolComponent {
-        if let Some(pool) = pools.get(&encoded_hash) {
+            id: pool.address.clone(),
-            let component = ProtocolComponent {
+            tx: Some(Transaction {
-                id: pool.address.clone(),
+                to: tx.to.clone(),
-                tx: Some(Transaction {
+                from: tx.from.clone(),
-                    to: tx.to.clone(),
+                hash: tx.hash.clone(),
-                    from: tx.from.clone(),
+                index: tx.index.into(),
-                    hash: tx.hash.clone(),
+            }),
-                    index: tx.index.into(),
+            tokens: pool
-                }),
+                .tokens
-                tokens: pool
-                    .tokens
-                    .clone()
-                    .into_iter()
-                    .map(|token| Result::Ok(hex::decode(token)?))
-                    .collect::<Result<Vec<_>>>()
-                    .with_context(|| "Token addresses were not formatted properly")?,
-                static_att: zip(
-                    pool.attribute_keys.clone().into_iter(),
-                    pool.attribute_vals.clone().into_iter(),
-                )
                 .clone()
-                .map(|(key, value)| Attribute {
+                .into_iter()
-                    name: key,
+                .map(|token| Result::Ok(hex::decode(token)?))
-                    value: value.into(),
+                .collect::<Result<Vec<_>>>()
-                    change: ChangeType::Creation.into(),
+                .with_context(|| "Token addresses were not formatted properly")?,
-                })
+            static_att: zip(
-                .collect::<Vec<_>>(),
+                pool.attribute_keys
-                contracts: vec![hex::decode(pool.address.clone())
+                    .clone()
-                    .with_context(|| "Pool address was not formatted properly")?],
+                    .unwrap_or(vec![])
+                    .into_iter(),
+                pool.attribute_vals
+                    .clone()
+                    .unwrap_or(vec![])
+                    .into_iter(),
+            )
+            .clone()
+            .map(|(key, value)| Attribute {
+                name: key,
+                value: value.into(),
                 change: ChangeType::Creation.into(),
-                protocol_type: Some(ProtocolType {
+            })
-                    name: "curve_pool".into(),
+            .collect::<Vec<_>>(),
-                    financial_type: FinancialType::Swap.into(),
+            contracts: vec![hex::decode(pool.address.clone())
-                    attribute_schema: Vec::new(),
+                .with_context(|| "Pool address was not formatted properly")?],
-                    implementation_type: ImplementationType::Vm.into(),
+            change: ChangeType::Creation.into(),
-                }),
+            protocol_type: Some(ProtocolType {
-            };
+                name: "curve_pool".into(),
-            components.push(component);
+                financial_type: FinancialType::Swap.into(),
-        }
+                attribute_schema: Vec::new(),
+                implementation_type: ImplementationType::Vm.into(),
+            }),
+        }))
+    } else {
+        Ok(None)
     }
-    Ok(components)
 }
 
 fn parse_params(params: &String) -> Result<HashMap<String, PoolQueryParams>, anyhow::Error> {
@@ -119,8 +122,8 @@ mod tests {
                         "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48".to_string(),
                         "0xdac17f958d2ee523a2206206994597c13d831ec7".to_string(),
                     ],
-                    attribute_keys: vec!["key1".to_string()],
+                    attribute_keys: Some(vec!["key1".to_string()]),
-                    attribute_vals: vec!["val1".to_string()],
+                    attribute_vals: Some(vec!["val1".to_string()]),
                 },
             );
             map

substreams/ethereum-curve/substreams.yaml

@@ -19,7 +19,7 @@ binaries:
 modules:
   - name: map_components
     kind: map
-    initialBlock: 11942410
+    initialBlock: 10809473
     inputs:
       - params: string
       - source: sf.ethereum.type.v2.Block
@@ -28,7 +28,7 @@ modules:
 
   - name: store_components
     kind: store
-    initialBlock: 11942410
+    initialBlock: 10809473
     updatePolicy: add
     valueType: int64
     inputs:
@@ -36,7 +36,7 @@ modules:
 
   - name: store_component_tokens
     kind: store
-    initialBlock: 11942410
+    initialBlock: 10809473
     updatePolicy: set
     valueType: string
     inputs:
@@ -44,7 +44,7 @@ modules:
 
   - name: map_relative_balances
     kind: map
-    initialBlock: 11942410  # An arbitrary block that should change based on your requirements
+    initialBlock: 10809473  # An arbitrary block that should change based on your requirements
     inputs:
       - source: sf.ethereum.type.v2.Block
       - store: store_components
@@ -54,7 +54,7 @@ modules:
 
   - name: store_balances
     kind: store
-    initialBlock: 11942410
+    initialBlock: 10809473
     updatePolicy: add
     valueType: bigint
     inputs:
@@ -62,7 +62,7 @@ modules:
 
   - name: map_protocol_changes
     kind: map
-    initialBlock: 11942410
+    initialBlock: 10809473
     inputs:
       - source: sf.ethereum.type.v2.Block
       - map: map_components