From 5567a386196fbe6e7ff161661699c1ffabb2a2d4 Mon Sep 17 00:00:00 2001 From: Kyle Espinola Date: Fri, 6 Oct 2023 20:31:16 +0200 Subject: [PATCH] refactor: use hub-uploads for metadata json saving --- .env | 4 +- Cargo.lock | 20 +++ api/Cargo.toml | 2 +- .../tasks/metadata_json_upload_task.rs | 159 ++++++++---------- api/src/background_worker/tasks/mod.rs | 2 + api/src/{nft_storage.rs => hub_uploads.rs} | 71 +++----- api/src/lib.rs | 4 +- api/src/main.rs | 8 +- 8 files changed, 118 insertions(+), 152 deletions(-) rename api/src/{nft_storage.rs => hub_uploads.rs} (53%) diff --git a/.env b/.env index ba460bd..10b52df 100644 --- a/.env +++ b/.env @@ -5,9 +5,7 @@ SOLANA_ENDPOINT=https://api.devnet.solana.com KAFKA_BROKERS=127.0.0.1:9092 KAFKA_SSL=false SOLANA_KEYPAIR_PATH=./keypair.json -NFT_STORAGE_API_ENDPOINT=https://api.nft.storage -IPFS_ENDPOINT=https://nftstorage.link/ipfs +HUB_UPLOADS_API_ENDPOINT=http://localhost:3000 CREDIT_SHEET=credits.toml ASSET_CDN=https://assets.holaplex.tools -NFT_STORAGE_AUTH_TOKEN="" REDIS_URL=redis://localhost:6379 \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 46b0024..cebd9c7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2074,6 +2074,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "mime_guess" +version = "2.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4192263c238a5f0d0c6bfd21f336a313a4ce1c450542449ca191bb657b4642ef" +dependencies = [ + "mime", + "unicase", +] + [[package]] name = "minimal-lexical" version = "0.2.1" @@ -3088,6 +3098,7 @@ dependencies = [ "js-sys", "log", "mime", + "mime_guess", "native-tls", "once_cell", "percent-encoding", @@ -4526,6 +4537,15 @@ dependencies = [ "version_check", ] +[[package]] +name = "unicase" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7d2d4dafb69621809a81864c9c1b864479e1235c0dd4e199924b9742439ed89" +dependencies = [ + "version_check", +] + [[package]] name = "unicode-bidi" version = "0.3.13" diff --git a/api/Cargo.toml b/api/Cargo.toml index 421db1c..3baa4d7 100644 --- a/api/Cargo.toml +++ b/api/Cargo.toml @@ -35,7 +35,7 @@ poem = { version = "1.3.50", features = ["anyhow", "test"] } async-graphql-poem = "5.0.3" prost = "0.11.9" prost-types = "0.11.9" -reqwest = { version = "0.11.14", features = ["json"] } +reqwest = { version = "0.11.14", features = ["json", "multipart"] } async-trait = "0.1.68" strum = { version = "0.24.1", features = ["derive"] } diff --git a/api/src/background_worker/tasks/metadata_json_upload_task.rs b/api/src/background_worker/tasks/metadata_json_upload_task.rs index e162949..f348c78 100644 --- a/api/src/background_worker/tasks/metadata_json_upload_task.rs +++ b/api/src/background_worker/tasks/metadata_json_upload_task.rs @@ -11,8 +11,8 @@ use crate::{ collection_creators, collection_mints, collections, drops, metadata_jsons, mint_creators, sea_orm_active_enums::Blockchain as BlockchainEnum, update_histories, }, + hub_uploads::{HubUploadClient, UploadResponse}, mutations::collection::fetch_owner, - nft_storage::NftStorageClient, objects::MetadataJsonInput, proto::{ CreateEditionTransaction, EditionInfo, MasterEdition, MetaplexMasterEditionTransaction, @@ -27,7 +27,7 @@ trait After { &self, db: Connection, context: Context, - identifier: String, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError>; } @@ -42,14 +42,8 @@ impl After for CreateDrop { &self, db: Connection, context: Context, - identifier: String, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError> { - let metadata_uri = context - .nft_storage - .ipfs_endpoint - .join(&identifier)? - .to_string(); - let conn = db.get(); let (drop, collection) = drops::Entity::find_by_id_with_collection(self.drop_id) .one(conn) @@ -75,10 +69,13 @@ impl After for CreateDrop { let mut metadata_json_am: metadata_jsons::ActiveModel = metadata_json.clone().into(); - metadata_json_am.uri = Set(Some(metadata_uri.clone())); - metadata_json_am.identifier = Set(Some(identifier.clone())); + metadata_json_am.uri = Set(Some(upload_response.uri)); + metadata_json_am.identifier = Set(Some(upload_response.cid)); - metadata_json_am.update(conn).await?; + let metadata_json = metadata_json_am.update(conn).await?; + let metadata_uri = metadata_json + .uri + .ok_or(BackgroundTaskError::NoMetadataUri)?; let event_key = NftEventKey { id: collection.id.to_string(), @@ -98,9 +95,9 @@ impl After for CreateDrop { master_edition: Some(MasterEdition { owner_address, supply, + metadata_uri, name: metadata_json.name, symbol: metadata_json.symbol, - metadata_uri, seller_fee_basis_points: seller_fee_basis_points.into(), creators: creators.into_iter().map(Into::into).collect(), }), @@ -149,14 +146,8 @@ impl After for MintToCollection { &self, db: Connection, context: Context, - identifier: String, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError> { - let metadata_uri = context - .nft_storage - .ipfs_endpoint - .join(&identifier)? - .to_string(); - let conn = db.get(); let (collection_mint, collection) = collection_mints::Entity::find_by_id_with_collection(self.collection_mint_id) @@ -182,10 +173,13 @@ impl After for MintToCollection { let mut metadata_json_am: metadata_jsons::ActiveModel = metadata_json.clone().into(); - metadata_json_am.uri = Set(Some(metadata_uri.clone())); - metadata_json_am.identifier = Set(Some(identifier.clone())); + metadata_json_am.uri = Set(Some(upload_response.uri)); + metadata_json_am.identifier = Set(Some(upload_response.cid)); - metadata_json_am.update(conn).await?; + let metadata_json = metadata_json_am.update(conn).await?; + let metadata_uri = metadata_json + .uri + .ok_or(BackgroundTaskError::NoMetadataUri)?; let event_key = NftEventKey { id: collection_mint.id.to_string(), @@ -204,9 +198,9 @@ impl After for MintToCollection { .mint_to_collection(event_key, MintMetaplexMetadataTransaction { metadata: Some(MetaplexMetadata { owner_address, + metadata_uri, name: metadata_json.name, symbol: metadata_json.symbol, - metadata_uri, seller_fee_basis_points: seller_fee_basis_points.into(), creators: creators.into_iter().map(Into::into).collect(), }), @@ -236,14 +230,8 @@ impl After for CreateCollection { &self, db: Connection, context: Context, - identifier: String, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError> { - let metadata_uri = context - .nft_storage - .ipfs_endpoint - .join(&identifier)? - .to_string(); - let conn = db.get(); let collection = collections::Entity::find_by_id(self.collection_id) @@ -268,10 +256,13 @@ impl After for CreateCollection { let mut metadata_json_am: metadata_jsons::ActiveModel = metadata_json.clone().into(); - metadata_json_am.uri = Set(Some(metadata_uri.clone())); - metadata_json_am.identifier = Set(Some(identifier.clone())); + metadata_json_am.uri = Set(Some(upload_response.uri)); + metadata_json_am.identifier = Set(Some(upload_response.cid)); - metadata_json_am.update(conn).await?; + let metadata_json = metadata_json_am.update(conn).await?; + let metadata_uri = metadata_json + .uri + .ok_or(BackgroundTaskError::NoMetadataUri)?; let event_key = NftEventKey { id: collection.id.to_string(), @@ -287,10 +278,10 @@ impl After for CreateCollection { .create_collection(event_key, MetaplexMasterEditionTransaction { master_edition: Some(MasterEdition { owner_address, + metadata_uri, supply: Some(0), name: metadata_json.name, symbol: metadata_json.symbol, - metadata_uri, seller_fee_basis_points: seller_fee_basis_points.into(), creators: creators.into_iter().map(Into::into).collect(), }), @@ -317,15 +308,9 @@ impl After for QueueMintToDrop { async fn after( &self, db: Connection, - context: Context, - identifier: String, + _context: Context, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError> { - let metadata_uri = context - .nft_storage - .ipfs_endpoint - .join(&identifier)? - .to_string(); - let conn = db.get(); let metadata_json = metadata_jsons::Entity::find_by_id(self.collection_mint_id) @@ -335,8 +320,8 @@ impl After for QueueMintToDrop { let mut metadata_json_am: metadata_jsons::ActiveModel = metadata_json.clone().into(); - metadata_json_am.uri = Set(Some(metadata_uri.clone())); - metadata_json_am.identifier = Set(Some(identifier.clone())); + metadata_json_am.uri = Set(Some(upload_response.uri)); + metadata_json_am.identifier = Set(Some(upload_response.cid)); metadata_json_am.update(conn).await?; @@ -355,14 +340,8 @@ impl After for UpdateMint { &self, db: Connection, context: Context, - identifier: String, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError> { - let metadata_uri = context - .nft_storage - .ipfs_endpoint - .join(&identifier)? - .to_string(); - let conn = db.get(); let update_history = update_histories::Entity::find() @@ -394,10 +373,13 @@ impl After for UpdateMint { let mut metadata_json_am: metadata_jsons::ActiveModel = metadata_json.clone().into(); - metadata_json_am.uri = Set(Some(metadata_uri.clone())); - metadata_json_am.identifier = Set(Some(identifier.clone())); + metadata_json_am.uri = Set(Some(upload_response.uri)); + metadata_json_am.identifier = Set(Some(upload_response.cid)); - metadata_json_am.update(conn).await?; + let metadata_json = metadata_json_am.update(conn).await?; + let metadata_uri = metadata_json + .uri + .ok_or(BackgroundTaskError::NoMetadataUri)?; match collection.blockchain { BlockchainEnum::Solana => { @@ -413,9 +395,9 @@ impl After for UpdateMint { UpdateSolanaMintPayload { metadata: Some(MetaplexMetadata { owner_address, + metadata_uri, name: metadata_json.name, symbol: metadata_json.symbol, - metadata_uri, seller_fee_basis_points: collection.seller_fee_basis_points.into(), creators: creators.into_iter().map(Into::into).collect(), }), @@ -446,14 +428,8 @@ impl After for PatchCollection { &self, db: Connection, context: Context, - identifier: String, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError> { - let metadata_uri = context - .nft_storage - .ipfs_endpoint - .join(&identifier)? - .to_string(); - let conn = db.get(); let collection = collections::Entity::find_by_id(self.collection_id) @@ -478,10 +454,13 @@ impl After for PatchCollection { let mut metadata_json_am: metadata_jsons::ActiveModel = metadata_json.clone().into(); - metadata_json_am.uri = Set(Some(metadata_uri.clone())); - metadata_json_am.identifier = Set(Some(identifier.clone())); + metadata_json_am.uri = Set(Some(upload_response.uri)); + metadata_json_am.identifier = Set(Some(upload_response.cid)); - metadata_json_am.update(conn).await?; + let metadata_json = metadata_json_am.update(conn).await?; + let metadata_uri = metadata_json + .uri + .ok_or(BackgroundTaskError::NoMetadataUri)?; let event_key = NftEventKey { id: collection.id.to_string(), @@ -497,10 +476,10 @@ impl After for PatchCollection { .update_collection(event_key, MetaplexMasterEditionTransaction { master_edition: Some(MasterEdition { owner_address, + metadata_uri, supply: Some(0), name: metadata_json.name, symbol: metadata_json.symbol, - metadata_uri, seller_fee_basis_points: seller_fee_basis_points.into(), creators: creators.into_iter().map(Into::into).collect(), }), @@ -528,14 +507,8 @@ impl After for PatchDrop { &self, db: Connection, context: Context, - identifier: String, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError> { - let metadata_uri = context - .nft_storage - .ipfs_endpoint - .join(&identifier)? - .to_string(); - let conn = db.get(); let (drop, collection) = drops::Entity::find_by_id_with_collection(self.drop_id) @@ -562,10 +535,13 @@ impl After for PatchDrop { let mut metadata_json_am: metadata_jsons::ActiveModel = metadata_json.clone().into(); - metadata_json_am.uri = Set(Some(metadata_uri.clone())); - metadata_json_am.identifier = Set(Some(identifier.clone())); + metadata_json_am.uri = Set(Some(upload_response.uri)); + metadata_json_am.identifier = Set(Some(upload_response.cid)); - metadata_json_am.update(conn).await?; + let metadata_json = metadata_json_am.update(conn).await?; + let metadata_uri = metadata_json + .uri + .ok_or(BackgroundTaskError::NoMetadataUri)?; let event_key = NftEventKey { id: collection.id.to_string(), @@ -584,10 +560,10 @@ impl After for PatchDrop { MetaplexMasterEditionTransaction { master_edition: Some(MasterEdition { owner_address, + metadata_uri, supply: collection.supply.map(TryInto::try_into).transpose()?, name: metadata_json.name, symbol: metadata_json.symbol, - metadata_uri, seller_fee_basis_points: seller_fee_basis_points.into(), creators: creators.into_iter().map(Into::into).collect(), }), @@ -640,16 +616,16 @@ impl After for Caller { &self, db: Connection, context: Context, - identifier: String, + upload_response: UploadResponse, ) -> Result<(), BackgroundTaskError> { match self { - Self::CreateDrop(inner) => inner.after(db, context, identifier).await, - Self::MintToCollection(inner) => inner.after(db, context, identifier).await, - Self::CreateCollection(inner) => inner.after(db, context, identifier).await, - Self::QueueMintToDrop(inner) => inner.after(db, context, identifier).await, - Self::UpdateMint(inner) => inner.after(db, context, identifier).await, - Self::PatchCollection(inner) => inner.after(db, context, identifier).await, - Self::PatchDrop(inner) => inner.after(db, context, identifier).await, + Self::CreateDrop(inner) => inner.after(db, context, upload_response).await, + Self::MintToCollection(inner) => inner.after(db, context, upload_response).await, + Self::CreateCollection(inner) => inner.after(db, context, upload_response).await, + Self::QueueMintToDrop(inner) => inner.after(db, context, upload_response).await, + Self::UpdateMint(inner) => inner.after(db, context, upload_response).await, + Self::PatchCollection(inner) => inner.after(db, context, upload_response).await, + Self::PatchDrop(inner) => inner.after(db, context, upload_response).await, } } } @@ -672,16 +648,16 @@ impl MetadataJsonUploadTask { #[derive(Clone, Debug)] pub struct Context { - nft_storage: NftStorageClient, + hub_uploads: HubUploadClient, solana: Solana, polygon: Polygon, } impl Context { #[must_use] - pub fn new(nft_storage: NftStorageClient, solana: Solana, polygon: Polygon) -> Self { + pub fn new(hub_uploads: HubUploadClient, solana: Solana, polygon: Polygon) -> Self { Self { - nft_storage, + hub_uploads, solana, polygon, } @@ -699,10 +675,9 @@ impl BackgroundTask for MetadataJsonUploadTask { } async fn process(&self, db: Connection, context: Context) -> Result<(), BackgroundTaskError> { - let response = context.nft_storage.upload(&self.metadata_json).await?; - let cid = response.value.cid; + let response = context.hub_uploads.upload(&self.metadata_json).await?; - self.caller.after(db, context.clone(), cid).await?; + self.caller.after(db, context.clone(), response).await?; Ok(()) } diff --git a/api/src/background_worker/tasks/mod.rs b/api/src/background_worker/tasks/mod.rs index a113bfb..48f6b0d 100644 --- a/api/src/background_worker/tasks/mod.rs +++ b/api/src/background_worker/tasks/mod.rs @@ -30,6 +30,8 @@ pub enum BackgroundTaskError { Conversion(#[from] std::convert::Infallible), #[error("No creator")] NoCreator, + #[error("No metadata json uri")] + NoMetadataUri, } #[async_trait::async_trait] diff --git a/api/src/nft_storage.rs b/api/src/hub_uploads.rs similarity index 53% rename from api/src/nft_storage.rs rename to api/src/hub_uploads.rs index ab115fd..9d5ceb9 100644 --- a/api/src/nft_storage.rs +++ b/api/src/hub_uploads.rs @@ -1,55 +1,44 @@ -use std::collections::HashMap; - use hub_core::{ anyhow::Result, backon::{ExponentialBuilder, Retryable}, clap, prelude::*, }; -use reqwest::Response; +use reqwest::{ + multipart::{Form, Part}, + Response, +}; use serde::{Deserialize, Serialize}; -use serde_json::Value; /// Arguments for establishing a nft storage connection #[derive(Debug, clap::Args)] -pub struct NftStorageArgs { - #[arg(long, env)] - pub nft_storage_api_endpoint: String, - #[arg(long, env)] - pub nft_storage_auth_token: String, +pub struct HubUploadArgs { #[arg(long, env)] - pub ipfs_endpoint: String, + pub hub_uploads_api_endpoint: String, } #[derive(Debug, Clone)] -pub struct NftStorageClient { +pub struct HubUploadClient { http: reqwest::Client, - auth: String, pub api_base_url: Url, - pub ipfs_endpoint: Url, } -impl NftStorageClient { +impl HubUploadClient { /// Returns the `NftStorage` client /// /// # Errors /// if http client fails to build or url parsing fails - pub fn new(args: NftStorageArgs) -> Result { - let NftStorageArgs { - nft_storage_api_endpoint, - nft_storage_auth_token, - ipfs_endpoint, + pub fn new(args: HubUploadArgs) -> Result { + let HubUploadArgs { + hub_uploads_api_endpoint, } = args; - let api_base_url = Url::parse(&nft_storage_api_endpoint) + let api_base_url = Url::parse(&hub_uploads_api_endpoint) .context("failed to parse nft storage base url")?; - let ipfs_endpoint = Url::parse(&ipfs_endpoint)?; Ok(Self { http: reqwest::Client::new(), - auth: nft_storage_auth_token, api_base_url, - ipfs_endpoint, }) } @@ -57,13 +46,17 @@ impl NftStorageClient { /// /// # Errors /// Post request can fail if the auth token/payload is invalid or the api is down - pub async fn post(&self, endpoint: String, body: impl Serialize) -> Result { + async fn post(&self, endpoint: String, body: impl Serialize) -> Result { let url = self.api_base_url.join(&endpoint)?; + let serialized_body = serde_json::to_vec(&body).context("failed to serialize body")?; + let part = Part::bytes(serialized_body).file_name("file_name.extension"); + + let form = Form::new().part("file", part); + self.http .post(url) - .bearer_auth(&self.auth) - .json(&body) + .multipart(form) .send() .await .context("failed to send post request") @@ -74,7 +67,7 @@ impl NftStorageClient { /// # Errors /// If the upload fails pub async fn upload(&self, data: &impl Serialize) -> Result { - let post = || self.post("/upload".to_string(), data); + let post = || self.post("/uploads".to_string(), data); post.retry( &ExponentialBuilder::default() @@ -91,28 +84,6 @@ impl NftStorageClient { #[derive(Serialize, Deserialize, Clone, Debug)] pub struct UploadResponse { - pub ok: bool, - pub value: Nft, -} - -#[derive(Serialize, Deserialize, Clone, Debug)] -pub struct Nft { - pub cid: String, - pub size: u64, - pub created: String, - pub r#type: String, - #[serde(flatten)] - extra: HashMap, -} - -#[derive(Serialize, Deserialize, Clone, Debug)] - -pub struct Pin { + pub uri: String, pub cid: String, - pub name: String, - pub status: String, - pub created: String, - pub size: String, - #[serde(flatten)] - extra: HashMap, } diff --git a/api/src/lib.rs b/api/src/lib.rs index 7d823ea..9a3c1ee 100644 --- a/api/src/lib.rs +++ b/api/src/lib.rs @@ -9,9 +9,9 @@ pub mod db; pub mod entities; pub mod events; pub mod handlers; +pub mod hub_uploads; pub mod metrics; pub mod mutations; -pub mod nft_storage; pub mod objects; pub mod queries; @@ -114,7 +114,7 @@ pub struct Args { pub db: db::DbArgs, #[command(flatten)] - pub nft_storage: nft_storage::NftStorageArgs, + pub hub_uploads: hub_uploads::HubUploadArgs, #[arg(long, env)] pub redis_url: String, diff --git a/api/src/main.rs b/api/src/main.rs index 9cdb0b9..0f8f2a1 100644 --- a/api/src/main.rs +++ b/api/src/main.rs @@ -11,8 +11,8 @@ use holaplex_hub_nfts::{ db::Connection, events, handlers::{graphql_handler, health, metrics_handler, playground}, + hub_uploads::HubUploadClient, metrics::Metrics, - nft_storage::NftStorageClient, proto, Actions, AppState, Args, Services, }; use hub_core::{prelude::*, tokio}; @@ -28,7 +28,7 @@ pub fn main() { let Args { port, db, - nft_storage, + hub_uploads, redis_url, } = args; @@ -45,7 +45,7 @@ pub fn main() { .build::() .await?; let credits = common.credits_cfg.build::().await?; - let nft_storage = NftStorageClient::new(nft_storage)?; + let hub_uploads = HubUploadClient::new(hub_uploads)?; let metrics = Metrics::new()?; @@ -62,7 +62,7 @@ pub fn main() { let redis_client = RedisClient::open(redis_url)?; let metadata_json_upload_task_context = - MetadataJsonUploadContext::new(nft_storage, solana.clone(), polygon.clone()); + MetadataJsonUploadContext::new(hub_uploads, solana.clone(), polygon.clone()); let job_queue = JobQueue::new(redis_client, connection.clone()); let worker = Worker::::new(