Implement poll-based event subscription for Starknet (#221)

* Add receipt block info to MessageResponse

* Managed to poll SendPacket events successfully

* Add StarknetEventEncoding field

* Implement HasEncoding<AsStarknetEvent> for StarknetChain

* Implementing CanPollStarknetIbcEvents

* Finish draft for CanPollStarknetIbcEvents

* Define CanQueryBlockEvents component

* Drafting Starknet event stream

* Done initial implementation of CanCreateStarknetSubscription

* Multiplex event subscription

* Add wait and retry block events querier

* Implement CanQueryBlockEvents and CanCreateStarknetSubscription for StarknetChain

* Fix clippy

* Remove sleep in test

* Add address argument for get_block_events

* Rename providers

* Implement StarknetEvent::from_emitted_event

* Implement GetStarknetBlockEvents

relayer/crates/cairo-encoding-components/src/types/as_starknet_event.rs

@@ -0,0 +1 @@
+pub struct AsStarknetEvent;

relayer/crates/cairo-encoding-components/src/types/mod.rs

@@ -1,2 +1,3 @@
 pub mod as_felt;
+pub mod as_starknet_event;
 pub mod nat;

relayer/crates/starknet-chain-components/Cargo.toml

@@ -14,6 +14,7 @@ cgp                                 = { workspace = true }
 hermes-relayer-components           = { workspace = true }
 hermes-test-components              = { workspace = true }
 hermes-runtime-components           = { workspace = true }
+hermes-async-runtime-components     = { workspace = true }
 hermes-chain-components             = { workspace = true }
 hermes-chain-type-components        = { workspace = true }
 hermes-encoding-components          = { workspace = true }
@@ -35,4 +36,5 @@ serde                       = { workspace = true }
 serde_json                  = { workspace = true }
 starknet                    = { workspace = true }
 tonic                       = { workspace = true }
+futures                     = { workspace = true }
 crypto-bigint               = "0.5.5"

relayer/crates/starknet-chain-components/src/components/chain.rs

@@ -73,6 +73,7 @@ use crate::impls::payload_builders::create_client::BuildStarknetCreateClientPayl
 use crate::impls::payload_builders::update_client::BuildStarknetUpdateClientPayload;
 use crate::impls::queries::ack_commitment::QueryStarknetAckCommitment;
 use crate::impls::queries::balance::QueryStarknetWalletBalance;
+use crate::impls::queries::block_events::default::DefaultQueryBlockEvents;
 use crate::impls::queries::channel_end::QueryChannelEndFromStarknet;
 use crate::impls::queries::client_state::QueryCometClientState;
 use crate::impls::queries::connection_end::QueryConnectionEndFromStarknet;
@@ -111,6 +112,7 @@ pub use crate::traits::contract::invoke::ContractInvokerComponent;
 pub use crate::traits::contract::message::InvokeContractMessageBuilderComponent;
 pub use crate::traits::messages::transfer::TransferTokenMessageBuilderComponent;
 pub use crate::traits::queries::address::ContractAddressQuerierComponent;
+use crate::traits::queries::block_events::BlockEventsQuerierComponent;
 pub use crate::traits::queries::token_balance::TokenBalanceQuerierComponent;
 pub use crate::traits::transfer::TokenTransferComponent;
 pub use crate::traits::types::blob::BlobTypeComponent;
@@ -229,6 +231,8 @@ cgp_preset! {
             ReadPacketSrcStarknetFields,
         ChainStatusQuerierComponent:
             QueryStarknetChainStatus,
+        BlockEventsQuerierComponent:
+            DefaultQueryBlockEvents,
         MessageSenderComponent:
             SendCallMessages,
         TxSubmitterComponent:

relayer/crates/starknet-chain-components/src/components/starknet_to_cosmos.rs

@@ -94,6 +94,5 @@ cgp_preset! {
             PacketDstChannelIdGetterComponent,
         ]:
             ReadPacketDstStarknetFields,
-
     }
 }

relayer/crates/starknet-chain-components/src/impls/encoding/option.rs

