feat(starknet): firehose filter

Author: xJonathanLEI

chain/starknet/src/adapter.rs

@@ -3,7 +3,20 @@ use std::collections::{hash_map::Entry, HashMap};
 use graph::{
     blockchain::{EmptyNodeCapabilities, TriggerFilter as TriggerFilterTrait},
     components::store::BlockNumber,
+    firehose::{
+        BlockHeaderOnly as FirehoseFilterBlockHeaderOnly, BlockRange as FirehoseFilterBlockRange,
+        ContractEventFilter as FirehoseFilterContractEventFilter,
+        TopicWithRanges as FirehoseFilterTopicWithRanges,
+        TransactionEventFilter as FirehoseFilterTransactionEventFilter,
+    },
 };
+use prost::Message;
+use prost_types::Any;
+
+const BLOCK_HEADER_ONLY_TYPE_URL: &str =
+    "type.googleapis.com/zklend.starknet.transform.v1.BlockHeaderOnly";
+const TRANSACTION_EVENT_FILTER_TYPE_URL: &str =
+    "type.googleapis.com/zklend.starknet.transform.v1.TransactionEventFilter";
 
 use crate::{
     data_source::{DataSource, DataSourceTemplate},
@@ -50,7 +63,43 @@ impl TriggerFilterTrait<Chain> for TriggerFilter {
     }
 
     fn to_firehose_filter(self) -> Vec<prost_types::Any> {
-        todo!()
+        // An empty event filter list means that the subgraph is not interested in events at all.
+        // So we can stream just header-only blocks.
+        if self.event.is_empty() {
+            return vec![Any {
+                type_url: BLOCK_HEADER_ONLY_TYPE_URL.into(),
+                value: FirehoseFilterBlockHeaderOnly {}.encode_to_vec(),
+            }];
+        }
+
+        let event_filters = self
+            .event
+            .contract_addresses
+            .iter()
+            .map(
+                |(contract_address, topic_with_ranges)| FirehoseFilterContractEventFilter {
+                    contract_address: contract_address.into(),
+                    topics: topic_with_ranges
+                        .iter()
+                        .map(|(topic, ranges)| FirehoseFilterTopicWithRanges {
+                            topic: topic.into(),
+                            block_ranges: ranges
+                                .iter()
+                                .map(|range| FirehoseFilterBlockRange {
+                                    start_block: range.start_block as u64,
+                                    end_block: range.end_block.unwrap_or_default() as u64,
+                                })
+                                .collect(),
+                        })
+                        .collect(),
+                },
+            )
+            .collect();
+
+        vec![Any {
+            type_url: TRANSACTION_EVENT_FILTER_TYPE_URL.into(),
+            value: FirehoseFilterTransactionEventFilter { event_filters }.encode_to_vec(),
+        }]
     }
 }
 

chain/starknet/src/felt.rs

@@ -63,6 +63,12 @@ impl FromStr for Felt {
     }
 }
 
+impl From<&Felt> for Vec<u8> {
+    fn from(value: &Felt) -> Self {
+        value.0.to_vec()
+    }
+}
+
 impl<'de> Deserialize<'de> for Felt {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where

graph/build.rs

@@ -8,6 +8,7 @@ fn main() {
                 "proto/ethereum/transforms.proto",
                 "proto/near/transforms.proto",
                 "proto/cosmos/transforms.proto",
+                "proto/starknet/transforms.proto",
             ],
             &["proto"],
         )

graph/proto/starknet/transforms.proto

@@ -0,0 +1,36 @@
+syntax = "proto3";
+
+package zklend.starknet.transform.v1;
+
+option go_package = "github.com/starknet-graph/firehose-starknet/types/pb/zklend/starknet/type/v1;pbtransform";
+
+// Stream block headers only. The `transactions` field is always empty.
+message BlockHeaderOnly {}
+
+// Stream every single block, but each block will only contain transactions that match with `event_filters`.
+// A TransactionEventFilter message with an empty `event_filters` is invalid. Do not send any filter instead
+// if you wish to receive full blocks.
+message TransactionEventFilter {
+  repeated ContractEventFilter event_filters = 1;
+}
+
+// Only include transactions which emit at least one event that *BOTH*
+// * is emitted by `contract_address`
+// * matches with at least one topic in `topics`
+message ContractEventFilter {
+  bytes contract_address = 1;
+  repeated TopicWithRanges topics = 2;
+}
+
+// Matches events whose `keys[0]` equals `topic`, *AND* in any of the `block_ranges`.
+message TopicWithRanges {
+  bytes topic = 1;
+  repeated BlockRange block_ranges = 2;
+}
+
+// A range of blocks. Both `start_block` and `end_block` are inclusive. When `end_block` is `0`, it means that any
+// block height >= `start_block` is matched.
+message BlockRange {
+  uint64 start_block = 1;
+  uint64 end_block = 2;
+}

graph/src/firehose/codec.rs

@@ -14,7 +14,12 @@ mod pbnear;
 #[path = "sf.cosmos.transform.v1.rs"]
 mod pbcosmos;
 
+#[rustfmt::skip]
+#[path = "zklend.starknet.transform.v1.rs"]
+mod pbstarknet;
+
 pub use pbcosmos::*;
 pub use pbethereum::*;
 pub use pbfirehose::*;
 pub use pbnear::*;
+pub use pbstarknet::*;

graph/src/firehose/zklend.starknet.transform.v1.rs

@@ -0,0 +1,43 @@
+/// Stream block headers only. The `transactions` field is always empty.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct BlockHeaderOnly {}
+/// Stream every single block, but each block will only contain transactions that match with `event_filters`.
+/// A TransactionEventFilter message with an empty `event_filters` is invalid. Do not send any filter instead
+/// if you wish to receive full blocks.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TransactionEventFilter {
+    #[prost(message, repeated, tag = "1")]
+    pub event_filters: ::prost::alloc::vec::Vec<ContractEventFilter>,
+}
+/// Only include transactions which emit at least one event that *BOTH*
+/// * is emitted by `contract_address`
+/// * matches with at least one topic in `topics`
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ContractEventFilter {
+    #[prost(bytes = "vec", tag = "1")]
+    pub contract_address: ::prost::alloc::vec::Vec<u8>,
+    #[prost(message, repeated, tag = "2")]
+    pub topics: ::prost::alloc::vec::Vec<TopicWithRanges>,
+}
+/// Matches events whose `keys\[0\]` equals `topic`, *AND* in any of the `block_ranges`.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TopicWithRanges {
+    #[prost(bytes = "vec", tag = "1")]
+    pub topic: ::prost::alloc::vec::Vec<u8>,
+    #[prost(message, repeated, tag = "2")]
+    pub block_ranges: ::prost::alloc::vec::Vec<BlockRange>,
+}
+/// A range of blocks. Both `start_block` and `end_block` are inclusive. When `end_block` is `0`, it means that any
+/// block height >= `start_block` is matched.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct BlockRange {
+    #[prost(uint64, tag = "1")]
+    pub start_block: u64,
+    #[prost(uint64, tag = "2")]
+    pub end_block: u64,
+}