feat: process client stream

entities/src/models.rs

@@ -40,7 +40,7 @@ pub struct Creator {
     pub creator_share: u8,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
 pub struct CompleteAssetDetails {
     // From AssetStaticDetails
     pub pubkey: Pubkey,

nft_ingester/Cargo.toml

@@ -51,6 +51,11 @@ libc = "0.2.144"
 mockall = "0.11.4"
 entities = { path = "../entities" }
 mpl-token-metadata = "1.13.2"
+futures = "0.3.29"
+async-trait = "0.1.74"
+
+[dev-dependencies]
+tempfile = "3.8.1"
 
 [dependencies.utils]
 version = "0.1.8"

nft_ingester/src/gapfiller.rs

@@ -1,11 +1,45 @@
 use crate::error::IngesterError;
+use async_trait::async_trait;
 use entities::models::CompleteAssetDetails;
 use entities::models::Updated;
+use futures::{Stream, StreamExt};
+use log::error;
+use mockall::automock;
 use rocks_db::asset::{AssetCollection, AssetLeaf};
 use rocks_db::{AssetAuthority, AssetDynamicDetails, AssetOwner, AssetStaticDetails, Storage};
 use serde_json::json;
+use std::pin::Pin;
 use std::sync::Arc;
 
+pub type AsyncError = Box<dyn std::error::Error + Send + Sync>;
+pub type AssetDetailsStream =
+    Pin<Box<dyn Stream<Item = Result<CompleteAssetDetails, AsyncError>> + Send + Sync>>;
+
+#[automock]
+#[async_trait]
+pub trait AssetDetailsStreamer: Send + Sync {
+    async fn get_asset_details_stream_in_range(
+        &self,
+        start_slot: u64,
+        end_slot: u64,
+    ) -> Result<AssetDetailsStream, AsyncError>;
+}
+
+pub async fn process_asset_details_stream(storage: Arc<Storage>, mut stream: AssetDetailsStream) {
+    while let Some(result) = stream.next().await {
+        match result {
+            Ok(details) => {
+                if let Some(e) = insert_gaped_data(storage.clone(), details).err() {
+                    error!("Error processing gaped data: {}", e)
+                }
+            }
+            Err(e) => {
+                error!("Error processing stream item: {}", e);
+            }
+        }
+    }
+}
+
 pub fn insert_gaped_data(
     rocks_storage: Arc<Storage>,
     data: CompleteAssetDetails,

nft_ingester/tests/gapfiller_tests.rs

@@ -0,0 +1,52 @@
+use entities::models::{CompleteAssetDetails, Updated};
+use futures::stream;
+use nft_ingester::gapfiller::{process_asset_details_stream, AsyncError};
+use solana_sdk::pubkey::Pubkey;
+use std::sync::Arc;
+use tempfile::TempDir;
+
+use rocks_db::Storage;
+
+fn create_test_complete_asset_details(pubkey: Pubkey) -> CompleteAssetDetails {
+    CompleteAssetDetails {
+        pubkey,
+        supply: Some(Updated::new(1, None, 10)),
+        ..Default::default()
+    }
+}
+
+#[tokio::test]
+async fn test_process_asset_details_stream() {
+    let temp_dir = TempDir::new().expect("Failed to create a temporary directory");
+    let storage = Arc::new(
+        Storage::open(temp_dir.path().to_str().unwrap()).expect("Failed to create a database"),
+    );
+
+    let first_key = Pubkey::new_unique();
+    let second_key = Pubkey::new_unique();
+
+    let details1 = create_test_complete_asset_details(first_key.clone());
+    let details2 = create_test_complete_asset_details(second_key.clone());
+
+    let stream = stream::iter(vec![
+        Ok(details1),
+        Ok(details2),
+        Err(AsyncError::from("test error")),
+    ]);
+
+    process_asset_details_stream(storage.clone(), Box::pin(stream)).await;
+
+    let selected_data = storage
+        .asset_dynamic_data
+        .get(first_key.clone())
+        .unwrap()
+        .unwrap();
+    assert_eq!(selected_data.supply, Some(Updated::new(1, None, 10)));
+
+    let selected_data = storage
+        .asset_dynamic_data
+        .get(second_key.clone())
+        .unwrap()
+        .unwrap();
+    assert_eq!(selected_data.supply, Some(Updated::new(1, None, 10)));
+}