@@ -21,11 +21,16 @@ where
     ) -> Result<Option<Value>, Encoding::Error> {
         let class_hashes = encoding.get_field(PhantomData);
 
-        if class_hashes.contains(&event.class_hash) {
-            let value = encoding.decode(event)?;
-            Ok(Some(value))
-        } else {
-            Ok(None)
+        match &event.class_hash {
+            Some(class_hash) => {
+                if class_hashes.contains(class_hash) {
+                    let value = encoding.decode(event)?;
+                    Ok(Some(value))
+                } else {
+                    Ok(None)
+                }
+            }
+            None => Ok(None),
         }
     }
 }

relayer/crates/starknet-chain-components/src/impls/mod.rs

@@ -14,6 +14,7 @@ pub mod queries;
 pub mod send_message;
 pub mod starknet_to_cosmos;
 pub mod submit_tx;
+pub mod subscription;
 pub mod transfer;
 pub mod tx_response;
 pub mod types;

relayer/crates/starknet-chain-components/src/impls/queries/block_events/default.rs

@@ -0,0 +1,38 @@
+use cgp::prelude::CanRaiseAsyncError;
+use hermes_chain_components::traits::queries::chain_status::CanQueryChainHeight;
+use hermes_chain_components::traits::types::event::HasEventType;
+use hermes_chain_components::traits::types::height::HasHeightType;
+use hermes_chain_type_components::traits::types::address::HasAddressType;
+use hermes_runtime_components::traits::runtime::HasRuntime;
+use hermes_runtime_components::traits::sleep::CanSleep;
+use starknet::core::types::Felt;
+use starknet::providers::ProviderError;
+
+use crate::impls::queries::block_events::get::GetStarknetBlockEvents;
+use crate::impls::queries::block_events::retry::RetryQueryBlockEvents;
+use crate::impls::queries::block_events::wait::WaitBlockHeightAndQueryEvents;
+use crate::traits::provider::HasStarknetProvider;
+use crate::traits::queries::block_events::BlockEventsQuerier;
+use crate::types::event::StarknetEvent;
+
+pub struct DefaultQueryBlockEvents;
+
+impl<Chain> BlockEventsQuerier<Chain> for DefaultQueryBlockEvents
+where
+    Chain: HasRuntime
+        + HasHeightType<Height = u64>
+        + HasEventType<Event = StarknetEvent>
+        + HasAddressType<Address = Felt>
+        + HasStarknetProvider
+        + CanQueryChainHeight
+        + CanRaiseAsyncError<ProviderError>,
+    Chain::Runtime: CanSleep,
+{
+    async fn query_block_events(
+        chain: &Chain,
+        height: &Chain::Height,
+        address: &Chain::Address,
+    ) -> Result<Vec<Chain::Event>, Chain::Error> {
+        <RetryQueryBlockEvents<WaitBlockHeightAndQueryEvents<GetStarknetBlockEvents>>>::query_block_events(chain, height, address).await
+    }
+}

relayer/crates/starknet-chain-components/src/impls/queries/block_events/get.rs

@@ -0,0 +1,51 @@
+use cgp::prelude::*;
+use hermes_chain_components::traits::types::event::HasEventType;
+use hermes_chain_components::traits::types::height::HasHeightType;
+use hermes_chain_type_components::traits::types::address::HasAddressType;
+use starknet::core::types::{BlockId, EventFilter, Felt};
+use starknet::providers::{Provider, ProviderError};
+
+use crate::traits::provider::HasStarknetProvider;
+use crate::traits::queries::block_events::BlockEventsQuerier;
+use crate::types::event::StarknetEvent;
+
+pub struct GetStarknetBlockEvents;
+
+impl<Chain> BlockEventsQuerier<Chain> for GetStarknetBlockEvents
+where
+    Chain: HasHeightType<Height = u64>
+        + HasEventType<Event = StarknetEvent>
+        + HasAddressType<Address = Felt>
+        + HasStarknetProvider
+        + CanRaiseAsyncError<ProviderError>,
+{
+    async fn query_block_events(
+        chain: &Chain,
+        height: &u64,
+        address: &Felt,
+    ) -> Result<Vec<StarknetEvent>, Chain::Error> {
+        let provider = chain.provider();
+
+        let raw_events = provider
+            .get_events(
+                EventFilter {
+                    from_block: Some(BlockId::Number(*height)),
+                    to_block: Some(BlockId::Number(*height)),
+                    address: Some(*address),
+                    keys: None,
+                },
+                None,
+                1000,
+            )
+            .await
+            .map_err(Chain::raise_error)?;
+
+        let events = raw_events
+            .events
+            .into_iter()
+            .map(StarknetEvent::from)
+            .collect();
+
+        Ok(events)
+    }
+}

