reffactoring

src/lib.rs

@@ -280,11 +280,17 @@ extern crate derive_builder;
 pub use near_indexer_primitives;
 
 pub use aws_credential_types::Credentials;
-pub use providers::fastnear::client as fastnear_client;
+
+mod providers;
+
+pub use providers::fastnear;
+pub use providers::s3;
+
+pub use providers::fastnear::client::FastNearClient;
 pub use providers::fastnear::types::{FastNearConfig, FastNearConfigBuilder};
-pub use providers::s3::client as s3_client;
+
+pub use providers::s3::client::LakeS3Client;
 pub use providers::s3::types::{LakeConfig, LakeConfigBuilder};
-pub mod providers;
 
 pub(crate) const LAKE_FRAMEWORK: &str = "near_lake_framework";
 
@@ -317,11 +323,10 @@ pub fn streamer<T: Into<providers::NearLakeFrameworkConfig>>(
     let (sender, receiver) = tokio::sync::mpsc::channel(config.blocks_preload_pool_size());
     match config {
         providers::NearLakeFrameworkConfig::Lake(config) => {
-            (tokio::spawn(providers::s3::start(sender, config)), receiver)
+            (tokio::spawn(s3::start(sender, config)), receiver)
+        }
+        providers::NearLakeFrameworkConfig::FastNear(config) => {
+            (tokio::spawn(fastnear::start(sender, config)), receiver)
         }
-        providers::NearLakeFrameworkConfig::FastNear(config) => (
-            tokio::spawn(providers::fastnear::start(sender, config)),
-            receiver,
-        ),
     }
 }

src/providers/fastnear/fetchers.rs

@@ -21,6 +21,21 @@ pub async fn fetch_optimistic_block(
         .expect("Failed to fetch optimistic block")
 }
 
