diff --git a/Cargo.lock b/Cargo.lock index 8363893b08..272d334a0b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9904,6 +9904,7 @@ dependencies = [ "billing_component", "bootstrap", "bucket_paths", + "bytes 1.10.1", "chrono", "cloud_storage", "collections", @@ -10819,11 +10820,10 @@ dependencies = [ [[package]] name = "tempfile" -version = "3.18.0" +version = "3.20.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2c317e0a526ee6120d8dabad239c8dadca62b24b6f168914bbbc8e2fb1f0e567" +checksum = "e8a64e3985349f2441a1a9ef0b853f869006c3855f2cda6862a94d26ebb9d6a1" dependencies = [ - "cfg-if 1.0.0", "fastrand 2.3.0", "getrandom 0.3.1", "once_cell", diff --git a/Cargo.toml b/Cargo.toml index cfcd837e2d..49125c675a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -182,6 +182,7 @@ actix-web-actors = "4.3.1+deprecated" actix-web-lab = "0.20.2" anyhow = "1.0.98" base64 = "0.22.1" +bytes = "1.10.1" chrono = { version = "0.4.41", features = ["serde"] } cookie = "0.18.1" futures = { version = "0.3.31" } @@ -198,6 +199,8 @@ serde_json = "1.0.140" serde_yaml = "0.9.34+deprecated" sha3 = "0.10.8" tauri = { version = "2.4.1", features = ["unstable", "devtools"] } # NB: "devtools" prevents App Store distribution +tempdir = "0.3.7" +tempfile = "3.20.0" thiserror = "2.0.12" tokio = { version = "1.45.1", features = ["macros"] } toml = "0.8.22" diff --git a/crates/desktop/artcraft/Cargo.toml b/crates/desktop/artcraft/Cargo.toml index 0b48f355a8..a65a686202 100644 --- a/crates/desktop/artcraft/Cargo.toml +++ b/crates/desktop/artcraft/Cargo.toml @@ -54,8 +54,8 @@ serde_json = { workspace = true } tauri = { workspace = true } tauri-plugin-log = "2.3.1" tauri-plugin-upload = "2.2.1" -tempdir = "0.3.7" -tempfile = "3.18.0" +tempdir = { workspace = true } +tempfile = { workspace = true } tokio = { version = "1.42", features = ["rt", "rt-multi-thread", "fs"] } tokio-util = { version = "0.7", features = ["codec"] } url = { workspace = true } diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/web/get_inference_job_by_fal_id.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/web/get_inference_job_by_fal_id.rs new file mode 100644 index 0000000000..f71b9dc9c2 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/web/get_inference_job_by_fal_id.rs @@ -0,0 +1,133 @@ +use anyhow::anyhow; +use chrono::{DateTime, NaiveDateTime, TimeZone, Utc}; +use log::warn; +use sqlx::pool::PoolConnection; +use sqlx::{MySql, MySqlPool}; + +use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; +use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; +use enums::by_table::generic_inference_jobs::inference_job_external_third_party::InferenceJobExternalThirdParty; +use enums::by_table::generic_inference_jobs::inference_job_product_category::InferenceJobProductCategory; +use enums::common::job_status_plus::JobStatusPlus; +use enums::no_table::style_transfer::style_transfer_name::StyleTransferName; +use errors::AnyhowResult; +use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use tokens::tokens::users::UserToken; + +use crate::helpers::boolean_converters::i8_to_bool; +use crate::payloads::generic_inference_args::generic_inference_args::{GenericInferenceArgs, PolymorphicInferenceArgs}; + +#[derive(Debug, Default)] +pub struct FalJobDetails { + pub job_token: InferenceJobToken, + + pub status: JobStatusPlus, + + pub external_third_party: InferenceJobExternalThirdParty, + pub external_third_party_id: String, + + pub maybe_creator_user_token: Option, + pub maybe_creator_anonymous_visitor_token: Option, + pub creator_ip_address: String, + + pub created_at: DateTime, + pub updated_at: DateTime, +} + +/// Returns Ok(None) when the record cannot be found. +pub async fn get_inference_job_by_fal_id(fal_id: &str, mysql_pool: &MySqlPool) + -> AnyhowResult> +{ + let mut connection = mysql_pool.acquire().await?; + get_inference_job_by_fal_id_from_connection(fal_id, &mut connection).await +} + + +/// Returns Ok(None) when the record cannot be found. +pub async fn get_inference_job_by_fal_id_from_connection(fal_id: &str, mysql_connection: &mut PoolConnection) + -> AnyhowResult> +{ + // NB(bt): jobs.uuid_idempotency_token is the current way to reconstruct the hash of the + // TTS result since we don't store a bucket hash on the table. This is an ugly hack :( + // TODO(bt,2023-10-12): ^^^ Is this comment still accurate? I don't see that field referenced below. + + let maybe_status = sqlx::query_as!( + RawJobRecord, + r#" +SELECT + jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`, + + jobs.status as `status: enums::common::job_status_plus::JobStatusPlus`, + + jobs.maybe_external_third_party as `external_third_party: enums::by_table::generic_inference_jobs::inference_job_external_third_party::InferenceJobExternalThirdParty`, + jobs.maybe_external_third_party_id as `external_third_party_id`, + + jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`, + jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`, + jobs.creator_ip_address, + + jobs.created_at, + jobs.updated_at + +FROM generic_inference_jobs as jobs + +WHERE jobs.maybe_external_third_party = ? +AND jobs.maybe_external_third_party_id = ? + "#, + InferenceJobExternalThirdParty::Fal.to_str(), + fal_id + ) + .fetch_one(&mut **mysql_connection) + .await; + + let record = match maybe_status { + Ok(record) => record, + Err(err) => match err { + sqlx::Error::RowNotFound => return Ok(None), + _ => { + warn!("error querying job record: {:?}", err); + return Err(anyhow!("error querying job record: {:?}", err)); + } + } + }; + + let record = raw_record_to_public_result(record)?; + + Ok(Some(record)) +} + +fn raw_record_to_public_result(record: RawJobRecord) -> AnyhowResult { + Ok(FalJobDetails { + job_token: record.job_token, + status: record.status, + external_third_party: record.external_third_party.ok_or_else(|| anyhow!("missing external_third_party"))?, + external_third_party_id: record.external_third_party_id.ok_or_else(|| anyhow!("missing external_third_party_id"))?, + maybe_creator_user_token: record.maybe_creator_user_token, + maybe_creator_anonymous_visitor_token: record.maybe_creator_anonymous_visitor_token, + creator_ip_address: record.creator_ip_address, + created_at: record.created_at, + updated_at: record.updated_at, + }) +} + +#[derive(Debug, Default)] +struct RawJobRecord { + pub job_token: InferenceJobToken, + + pub status: JobStatusPlus, + + // NB: Nullable, but required to be filled for this query + pub external_third_party: Option, + + // NB: Nullable, but required to be filled for this query + pub external_third_party_id: Option, + + pub maybe_creator_user_token: Option, + pub maybe_creator_anonymous_visitor_token: Option, + pub creator_ip_address: String, + + pub created_at: DateTime, + pub updated_at: DateTime, +} + diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/web/mod.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/web/mod.rs index 9ab33d089e..895bdce65d 100644 --- a/crates/schema/database/mysql_queries/src/queries/generic_inference/web/mod.rs +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/web/mod.rs @@ -9,4 +9,5 @@ pub mod kill_generic_inference_jobs; pub mod kill_jobs_in_development; pub mod list_session_jobs; pub mod mark_generic_inference_job_cancelled_by_user; -pub mod mark_generic_inference_job_successfully_done_by_token; \ No newline at end of file +pub mod mark_generic_inference_job_successfully_done_by_token; +pub mod get_inference_job_by_fal_id; \ No newline at end of file diff --git a/crates/schema/public/enums/src/by_table/generic_inference_jobs/inference_job_external_third_party.rs b/crates/schema/public/enums/src/by_table/generic_inference_jobs/inference_job_external_third_party.rs index 6036f181cf..8152544585 100644 --- a/crates/schema/public/enums/src/by_table/generic_inference_jobs/inference_job_external_third_party.rs +++ b/crates/schema/public/enums/src/by_table/generic_inference_jobs/inference_job_external_third_party.rs @@ -9,11 +9,12 @@ use strum::EnumIter; /// /// YOU CAN ADD NEW VALUES, BUT DO NOT CHANGE EXISTING VALUES WITHOUT A MIGRATION STRATEGY. #[cfg_attr(test, derive(EnumIter, EnumCount))] -#[derive(Clone, Copy, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize)] +#[derive(Clone, Copy, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize, Default)] #[serde(rename_all = "snake_case")] pub enum InferenceJobExternalThirdParty { /// Fal jobs #[serde(rename = "fal")] + #[default] Fal, } diff --git a/crates/schema/public/enums/src/by_table/media_files/media_file_type.rs b/crates/schema/public/enums/src/by_table/media_files/media_file_type.rs index 4f8ee74a3a..68df27a89d 100644 --- a/crates/schema/public/enums/src/by_table/media_files/media_file_type.rs +++ b/crates/schema/public/enums/src/by_table/media_files/media_file_type.rs @@ -103,6 +103,24 @@ impl MediaFileType { matches!(self, Self::Jpg | Self::Png | Self::Image) } + /// Returns the `MediaFileType` if the mime type matches one of the known types. + /// This is not exhaustive. + /// (NB: In general we shouldn't keep much application logic in the `enums` crate, + /// but this is a good case for it.) + pub fn try_from_mime_type(mime_type: &str) -> Option { + match mime_type { + "image/jpeg" => Some(Self::Jpg), + "image/png" => Some(Self::Png), + //"audio/wav" => Some(Self::Wav), + //"audio/mpeg" => Some(Self::Mp3), + //"image/gif" => Some(Self::Gif), + //"video/mp4" => Some(Self::Mp4), + //"model/gltf-binary" => Some(Self::Glb), + //"model/gltf+json" => Some(Self::Gltf), + _ => None, + } + } + pub fn to_str(&self) -> &'static str { match self { Self::Audio => "audio", diff --git a/crates/service/web/storyteller_web/Cargo.toml b/crates/service/web/storyteller_web/Cargo.toml index a9e315fb16..aa01448628 100644 --- a/crates/service/web/storyteller_web/Cargo.toml +++ b/crates/service/web/storyteller_web/Cargo.toml @@ -124,7 +124,9 @@ elasticsearch = "8.5.0-alpha.1" # Files infer = "0.16.0" -tempdir = "0.3.7" +bytes = { workspace = true } +tempdir = { workspace = true } +tempfile = { workspace = true } # The "never-encode-slash" feature seems to have been removed: # https://github.com/durch/rust-s3/pull/254 diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal_webhook_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal_webhook_handler.rs index 255b8b9fdc..c77aa37f0c 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal_webhook_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal_webhook_handler.rs @@ -11,10 +11,10 @@ use anyhow::Error; use errors::AnyhowResult; use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess; use http_server_common::response::serialize_as_json_error::serialize_as_json_error; -use log::info; +use log::{info, warn}; use serde_json::Value; use utoipa::ToSchema; - +use mysql_queries::queries::generic_inference::web::get_inference_job_by_fal_id::get_inference_job_by_fal_id; // 1. tauri --> hit endpoint to enqueue // // 2. webhook @@ -111,10 +111,35 @@ pub async fn fal_webhook_handler( ) -> Result, FalWebhookError> { info!("Received FAL webhook body: {:?}", request); + + let request_id = request.request_id + .as_deref() + .ok_or_else(|| FalWebhookError::BadInput("Missing request_id".to_string()))?; + + let payload = request.payload + .as_ref() + .ok_or_else(|| FalWebhookError::BadInput("Missing payload".to_string()))?; + + let db_result = get_inference_job_by_fal_id( + request_id, + &server_state.mysql_pool, + ).await; + + let job = match db_result { + Ok(Some(record)) => record, + Ok(None) => { + warn!("Could not find job record by fal request id: {:?} ; request payload = {:?}", request_id, request); + return Err(FalWebhookError::NotFound) + }, + Err(err) => { + warn!("Error querying job record: {:?}", err); + return Err(FalWebhookError::ServerError); + } + }; - if let Some(payload) = request.payload.as_ref() { - if let Some(payload_obj) = payload.as_object() { - handle_image_payload(payload_obj).await?; + if let Some(payload_obj) = payload.as_object() { + if payload_obj.contains_key("image") { + handle_image_payload(payload_obj, &job, &server_state).await?; } } diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/handle_image_payload.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/handle_image_payload.rs index 129c56f66b..86ba6d7df6 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/handle_image_payload.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/handle_image_payload.rs @@ -1,7 +1,25 @@ +use crate::http_server::endpoints::media_files::upload::upload_error::MediaFileUploadError; +use crate::state::server_state::ServerState; +use crate::util::http_download_url_to_bytes::http_download_url_to_bytes; +use crate::util::http_download_url_to_tempfile::http_download_url_to_tempfile; +use actix_web::web; use anyhow::anyhow; -use log::info; -use serde_json::{Map, Value}; +use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath; +use enums::by_table::media_files::media_file_class::MediaFileClass; +use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory; +use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory; +use enums::by_table::media_files::media_file_type::MediaFileType; use errors::AnyhowResult; +use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes; +use images::image_info::image_info::ImageInfo; +use log::{info, warn}; +use mimetypes::mimetype_info::mimetype_info::MimetypeInfo; +use mysql_queries::queries::generic_inference::web::get_inference_job_by_fal_id::FalJobDetails; +use mysql_queries::queries::media_files::create::insert_builder::media_file_insert_builder::MediaFileInsertBuilder; +use serde_json::{Map, Value}; +use std::sync::Arc; + +const PREFIX : Option<&str> = Some("image_"); #[derive(Deserialize, Debug)] pub struct FalWebhookImage { @@ -14,7 +32,9 @@ pub struct FalWebhookImage { } pub async fn handle_image_payload( - payload: &Map, + payload: &Map, + job: &FalJobDetails, + server_state: &ServerState, ) -> AnyhowResult<()> { let image_value = payload.get("image") @@ -24,5 +44,62 @@ pub async fn handle_image_payload( info!("Image payload received: {:?}", image); + let image_url = image.url + .as_deref() + .ok_or_else(|| anyhow!("no `url` in image payload"))?; + + //let mime_type = image.content_type + // .as_deref() + // .ok_or_else(|| anyhow!("no `content_type` in image payload"))?; + + let file_bytes = http_download_url_to_bytes(image_url) + .await + .map_err(|e| anyhow!("Failed to download image: {:?}", e))?; + + let mimetype_info = MimetypeInfo::get_for_bytes(&file_bytes) + .ok_or_else(|| anyhow!("Failed to get mimetype info"))?; + + let mime_type = mimetype_info.mime_type(); + + let media_file_type = MediaFileType::try_from_mime_type(mime_type) + .ok_or_else(|| anyhow!("Unsupported media file type: {}", mime_type))?; + + let extension_with_period = mimetype_info.file_extension() + .map(|ext| ext.extension_with_period()) + .ok_or_else(|| anyhow!("Failed to get file extension from mimetype info"))?; + + let file_size_bytes = file_bytes.len(); + let file_hash = sha256_hash_bytes(&file_bytes)?; + let image_info = ImageInfo::decode_image_from_bytes(&file_bytes)?; + + let public_upload_path = MediaFileBucketPath::generate_new(PREFIX, Some(&extension_with_period)); + + info!("Uploading media to bucket path: {}", public_upload_path.get_full_object_path_str()); + + server_state.public_bucket_client.upload_file_with_content_type_process( + public_upload_path.get_full_object_path_str(), + file_bytes.as_ref(), + &mime_type) + .await?; + + let media_token = MediaFileInsertBuilder::new() + .maybe_creator_user(job.maybe_creator_user_token.as_ref()) + .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref()) + .creator_ip_address(&job.creator_ip_address) + .public_bucket_directory_hash(&public_upload_path) + .media_file_class(MediaFileClass::Image) + .media_file_type(media_file_type) + .media_file_origin_category(MediaFileOriginCategory::Inference) + //.media_file_origin_product_category(MediaFileOriginProductCategory::Unknown) + .mime_type(mime_type) + .file_size_bytes(file_size_bytes as u64) + .frame_width(image_info.width()) + .frame_height(image_info.height()) + .checksum_sha2(&file_hash) + .insert_pool(&server_state.mysql_pool) + .await?; + + info!("Media file uploaded with token: {}", media_token); + Ok(()) } diff --git a/crates/service/web/storyteller_web/src/util/http_download_url_to_bytes.rs b/crates/service/web/storyteller_web/src/util/http_download_url_to_bytes.rs new file mode 100644 index 0000000000..dc057b5489 --- /dev/null +++ b/crates/service/web/storyteller_web/src/util/http_download_url_to_bytes.rs @@ -0,0 +1,24 @@ +use bytes::Bytes; +use errors::AnyhowResult; +use reqwest::{Client, IntoUrl}; + +const USER_AGENT: &str = "storyteller-web/1.0"; + +// TODO(bt, 2025-06-03): Don't load the entire file into memory! + +/// Downloads a (binary) file to memory. Good for images, etc. Not great for large files. +pub async fn http_download_url_to_bytes(url: U) -> AnyhowResult { + let client = Client::builder() + .gzip(true) + .build()?; + + let response = client.get(url) // NB: No IntoUrl for &Url. + .header("User-Agent", USER_AGENT) + .header("Accept", "*/*") + .send() + .await?; + + let bytes = response.bytes().await?; + + Ok(bytes) +} diff --git a/crates/service/web/storyteller_web/src/util/http_download_url_to_tempfile.rs b/crates/service/web/storyteller_web/src/util/http_download_url_to_tempfile.rs new file mode 100644 index 0000000000..3d3d470b57 --- /dev/null +++ b/crates/service/web/storyteller_web/src/util/http_download_url_to_tempfile.rs @@ -0,0 +1,14 @@ +use crate::util::http_download_url_to_bytes::http_download_url_to_bytes; +use errors::AnyhowResult; +use reqwest::IntoUrl; +use std::io::Write; +use tempfile::NamedTempFile; + +const USER_AGENT: &str = "storyteller-web/1.0"; + +/// Downloads a (binary) file to a filesystem path. Good for images, etc. Not great for large files. +pub async fn http_download_url_to_tempfile(url: U, temp_file: &mut NamedTempFile) -> AnyhowResult<()> { + let bytes = http_download_url_to_bytes(url).await?; + temp_file.write_all(&bytes)?; + Ok(()) +} diff --git a/crates/service/web/storyteller_web/src/util/mod.rs b/crates/service/web/storyteller_web/src/util/mod.rs index 3dd65224f6..ad33881fae 100644 --- a/crates/service/web/storyteller_web/src/util/mod.rs +++ b/crates/service/web/storyteller_web/src/util/mod.rs @@ -9,6 +9,8 @@ pub mod email_to_gravatar; pub mod encrypted_sort_id; pub mod enroll_in_studio; pub mod generate_random_username; +pub mod http_download_url_to_bytes; +pub mod http_download_url_to_tempfile; pub mod placeholder_images; pub mod read_toml_file_to_struct; pub mod title_to_url_slug;