relayer/crates/starknet-chain-components/src/impls/queries/block_events/mod.rs

@@ -0,0 +1,5 @@
+pub mod default;
+pub mod get;
+pub mod retry;
+pub mod traces;
+pub mod wait;

relayer/crates/starknet-chain-components/src/impls/queries/block_events/retry.rs

@@ -0,0 +1,41 @@
+use core::marker::PhantomData;
+use core::time::Duration;
+
+use cgp::prelude::HasAsyncErrorType;
+use hermes_chain_components::traits::types::event::HasEventType;
+use hermes_chain_components::traits::types::height::HasHeightType;
+use hermes_chain_type_components::traits::types::address::HasAddressType;
+use hermes_runtime_components::traits::runtime::HasRuntime;
+use hermes_runtime_components::traits::sleep::CanSleep;
+
+use crate::traits::queries::block_events::BlockEventsQuerier;
+
+pub struct RetryQueryBlockEvents<InQuerier>(pub PhantomData<InQuerier>);
+
+impl<Chain, InQuerier> BlockEventsQuerier<Chain> for RetryQueryBlockEvents<InQuerier>
+where
+    Chain: HasRuntime + HasHeightType + HasAddressType + HasEventType + HasAsyncErrorType,
+    InQuerier: BlockEventsQuerier<Chain>,
+    Chain::Runtime: CanSleep,
+{
+    async fn query_block_events(
+        chain: &Chain,
+        height: &Chain::Height,
+        address: &Chain::Address,
+    ) -> Result<Vec<Chain::Event>, Chain::Error> {
+        let runtime = chain.runtime();
+        let mut sleep_time = Duration::from_millis(500);
+
+        for _ in 0..5 {
+            let res = InQuerier::query_block_events(chain, height, address).await;
+            if let Ok(events) = res {
+                return Ok(events);
+            }
+
+            runtime.sleep(sleep_time).await;
+            sleep_time *= 2;
+        }
+
+        InQuerier::query_block_events(chain, height, address).await
+    }
+}

relayer/crates/starknet-chain-components/src/impls/queries/block_events/traces.rs

@@ -0,0 +1,55 @@
+use cgp::prelude::*;
+use hermes_chain_components::traits::types::event::HasEventType;
+use hermes_chain_components::traits::types::height::HasHeightType;
+use hermes_chain_type_components::traits::types::address::HasAddressType;
+use starknet::core::types::{BlockId, ExecuteInvocation, Felt, TransactionTrace};
+use starknet::providers::{Provider, ProviderError};
+
+use crate::impls::send_message::extract_events_from_function_invocation;
+use crate::traits::provider::HasStarknetProvider;
+use crate::traits::queries::block_events::BlockEventsQuerier;
+use crate::types::event::StarknetEvent;
+
+pub struct QueryStarknetBlockEventsFromTraces;
+
+impl<Chain> BlockEventsQuerier<Chain> for QueryStarknetBlockEventsFromTraces
+where
+    Chain: HasHeightType<Height = u64>
+        + HasEventType<Event = StarknetEvent>
+        + HasAddressType<Address = Felt>
+        + HasStarknetProvider
+        + CanRaiseAsyncError<ProviderError>,
+{
+    async fn query_block_events(
+        chain: &Chain,
+        height: &u64,
+        address: &Felt,
+    ) -> Result<Vec<StarknetEvent>, Chain::Error> {
+        let provider = chain.provider();
+
+        let traces = provider
+            .trace_block_transactions(BlockId::Number(*height))
+            .await
+            .map_err(Chain::raise_error)?;
+
+        let events: Vec<StarknetEvent> = traces
+            .into_iter()
+            .filter_map(|trace| match trace.trace_root {
+                TransactionTrace::Invoke(invoke) => match invoke.execute_invocation {
+                    ExecuteInvocation::Success(invoke) => {
+                        if &invoke.contract_address == address {
+                            Some(extract_events_from_function_invocation(invoke))
+                        } else {
+                            None
+                        }
+                    }
+                    _ => None,
+                },
+                _ => None,
+            })
+            .flatten()
+            .collect();
+
+        Ok(events)
+    }
+}

