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"
+interface = { path = "../interface" }
+
+[dev-dependencies]
+tempfile = "3.8.1"
 
 [dependencies.utils]
 version = "0.1.8"

nft_ingester/src/gapfiller.rs

@@ -1,11 +1,29 @@
 use crate::error::IngesterError;
 use entities::models::CompleteAssetDetails;
 use entities::models::Updated;
+use futures::StreamExt;
+use interface::AssetDetailsStream;
+use log::error;
 use rocks_db::asset::{AssetCollection, AssetLeaf};
 use rocks_db::{AssetAuthority, AssetDynamicDetails, AssetOwner, AssetStaticDetails, Storage};
 use serde_json::json;
 use std::sync::Arc;
 
+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,53 @@
+use entities::models::{CompleteAssetDetails, Updated};
+use futures::stream;
+use interface::AsyncError;
+use nft_ingester::gapfiller::process_asset_details_stream;
+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)));
+}