feat(katana): forked events

crates/katana/primitives/src/event.rs

@@ -12,6 +12,47 @@ pub struct OrderedEvent {
     pub data: Vec<Felt>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum MaybeForkedContinuationToken {
+    Token(ContinuationToken),
+    Forked(String),
+}
+
+impl MaybeForkedContinuationToken {
+    pub fn parse(value: &str) -> Result<Self, ContinuationTokenError> {
+        const FORKED_TOKEN_PREFIX: &str = "FK_";
+        if let Some(token) = value.strip_prefix(FORKED_TOKEN_PREFIX) {
+            Ok(MaybeForkedContinuationToken::Forked(token.to_string()))
+        } else {
+            let token = ContinuationToken::parse(value)?;
+            Ok(MaybeForkedContinuationToken::Token(token))
+        }
+    }
+
+    pub fn to_token(self) -> Option<ContinuationToken> {
+        match self {
+            MaybeForkedContinuationToken::Token(token) => Some(token),
+            _ => None,
+        }
+    }
+}
+
+impl std::fmt::Display for MaybeForkedContinuationToken {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            MaybeForkedContinuationToken::Token(token) => write!(f, "{token}"),
+            MaybeForkedContinuationToken::Forked(token) => write!(f, "FK_{token}"),
+        }
+    }
+}
+
+impl<'de> ::serde::Deserialize<'de> for MaybeForkedContinuationToken {
+    fn deserialize<D: ::serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        let s = String::deserialize(deserializer)?;
+        Self::parse(&s).map_err(::serde::de::Error::custom)
+    }
+}
+
 /// Represents a continuation token for implementing paging in event queries.
 ///
 /// This struct stores the necessary information to resume fetching events
@@ -20,7 +61,7 @@ pub struct OrderedEvent {
 ///
 /// There JSON-RPC specification does not specify the format of the continuation token,
 /// so how the node should handle it is implementation specific.
-#[derive(PartialEq, Eq, Debug, Default)]
+#[derive(PartialEq, Eq, Debug, Clone, Default)]
 pub struct ContinuationToken {
     /// The block number to continue from.
     pub block_n: u64,

crates/katana/rpc/rpc/Cargo.toml

@@ -45,5 +45,6 @@ rand.workspace = true
 rstest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+similar-asserts.workspace = true
 tempfile.workspace = true
 tokio.workspace = true

crates/katana/rpc/rpc/src/starknet/forking.rs

@@ -1,13 +1,16 @@
 use katana_primitives::block::{BlockIdOrTag, BlockNumber};
+use katana_primitives::contract::ContractAddress;
 use katana_primitives::transaction::TxHash;
+use katana_primitives::Felt;
 use katana_rpc_types::block::{
     MaybePendingBlockWithReceipts, MaybePendingBlockWithTxHashes, MaybePendingBlockWithTxs,
 };
 use katana_rpc_types::error::starknet::StarknetApiError;
+use katana_rpc_types::event::EventsPage;
 use katana_rpc_types::receipt::TxReceiptWithBlockInfo;
 use katana_rpc_types::state_update::MaybePendingStateUpdate;
 use katana_rpc_types::transaction::Tx;
-use starknet::core::types::TransactionStatus;
+use starknet::core::types::{EventFilter, TransactionStatus};
 use starknet::providers::jsonrpc::HttpTransport;
 use starknet::providers::{JsonRpcClient, Provider, ProviderError};
 use url::Url;
@@ -20,6 +23,12 @@ pub enum Error {
 
     #[error("Block out of range")]
     BlockOutOfRange,
+
+    #[error("Not allowed to use block tag as a block identifier")]
+    BlockTagNotAllowed,
+
+    #[error("Unexpected pending data")]
+    UnexpectedPendingData,
 }
 
 #[derive(Debug)]
@@ -108,7 +117,7 @@ impl<P: Provider> ForkedClient<P> {
                         block.block_number
                     }
                     starknet::core::types::MaybePendingBlockWithTxHashes::PendingBlock(_) => {
-                        panic!("shouldn't be possible to be pending")
+                        return Err(Error::UnexpectedPendingData);
                     }
                 };
 
@@ -119,9 +128,7 @@ impl<P: Provider> ForkedClient<P> {
                 Ok(tx?.into())
             }
 
-            BlockIdOrTag::Tag(_) => {
-                panic!("shouldn't be possible to be tag")
-            }
+            BlockIdOrTag::Tag(_) => Err(Error::BlockTagNotAllowed),
         }
     }
 
@@ -141,7 +148,7 @@ impl<P: Provider> ForkedClient<P> {
             }
 
             starknet::core::types::MaybePendingBlockWithTxs::PendingBlock(_) => {
-                panic!("shouldn't be possible to be pending")
+                Err(Error::UnexpectedPendingData)
             }
         }
     }
@@ -159,7 +166,7 @@ impl<P: Provider> ForkedClient<P> {
                 }
             }
             starknet::core::types::MaybePendingBlockWithReceipts::PendingBlock(_) => {
-                panic!("shouldn't be possible to be pending")
+                return Err(Error::UnexpectedPendingData);
             }
         }
 
@@ -179,7 +186,7 @@ impl<P: Provider> ForkedClient<P> {
                 }
             }
             starknet::core::types::MaybePendingBlockWithTxHashes::PendingBlock(_) => {
-                panic!("shouldn't be possible to be pending")
+                return Err(Error::UnexpectedPendingData);
             }
         }
 
@@ -201,7 +208,7 @@ impl<P: Provider> ForkedClient<P> {
                 }
             }
             BlockIdOrTag::Tag(_) => {
-                panic!("shouldn't be possible to be tag")
+                return Err(Error::BlockTagNotAllowed);
             }
             _ => {}
         }
@@ -228,21 +235,46 @@ impl<P: Provider> ForkedClient<P> {
                 }
             }
             BlockIdOrTag::Tag(_) => {
-                panic!("shouldn't be possible to be tag")
+                return Err(Error::BlockTagNotAllowed);
             }
             _ => {}
         }
 
         let state_update = self.provider.get_state_update(block_id).await?;
         Ok(state_update.into())
     }
+
+    // NOTE(kariy): The reason why I don't just use EventFilter as a param, bcs i wanna make sure
+    // the from/to blocks are not None. maybe should do the same for other methods that accept a
+    // BlockId in some way?
+    pub async fn get_events(
+        &self,
+        from: BlockNumber,
+        to: BlockNumber,
+        address: Option<ContractAddress>,
+        keys: Option<Vec<Vec<Felt>>>,
+        continuation_token: Option<String>,
+        chunk_size: u64,
+    ) -> Result<EventsPage, Error> {
+        let from_block = Some(BlockIdOrTag::Number(from));
+        let to_block = Some(BlockIdOrTag::Number(to));
+        let address = address.map(Felt::from);
+        let filter = EventFilter { from_block, to_block, address, keys };
+
+        let events = self.provider.get_events(filter, continuation_token, chunk_size).await?;
+
+        Ok(events)
+    }
 }
 
 impl From<Error> for StarknetApiError {
     fn from(value: Error) -> Self {
         match value {
             Error::Provider(provider_error) => provider_error.into(),
             Error::BlockOutOfRange => StarknetApiError::BlockNotFound,
+            Error::BlockTagNotAllowed | Error::UnexpectedPendingData => {
+                StarknetApiError::UnexpectedError { reason: value.to_string() }
+            }
         }
     }
 }