handle re-upload
This commit is contained in:
parent
b61923ec8d
commit
b6746cf50b
|
|
@ -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() });
|
||||
|
|
|
|||
|
|
@ -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<HashM
|
|||
}
|
||||
|
||||
/// Build, finalize (uploading references) and send an image request.
|
||||
///
|
||||
/// References the account already uploaded are reused from the ledger. If
|
||||
/// Higgsfield then rejects the job because a reused media id is no good
|
||||
/// (deleted or flagged since), those ledger entries are dropped and the
|
||||
/// request is finalized and sent once more with fresh uploads.
|
||||
pub async fn send_higgsfield_image_request(
|
||||
maybe_app: Option<&AppHandle>,
|
||||
credential: &AuthCredential,
|
||||
builder: GenerateImageRequestBuilder,
|
||||
client: &RouterClient,
|
||||
media_url_map: &HashMap<MediaFileToken, String>,
|
||||
upload_cache: &dyn AssetUploadCache,
|
||||
upload_cache: &AccountUploadCache,
|
||||
) -> Result<GenerateImageResponse, GenerateError> {
|
||||
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<MediaFileToken, String>,
|
||||
upload_cache: &AccountUploadCache,
|
||||
notices: &HiggsfieldUploadNotices,
|
||||
) -> Result<ImageGenerationRequest, GenerateError> {
|
||||
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<MediaFileToken, String>,
|
||||
upload_cache: &dyn AssetUploadCache,
|
||||
upload_cache: &AccountUploadCache,
|
||||
) -> Result<GenerateVideoResponse, GenerateError> {
|
||||
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<MediaFileToken, String>,
|
||||
upload_cache: &AccountUploadCache,
|
||||
notices: &HiggsfieldUploadNotices,
|
||||
) -> Result<VideoGenerationRequest, GenerateError> {
|
||||
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())
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Mutex<Vec<String>>>,
|
||||
}
|
||||
|
||||
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<String>) -> 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<String> = 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();
|
||||
|
|
|
|||
|
|
@ -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<AppHandle>,
|
||||
progress: Mutex<Option<ProgressNotice>>,
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
pub mod higgsfield_session_from_credential;
|
||||
|
||||
pub mod mark_higgsfield_relogin_required;
|
||||
pub mod higgsfield_upload_notices;
|
||||
|
|
|
|||
|
|
@ -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::<LocalFilesDatabase>()
|
||||
.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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String>,
|
||||
pub origin: AssetUploadOrigin,
|
||||
}
|
||||
|
||||
impl CachedAssetUpload {
|
||||
/// Bytes we uploaded.
|
||||
pub fn uploaded(service_id: impl Into<String>, maybe_service_url: Option<String>) -> 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<String>, result_url: impl Into<String>) -> 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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<MediaStatusFamily> {
|
||||
|
|
@ -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<MediaInput> {
|
||||
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<MediaFileToken, String>>,
|
||||
/// 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<MediaInput> {
|
||||
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());
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
@ -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<String>,
|
||||
pub file_size_bytes: Option<i64>,
|
||||
pub uploaded_at: DateTime<Utc>,
|
||||
|
|
|
|||
|
|
@ -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<i64>,
|
||||
}
|
||||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
Loading…
Reference in New Issue