+/// Fetches the optimistic block from the fastenar
+/// This function is used to fetch the optimistic block by height
+/// This function will be using endpoint `/v0/block_opt/:block_height`
+/// This would be waiting some time until the optimistic block is available
+/// Returns `near_indexer_primitives::StreamerMessage`
+pub async fn fetch_optimistic_block_by_height(
+    client: &FastNearClient,
+    block_height: types::BlockHeight,
+) -> near_indexer_primitives::StreamerMessage {
+    client
+        .fetch_until_success(&format!("/v0/block_opt/{}", block_height))
+        .await
+        .expect("Failed to fetch optimistic block")
+}
+
 /// Fetches the genesis block from the fastenar
 /// Returns `near_indexer_primitives::StreamerMessage`
 pub async fn fetch_first_block(

src/providers/fastnear/mod.rs

@@ -12,7 +12,7 @@ pub async fn start(
     blocks_sink: tokio::sync::mpsc::Sender<near_indexer_primitives::StreamerMessage>,
     config: types::FastNearConfig,
 ) -> anyhow::Result<()> {
-    let client = config.client();
+    let client = config.client().await;
     let max_num_threads = config.num_threads;
     let next_sink_block =
         std::sync::Arc::new(std::sync::atomic::AtomicU64::new(config.start_block_height));

src/providers/fastnear/types.rs

@@ -32,8 +32,8 @@ pub struct FastNearConfig {
 }
 
 impl FastNearConfig {
-    pub fn client(&self) -> crate::fastnear_client::FastNearClient {
-        crate::fastnear_client::FastNearClient::new(self.endpoint.clone())
+    pub async fn client(&self) -> super::client::FastNearClient {
+        super::client::FastNearClient::new(self.endpoint.clone())
     }
 }
 

src/providers/s3/client.rs

@@ -80,3 +80,151 @@ pub trait S3Client: Send + Sync {
         start_after_prefix: &str,
     ) -> Result<Vec<String>, ListCommonPrefixesError>;
 }
+
+#[derive(Clone, Debug)]
+pub struct LakeS3Client {
+    s3: aws_sdk_s3::Client,
+}
+
+impl LakeS3Client {
+    pub fn new(s3: aws_sdk_s3::Client) -> Self {
+        Self { s3 }
+    }
+
+    pub fn from_conf(config: aws_sdk_s3::config::Config) -> Self {
+        let s3_client = aws_sdk_s3::Client::from_conf(config);
+
+        Self { s3: s3_client }
+    }
+}
+
+#[async_trait]
+impl S3Client for LakeS3Client {
+    async fn get_object_bytes(
+        &self,
+        bucket: &str,
+        prefix: &str,
+    ) -> Result<Vec<u8>, GetObjectBytesError> {
+        let object = self
+            .s3
+            .get_object()
+            .bucket(bucket)
+            .key(prefix)
+            .request_payer(aws_sdk_s3::types::RequestPayer::Requester)
+            .send()
+            .await?;
+
+        let bytes = object.body.collect().await?.into_bytes().to_vec();
+
+        Ok(bytes)
+    }
+
+    async fn list_common_prefixes(
+        &self,
+        bucket: &str,
+        start_after_prefix: &str,
+    ) -> Result<Vec<String>, ListCommonPrefixesError> {
+        let response = self
+            .s3
+            .list_objects_v2()
+            .max_keys(1000) // 1000 is the default and max value for this parameter
+            .delimiter("/".to_string())
+            .start_after(start_after_prefix)
+            .request_payer(aws_sdk_s3::types::RequestPayer::Requester)
+            .bucket(bucket)
+            .send()
+            .await?;
+
+        let prefixes = match response.common_prefixes {
+            None => vec![],
+            Some(common_prefixes) => common_prefixes
+                .into_iter()
+                .filter_map(|common_prefix| common_prefix.prefix)
+                .collect::<Vec<String>>()
+                .into_iter()
+                .filter_map(|prefix_string| prefix_string.split('/').next().map(String::from))
+                .collect(),
+        };
+
+        Ok(prefixes)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    use std::sync::Arc;
+
+    use crate::providers::s3::fetchers::fetch_streamer_message;
+    use async_trait::async_trait;
+
+    #[derive(Clone, Debug)]
+    pub struct LakeS3Client {}
+
+    #[async_trait]
+    impl S3Client for LakeS3Client {
+        async fn get_object_bytes(
+            &self,
+            _bucket: &str,
+            prefix: &str,
+        ) -> Result<Vec<u8>, GetObjectBytesError> {
+            let path = format!("{}/blocks/{}", env!("CARGO_MANIFEST_DIR"), prefix);
+            tokio::fs::read(path)
+                .await
+                .map_err(|e| GetObjectBytesError(Arc::new(e)))
+        }
+
+        async fn list_common_prefixes(
+            &self,
+            _bucket: &str,
+            _start_after: &str,
+        ) -> Result<Vec<String>, ListCommonPrefixesError> {
+            Ok(Vec::new())
+        }
+    }
+
+    #[tokio::test]
+    async fn deserializes_meta_transactions() {
+        let lake_client = LakeS3Client {};
+        let streamer_message =
+            fetch_streamer_message(&lake_client, "near-lake-data-mainnet", 879765)
+                .await
+                .unwrap();
+
+        let delegate_action = &streamer_message.shards[0]
+            .chunk
+            .as_ref()
+            .unwrap()
+            .transactions[0]
+            .transaction
+            .actions[0];
+
+        assert_eq!(
+            serde_json::to_value(delegate_action).unwrap(),
+            serde_json::json!({
+                "Delegate": {
+                    "delegate_action": {
+                        "sender_id": "test.near",
+                        "receiver_id": "test.near",
+                        "actions": [
+                          {
+                            "AddKey": {
+                              "public_key": "ed25519:CnQMksXTTtn81WdDujsEMQgKUMkFvDJaAjDeDLTxVrsg",
+                              "access_key": {
+                                "nonce": 0,
+                                "permission": "FullAccess"
+                              }
+                            }
+                          }
+                        ],
+                        "nonce": 879546,
+                        "max_block_height": 100,
+                        "public_key": "ed25519:8Rn4FJeeRYcrLbcrAQNFVgvbZ2FCEQjgydbXwqBwF1ib"
+                    },
+                    "signature": "ed25519:25uGrsJNU3fVgUpPad3rGJRy2XQum8gJxLRjKFCbd7gymXwUxQ9r3tuyBCD6To7SX5oSJ2ScJZejwqK1ju8WdZfS"
+                }
+            })
+        );
+    }
+}