Get metadata json for collection & mints from URI

consumer/Cargo.toml

@@ -33,6 +33,7 @@ holaplex-hub-nfts-solana-entity = { path = "../entity" }
 jsonrpsee = { version = "0.18.2", features = ["macros", "http-client"] }
 bs58 = "0.5.0"
 rand = "0.8.5" 
+tokio-retry = "0.3.0"
 
 [dependencies.hub-core]
 package = "holaplex-hub-core"

consumer/src/asset_api.rs

@@ -168,6 +168,11 @@ pub struct Metadata {
     pub description: Option<String>,
     pub name: String,
     pub symbol: Option<String>,
+    pub image: Option<String>,
+    pub external_url: Option<String>,
+    pub animation_url: Option<String>,
+    #[serde(flatten)]
+    extra: HashMap<String, serde_json::Value>,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

consumer/src/events.rs

@@ -675,7 +675,6 @@ impl Processor {
             mint: tx.addresses.mint.to_string(),
             created_at: Utc::now().naive_utc(),
             associated_token_account: tx.addresses.associated_token_account.to_string(),
-            ..Default::default()
         };
 
         CollectionMint::create(&self.db, collection_mint).await?;

consumer/src/import.rs

@@ -9,15 +9,21 @@ use holaplex_hub_nfts_solana_core::{
     Collection, Services,
 };
 use holaplex_hub_nfts_solana_entity::{collection_mints, collections, prelude::CollectionMints};
-use hub_core::{chrono::Utc, prelude::*, producer::Producer, util::DebugShim, uuid::Uuid};
+use hub_core::{
+    chrono::Utc, futures_util::stream, prelude::*, producer::Producer, reqwest, util::DebugShim,
+    uuid::Uuid,
+};
 use mpl_token_metadata::pda::{find_master_edition_account, find_metadata_account};
 use spl_associated_token_account::get_associated_token_address;
+use tokio_retry::{strategy::ExponentialBackoff, Retry};
 
 use crate::{
     asset_api::{self, Asset, RpcClient},
     solana::Solana,
 };
 
+const CONCURRENT_REQUESTS: usize = 64;
+
 // TODO: could this just be a newtype over events::Processor?
 #[derive(Debug, Clone)]
 pub struct Processor {
@@ -35,7 +41,7 @@ impl Processor {
         }
     }
 
-    pub async fn process(&self, msg: &Services) -> Result<Option<()>> {
+    pub async fn process(self, msg: &Services) -> Result<Option<()>> {
         match msg {
             Services::Nfts(key, msg) => {
                 let key = SolanaNftEventKey::from(key.clone());
@@ -52,7 +58,7 @@ impl Processor {
     }
 
     async fn process_import(
-        &self,
+        self,
         SolanaNftEventKey {
             id,
             project_id,
@@ -70,6 +76,7 @@ impl Processor {
         let collection = rpc.get_asset(&mint_address).await?;
 
         let collection_model = Collection::find_by_id(db, id.parse()?).await?;
+
         if let Some(collection_model) = collection_model {
             info!(
                 "Deleting already indexed collection: {:?}",
@@ -90,6 +97,7 @@ impl Processor {
                 .await?;
 
             let mut mints: Vec<collection_mints::ActiveModel> = Vec::new();
+            let mut futures = Vec::new();
 
             for asset in result.items {
                 let project_id = project_id.clone();
@@ -101,11 +109,22 @@ impl Processor {
                 }
 
                 info!("Importing mint: {:?}", asset.id.to_string());
-                let model = self
-                    .collection_mint_event(project_id, user_id, collection_model.id, asset)
-                    .await?;
 
-                mints.push(model.into());
+                let future = self.clone().collection_mint_event(
+                    project_id,
+                    user_id,
+                    collection_model.id,
+                    asset,
+                );
+                futures.push(future);
+            }
+
+            let mut buffered = stream::iter(futures).buffer_unordered(CONCURRENT_REQUESTS);
+            while let Some(result) = buffered.next().await {
+                match result {
+                    Ok(model) => mints.push(model),
+                    Err(e) => bail!("Error: {}", e),
+                }
             }
 
             CollectionMints::insert_many(mints)
@@ -132,20 +151,20 @@ impl Processor {
         let owner = collection.ownership.owner.try_into()?;
         let mint = collection.id.try_into()?;
         let seller_fee_basis_points = collection.royalty.basis_points;
-        let metadata = collection.content.metadata;
+
+        let json_uri = collection.content.json_uri.clone();
+
+        let json_metadata = Self::get_metadata_json(json_uri.clone()).await?;
+
         let files: Vec<File> = collection
             .content
             .files
             .map(|fs| fs.iter().map(Into::into).collect())
             .unwrap_or_default();
 
-        let image = files
-            .iter()
-            .find(|f| f.mime.as_ref().map_or(false, |m| m.contains("image")))
-            .map(|f| f.uri.clone())
-            .unwrap_or_default();
+        let image = json_metadata.image.unwrap_or_default();
 
-        let attributes = metadata
+        let attributes = json_metadata
             .attributes
             .clone()
             .map(|attributes| attributes.iter().map(Into::into).collect::<Vec<_>>())
@@ -192,9 +211,9 @@ impl Processor {
                             seller_fee_basis_points,
                             creators,
                             metadata: Some(Metadata {
-                                name: metadata.name,
-                                description: metadata.description,
-                                symbol: metadata.symbol.unwrap_or_default(),
+                                name: json_metadata.name,
+                                description: json_metadata.description,
+                                symbol: json_metadata.symbol.unwrap_or_default(),
                                 attributes,
                                 uri: collection.content.json_uri,
                                 image,
@@ -215,41 +234,57 @@ impl Processor {
         Ok(collection_model)
     }
 
+    async fn get_metadata_json(uri: String) -> Result<asset_api::Metadata> {
+        let json_metadata_fut = || async {
+            reqwest::get(uri.clone())
+                .await?
+                .json::<asset_api::Metadata>()
+                .await
+        };
+
+        let json_metadata = Retry::spawn(
+            ExponentialBackoff::from_millis(20).take(10),
+            json_metadata_fut,
+        )
+        .await?;
+
+        Ok(json_metadata)
+    }
+
     async fn collection_mint_event(
-        &self,
+        self,
         project_id: String,
         user_id: String,
         collection: Uuid,
         asset: Asset,
-    ) -> Result<collection_mints::Model> {
+    ) -> Result<collection_mints::ActiveModel> {
         let producer = self.producer.clone();
         let owner = asset.ownership.owner.try_into()?;
         let mint = asset.id.try_into()?;
         let ata = get_associated_token_address(&owner, &mint);
         let seller_fee_basis_points = asset.royalty.basis_points;
-        let metadata = asset.content.metadata;
+
         let update_authority = asset
             .authorities
             .get(0)
             .context("Invalid index")?
             .address
             .clone();
 
-        let files = asset
+        let json_metadata = Self::get_metadata_json(asset.content.json_uri.clone()).await?;
+
+        let files: Vec<File> = asset
             .content
             .files
-            .map(|fs| fs.iter().map(Into::into).collect::<Vec<File>>())
+            .map(|fs| fs.iter().map(Into::into).collect())
             .unwrap_or_default();
 
-        let image = files
-            .iter()
-            .find(|f| f.mime.as_ref().map_or(false, |m| m.contains("image")))
-            .map(|f| f.uri.clone())
-            .unwrap_or_default();
+        let image = json_metadata.image.unwrap_or_default();
 
-        let attributes = metadata
+        let attributes = json_metadata
             .attributes
-            .map(|attributes| attributes.iter().map(Into::into).collect())
+            .clone()
+            .map(|attributes| attributes.iter().map(Into::into).collect::<Vec<_>>())
             .unwrap_or_default();
 
         let creators = asset
@@ -285,9 +320,9 @@ impl Processor {
                         compressed: asset.compression.compressed,
                         creators,
                         metadata: Some(Metadata {
-                            name: metadata.name,
-                            description: metadata.description,
-                            symbol: metadata.symbol.unwrap_or_default(),
+                            name: json_metadata.name,
+                            description: json_metadata.description,
+                            symbol: json_metadata.symbol.unwrap_or_default(),
                             attributes,
                             uri: asset.content.json_uri,
                             image,
@@ -304,7 +339,7 @@ impl Processor {
             )
             .await?;
 
-        Ok(mint_model)
+        Ok(mint_model.into())
     }
 }