Deduplicate index keys

ethereum-common/src/calls.rs

@@ -1,9 +1,11 @@
+use std::collections::HashSet;
+
 use crate::pb::sf::substreams::ethereum::v1::{Call, Calls};
-use substreams::pb::sf::substreams::index::v1::Keys;
 use crate::pb::sf::substreams::v1::Clock;
 use anyhow::Ok;
 use substreams::errors::Error;
 use substreams::matches_keys_in_parsed_expr;
+use substreams::pb::sf::substreams::index::v1::Keys;
 use substreams::Hex;
 use substreams_ethereum::pb::eth::v2::Block;
 
@@ -36,16 +38,16 @@ fn all_calls(blk: Block) -> Result<Calls, Error> {
 
 #[substreams::handlers::map]
 fn index_calls(calls: Calls) -> Result<Keys, Error> {
-    let mut keys = Keys::default();
+    let keys: Vec<_> = calls
+        .calls
+        .iter()
+        .filter_map(|call| call.call.as_ref())
+        .flat_map(call_keys)
+        .collect::<HashSet<_>>()
+        .into_iter()
+        .collect();
 
-    calls.calls.into_iter().for_each(|call| {
-        if let Some(call) = &call.call {
-            call_keys(call).into_iter().for_each(|k| {
-                keys.keys.push(k);
-            });
-        }
-    });
-    Ok(keys)
+    Ok(Keys { keys })
 }
 
 #[substreams::handlers::map]
@@ -68,7 +70,10 @@ fn filtered_calls(query: String, calls: Calls) -> Result<Calls, Error> {
     })
 }
 
-pub fn call_matches(call: &substreams_ethereum::pb::eth::v2::Call, query: &str) -> Result<bool, Error> {
+pub fn call_matches(
+    call: &substreams_ethereum::pb::eth::v2::Call,
+    query: &str,
+) -> Result<bool, Error> {
     matches_keys_in_parsed_expr(&call_keys(call), query)
 }
 

ethereum-common/src/combined.rs

@@ -3,35 +3,38 @@ use crate::events::*;
 use crate::pb::sf::substreams::ethereum::v1::{
     Call, Calls, Event, Events, EventsAndCalls, Transaction, Transactions,
 };
-use substreams::pb::sf::substreams::index::v1::Keys;
 use crate::pb::sf::substreams::v1::Clock;
 use anyhow::Ok;
 use std::collections::HashMap;
+use std::collections::HashSet;
 use substreams::errors::Error;
+use substreams::pb::sf::substreams::index::v1::Keys;
 use substreams::Hex;
 use substreams_ethereum::pb::eth::v2::{Block, Call as ethCall, Log};
 
 #[substreams::handlers::map]
+
 fn index_events_and_calls(events: Events, calls: Calls) -> Result<Keys, Error> {
-    let mut keys = Keys::default();
+    let event_keys: HashSet<_> = events
+        .events
+        .iter()
+        .filter_map(|e| e.log.as_ref())
+        .flat_map(evt_keys)
+        .collect();
 
-    events.events.into_iter().for_each(|e| {
-        if let Some(log) = e.log {
-            evt_keys(&log).into_iter().for_each(|k| {
-                keys.keys.push(k);
-            });
-        }
-    });
+    let call_keys: HashSet<_> = calls
+        .calls
+        .iter()
+        .filter_map(|call| call.call.as_ref())
+        .flat_map(call_keys)
+        .collect();
 
-    calls.calls.into_iter().for_each(|call| {
-        if let Some(call) = &call.call {
-            call_keys(call).into_iter().for_each(|k| {
-                keys.keys.push(k);
-            });
-        }
-    });
+    let keys: Vec<_> = event_keys
+        .into_iter()
+        .chain(call_keys.into_iter())
+        .collect();
 
-    Ok(keys)
+    Ok(Keys { keys })
 }
 
 #[substreams::handlers::map]

ethereum-common/src/events.rs

@@ -1,9 +1,11 @@
+use std::collections::HashSet;
+
 use crate::pb::sf::substreams::ethereum::v1::{Event, Events};
-use substreams::pb::sf::substreams::index::v1::Keys;
 use crate::pb::sf::substreams::v1::Clock;
 use anyhow::Ok;
 use substreams::errors::Error;
 use substreams::matches_keys_in_parsed_expr;
+use substreams::pb::sf::substreams::index::v1::Keys;
 use substreams::Hex;
 use substreams_ethereum::pb::eth::v2::Block;
 
@@ -37,17 +39,16 @@ fn all_events(blk: Block) -> Result<Events, Error> {
 
 #[substreams::handlers::map]
 fn index_events(events: Events) -> Result<Keys, Error> {
-    let mut keys = Keys::default();
-
-    events.events.into_iter().for_each(|e| {
-        if let Some(log) = e.log {
-            evt_keys(&log).into_iter().for_each(|k| {
-                keys.keys.push(k);
-            });
-        }
-    });
+    let keys: Vec<_> = events
+        .events
+        .iter()
+        .filter_map(|event| event.log.as_ref())
+        .flat_map(evt_keys)
+        .collect::<HashSet<_>>()
+        .into_iter()
+        .collect();
 
-    Ok(keys)
+    Ok(Keys { keys })
 }
 
 #[substreams::handlers::map]
@@ -84,6 +85,9 @@ pub fn evt_keys(log: &substreams_ethereum::pb::eth::v2::Log) -> Vec<String> {
     keys
 }
 
-pub fn evt_matches(log: &substreams_ethereum::pb::eth::v2::Log, query: &str) -> Result<bool, Error> {
+pub fn evt_matches(
+    log: &substreams_ethereum::pb::eth::v2::Log,
+    query: &str,
+) -> Result<bool, Error> {
     matches_keys_in_parsed_expr(&evt_keys(log), query)
 }