diff --git a/crates/api_clients/higgsfield_client/src/error/higgsfield_error.rs b/crates/api_clients/higgsfield_client/src/error/higgsfield_error.rs index 58b36f5f..345d1f64 100644 --- a/crates/api_clients/higgsfield_client/src/error/higgsfield_error.rs +++ b/crates/api_clients/higgsfield_client/src/error/higgsfield_error.rs @@ -69,6 +69,21 @@ impl HiggsfieldError { } } + /// The server refused a job because a referenced media input is unknown + /// or unusable (`404 "Media input not found"` and its 400/422 cousins). + /// Happens when a previously uploaded reference has since been deleted or + /// flagged: forget the cached upload and upload afresh. + pub fn is_media_input_rejected(&self) -> bool { + let text = match self { + Self::Api(HiggsfieldApiError::NotFound { raw_http_body }) => raw_http_body, + Self::Api(HiggsfieldApiError::BadRequest { reason, .. }) => reason, + Self::Api(HiggsfieldApiError::UnprocessableEntity { reason, .. }) => reason, + _ => return false, + }; + let text = text.to_ascii_lowercase(); + text.contains("media") && (text.contains("not found") || text.contains("input")) + } + /// Whether the session itself is the problem (expired token, dead session, /// bot protection, unusable cookies) — some form of re-authentication is /// needed before retrying. Union of [`Self::is_token_rejected`], @@ -121,6 +136,18 @@ mod tests { assert!(timed_out.user_facing_rejection().unwrap().contains("didn't finish in time")); } + #[test] + fn media_input_rejections_are_recognised_by_status_and_body() { + let not_found = HiggsfieldError::Api(HiggsfieldApiError::NotFound { raw_http_body: r#"{"detail":"Media input not found"}"#.to_string() }); + assert!(not_found.is_media_input_rejected()); + let unprocessable = HiggsfieldError::Api(HiggsfieldApiError::UnprocessableEntity { reason: "Invalid media input".to_string(), raw_http_body: String::new() }); + assert!(unprocessable.is_media_input_rejected()); + let job_missing = HiggsfieldError::Api(HiggsfieldApiError::NotFound { raw_http_body: r#"{"detail":"Job not found"}"#.to_string() }); + assert!(!job_missing.is_media_input_rejected(), "a missing job is not a media problem"); + let dead = HiggsfieldError::Api(HiggsfieldApiError::NoActiveSession { raw_http_body: String::new() }); + assert!(!dead.is_media_input_rejected()); + } + #[test] fn other_errors_are_not_user_facing_rejections() { let dead = HiggsfieldError::Api(HiggsfieldApiError::NoActiveSession { raw_http_body: String::new() }); diff --git a/crates/artcraftx/src/commands/generate/common/higgsfield_generation.rs b/crates/artcraftx/src/commands/generate/common/higgsfield_generation.rs index 179c6b04..c317a600 100644 --- a/crates/artcraftx/src/commands/generate/common/higgsfield_generation.rs +++ b/crates/artcraftx/src/commands/generate/common/higgsfield_generation.rs @@ -8,7 +8,6 @@ use std::collections::HashMap; use artcraft_client::utils::api_host::ApiHost; use log::{error, info, warn}; -use router::api::asset_upload_cache::AssetUploadCache; use router::client::router_client::RouterClient; use router::client::router_higgsfield_client::RouterHiggsfieldClient; use router::errors::artcraft_router_error::ArtcraftRouterError; @@ -16,16 +15,23 @@ use router::errors::provider_error::ProviderError; use router::generate::generate_image::generate_image_request_builder::GenerateImageRequestBuilder; use router::generate::generate_image::generate_image_response::GenerateImageResponse; use router::generate::generate_image::image_generation_draft_context::ImageGenerationDraftContext; +use router::generate::generate_image::image_generation_draft::ImageGenerationDraftRequest; use router::generate::generate_image::image_generation_draft_or_request::ImageGenerationDraftOrRequest; +use router::generate::generate_image::image_generation_request::ImageGenerationRequest; use router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder; use router::generate::generate_video::generate_video_response::GenerateVideoResponse; use router::generate::generate_video::video_generation_draft_context::VideoGenerationDraftContext; +use router::generate::generate_video::video_generation_draft::VideoGenerationDraftRequest; use router::generate::generate_video::video_generation_draft_or_request::VideoGenerationDraftOrRequest; +use router::generate::generate_video::video_generation_request::VideoGenerationRequest; use sqlite_identifiers::ids::media_file_token::MediaFileToken; +use tauri::AppHandle; use crate::commands::generate::generate_error::{CredentialProblemReason, GenerateError}; use crate::commands::generate::generate_image::utils::map_media_files_to_urls::map_media_file_tokens_to_cdn_urls; use crate::credentials::auth_credential::AuthCredential; +use crate::services::asset_uploads::asset_upload_ledger::AccountUploadCache; +use crate::services::higgsfield::higgsfield_upload_notices::HiggsfieldUploadNotices; use crate::services::higgsfield::higgsfield_session_from_credential::higgsfield_session_from_credential; /// Batch job ids are stored on the task as one comma-separated @@ -94,73 +100,140 @@ pub async fn higgsfield_media_url_map(tokens: &[MediaFileToken]) -> Result, credential: &AuthCredential, builder: GenerateImageRequestBuilder, client: &RouterClient, media_url_map: &HashMap, - upload_cache: &dyn AssetUploadCache, + upload_cache: &AccountUploadCache, ) -> Result { - let request = match builder.build2().map_err(|err| { + let draft = match builder.build2().map_err(|err| { warn!("Could not build Higgsfield image request: {:?}", err); GenerateError::from(err) })? { - ImageGenerationDraftOrRequest::Request(request) => request, - ImageGenerationDraftOrRequest::Draft(draft) => { - info!("Higgsfield image request has references; uploading them first"); - let context = ImageGenerationDraftContext { - client: Some(client), - media_file_to_artcraft_url_map: Some(media_url_map), - asset_upload_cache: Some(upload_cache), - }; - draft.finalize(context).await.map_err(|err| { - warn!("Could not upload references to Higgsfield: {:?}", err); - higgsfield_error_to_generate_error(credential, err) - })? + ImageGenerationDraftOrRequest::Request(request) => { + return request.send_request(client).await + .map_err(|err| higgsfield_error_to_generate_error(credential, err)); } + ImageGenerationDraftOrRequest::Draft(draft) => draft, }; - request.send_request(client).await.map_err(|err| { - warn!("Higgsfield image generation failed: {:?}", err); + info!("Higgsfield image request has references; uploading them first"); + // Reuse flashes a notice; a real upload keeps a progress toast up until + // `notices` drops at the end of this function (retry included). + let notices = HiggsfieldUploadNotices::new(maybe_app); + let request = finalize_image_draft(credential, draft.clone(), client, media_url_map, upload_cache, ¬ices).await?; + match request.send_request(client).await { + Ok(response) => Ok(response), + Err(err) if should_retry_with_fresh_uploads(&err, upload_cache) => { + let forgotten = upload_cache.forget_reused().await; + warn!("Higgsfield rejected reused reference media ({:?}); forgot {} ledger entries and uploading afresh", err, forgotten); + let request = finalize_image_draft(credential, draft, client, media_url_map, upload_cache, ¬ices).await?; + request.send_request(client).await + .map_err(|err| higgsfield_error_to_generate_error(credential, err)) + } + Err(err) => Err(higgsfield_error_to_generate_error(credential, err)), + } +} + +async fn finalize_image_draft( + credential: &AuthCredential, + draft: ImageGenerationDraftRequest, + client: &RouterClient, + media_url_map: &HashMap, + upload_cache: &AccountUploadCache, + notices: &HiggsfieldUploadNotices, +) -> Result { + let context = ImageGenerationDraftContext { + client: Some(client), + media_file_to_artcraft_url_map: Some(media_url_map), + asset_upload_cache: Some(upload_cache), + asset_upload_observer: Some(notices), + }; + draft.finalize(context).await.map_err(|err| { + warn!("Could not upload references to Higgsfield: {:?}", err); higgsfield_error_to_generate_error(credential, err) }) } /// Build, finalize (uploading keyframes and references) and send a video -/// request. +/// request. Reused uploads that Higgsfield rejects at enqueue time are +/// forgotten and uploaded afresh, once — see [`send_higgsfield_image_request`]. pub async fn send_higgsfield_video_request( + maybe_app: Option<&AppHandle>, credential: &AuthCredential, builder: GenerateVideoRequestBuilder, client: &RouterClient, media_url_map: &HashMap, - upload_cache: &dyn AssetUploadCache, + upload_cache: &AccountUploadCache, ) -> Result { - let request = match builder.build2().map_err(|err| { + let draft = match builder.build2().map_err(|err| { warn!("Could not build Higgsfield video request: {:?}", err); GenerateError::from(err) })? { - VideoGenerationDraftOrRequest::Request(request) => request, - VideoGenerationDraftOrRequest::Draft(draft) => { - info!("Higgsfield video request has media; uploading it first"); - let context = VideoGenerationDraftContext { - client: Some(client), - media_file_to_artcraft_url_map: Some(media_url_map), - character_token_to_kinovi_id_map: None, - asset_upload_cache: Some(upload_cache), - }; - draft.finalize(context).await.map_err(|err| { - warn!("Could not upload media to Higgsfield: {:?}", err); - higgsfield_error_to_generate_error(credential, err) - })? + VideoGenerationDraftOrRequest::Request(request) => { + return request.send_request(client).await + .map_err(|err| higgsfield_error_to_generate_error(credential, err)); } + VideoGenerationDraftOrRequest::Draft(draft) => draft, }; - request.send_request(client).await.map_err(|err| { - warn!("Higgsfield video generation failed: {:?}", err); + info!("Higgsfield video request has media; uploading it first"); + // Reuse flashes a notice; a real upload keeps a progress toast up until + // `notices` drops at the end of this function (retry included). + let notices = HiggsfieldUploadNotices::new(maybe_app); + let request = finalize_video_draft(credential, draft.clone(), client, media_url_map, upload_cache, ¬ices).await?; + match request.send_request(client).await { + Ok(response) => Ok(response), + Err(err) if should_retry_with_fresh_uploads(&err, upload_cache) => { + let forgotten = upload_cache.forget_reused().await; + warn!("Higgsfield rejected reused reference media ({:?}); forgot {} ledger entries and uploading afresh", err, forgotten); + let request = finalize_video_draft(credential, draft, client, media_url_map, upload_cache, ¬ices).await?; + request.send_request(client).await + .map_err(|err| higgsfield_error_to_generate_error(credential, err)) + } + Err(err) => Err(higgsfield_error_to_generate_error(credential, err)), + } +} + +async fn finalize_video_draft( + credential: &AuthCredential, + draft: VideoGenerationDraftRequest, + client: &RouterClient, + media_url_map: &HashMap, + upload_cache: &AccountUploadCache, + notices: &HiggsfieldUploadNotices, +) -> Result { + let context = VideoGenerationDraftContext { + client: Some(client), + media_file_to_artcraft_url_map: Some(media_url_map), + character_token_to_kinovi_id_map: None, + asset_upload_cache: Some(upload_cache), + asset_upload_observer: Some(notices), + }; + draft.finalize(context).await.map_err(|err| { + warn!("Could not upload media to Higgsfield: {:?}", err); higgsfield_error_to_generate_error(credential, err) }) } +/// Retry with fresh uploads only when Higgsfield rejected a media input and +/// this request actually reused something from the ledger; otherwise the +/// media was uploaded just now and re-uploading would change nothing. +fn should_retry_with_fresh_uploads(err: &ArtcraftRouterError, upload_cache: &AccountUploadCache) -> bool { + let media_rejected = matches!( + err, + ArtcraftRouterError::Provider(ProviderError::Higgsfield(higgsfield_err)) if higgsfield_err.is_media_input_rejected() + ); + media_rejected && upload_cache.reused_count() > 0 +} + /// One `provider_job_id` for a Higgsfield job set. pub fn join_higgsfield_job_ids(job_ids: &[String]) -> String { job_ids.join(&HIGGSFIELD_JOB_ID_SEPARATOR.to_string()) diff --git a/crates/artcraftx/src/commands/generate/generate_image/enqueue_image_generation.rs b/crates/artcraftx/src/commands/generate/generate_image/enqueue_image_generation.rs index b018c30a..4edd361e 100644 --- a/crates/artcraftx/src/commands/generate/generate_image/enqueue_image_generation.rs +++ b/crates/artcraftx/src/commands/generate/generate_image/enqueue_image_generation.rs @@ -101,7 +101,7 @@ pub async fn enqueue_image_generation( } GenerationSource::HiggsfieldCookies | GenerationSource::Higgsfield => { - enqueue_via_higgsfield(request, &credential, ledger).await + enqueue_via_higgsfield(Some(app), request, &credential, ledger).await } other => Err(credential_not_usable( diff --git a/crates/artcraftx/src/commands/generate/generate_image/higgsfield/enqueue_via_higgsfield.rs b/crates/artcraftx/src/commands/generate/generate_image/higgsfield/enqueue_via_higgsfield.rs index 258e2957..81249450 100644 --- a/crates/artcraftx/src/commands/generate/generate_image/higgsfield/enqueue_via_higgsfield.rs +++ b/crates/artcraftx/src/commands/generate/generate_image/higgsfield/enqueue_via_higgsfield.rs @@ -23,6 +23,7 @@ use crate::commands::generate::task_enqueue_success::TaskEnqueueSuccess; use crate::commands::utils::api_adapters::models::image::tauri_image_model_to_generation_model::tauri_image_model_to_generation_model; use crate::commands::utils::api_adapters::models::image::tauri_image_model_to_router_model::tauri_image_model_to_router_model; use crate::credentials::auth_credential::AuthCredential; +use tauri::AppHandle; use crate::services::asset_uploads::asset_upload_ledger::AssetUploadLedger; /// Enqueue via the router's first-party Higgsfield provider. @@ -34,6 +35,7 @@ use crate::services::asset_uploads::asset_upload_ledger::AssetUploadLedger; /// re-uploads them. The returned task carries every job id of the /// Higgsfield job set so the poller can follow a batch. pub async fn enqueue_via_higgsfield( + maybe_app: Option<&AppHandle>, request: &TauriGenerateImageRequest, credential: &AuthCredential, ledger: &AssetUploadLedger, @@ -73,7 +75,7 @@ pub async fn enqueue_via_higgsfield( let client = higgsfield_router_client(credential)?; // References this account already uploaded are reused by media id. let upload_cache = ledger.for_higgsfield_credential(&credential.id); - let response = send_higgsfield_image_request(credential, router_request, &client, &media_url_map, &upload_cache).await?; + let response = send_higgsfield_image_request(maybe_app, credential, router_request, &client, &media_url_map, &upload_cache).await?; let payload = response .get_higgsfield_payload() @@ -162,7 +164,7 @@ mod live_higgsfield_image_tests { ..Default::default() }; - let success = enqueue_via_higgsfield(&request, &credential, &ledger().await).await.expect("enqueue should succeed"); + let success = enqueue_via_higgsfield(None, &request, &credential, &ledger().await).await.expect("enqueue should succeed"); println!("[live] Higgsfield image enqueued: provider_job_id={:?}", success.provider_job_id); assert_eq!(success.provider, GenerationSource::Higgsfield); assert!(success.provider_job_id.is_some()); @@ -181,7 +183,7 @@ mod live_higgsfield_image_tests { ..Default::default() }; - let success = enqueue_via_higgsfield(&request, &credential, &ledger().await).await.expect("enqueue should succeed"); + let success = enqueue_via_higgsfield(None, &request, &credential, &ledger().await).await.expect("enqueue should succeed"); println!("[live] Higgsfield image edit enqueued: provider_job_id={:?}", success.provider_job_id); assert!(success.provider_job_id.is_some()); } diff --git a/crates/artcraftx/src/commands/generate/generate_video/handle_credential_router.rs b/crates/artcraftx/src/commands/generate/generate_video/handle_credential_router.rs index 0b20074b..8883ab28 100644 --- a/crates/artcraftx/src/commands/generate/generate_video/handle_credential_router.rs +++ b/crates/artcraftx/src/commands/generate/generate_video/handle_credential_router.rs @@ -11,7 +11,6 @@ use crate::commands::generate::generate_video::request::TauriGenerateVideoReques use crate::utils::services::artcraft_api_host::maybe_artcraft_api_host_for_service; use crate::credentials::auth_credential::AuthCredential; use core_types::enums::generation_source::GenerationSource; -use crate::events::notice_events::progress_notice_event::ProgressNotice; use crate::services::asset_uploads::asset_upload_ledger::AssetUploadLedger; use crate::state::data_dir::app_data_root::AppDataRoot; use tauri::AppHandle; @@ -45,15 +44,10 @@ pub async fn handle_credential_router( } GenerationSource::HiggsfieldCookies | GenerationSource::Higgsfield => { - // Reference media triggers Higgsfield's mandatory IP/likeness scan, - // which routinely takes 15-30s before the job can even enqueue. Keep - // a progress toast up for as long as the router call runs; the guard - // takes it down on success and failure alike. - let has_media = request.media_sources().iter().next().is_some(); - let _ip_check_notice = maybe_app.filter(|_| has_media).map(|app| { - ProgressNotice::start(app, "Higgsfield is checking your media for Intellectual Property and Likeness") - }); - handle_higgsfield_video_via_router(request, &credential, ledger).await + // Reference media is uploaded (with Higgsfield's 15-30s IP/likeness + // scan) or reused from the ledger; the notices for either come from + // the router's upload observer, so nothing is announced up front. + handle_higgsfield_video_via_router(maybe_app, request, &credential, ledger).await } other => Err(credential_not_usable( &credential, diff --git a/crates/artcraftx/src/commands/generate/generate_video/higgsfield/handle_higgsfield_video_via_router.rs b/crates/artcraftx/src/commands/generate/generate_video/higgsfield/handle_higgsfield_video_via_router.rs index 7bf51bc9..ccf84e96 100644 --- a/crates/artcraftx/src/commands/generate/generate_video/higgsfield/handle_higgsfield_video_via_router.rs +++ b/crates/artcraftx/src/commands/generate/generate_video/higgsfield/handle_higgsfield_video_via_router.rs @@ -7,6 +7,7 @@ use router::api::router_provider::RouterProvider; use router::client::request_mismatch_mitigation_strategy::RequestMismatchMitigationStrategy; use router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder; use sqlite_identifiers::enums::task_type::TaskType; +use tauri::AppHandle; use crate::commands::generate::common::higgsfield_generation::{ higgsfield_media_url_map, higgsfield_router_client, join_higgsfield_job_ids, send_higgsfield_video_request, @@ -32,6 +33,7 @@ use crate::services::asset_uploads::asset_upload_ledger::AssetUploadLedger; /// task carries every job id of the Higgsfield job set so the poller can /// follow a batch. pub async fn handle_higgsfield_video_via_router( + maybe_app: Option<&AppHandle>, request: &TauriGenerateVideoRequest, credential: &AuthCredential, ledger: &AssetUploadLedger, @@ -69,7 +71,7 @@ pub async fn handle_higgsfield_video_via_router( let client = higgsfield_router_client(credential)?; // References this account already uploaded are reused by media id. let upload_cache = ledger.for_higgsfield_credential(&credential.id); - let response = send_higgsfield_video_request(credential, router_request, &client, &media_url_map, &upload_cache).await?; + let response = send_higgsfield_video_request(maybe_app, credential, router_request, &client, &media_url_map, &upload_cache).await?; let payload = response .get_higgsfield_payload() @@ -142,7 +144,7 @@ mod live_higgsfield_video_tests { ..Default::default() }; - let success = handle_higgsfield_video_via_router(&request, &credential, &test_ledger().await).await.expect("enqueue should succeed"); + let success = handle_higgsfield_video_via_router(None, &request, &credential, &test_ledger().await).await.expect("enqueue should succeed"); println!("[live] Higgsfield video enqueued: provider_job_id={:?}", success.provider_job_id); assert_eq!(success.provider, GenerationSource::Higgsfield); assert!(success.provider_job_id.is_some()); @@ -162,7 +164,7 @@ mod live_higgsfield_video_tests { ..Default::default() }; - let success = handle_higgsfield_video_via_router(&request, &credential, &test_ledger().await).await.expect("enqueue should succeed"); + let success = handle_higgsfield_video_via_router(None, &request, &credential, &test_ledger().await).await.expect("enqueue should succeed"); println!("[live] Higgsfield image-to-video enqueued: provider_job_id={:?}", success.provider_job_id); assert!(success.provider_job_id.is_some()); } diff --git a/crates/artcraftx/src/services/asset_uploads/asset_upload_ledger.rs b/crates/artcraftx/src/services/asset_uploads/asset_upload_ledger.rs index 8b1a700e..d398fbda 100644 --- a/crates/artcraftx/src/services/asset_uploads/asset_upload_ledger.rs +++ b/crates/artcraftx/src/services/asset_uploads/asset_upload_ledger.rs @@ -20,6 +20,7 @@ use std::future::Future; use std::path::Path; +use std::sync::{Arc, Mutex}; use artcraft_client::credentials::storyteller_credential_set::StorytellerCredentialSet; use artcraft_client::endpoints::media_files::get_media_file::get_media_file; @@ -34,7 +35,7 @@ use local_files_database::queries::get_asset_upload::{get_asset_upload, GetAsset use local_files_database::queries::touch_asset_upload_reused::touch_asset_upload_reused; use local_files_database::queries::upsert_asset_upload::{upsert_asset_upload, UpsertAssetUploadArgs}; use log::{info, warn}; -use router::api::asset_upload_cache::{AssetUploadCache, CachedAssetUpload}; +use router::api::asset_upload_cache::{AssetUploadCache, AssetUploadOrigin, CachedAssetUpload}; use sqlite_identifiers::ids::media_file_token::MediaFileToken; use crate::services::local_files::file_hash_service::hash_local_file_memoized; @@ -56,6 +57,10 @@ pub struct AccountUploadCache { ledger: AssetUploadLedger, service: GenerationSource, account_id: String, + /// Hashes whose upload this request skipped thanks to a ledger hit. If the + /// provider then rejects the request over its media, these are the entries + /// to drop before uploading afresh (see `forget_reused`). + reused_this_request: Arc>>, } impl AssetUploadLedger { @@ -65,7 +70,12 @@ impl AssetUploadLedger { /// This account's uploads on `service`. pub fn for_account(&self, service: GenerationSource, account_id: impl Into) -> AccountUploadCache { - AccountUploadCache { ledger: self.clone(), service, account_id: account_id.into() } + AccountUploadCache { + ledger: self.clone(), + service, + account_id: account_id.into(), + reused_this_request: Arc::new(Mutex::new(Vec::new())), + } } /// A stored Higgsfield credential's uploads. @@ -110,6 +120,7 @@ impl AssetUploadLedger { account_id, file_hash_blake3: file_hash, service_id: &upload.service_id, + origin: upload.origin.as_str(), maybe_service_url: upload.maybe_service_url.as_deref(), maybe_file_size_bytes: file_size_bytes.map(|size| size as i64), }).await; @@ -137,6 +148,24 @@ impl AccountUploadCache { self.service } + /// How many uploads this request has skipped so far via ledger hits. + pub fn reused_count(&self) -> usize { + self.reused_this_request.lock().map(|reused| reused.len()).unwrap_or(0) + } + + /// Drop every ledger entry this request reused, so the next finalize + /// uploads those files afresh. Returns how many were forgotten. + pub async fn forget_reused(&self) -> usize { + let hashes: Vec = match self.reused_this_request.lock() { + Ok(mut reused) => reused.drain(..).collect(), + Err(_) => Vec::new(), + }; + for hash in &hashes { + self.ledger.forget(self.service, &self.account_id, hash).await; + } + hashes.len() + } + pub fn account_id(&self) -> &str { &self.account_id } @@ -176,10 +205,7 @@ impl AccountUploadCache { if let Some(hash) = maybe_hash.as_deref() { let file_size_bytes = std::fs::metadata(path).ok().map(|metadata| metadata.len()); - self.record_upload(hash, file_size_bytes.unwrap_or(0), &CachedAssetUpload { - service_id: token.as_str().to_string(), - maybe_service_url: None, - }).await; + self.record_upload(hash, file_size_bytes.unwrap_or(0), &CachedAssetUpload::uploaded(token.as_str().to_string(), None)).await; } Ok(token) } @@ -197,10 +223,7 @@ impl AccountUploadCache { return Ok(token); } let token = upload().await?; - self.record_upload(&hash, bytes.len() as u64, &CachedAssetUpload { - service_id: token.as_str().to_string(), - maybe_service_url: None, - }).await; + self.record_upload(&hash, bytes.len() as u64, &CachedAssetUpload::uploaded(token.as_str().to_string(), None)).await; Ok(token) } @@ -227,6 +250,7 @@ impl AssetUploadCache for AccountUploadCache { .map(|record| CachedAssetUpload { service_id: record.service_id, maybe_service_url: record.service_url, + origin: AssetUploadOrigin::parse(&record.origin), }) } @@ -235,6 +259,9 @@ impl AssetUploadCache for AccountUploadCache { } async fn note_reused(&self, file_hash_blake3: &str) { + if let Ok(mut reused) = self.reused_this_request.lock() { + reused.push(file_hash_blake3.to_string()); + } self.ledger.note_reused(self.service, &self.account_id, file_hash_blake3).await } @@ -282,7 +309,7 @@ mod tests { let other = ledger.for_higgsfield_credential(&CredentialId::from_trusted("credential_other")); assert!(cache.find_upload("hash1").await.is_none()); - let upload = CachedAssetUpload { service_id: "media-1".into(), maybe_service_url: Some("https://cdn/x".into()) }; + let upload = CachedAssetUpload::uploaded("media-1", Some("https://cdn/x".into())); cache.record_upload("hash1", 10, &upload).await; assert_eq!(cache.find_upload("hash1").await, Some(upload)); assert!(other.find_upload("hash1").await.is_none(), "accounts don't share uploads"); @@ -294,6 +321,41 @@ mod tests { assert!(cache.find_upload("hash1").await.is_none()); } + #[tokio::test] + async fn generation_results_keep_their_origin() { + let dir = tempfile::tempdir().unwrap(); + let root = AppDataRoot::create_existing(dir.path().join("root")).unwrap(); + let ledger = AssetUploadLedger::new(LocalFilesDatabase::connect(&root).await.unwrap()); + let cache = ledger.for_higgsfield_credential(&CredentialId::from_trusted("credential_test")); + let result = CachedAssetUpload::generation_result("job-1", "https://cdn/result.png"); + cache.record_upload("hash-result", 10, &result).await; + assert_eq!(cache.find_upload("hash-result").await, Some(result)); + let plain = CachedAssetUpload::uploaded("media-1", Some("https://cdn/x".into())); + cache.record_upload("hash-upload", 10, &plain).await; + assert_eq!(cache.find_upload("hash-upload").await.unwrap().origin, AssetUploadOrigin::Upload); + } + + #[tokio::test] + async fn forget_reused_drops_only_what_this_request_reused() { + let dir = tempfile::tempdir().unwrap(); + let root = AppDataRoot::create_existing(dir.path().join("root")).unwrap(); + let ledger = AssetUploadLedger::new(LocalFilesDatabase::connect(&root).await.unwrap()); + let cache = ledger.for_higgsfield_credential(&CredentialId::from_trusted("credential_test")); + let upload = |id: &str| CachedAssetUpload::uploaded(id, Some("https://cdn/x".into())); + cache.record_upload("reused", 10, &upload("media-reused")).await; + cache.record_upload("fresh", 10, &upload("media-fresh")).await; + assert_eq!(cache.reused_count(), 0); + + cache.note_reused("reused").await; + assert_eq!(cache.reused_count(), 1); + + assert_eq!(cache.forget_reused().await, 1); + assert_eq!(cache.reused_count(), 0, "cleared for the retry"); + assert!(cache.find_upload("reused").await.is_none(), "the reused entry is gone"); + assert!(cache.find_upload("fresh").await.is_some(), "an entry that wasn't reused survives"); + assert_eq!(cache.forget_reused().await, 0, "nothing left to forget"); + } + #[tokio::test] async fn artcraft_file_helper_uploads_on_miss_and_records() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/artcraftx/src/services/higgsfield/higgsfield_upload_notices.rs b/crates/artcraftx/src/services/higgsfield/higgsfield_upload_notices.rs new file mode 100644 index 00000000..cfa1252c --- /dev/null +++ b/crates/artcraftx/src/services/higgsfield/higgsfield_upload_notices.rs @@ -0,0 +1,59 @@ +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Mutex; + +use router::api::asset_upload_observer::AssetUploadObserver; +use tauri::AppHandle; + +use crate::events::basic_sendable_event_trait::BasicSendableEvent; +use crate::events::notice_events::flash_notice_event::FlashNoticeEvent; +use crate::events::notice_events::progress_notice_event::ProgressNotice; + +const IP_CHECK_MESSAGE: &str = "Higgsfield is checking your media for Intellectual Property and Likeness"; +const UPLOADING_MESSAGE: &str = "Uploading your media to Higgsfield"; +const REUSED_MESSAGE: &str = "Reusing media Higgsfield already has (no upload or new IP check needed)"; + +/// Turns the router's per-file upload decisions into frontend notices for +/// one Higgsfield request: +/// +/// - the first real upload puts up a progress toast (the IP/likeness check +/// routinely takes 15-30s), which stays up until this value is dropped — +/// so hold it across the enqueue, including retries; +/// - a reuse flashes a plain, progress-free notice once, since nothing is +/// being waited on. +/// +/// Without an app handle (tests, headless callers) it observes silently. +pub struct HiggsfieldUploadNotices { + maybe_app: Option, + progress: Mutex>, + reuse_announced: AtomicBool, +} + +impl HiggsfieldUploadNotices { + pub fn new(maybe_app: Option<&AppHandle>) -> Self { + Self { + maybe_app: maybe_app.cloned(), + progress: Mutex::new(None), + reuse_announced: AtomicBool::new(false), + } + } +} + +impl AssetUploadObserver for HiggsfieldUploadNotices { + fn on_reused(&self, _kind: &str, _description: &str) { + let Some(app) = &self.maybe_app else { return }; + if self.reuse_announced.swap(true, Ordering::SeqCst) { + return; + } + FlashNoticeEvent::new(REUSED_MESSAGE).send_infallible(app); + } + + fn on_uploading(&self, _kind: &str, _byte_count: u64, ip_check: bool) { + let Some(app) = &self.maybe_app else { return }; + let Ok(mut progress) = self.progress.lock() else { return }; + if progress.is_some() { + return; // One toast covers every upload of the request. + } + let message = if ip_check { IP_CHECK_MESSAGE } else { UPLOADING_MESSAGE }; + *progress = Some(ProgressNotice::start(app, message)); + } +} diff --git a/crates/artcraftx/src/services/higgsfield/mod.rs b/crates/artcraftx/src/services/higgsfield/mod.rs index f5fb8fad..66c45461 100644 --- a/crates/artcraftx/src/services/higgsfield/mod.rs +++ b/crates/artcraftx/src/services/higgsfield/mod.rs @@ -1,3 +1,4 @@ pub mod higgsfield_session_from_credential; pub mod mark_higgsfield_relogin_required; +pub mod higgsfield_upload_notices; diff --git a/crates/artcraftx/src/threads/third_party_task_polling_thread/handlers/higgsfield/handle_higgsfield_complete.rs b/crates/artcraftx/src/threads/third_party_task_polling_thread/handlers/higgsfield/handle_higgsfield_complete.rs index 474f455e..ab272fc9 100644 --- a/crates/artcraftx/src/threads/third_party_task_polling_thread/handlers/higgsfield/handle_higgsfield_complete.rs +++ b/crates/artcraftx/src/threads/third_party_task_polling_thread/handlers/higgsfield/handle_higgsfield_complete.rs @@ -2,7 +2,7 @@ use artcraft_client::enums::common::generation::common_model_type::CommonModelTy use core_types::enums::generation_source::GenerationSource; use errors::AnyhowResult; use higgsfield_client::endpoints::jobs::job_status::JobStatusResponse; -use log::{error, info}; +use log::{error, info, warn}; use sqlite_database::queries::task::Task; use sqlite_identifiers::enums::task_media_file_class::TaskMediaFileClass; use sqlite_identifiers::enums::task_model_type::TaskModelType; @@ -15,7 +15,14 @@ use crate::state::data_dir::app_data_root::AppDataRoot; use crate::state::database::task_database::TaskDatabase; use crate::threads::task_completion::complete_task_with_local_files::{complete_task_with_local_files, CompleteTaskArgs}; use crate::threads::task_completion::upload_results_to_artcraft::CompletionPrompt; +use crate::services::asset_uploads::asset_upload_ledger::{AccountUploadCache, AssetUploadLedger}; +use crate::state::database::local_files_database::LocalFilesDatabase; use crate::utils::download::download_url_to_temp_dir::download_url_to_temp_dir; +use core_types::identifiers::credential_id::CredentialId; +use file_hashing::hash_file_blake3; +use router::api::asset_upload_cache::{AssetUploadCache, CachedAssetUpload}; +use std::path::Path; +use tauri::Manager; /// Download filename slug when the task has no model type recorded. const HIGGSFIELD_FALLBACK_MODEL_SLUG: &str = "higgsfield"; @@ -28,6 +35,7 @@ pub async fn handle_higgsfield_complete( app_preferences: &AppPreferencesManager, task_database: &TaskDatabase, storyteller_creds_manager: &StorytellerCredentialManager, + credential_id: &str, task: &Task, jobs: &[JobStatusResponse], ) { @@ -39,6 +47,7 @@ pub async fn handle_higgsfield_complete( app_preferences, task_database, storyteller_creds_manager, + credential_id, task, jobs, ).await; @@ -54,16 +63,28 @@ async fn handle_higgsfield_complete_inner( app_preferences: &AppPreferencesManager, task_database: &TaskDatabase, storyteller_creds_manager: &StorytellerCredentialManager, + credential_id: &str, task: &Task, jobs: &[JobStatusResponse], ) -> AnyhowResult<()> { + // Higgsfield already holds these files: remember each one under the + // account that generated it, so using the download as a reference later + // passes the job id instead of uploading (and IP-checking) it again. + let maybe_ledger = app_handle.try_state::() + .map(|database| AssetUploadLedger::new(database.inner().clone())) + .map(|ledger| ledger.for_higgsfield_credential(&CredentialId::from_trusted(credential_id))); + // NB: `NamedTempFile`s delete themselves on drop, so keep them alive until // the completion routine has copied and uploaded them. let mut temp_files = Vec::with_capacity(jobs.len()); for (index, job) in jobs.iter().enumerate() { let url = job.result_url().expect("caller passes only jobs with a result URL"); info!("[HiggsfieldComplete] Downloading result {} of {} for task {} ...", index + 1, jobs.len(), task.id.as_str()); - temp_files.push(download_url_to_temp_dir(url, app_data_root).await?); + let temp_file = download_url_to_temp_dir(url, app_data_root).await?; + if let Some(ledger) = &maybe_ledger { + record_generation_result(ledger, job, temp_file.path()).await; + } + temp_files.push(temp_file); } let local_files = temp_files.iter() .map(|file| file.path().to_path_buf()) @@ -102,6 +123,21 @@ async fn handle_higgsfield_complete_inner( Ok(()) } +/// Ledger a downloaded result as content Higgsfield generated (job id + +/// result URL). Best-effort: a hashing or database problem only costs a +/// future re-upload. +async fn record_generation_result(ledger: &AccountUploadCache, job: &JobStatusResponse, path: &Path) { + let Some(url) = job.result_url() else { return }; + let (hash, size) = match (hash_file_blake3(path), std::fs::metadata(path)) { + (Ok(hash), Ok(metadata)) => (hash, metadata.len()), + (Err(err), _) | (_, Err(err)) => { + warn!("[HiggsfieldComplete] Could not hash downloaded result of job {}: {}", job.id, err); + return; + } + }; + ledger.record_upload(&hash, size, &CachedAssetUpload::generation_result(job.id.as_str(), url)).await; +} + fn media_class_for(task_type: TaskType) -> TaskMediaFileClass { match task_type { TaskType::VideoGeneration => TaskMediaFileClass::Video, diff --git a/crates/artcraftx/src/threads/third_party_task_polling_thread/handlers/higgsfield/poll_higgsfield_tasks.rs b/crates/artcraftx/src/threads/third_party_task_polling_thread/handlers/higgsfield/poll_higgsfield_tasks.rs index 9ad2fe0b..0afd8d25 100644 --- a/crates/artcraftx/src/threads/third_party_task_polling_thread/handlers/higgsfield/poll_higgsfield_tasks.rs +++ b/crates/artcraftx/src/threads/third_party_task_polling_thread/handlers/higgsfield/poll_higgsfield_tasks.rs @@ -96,7 +96,7 @@ async fn poll_single_task( return Ok(()); } - let Some((session, progress)) = find_job_set(app_handle, app_data_root, sessions, &job_ids, task).await else { + let Some((credential_id, session, progress)) = find_job_set(app_handle, app_data_root, sessions, &job_ids, task).await else { // No account knows the jobs. Leave the task pending: the account may be // temporarily unreachable (expired session, network) and come back. return Ok(()); @@ -170,6 +170,7 @@ async fn poll_single_task( app_preferences, task_database, storyteller_creds_manager, + credential_id, task, &finished, ).await; @@ -184,7 +185,7 @@ async fn find_job_set<'a>( sessions: &'a [(String, HiggsfieldSession)], job_ids: &[JobId], task: &Task, -) -> Option<(&'a HiggsfieldSession, JobSetProgress)> { +) -> Option<(&'a str, &'a HiggsfieldSession, JobSetProgress)> { for (credential_id, session) in sessions { let batch = match session.job_status_batch(job_ids).await { Ok(batch) => batch, @@ -212,15 +213,15 @@ async fn find_job_set<'a>( let status = match statuses.get(job_id) { Some(status) => (*status).clone(), None if batch.missing.contains(job_id) => JobStatus::Other("missing".to_string()), - None => return Some((session, JobSetProgress::Pending)), + None => return Some((credential_id, session, JobSetProgress::Pending)), }; if !status.is_terminal() && !batch.missing.contains(job_id) { info!("[HiggsfieldPolling] Task {} job {} is {}", task.id.as_str(), job_id, status); - return Some((session, JobSetProgress::Pending)); + return Some((credential_id, session, JobSetProgress::Pending)); } outcomes.push(JobOutcome { job_id: job_id.clone(), status }); } - return Some((session, JobSetProgress::Done(outcomes))); + return Some((credential_id, session, JobSetProgress::Done(outcomes))); } None } diff --git a/crates/router/src/api/asset_upload_cache.rs b/crates/router/src/api/asset_upload_cache.rs index 6407c091..aff8d844 100644 --- a/crates/router/src/api/asset_upload_cache.rs +++ b/crates/router/src/api/asset_upload_cache.rs @@ -12,13 +12,57 @@ use async_trait::async_trait; -/// A previously uploaded copy of some bytes, as the provider knows it. +/// How the provider came to hold some content. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum AssetUploadOrigin { + /// We uploaded the bytes; `service_id` is the provider's media id. + Upload, + /// The provider generated the file itself and we downloaded it; + /// `service_id` is the job id, and the content can be referenced as a + /// previous generation without uploading it (or re-running an IP check). + GenerationResult, +} + +impl AssetUploadOrigin { + /// The stored form. + pub fn as_str(self) -> &'static str { + match self { + Self::Upload => "upload", + Self::GenerationResult => "generation_result", + } + } + + /// Unknown values read as plain uploads (the conservative choice: they get + /// verified as media before reuse). + pub fn parse(value: &str) -> Self { + match value { + "generation_result" => Self::GenerationResult, + _ => Self::Upload, + } + } +} + +/// A copy of some bytes the provider already holds, as the provider knows it. #[derive(Clone, Debug, PartialEq, Eq)] pub struct CachedAssetUpload { - /// The provider's own id for the asset (e.g. a Higgsfield media id). + /// The provider's own id for the asset: a media id for an upload, a job + /// id for a generation result. pub service_id: String, /// Where the provider serves it, when it publishes a URL. pub maybe_service_url: Option, + pub origin: AssetUploadOrigin, +} + +impl CachedAssetUpload { + /// Bytes we uploaded. + pub fn uploaded(service_id: impl Into, maybe_service_url: Option) -> Self { + Self { service_id: service_id.into(), maybe_service_url, origin: AssetUploadOrigin::Upload } + } + + /// A generation the provider produced, by job id and result URL. + pub fn generation_result(job_id: impl Into, result_url: impl Into) -> Self { + Self { service_id: job_id.into(), maybe_service_url: Some(result_url.into()), origin: AssetUploadOrigin::GenerationResult } + } } /// One account's upload history on one provider. Implementations must not diff --git a/crates/router/src/api/asset_upload_observer.rs b/crates/router/src/api/asset_upload_observer.rs new file mode 100644 index 00000000..a59d6439 --- /dev/null +++ b/crates/router/src/api/asset_upload_observer.rs @@ -0,0 +1,19 @@ +//! Let the caller watch reference-media uploads as they happen. +//! +//! The router decides per file whether to reuse a provider's existing copy +//! (see [`crate::api::asset_upload_cache`]) or upload it — and the user +//! experience differs: a reuse is instant, an upload with an IP check is a +//! long wait. The caller learns which one happened through this trait, at +//! the moment it happens, so it can show the right notice. + +/// Observes upload decisions for one request. Implementations must be +/// cheap and must not fail; they only inform the UI. +pub trait AssetUploadObserver: Send + Sync { + /// A file was not uploaded because the provider already holds a verified + /// copy. `kind` is e.g. "image" / "video" / "audio". + fn on_reused(&self, kind: &str, description: &str); + + /// A fresh upload is starting. `ip_check` means the provider will also run + /// (and the router will wait for) its intellectual-property check. + fn on_uploading(&self, kind: &str, byte_count: u64, ip_check: bool); +} diff --git a/crates/router/src/api/mod.rs b/crates/router/src/api/mod.rs index 47a04a9e..b2ca4395 100644 --- a/crates/router/src/api/mod.rs +++ b/crates/router/src/api/mod.rs @@ -1,4 +1,5 @@ pub mod asset_upload_cache; +pub mod asset_upload_observer; pub mod audio_list_ref; pub mod audio_ref; pub mod character_list_ref; diff --git a/crates/router/src/generate/generate_image/image_generation_draft_context.rs b/crates/router/src/generate/generate_image/image_generation_draft_context.rs index 37155d4c..92704583 100644 --- a/crates/router/src/generate/generate_image/image_generation_draft_context.rs +++ b/crates/router/src/generate/generate_image/image_generation_draft_context.rs @@ -4,6 +4,7 @@ use std::fmt::Debug; use sqlite_identifiers::ids::media_file_token::MediaFileToken; use crate::api::asset_upload_cache::AssetUploadCache; +use crate::api::asset_upload_observer::AssetUploadObserver; use crate::client::router_client::RouterClient; use crate::client::router_higgsfield_client::RouterHiggsfieldClient; use crate::client::router_seedance2pro_client::RouterSeedance2ProClient; @@ -33,6 +34,10 @@ pub struct ImageGenerationDraftContext<'a> { /// Optional: the account's upload history on the target provider, so /// media it already holds isn't uploaded again. pub asset_upload_cache: Option<&'a dyn AssetUploadCache>, + + /// Optional: told whether each reference was reused or uploaded, so the + /// caller can show the right notice. + pub asset_upload_observer: Option<&'a dyn AssetUploadObserver>, } impl<'a> ImageGenerationDraftContext<'a> { @@ -55,6 +60,7 @@ impl Debug for ImageGenerationDraftContext<'_> { .field("client", &self.client.is_some()) .field("media_file_to_artcraft_url_map", &self.media_file_to_artcraft_url_map.map(|m| m.len())) .field("asset_upload_cache", &self.asset_upload_cache.is_some()) + .field("asset_upload_observer", &self.asset_upload_observer.is_some()) .finish() } } diff --git a/crates/router/src/generate/generate_image/providers/higgsfield/draft.rs b/crates/router/src/generate/generate_image/providers/higgsfield/draft.rs index 701d22b3..475eef12 100644 --- a/crates/router/src/generate/generate_image/providers/higgsfield/draft.rs +++ b/crates/router/src/generate/generate_image/providers/higgsfield/draft.rs @@ -37,6 +37,7 @@ impl HiggsfieldImageDraftState { session: &client.session, maybe_map: draft_context.media_file_to_artcraft_url_map, maybe_cache: draft_context.asset_upload_cache, + maybe_observer: draft_context.asset_upload_observer, }; let mut reference_images = Vec::new(); diff --git a/crates/router/src/generate/generate_image/providers/kinovi/midjourney_7/request.rs b/crates/router/src/generate/generate_image/providers/kinovi/midjourney_7/request.rs index 021c700c..11ddfae1 100644 --- a/crates/router/src/generate/generate_image/providers/kinovi/midjourney_7/request.rs +++ b/crates/router/src/generate/generate_image/providers/kinovi/midjourney_7/request.rs @@ -130,6 +130,7 @@ mod tests { client: Some(&client), media_file_to_artcraft_url_map: Some(&map), asset_upload_cache: None, + asset_upload_observer: None, }; let request = ImageGenerationDraftRequest::KinoviMidjourney7(draft) diff --git a/crates/router/src/generate/generate_image/providers/kinovi/midjourney_7_niji/request.rs b/crates/router/src/generate/generate_image/providers/kinovi/midjourney_7_niji/request.rs index 47c7eef3..cbbbcb42 100644 --- a/crates/router/src/generate/generate_image/providers/kinovi/midjourney_7_niji/request.rs +++ b/crates/router/src/generate/generate_image/providers/kinovi/midjourney_7_niji/request.rs @@ -128,6 +128,7 @@ mod tests { client: Some(&client), media_file_to_artcraft_url_map: Some(&map), asset_upload_cache: None, + asset_upload_observer: None, }; let request = ImageGenerationDraftRequest::KinoviMidjourney7Niji(draft) diff --git a/crates/router/src/generate/generate_image/providers/kinovi/midjourney_8/request.rs b/crates/router/src/generate/generate_image/providers/kinovi/midjourney_8/request.rs index 97107ef0..e5ae9eaf 100644 --- a/crates/router/src/generate/generate_image/providers/kinovi/midjourney_8/request.rs +++ b/crates/router/src/generate/generate_image/providers/kinovi/midjourney_8/request.rs @@ -128,6 +128,7 @@ mod tests { client: Some(&client), media_file_to_artcraft_url_map: Some(&map), asset_upload_cache: None, + asset_upload_observer: None, }; let request = ImageGenerationDraftRequest::KinoviMidjourney8(draft) diff --git a/crates/router/src/generate/generate_image/providers/kinovi/seedream_5p0_pro/request.rs b/crates/router/src/generate/generate_image/providers/kinovi/seedream_5p0_pro/request.rs index 66f0d609..95b1d2d1 100644 --- a/crates/router/src/generate/generate_image/providers/kinovi/seedream_5p0_pro/request.rs +++ b/crates/router/src/generate/generate_image/providers/kinovi/seedream_5p0_pro/request.rs @@ -129,6 +129,7 @@ mod tests { client: Some(&client), media_file_to_artcraft_url_map: Some(&map), asset_upload_cache: None, + asset_upload_observer: None, }; let request = ImageGenerationDraftRequest::KinoviSeedream5p0Pro(draft) diff --git a/crates/router/src/generate/generate_video/providers/higgsfield/draft.rs b/crates/router/src/generate/generate_video/providers/higgsfield/draft.rs index 58b81834..27f1efd5 100644 --- a/crates/router/src/generate/generate_video/providers/higgsfield/draft.rs +++ b/crates/router/src/generate/generate_video/providers/higgsfield/draft.rs @@ -69,6 +69,7 @@ impl HiggsfieldVideoDraftState { session: &client.session, maybe_map: draft_context.media_file_to_artcraft_url_map, maybe_cache: draft_context.asset_upload_cache, + maybe_observer: draft_context.asset_upload_observer, }; let label = self.plan.model_label(); diff --git a/crates/router/src/generate/generate_video/video_generation_draft_context.rs b/crates/router/src/generate/generate_video/video_generation_draft_context.rs index f0bc54f6..51de5556 100644 --- a/crates/router/src/generate/generate_video/video_generation_draft_context.rs +++ b/crates/router/src/generate/generate_video/video_generation_draft_context.rs @@ -1,4 +1,5 @@ use crate::api::asset_upload_cache::AssetUploadCache; +use crate::api::asset_upload_observer::AssetUploadObserver; use crate::client::router_client::RouterClient; use crate::client::router_higgsfield_client::RouterHiggsfieldClient; use crate::client::router_seedance2pro_client::RouterSeedance2ProClient; @@ -25,6 +26,10 @@ pub struct VideoGenerationDraftContext<'a> { /// Optional: the account's upload history on the target provider, so /// media it already holds isn't uploaded again. pub asset_upload_cache: Option<&'a dyn AssetUploadCache>, + + /// Optional: told whether each reference was reused or uploaded, so the + /// caller can show the right notice. + pub asset_upload_observer: Option<&'a dyn AssetUploadObserver>, } impl <'a> VideoGenerationDraftContext<'a> { @@ -58,6 +63,7 @@ impl Debug for VideoGenerationDraftContext<'_> { .field("media_file_to_artcraft_url_map", &self.media_file_to_artcraft_url_map) .field("character_token_to_kinovi_id_map", &self.character_token_to_kinovi_id_map) .field("asset_upload_cache", &self.asset_upload_cache.is_some()) + .field("asset_upload_observer", &self.asset_upload_observer.is_some()) .finish() } } diff --git a/crates/router/src/utils/higgsfield_media.rs b/crates/router/src/utils/higgsfield_media.rs index 0c64f41d..304d3962 100644 --- a/crates/router/src/utils/higgsfield_media.rs +++ b/crates/router/src/utils/higgsfield_media.rs @@ -12,20 +12,20 @@ //! media status endpoint first, and any doubt means a fresh upload. use std::collections::HashMap; -use std::path::Path; use file_hashing::hash_bytes_blake3; use higgsfield_client::endpoints::media::get_media_status::MediaStatusFamily; use higgsfield_client::session::higgsfield_session::HiggsfieldSession; use higgsfield_client::session::upload_media::ReferenceMediaFile; use higgsfield_client::session::upload_source_guard::check_upload_source_url; -use higgsfield_client::types::ids::MediaId; +use higgsfield_client::types::ids::{JobId, MediaId}; use higgsfield_client::types::media_input::MediaInput; use higgsfield_client::types::media_mime_type::MediaMimeType; use log::{info, warn}; use sqlite_identifiers::ids::media_file_token::MediaFileToken; -use crate::api::asset_upload_cache::{AssetUploadCache, CachedAssetUpload}; +use crate::api::asset_upload_observer::AssetUploadObserver; +use crate::api::asset_upload_cache::{AssetUploadCache, AssetUploadOrigin, CachedAssetUpload}; use crate::api::audio_list_ref::AudioListRef; use crate::api::image_list_ref::ImageListRef; use crate::api::image_ref::ImageRef; @@ -48,6 +48,15 @@ pub enum HiggsfieldMediaKind { } impl HiggsfieldMediaKind { + /// Lower-case name for notices ("image" / "video" / "audio"). + pub fn label(self) -> &'static str { + match self { + Self::Image => "image", + Self::Video => "video", + Self::Audio => "audio", + } + } + /// The status endpoint family for verifying a previous upload. Audio has /// none, so cached audio ids are trusted as-is. fn status_family(self) -> Option { @@ -66,6 +75,16 @@ impl HiggsfieldMediaKind { } } + /// A previous Higgsfield generation used as this kind of reference. Audio + /// is never generated, so it has no job form. + fn job_input(self, job_id: MediaId, url: String) -> Option { + match self { + Self::Image => Some(MediaInput::from_image_job(job_id, url)), + Self::Video => Some(MediaInput::from_video_job(job_id, url)), + Self::Audio => None, + } + } + fn fallback_mime_type(self) -> MediaMimeType { match self { Self::Image => MediaMimeType::ImageJpeg, @@ -123,6 +142,8 @@ pub struct HiggsfieldUploadContext<'a> { pub maybe_map: Option<&'a HashMap>, /// The account's prior uploads; `None` uploads unconditionally. pub maybe_cache: Option<&'a dyn AssetUploadCache>, + /// Told whether each source was reused or uploaded. + pub maybe_observer: Option<&'a dyn AssetUploadObserver>, } /// Upload every source in `list`, in order. `None` / empty lists upload @@ -185,6 +206,9 @@ pub async fn upload_source_to_higgsfield( if let Some(reusable) = reusable_upload(context.session, cache, &file_hash, kind, ip_check).await { info!("Reusing Higgsfield {:?} reference {} for {} (already uploaded by this account)", kind, reusable.id, resolved.description); cache.note_reused(&file_hash).await; + if let Some(observer) = context.maybe_observer { + observer.on_reused(kind.label(), &resolved.description); + } return Ok(reusable); } } @@ -195,13 +219,16 @@ pub async fn upload_source_to_higgsfield( "Uploading {:?} reference to Higgsfield ({} bytes, {}, ip_check={})", kind, file.bytes.len(), file.mime_type, ip_check, ); + if let Some(observer) = context.maybe_observer { + observer.on_uploading(kind.label(), byte_count, ip_check); + } let uploaded = session_upload(context.session, file).await?; if let Some(cache) = context.maybe_cache { - cache.record_upload(&file_hash, byte_count, &CachedAssetUpload { - service_id: uploaded.id.as_str().to_string(), - maybe_service_url: Some(uploaded.url.clone()), - }).await; + cache.record_upload(&file_hash, byte_count, &CachedAssetUpload::uploaded( + uploaded.id.as_str(), + Some(uploaded.url.clone()), + )).await; } Ok(uploaded) } @@ -225,6 +252,10 @@ async fn reusable_upload( let cached = cache.find_upload(file_hash).await?; let media_id = MediaId::new(cached.service_id.clone()); + if cached.origin == AssetUploadOrigin::GenerationResult { + return reusable_generation_result(session, cache, file_hash, kind, media_id).await; + } + let Some(url) = cached.maybe_service_url.clone() else { warn!("Cached Higgsfield upload {} has no URL; uploading again", media_id); cache.forget_upload(file_hash).await; @@ -255,6 +286,37 @@ async fn reusable_upload( Some(kind.media_input(media_id, url)) } +/// A file Higgsfield generated itself (we downloaded the result) is used as a +/// "previous generation" reference by job id — no upload, and no IP check, +/// since it never left Higgsfield. The job is confirmed to still exist with +/// a result, and the fresh result URL is used (CDN URLs are time-limited). +async fn reusable_generation_result( + session: &HiggsfieldSession, + cache: &dyn AssetUploadCache, + file_hash: &str, + kind: HiggsfieldMediaKind, + job_id: MediaId, +) -> Option { + let job = match session.job_status(&JobId::new(job_id.as_str())).await { + Ok(job) => job, + Err(err) => { + warn!("Could not confirm Higgsfield generation {} still exists ({}); uploading instead", job_id, err); + cache.forget_upload(file_hash).await; + return None; + } + }; + let Some(url) = job.result_url() else { + warn!("Higgsfield generation {} no longer has a result; uploading instead", job_id); + cache.forget_upload(file_hash).await; + return None; + }; + let Some(input) = kind.job_input(job_id.clone(), url.to_string()) else { + info!("Higgsfield generation {} can't stand in for a {:?} reference; uploading instead", job_id, kind); + return None; + }; + Some(input) +} + /// Describe resolved bytes for upload: the MIME type is sniffed from the /// bytes, then guessed from the name hint (URL, path, or file name), then /// defaulted by `kind`. The resolved source URL (caller-supplied URLs only) @@ -476,7 +538,7 @@ mod tests { #[tokio::test] async fn audio_has_no_status_endpoint_so_the_cached_id_is_trusted() { - let cache = CannedCache::with(CachedAssetUpload { service_id: "media-audio".into(), maybe_service_url: Some("https://cdn/a.mp3".into()) }); + let cache = CannedCache::with(CachedAssetUpload::uploaded("media-audio", Some("https://cdn/a.mp3".into()))); let reused = reusable_upload(&unreachable_session(), &cache, "hash", HiggsfieldMediaKind::Audio, false).await.unwrap(); assert_eq!(reused.id.as_str(), "media-audio"); assert_eq!(reused.kind, MediaInputKind::AudioInput); @@ -486,14 +548,32 @@ mod tests { #[tokio::test] async fn an_entry_without_a_url_is_forgotten() { - let cache = CannedCache::with(CachedAssetUpload { service_id: "media-1".into(), maybe_service_url: None }); + let cache = CannedCache::with(CachedAssetUpload::uploaded("media-1", None)); assert!(reusable_upload(&unreachable_session(), &cache, "hash", HiggsfieldMediaKind::Audio, false).await.is_none()); assert_eq!(*cache.forgotten.lock().unwrap(), vec!["hash".to_string()]); } + #[tokio::test] + async fn a_generation_result_that_cannot_be_confirmed_means_a_fresh_upload() { + let cache = CannedCache::with(CachedAssetUpload::generation_result("job-1", "https://cdn/result.png")); + let result = reusable_upload(&unreachable_session(), &cache, "hash", HiggsfieldMediaKind::Image, true).await; + assert!(result.is_none()); + assert!(cache.forgotten.lock().unwrap().contains(&"hash".to_string())); + } + + #[test] + fn generation_results_become_job_references_by_kind() { + let image = HiggsfieldMediaKind::Image.job_input(MediaId::new("job-1"), "https://cdn/r.png".into()).unwrap(); + assert_eq!(image.kind, MediaInputKind::ImageJob); + assert_eq!(image.id.as_str(), "job-1"); + let video = HiggsfieldMediaKind::Video.job_input(MediaId::new("job-2"), "https://cdn/r.mp4".into()).unwrap(); + assert_eq!(video.kind, MediaInputKind::VideoJob); + assert!(HiggsfieldMediaKind::Audio.job_input(MediaId::new("job-3"), "https://cdn/r.mp3".into()).is_none()); + } + #[tokio::test] async fn a_status_check_that_fails_means_a_fresh_upload() { - let cache = CannedCache::with(CachedAssetUpload { service_id: "media-1".into(), maybe_service_url: Some("https://cdn/x.png".into()) }); + let cache = CannedCache::with(CachedAssetUpload::uploaded("media-1", Some("https://cdn/x.png".into()))); assert!(reusable_upload(&unreachable_session(), &cache, "hash", HiggsfieldMediaKind::Image, true).await.is_none()); assert_eq!(*cache.forgotten.lock().unwrap(), vec!["hash".to_string()]); assert!(cache.reused.lock().unwrap().is_empty()); diff --git a/crates/schema/local_files_database/migrations/20260907000003_add_asset_upload_origin.sql b/crates/schema/local_files_database/migrations/20260907000003_add_asset_upload_origin.sql new file mode 100644 index 00000000..733cb202 --- /dev/null +++ b/crates/schema/local_files_database/migrations/20260907000003_add_asset_upload_origin.sql @@ -0,0 +1,6 @@ +-- How the service came to hold the content. `upload`: we sent the bytes. +-- `generation_result`: the service produced the file itself (a generation +-- we downloaded), so `service_id` is its job id and the content can be +-- referenced as a previous generation without ever uploading it. + +ALTER TABLE asset_uploads ADD COLUMN origin TEXT NOT NULL DEFAULT 'upload'; diff --git a/crates/schema/local_files_database/src/queries/asset_upload_record.rs b/crates/schema/local_files_database/src/queries/asset_upload_record.rs index 7182e906..e6ef855b 100644 --- a/crates/schema/local_files_database/src/queries/asset_upload_record.rs +++ b/crates/schema/local_files_database/src/queries/asset_upload_record.rs @@ -6,8 +6,11 @@ pub struct AssetUploadRecord { pub service: String, pub account_id: String, pub file_hash_blake3: String, - /// The service's own primary key for the asset. + /// The service's own primary key for the asset: a media id for an upload, + /// a job id for a generation result (see `origin`). pub service_id: String, + /// `upload` or `generation_result`. + pub origin: String, pub service_url: Option, pub file_size_bytes: Option, pub uploaded_at: DateTime, diff --git a/crates/schema/local_files_database/src/queries/upsert_asset_upload.rs b/crates/schema/local_files_database/src/queries/upsert_asset_upload.rs index fd7ed461..802a59a6 100644 --- a/crates/schema/local_files_database/src/queries/upsert_asset_upload.rs +++ b/crates/schema/local_files_database/src/queries/upsert_asset_upload.rs @@ -9,6 +9,8 @@ pub struct UpsertAssetUploadArgs<'a> { pub account_id: &'a str, pub file_hash_blake3: &'a str, pub service_id: &'a str, + /// `upload` or `generation_result`. + pub origin: &'a str, pub maybe_service_url: Option<&'a str>, pub maybe_file_size_bytes: Option, } @@ -19,10 +21,11 @@ pub async fn upsert_asset_upload(args: UpsertAssetUploadArgs<'_>) -> Result<(), sqlx::query( r#" INSERT INTO asset_uploads - (service, account_id, file_hash_blake3, service_id, service_url, file_size_bytes, uploaded_at, last_reused_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL) + (service, account_id, file_hash_blake3, service_id, origin, service_url, file_size_bytes, uploaded_at, last_reused_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL) ON CONFLICT(service, account_id, file_hash_blake3) DO UPDATE SET service_id = excluded.service_id, + origin = excluded.origin, service_url = excluded.service_url, file_size_bytes = excluded.file_size_bytes, uploaded_at = excluded.uploaded_at, @@ -33,6 +36,7 @@ pub async fn upsert_asset_upload(args: UpsertAssetUploadArgs<'_>) -> Result<(), .bind(args.account_id) .bind(args.file_hash_blake3) .bind(args.service_id) + .bind(args.origin) .bind(args.maybe_service_url) .bind(args.maybe_file_size_bytes) .bind(Utc::now()) diff --git a/crates/schema/local_files_database/tests/asset_upload_queries.rs b/crates/schema/local_files_database/tests/asset_upload_queries.rs index c9974ec5..2e237e80 100644 --- a/crates/schema/local_files_database/tests/asset_upload_queries.rs +++ b/crates/schema/local_files_database/tests/asset_upload_queries.rs @@ -77,6 +77,7 @@ async fn record(db: &LocalFilesDbConnection, service: &str, account_id: &str, ha account_id, file_hash_blake3: hash, service_id, + origin: "upload", maybe_service_url: url, maybe_file_size_bytes: Some(1234), }).await.expect("upsert");