[NFT Metadata Crawler] Asset Uploader Throttler (#14904)

* yay

* upsert

* awesome

* AHHHHHHHHHHHHHHHHHHHHHHH

* boom

* boom

* lint

* lint

ecosystem/nft-metadata-crawler/Cargo.toml

@@ -37,6 +37,8 @@ futures = { workspace = true }
 google-cloud-storage = { workspace = true }
 image = { workspace = true }
 once_cell = { workspace = true }
+parking_lot = { workspace = true }
+rayon = { workspace = true }
 regex = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }

ecosystem/nft-metadata-crawler/migrations/2024-10-02-062003_add_asset_uploader_requests_table/up.sql

@@ -3,7 +3,7 @@ CREATE TABLE nft_metadata_crawler.asset_uploader_request_statuses (
   asset_uri VARCHAR NOT NULL,
   application_id UUID NOT NULL,
   status_code BIGINT NOT NULL DEFAULT 202,
-  error_message VARCHAR,
+  error_messages TEXT[],
   cdn_image_uri VARCHAR,
   num_failures BIGINT NOT NULL DEFAULT 0,
   request_received_at TIMESTAMP NOT NULL DEFAULT NOW(),

ecosystem/nft-metadata-crawler/src/asset_uploader/api/get_status.rs

@@ -32,7 +32,7 @@ pub fn get_status(
         } else {
             status_response.insert(row.asset_uri, GetStatusResponseSuccess::Error {
                 status_code: row.status_code as u16,
-                error_message: row.error_message,
+                error_message: row.error_messages,
             });
         };
     }

ecosystem/nft-metadata-crawler/src/asset_uploader/api/mod.rs

@@ -50,7 +50,7 @@ pub enum GetStatusResponseSuccess {
     },
     Error {
         status_code: u16,
-        error_message: Option<String>,
+        error_message: Option<Vec<Option<String>>>,
     },
 }
 

ecosystem/nft-metadata-crawler/src/asset_uploader/mod.rs

@@ -2,4 +2,5 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub mod api;
+pub mod throttler;
 pub mod worker;

ecosystem/nft-metadata-crawler/src/asset_uploader/throttler/config.rs

@@ -0,0 +1,33 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct AssetUploaderThrottlerConfig {
+    /// URI for the Asset Uploader Worker
+    pub asset_uploader_worker_uri: String,
+    /// Interval in seconds to poll Postgres to update upload queue
+    #[serde(default = "AssetUploaderThrottlerConfig::default_poll_interval_seconds")]
+    pub poll_interval_seconds: u64,
+    /// Maximum number of rows to poll from Postgres
+    #[serde(default = "AssetUploaderThrottlerConfig::default_poll_rows_limit")]
+    pub poll_rows_limit: u64,
+    /// Cloudflare Account Hash provided at the images home page used for generating the CDN image URLs
+    pub cloudflare_account_hash: String,
+    /// Cloudflare Image Delivery URL prefix provided at the images home page used for generating the CDN image URLs
+    pub cloudflare_image_delivery_prefix: String,
+    /// In addition to on the fly transformations, Cloudflare images can be returned in preset variants. This is the default variant used with the saved CDN image URLs.
+    pub cloudflare_default_variant: String,
+}
+
+impl AssetUploaderThrottlerConfig {
+    pub const fn default_poll_interval_seconds() -> u64 {
+        10
+    }
+
+    pub const fn default_poll_rows_limit() -> u64 {
+        600
+    }
+}