Initial pools.rs function for handling custom parameters

substreams/Cargo.lock

@@ -241,6 +241,8 @@ dependencies = [
  "num-bigint",
  "prost 0.11.9",
  "prost-types 0.12.3",
+ "serde",
+ "serde_qs",
  "substreams",
  "substreams-ethereum",
  "tycho-substreams",
@@ -572,6 +574,12 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "percent-encoding"
+version = "2.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
+
 [[package]]
 name = "petgraph"
 version = "0.6.4"
@@ -869,6 +877,17 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_qs"
+version = "0.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd34f36fe4c5ba9654417139a9b3a20d2e1de6012ee678ad14d240c22c78d8d6"
+dependencies = [
+ "percent-encoding",
+ "serde",
+ "thiserror",
+]
+
 [[package]]
 name = "sha3"
 version = "0.10.8"

substreams/ethereum-curve/Cargo.toml

@@ -20,6 +20,8 @@ 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"
 
 [build-dependencies]
 anyhow = "1"

substreams/ethereum-curve/src/lib.rs

@@ -1,3 +1,4 @@
 mod abi;
 pub mod modules;
 mod pool_factories;
+mod pools;

substreams/ethereum-curve/src/modules.rs

@@ -16,7 +16,7 @@ use substreams_ethereum::{block_view::LogView, pb::eth};
 
 use substreams_ethereum::Event;
 
-use crate::{abi, pool_factories};
+use crate::{abi, pool_factories, pools::emit_specific_pools};
 use tycho_substreams::{
     balances::store_balance_changes, contract::extract_contract_changes, prelude::*,
 };
@@ -42,14 +42,17 @@ fn tx_from_log(log: &LogView) -> Transaction {
 }
 
 #[substreams::handlers::map]
-pub fn map_components(block: eth::v2::Block) -> Result<BlockTransactionProtocolComponents> {
+pub fn map_components(
+    params: String,
+    block: eth::v2::Block,
+) -> Result<BlockTransactionProtocolComponents> {
     // Gather contract changes by indexing `PoolCreated` events and analysing the `Create` call
     // We store these as a hashmap by tx hash since we need to agg by tx hash later
     Ok(BlockTransactionProtocolComponents {
         tx_components: block
             .transactions()
             .filter_map(|tx| {
-                let components = tx
+                let mut components = tx
                     .logs_with_calls()
                     .filter(|(_, call)| !call.call.state_reverted)
                     .filter_map(|(log, call)| {
@@ -66,6 +69,8 @@ pub fn map_components(block: eth::v2::Block) -> Result<BlockTransactionProtocolC
                     })
                     .collect::<Vec<_>>();
 
+                components.extend(emit_specific_pools(&params, &block).expect("yur"));
+
                 if !components.is_empty() {
                     Some(TransactionProtocolComponents {
                         tx: Some(Transaction {

substreams/ethereum-curve/src/pool_factories.rs

@@ -98,7 +98,7 @@ pub fn address_map(
                     index: tx.index.into(),
                 }),
                 tokens: trimmed_coins,
-                contracts: vec![call_address.into(), pool_added.pool],
+                contracts: vec![pool_added.pool],
                 static_att: vec![
                     Attribute {
                         name: "pool_type".into(),
@@ -117,7 +117,12 @@ pub fn address_map(
                     },
                 ],
                 change: ChangeType::Creation.into(),
-                ..Default::default()
+                protocol_type: Some(ProtocolType {
+                    name: "curve_pool".into(),
+                    financial_type: FinancialType::Swap.into(),
+                    attribute_schema: Vec::new(),
+                    implementation_type: ImplementationType::Vm.into(),
+                }),
             })
         }
         MAIN_REGISTRY => {
@@ -186,7 +191,7 @@ pub fn address_map(
                     index: tx.index.into(),
                 }),
                 tokens: trimmed_coins,
-                contracts: vec![call_address.into(), pool_created.pool],
+                contracts: vec![pool_created.pool],
                 static_att: vec![
                     Attribute {
                         name: "pool_type".into(),
@@ -205,7 +210,12 @@ pub fn address_map(
                     },
                 ],
                 change: ChangeType::Creation.into(),
-                ..Default::default()
+                protocol_type: Some(ProtocolType {
+                    name: "curve_pool".into(),
+                    financial_type: FinancialType::Swap.into(),
+                    attribute_schema: Vec::new(),
+                    implementation_type: ImplementationType::Vm.into(),
+                }),
             })
         }
         CRYPTO_POOL_FACTORY => {
@@ -214,8 +224,10 @@ pub fn address_map(
             let deploy_call =
                 abi::crypto_pool_factory::functions::DeployPool::match_and_decode(call)?;
 
+            let component_id = &call.return_data[12..];
+
             Some(ProtocolComponent {
-                id: hex::encode(&call.return_data),
+                id: format!("0x{0}", hex::encode(component_id)),
                 tx: Some(Transaction {
                     to: tx.to.clone(),
                     from: tx.from.clone(),
@@ -223,7 +235,7 @@ pub fn address_map(
                     index: tx.index.into(),
                 }),
                 tokens: pool_added.coins.into(),
-                contracts: vec![call_address.into(), call.return_data.clone()],
+                contracts: vec![component_id.into()],
                 static_att: vec![
                     Attribute {
                         name: "pool_type".into(),
@@ -294,7 +306,12 @@ pub fn address_map(
                     },
                 ],
                 change: ChangeType::Creation.into(),
-                ..Default::default()
+                protocol_type: Some(ProtocolType {
+                    name: "curve_pool".into(),
+                    financial_type: FinancialType::Swap.into(),
+                    attribute_schema: Vec::new(),
+                    implementation_type: ImplementationType::Vm.into(),
+                }),
             })
         }
         META_POOL_FACTORY => {
@@ -333,8 +350,11 @@ pub fn address_map(
                                 call,
                             )
                         })?;
+
+                let component_id = &call.return_data[12..];
+
                 Some(ProtocolComponent {
-                    id: hex::encode(&call.return_data),
+                    id: format!("0x{0}", hex::encode(component_id)),
                     tx: Some(Transaction {
                         to: tx.to.clone(),
                         from: tx.from.clone(),
@@ -342,7 +362,7 @@ pub fn address_map(
                         index: tx.index.into(),
                     }),
                     tokens: pool_added.coins.into(),
-                    contracts: vec![call_address.into(), call.return_data.clone()],
+                    contracts: vec![component_id.into()],
                     static_att: vec![
                         Attribute {
                             name: "pool_type".into(),
@@ -366,7 +386,12 @@ pub fn address_map(
                         },
                     ],
                     change: ChangeType::Creation.into(),
-                    ..Default::default()
+                    protocol_type: Some(ProtocolType {
+                        name: "curve_pool".into(),
+                        financial_type: FinancialType::Swap.into(),
+                        attribute_schema: Vec::new(),
+                        implementation_type: ImplementationType::Vm.into(),
+                    }),
                 })
             } else if let Some(pool_added) =
                 abi::meta_pool_factory::events::MetaPoolDeployed::match_and_decode(log)
@@ -387,8 +412,11 @@ pub fn address_map(
                                 call,
                             )
                         })?;
+
+                let component_id = &call.return_data[12..];
+
                 Some(ProtocolComponent {
-                    id: hex::encode(&call.return_data),
+                    id: hex::encode(component_id),
                     tx: Some(Transaction {
                         to: tx.to.clone(),
                         from: tx.from.clone(),
@@ -396,18 +424,19 @@ pub fn address_map(
                         index: tx.index.into(),
                     }),
                     tokens: vec![pool_added.coin, add_pool.base_pool.clone()],
-                    contracts: vec![
-                        call_address.into(),
-                        call.return_data.clone(),
-                        add_pool.base_pool.clone(),
-                    ],
+                    contracts: vec![component_id.into(), add_pool.base_pool.clone()],
                     static_att: vec![Attribute {
                         name: "pool_type".into(),
                         value: "MetaPool".into(),
                         change: ChangeType::Creation.into(),
                     }],
                     change: ChangeType::Creation.into(),
-                    ..Default::default()
+                    protocol_type: Some(ProtocolType {
+                        name: "curve_pool".into(),
+                        financial_type: FinancialType::Swap.into(),
+                        attribute_schema: Vec::new(),
+                        implementation_type: ImplementationType::Vm.into(),
+                    }),
                 })
             } else {
                 None

substreams/ethereum-curve/src/pools.rs

@@ -0,0 +1,80 @@
+use std::collections::HashMap;
+
+use crate::abi::pool;
+use anyhow::{self, Context, Result};
+use ethabi::token;
+use serde::Deserialize;
+use serde_qs;
+use substreams_ethereum::{
+    block_view::LogView,
+    pb::{eth, eth::v2::Block},
+};
+use tycho_substreams::prelude::*;
+
+const PARAMS_SEPERATOR: &str = ",";
+
+#[derive(Debug, Deserialize)]
+struct PoolQueryParams {
+    address: String,
+    tx_hash: String,
+    tokens: Vec<String>,
+    attributes: Vec<(String, String)>,
+}
+
+pub fn emit_specific_pools(
+    params: &String,
+    block: &eth::v2::Block,
+) -> Result<Vec<ProtocolComponent>> {
+    let pools: HashMap<String, PoolQueryParams> = params
+        .split(PARAMS_SEPERATOR)
+        .map(|param| {
+            // TODO UNSAFE
+            let pool: PoolQueryParams = serde_qs::from_str(&param).unwrap();
+            (pool.tx_hash.clone(), pool)
+        })
+        .collect::<HashMap<_, _>>();
+
+    let mut components: Vec<ProtocolComponent> = vec![];
+
+    for tx in block.transactions() {
+        let encoded_hash = hex::encode(tx.hash.clone());
+        if let Some(pool) = pools.get(&encoded_hash) {
+            let component = ProtocolComponent {
+                id: pool.address.clone(),
+                tx: Some(Transaction {
+                    to: tx.to.clone(),
+                    from: tx.from.clone(),
+                    hash: tx.hash.clone(),
+                    index: tx.index.into(),
+                }),
+                tokens: pool
+                    .tokens
+                    .clone()
+                    .into_iter()
+                    .map(|token| Result::Ok(hex::decode(token)?))
+                    .collect::<Result<Vec<_>>>()
+                    .with_context(|| "")?,
+                static_att: pool
+                    .attributes
+                    .clone()
+                    .into_iter()
+                    .map(|attr| Attribute {
+                        name: attr.0,
+                        value: attr.1.into(),
+                        change: ChangeType::Creation.into(),
+                    })
+                    .collect::<Vec<_>>(),
+                contracts: vec![hex::decode(pool.address.clone()).with_context(|| "")?],
+                change: ChangeType::Creation.into(),
+                protocol_type: Some(ProtocolType {
+                    name: "curve_pool".into(),
+                    financial_type: FinancialType::Swap.into(),
+                    attribute_schema: Vec::new(),
+                    implementation_type: ImplementationType::Vm.into(),
+                }),
+            };
+            components.push(component);
+        }
+    }
+    Ok(components)
+}

substreams/ethereum-curve/substreams.yaml

@@ -19,7 +19,7 @@ binaries:
 modules:
   - name: map_components
     kind: map
-    initialBlock: 19128828
+    initialBlock: 18838383
     inputs:
       - source: sf.ethereum.type.v2.Block
     output:
@@ -27,7 +27,7 @@ modules:
 
   - name: store_components
     kind: store
-    initialBlock: 19128828
+    initialBlock: 18838383
     updatePolicy: add
     valueType: int64
     inputs:
@@ -35,7 +35,7 @@ modules:
 
   - name: store_component_tokens
     kind: store
-    initialBlock: 19128828
+    initialBlock: 18838383
     updatePolicy: set
     valueType: string
     inputs:
@@ -43,7 +43,7 @@ modules:
 
   - name: map_relative_balances
     kind: map
-    initialBlock: 19128828  # An arbitrary block that should change based on your requirements
+    initialBlock: 18838383  # An arbitrary block that should change based on your requirements
     inputs:
       - source: sf.ethereum.type.v2.Block
       - store: store_components
@@ -53,7 +53,7 @@ modules:
 
   - name: store_balances
     kind: store
-    initialBlock: 19128828
+    initialBlock: 18838383
     updatePolicy: add
     valueType: bigint
     inputs:
@@ -61,7 +61,7 @@ modules:
 
   - name: map_protocol_changes
     kind: map
-    initialBlock: 19128828
+    initialBlock: 18838383
     inputs:
       - source: sf.ethereum.type.v2.Block
       - map: map_components

substreams/ethereum-maverick/src/modules.rs

@@ -17,7 +17,7 @@ use itertools::Itertools;
 
 use substreams_ethereum::Event;
 
-use crate::{abi, pb};
+use crate::abi;
 
 use tycho_substreams::{
     balances::store_balance_changes, contract::extract_contract_changes, prelude::*,
@@ -95,7 +95,12 @@ pub fn map_components(block: eth::v2::Block) -> Result<BlockTransactionProtocolC
                                 },
                             ],
                             change: ChangeType::Creation.into(),
-                            ..Default::default()
+                            protocol_type: Some(ProtocolType {
+                                name: "maverick_pool".into(),
+                                financial_type: FinancialType::Swap.into(),
+                                attribute_schema: Vec::new(),
+                                implementation_type: ImplementationType::Vm.into(),
+                            }),
                         })
                     })
                     .collect::<Vec<_>>();