handle fal image download and bucket upload

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",

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"

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 }

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<UserToken>,
+  pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
+  pub creator_ip_address: String,
+
+  pub created_at: DateTime<Utc>,
+  pub updated_at: DateTime<Utc>,
+}
+
+/// 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<Option<FalJobDetails>>
+{
+  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<MySql>)
+  -> AnyhowResult<Option<FalJobDetails>>
+{
+  // 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<FalJobDetails> {
+  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<InferenceJobExternalThirdParty>,
+  
+  // NB: Nullable, but required to be filled for this query
+  pub external_third_party_id: Option<String>,
+
+  pub maybe_creator_user_token: Option<UserToken>,
+  pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
+  pub creator_ip_address: String,
+
+  pub created_at: DateTime<Utc>,
+  pub updated_at: DateTime<Utc>,
+}
+

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;
+pub mod mark_generic_inference_job_successfully_done_by_token;
+pub mod get_inference_job_by_fal_id;

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,
 }
 

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<Self> {
+    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",

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

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<Json<SimpleGenericJsonSuccess>, 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?;
     }
   }
 

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<String, Value>, 
+  payload: &Map<String, Value>,
+  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(())
 }

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<U: IntoUrl>(url: U) -> AnyhowResult<Bytes> {
+  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)
+}

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<U: IntoUrl>(url: U, temp_file: &mut NamedTempFile) -> AnyhowResult<()> {
+  let bytes = http_download_url_to_bytes(url).await?;
+  temp_file.write_all(&bytes)?;
+  Ok(())
+}

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;