relayer/crates/starknet-chain-components/src/impls/queries/block_events/wait.rs

@@ -0,0 +1,38 @@
+use core::marker::PhantomData;
+use core::time::Duration;
+
+use hermes_chain_components::traits::queries::chain_status::CanQueryChainHeight;
+use hermes_chain_components::traits::types::event::HasEventType;
+use hermes_chain_type_components::traits::types::address::HasAddressType;
+use hermes_runtime_components::traits::runtime::HasRuntime;
+use hermes_runtime_components::traits::sleep::CanSleep;
+
+use crate::traits::queries::block_events::BlockEventsQuerier;
+
+pub struct WaitBlockHeightAndQueryEvents<InQuerier>(pub PhantomData<InQuerier>);
+
+impl<Chain, InQuerier> BlockEventsQuerier<Chain> for WaitBlockHeightAndQueryEvents<InQuerier>
+where
+    Chain: HasRuntime + HasAddressType + HasEventType + CanQueryChainHeight,
+    InQuerier: BlockEventsQuerier<Chain>,
+    Chain::Runtime: CanSleep,
+{
+    async fn query_block_events(
+        chain: &Chain,
+        height: &Chain::Height,
+        address: &Chain::Address,
+    ) -> Result<Vec<Chain::Event>, Chain::Error> {
+        let runtime = chain.runtime();
+
+        loop {
+            let current_height = chain.query_chain_height().await?;
+            if &current_height >= height {
+                break;
+            } else {
+                runtime.sleep(Duration::from_millis(200)).await;
+            }
+        }
+
+        InQuerier::query_block_events(chain, height, address).await
+    }
+}

relayer/crates/starknet-chain-components/src/impls/queries/mod.rs

@@ -1,5 +1,6 @@
 pub mod ack_commitment;
 pub mod balance;
+pub mod block_events;
 pub mod channel_end;
 pub mod client_state;
 pub mod connection_end;

relayer/crates/starknet-chain-components/src/impls/send_message.rs

@@ -51,7 +51,7 @@ where
                     let message_responses = invocation
                         .calls
                         .into_iter()
-                        .map(extract_events_from_function_invocation)
+                        .map(extract_message_response_from_function_invocation)
                         .collect();
 
                     Ok(message_responses)
@@ -67,9 +67,18 @@ where
     }
 }
 
-pub fn extract_events_from_function_invocation(
+pub fn extract_message_response_from_function_invocation(
     invocation: FunctionInvocation,
 ) -> StarknetMessageResponse {
+    let result = invocation.result.clone();
+    let events = extract_events_from_function_invocation(invocation);
+
+    StarknetMessageResponse { result, events }
+}
+
+pub fn extract_events_from_function_invocation(
+    invocation: FunctionInvocation,
+) -> Vec<StarknetEvent> {
     let mut events: Vec<StarknetEvent> = invocation
         .events
         .into_iter()
@@ -83,14 +92,11 @@ pub fn extract_events_from_function_invocation(
         .collect();
 
     for inner in invocation.calls {
-        let mut message_response = extract_events_from_function_invocation(inner);
-        events.append(&mut message_response.events);
+        let mut in_events = extract_events_from_function_invocation(inner);
+        events.append(&mut in_events);
     }
 
-    StarknetMessageResponse {
-        result: invocation.result,
-        events,
-    }
+    events
 }
 
 impl Debug for UnexpectedTransactionTraceType {