feat(starknet): event trigger filter

chain/starknet/src/adapter.rs

@@ -1,18 +1,29 @@
+use std::collections::{hash_map::Entry, HashMap};
+
 use graph::{
     blockchain::{EmptyNodeCapabilities, TriggerFilter as TriggerFilterTrait},
     components::store::BlockNumber,
 };
 
 use crate::{
     data_source::{DataSource, DataSourceTemplate},
+    felt::Felt,
     Chain,
 };
 
+type TopicWithRanges = HashMap<Felt, Vec<BlockRange>>;
+
 #[derive(Default, Clone)]
 pub struct TriggerFilter {
+    pub(crate) event: StarknetEventFilter,
     pub(crate) block: StarknetBlockFilter,
 }
 
+#[derive(Default, Clone)]
+pub struct StarknetEventFilter {
+    pub contract_addresses: HashMap<Felt, TopicWithRanges>,
+}
+
 #[derive(Default, Clone)]
 pub struct StarknetBlockFilter {
     pub block_ranges: Vec<BlockRange>,
@@ -28,6 +39,8 @@ impl TriggerFilterTrait<Chain> for TriggerFilter {
     fn extend_with_template(&mut self, _data_source: impl Iterator<Item = DataSourceTemplate>) {}
 
     fn extend<'a>(&mut self, data_sources: impl Iterator<Item = &'a DataSource> + Clone) {
+        self.event
+            .extend(StarknetEventFilter::from_data_sources(data_sources.clone()));
         self.block
             .extend(StarknetBlockFilter::from_data_sources(data_sources));
     }
@@ -41,6 +54,86 @@ impl TriggerFilterTrait<Chain> for TriggerFilter {
     }
 }
 
+impl StarknetEventFilter {
+    pub fn from_data_sources<'a>(iter: impl IntoIterator<Item = &'a DataSource>) -> Self {
+        iter.into_iter()
+            // Using `filter_map` instead of `filter` to avoid having to unwrap source address in
+            // `fold` below.
+            .filter_map(|data_source| {
+                if data_source.mapping.event_handlers.is_empty() {
+                    None
+                } else {
+                    data_source
+                        .source
+                        .address
+                        .as_ref()
+                        .map(|source_address| (data_source, source_address.to_owned()))
+                }
+            })
+            .fold(
+                Self::default(),
+                |mut filter_opt, (data_source, source_address)| {
+                    filter_opt.extend(Self {
+                        contract_addresses: [(
+                            source_address,
+                            data_source
+                                .mapping
+                                .event_handlers
+                                .iter()
+                                .map(|event_handler| {
+                                    (
+                                        event_handler.event_selector.clone(),
+                                        vec![BlockRange {
+                                            start_block: data_source.source.start_block,
+                                            end_block: data_source.source.end_block,
+                                        }],
+                                    )
+                                })
+                                .collect(),
+                        )]
+                        .into_iter()
+                        .collect(),
+                    });
+                    filter_opt
+                },
+            )
+    }
+
+    pub fn extend(&mut self, other: StarknetEventFilter) {
+        if other.is_empty() {
+            return;
+        }
+
+        let StarknetEventFilter { contract_addresses } = other;
+
+        for (address, topic_with_ranges) in contract_addresses.into_iter() {
+            match self.contract_addresses.entry(address) {
+                Entry::Occupied(entry) => {
+                    let entry = entry.into_mut();
+                    for (topic, mut block_ranges) in topic_with_ranges.into_iter() {
+                        match entry.entry(topic) {
+                            Entry::Occupied(topic_entry) => {
+                                // TODO: merge overlapping block ranges
+                                topic_entry.into_mut().append(&mut block_ranges);
+                            }
+                            Entry::Vacant(topic_entry) => {
+                                topic_entry.insert(block_ranges);
+                            }
+                        }
+                    }
+                }
+                Entry::Vacant(entry) => {
+                    entry.insert(topic_with_ranges);
+                }
+            }
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.contract_addresses.is_empty()
+    }
+}
+
 impl StarknetBlockFilter {
     pub fn from_data_sources<'a>(iter: impl IntoIterator<Item = &'a DataSource>) -> Self {
         iter.into_iter()

chain/starknet/src/chain.rs

@@ -38,6 +38,7 @@ use crate::{
     data_source::{
         DataSource, DataSourceTemplate, UnresolvedDataSource, UnresolvedDataSourceTemplate,
     },
+    felt::Felt,
     trigger::{StarknetBlockTrigger, StarknetEventTrigger, StarknetTrigger},
 };
 
@@ -384,7 +385,6 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         panic!("Should never be called since not used by FirehoseBlockStream")
     }
 
-    #[allow(unused)]
     async fn triggers_in_block(
         &self,
         logger: &Logger,
@@ -402,12 +402,42 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
                 transaction
                     .events
                     .iter()
-                    .map(|event| {
-                        StarknetTrigger::Event(StarknetEventTrigger {
-                            event: Arc::new(event.clone()),
-                            block: shared_block.clone(),
-                            transaction: transaction.clone(),
-                        })
+                    .filter_map(|event| {
+                        let from_addr: Felt = event.from_addr.as_slice().try_into().ok()?;
+
+                        match filter.event.contract_addresses.get(&from_addr) {
+                            Some(entry) => {
+                                let event_topic: Felt =
+                                    event.keys.first()?.as_slice().try_into().ok()?;
+
+                                match entry.get(&event_topic) {
+                                    Some(block_ranges) => {
+                                        let block_matched = block_ranges.iter().any(|range| {
+                                            if block_height >= range.start_block {
+                                                match range.end_block {
+                                                    Some(end_block) => block_height < end_block,
+                                                    None => true,
+                                                }
+                                            } else {
+                                                false
+                                            }
+                                        });
+
+                                        if block_matched {
+                                            Some(StarknetTrigger::Event(StarknetEventTrigger {
+                                                event: Arc::new(event.clone()),
+                                                block: shared_block.clone(),
+                                                transaction: transaction.clone(),
+                                            }))
+                                        } else {
+                                            None
+                                        }
+                                    }
+                                    None => None,
+                                }
+                            }
+                            None => None,
+                        }
                     })
                     .collect()
             })

chain/starknet/src/felt.rs

@@ -3,12 +3,12 @@ use std::{
     str::FromStr,
 };
 
-use graph::anyhow;
+use graph::anyhow::{self, anyhow};
 use serde::{de::Visitor, Deserialize};
 
 /// Represents the primitive `FieldElement` type used in Starknet. Each `FieldElement` is 252-bit
 /// in size.
-#[derive(Clone, PartialEq, Eq)]
+#[derive(Hash, Clone, PartialEq, Eq)]
 pub struct Felt([u8; 32]);
 
 struct FeltVisitor;
@@ -25,6 +25,21 @@ impl From<[u8; 32]> for Felt {
     }
 }
 
+impl TryFrom<&[u8]> for Felt {
+    type Error = anyhow::Error;
+
+    fn try_from(value: &[u8]) -> Result<Self, Self::Error> {
+        if value.len() > 32 {
+            Err(anyhow!("slice too long"))
+        } else {
+            let mut buffer = [0u8; 32];
+            buffer[(32 - value.len())..].copy_from_slice(value);
+
+            Ok(buffer.into())
+        }
+    }
+}
+
 impl AsRef<[u8]> for Felt {
     fn as_ref(&self) -> &[u8] {
         &self.0