diff --git a/crates/api_clients/kinovi_web_client/src/requests/poll_orders/poll_orders.rs b/crates/api_clients/kinovi_web_client/src/requests/poll_orders/poll_orders.rs --- a/crates/api_clients/kinovi_web_client/src/requests/poll_orders/poll_orders.rs +++ b/crates/api_clients/kinovi_web_client/src/requests/poll_orders/poll_orders.rs @@ -36,7 +36,7 @@ pub struct PollOrdersResponse { // --- Public types --- -/// The status of one order (one generation task — video or image). +/// The status of one order (one generation task — video, image, or audio). #[derive(Debug, Clone)] pub struct OrderStatus { pub order_id: String, @@ -61,7 +61,7 @@ pub struct OrderStatus { /// Parsed `created_at` as a `DateTime`. `None` if the raw string could not be parsed. pub created_at_utc: Option>, - /// Whether this order produced an image or a video. `None` for older + /// Whether this order produced an image, video, or audio. `None` for older /// polling responses that didn't include the field — those came from the /// video-only era and can be treated as video by callers that need to. pub media_type: Option, @@ -119,14 +119,15 @@ pub struct MediaResult { } /// Media type of an order. Midjourney orders are `Image`; the various -/// Seedance/keyframe/reference flows are `Video`. +/// Seedance/keyframe/reference flows are `Video`; Suno orders are `Audio`. /// -/// `Unknown` covers response payloads that omit the field (older video -/// polling) or return an unrecognised value. +/// `Unknown` preserves unrecognised values. An omitted field is represented +/// by `None` on `OrderStatus`. #[derive(Debug, Clone, PartialEq)] pub enum OrderMediaType { Image, Video, + Audio, Unknown(String), } @@ -135,12 +136,14 @@ impl OrderMediaType { match s { "image" => Self::Image, "video" => Self::Video, + "audio" => Self::Audio, other => Self::Unknown(other.to_string()), } } pub fn is_image(&self) -> bool { matches!(self, Self::Image) } pub fn is_video(&self) -> bool { matches!(self, Self::Video) } + pub fn is_audio(&self) -> bool { matches!(self, Self::Audio) } } // --- Implementation --- @@ -275,6 +278,14 @@ mod tests { use errors::AnyhowResult; use log::LevelFilter; + #[test] + fn recognizes_audio_orders_without_changing_legacy_types() { + assert_eq!(OrderMediaType::from_str("audio"), OrderMediaType::Audio); + assert_eq!(OrderMediaType::from_str("image"), OrderMediaType::Image); + assert_eq!(OrderMediaType::from_str("video"), OrderMediaType::Video); + assert_eq!(OrderMediaType::from_str("future"), OrderMediaType::Unknown("future".into())); + } + fn test_session() -> AnyhowResult { let cookies = get_test_cookies()?; Ok(KinoviWebSession::from_cookies_string(cookies)) @@ -438,6 +449,28 @@ mod tests { }).collect() } + /// Synthetic contract fixture: audio clips need neither video dimensions + /// nor a top-level result URL when the results array is populated. + #[test] + fn audio_order_preserves_both_clips_without_dimensions() { + let body = r#"[{"result":{"data":{"json":{"orders":[{ + "orderId":"ord_suno_audio", + "resultUrl":null, + "taskStatus":"COMPLETED", + "results":[{"url":"https://example.test/one.mp3"},{"url":"https://example.test/two.mp3"}], + "failReason":null, + "createdAt":"2026-10-18T00:00:00Z", + "mediaType":"audio" + }],"nextCursor":null}}}}]"#; + let orders = parse_orders(body); + assert_eq!(orders.len(), 1); + assert_eq!(orders[0].media_type, Some(OrderMediaType::Audio)); + assert_eq!(orders[0].task_status, TaskStatus::Completed); + assert_eq!(orders[0].results.len(), 2); + assert_eq!(orders[0].results[1].url, "https://example.test/two.mp3"); + assert!(orders[0].results.iter().all(|clip| clip.maybe_width.is_none() && clip.maybe_height.is_none())); + } + /// Back-compat: when the raw payload omits `mediaType` (older video-only /// polling), `media_type` parses as `None` and the rest of the order /// continues to deserialise unchanged. @@ -555,12 +588,14 @@ mod tests { println!("Next cursor: {:?}", result.next_cursor); let mut images = 0; let mut videos = 0; + let mut audio = 0; let mut unknown = 0; let mut missing = 0; for order in &result.orders { match &order.media_type { Some(OrderMediaType::Image) => images += 1, Some(OrderMediaType::Video) => videos += 1, + Some(OrderMediaType::Audio) => audio += 1, Some(OrderMediaType::Unknown(_)) => unknown += 1, None => missing += 1, } @@ -569,8 +604,8 @@ mod tests { order.results.len(), order.result_url, order.fail_reason); } println!( - "media_type tally: image={}, video={}, unknown={}, missing={}", - images, videos, unknown, missing, + "media_type tally: image={}, video={}, audio={}, unknown={}, missing={}", + images, videos, audio, unknown, missing, ); assert_eq!(1, 2); // NB: Intentional failure to inspect output. Ok(()) diff --git a/crates/schema/database/mysql_queries/src/queries/media_files/create/insert_builder/media_file_insert_builder.rs b/crates/schema/database/mysql_queries/src/queries/media_files/create/insert_builder/media_file_insert_builder.rs --- a/crates/schema/database/mysql_queries/src/queries/media_files/create/insert_builder/media_file_insert_builder.rs +++ b/crates/schema/database/mysql_queries/src/queries/media_files/create/insert_builder/media_file_insert_builder.rs @@ -304,7 +304,10 @@ impl MediaFileInsertBuilder { self.insert_acquire(mysql_connection).await } - async fn insert_acquire<'a, A>(&'a self, connector: A) -> Result + /// Insert through a pool, connection, or existing transaction. Within a + /// transaction the underlying helper uses a savepoint; the caller controls + /// the final commit, including any associated job-status update. + pub async fn insert_acquire<'a, A>(&'a self, connector: A) -> Result where A: Acquire<'a, Database = MySql> + Send, >::Connection: Send, diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/mod.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/mod.rs --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/mod.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/mod.rs @@ -4,3 +4,4 @@ pub mod process_one_order; pub mod process_failed_job; pub mod process_successful_image_job; pub mod process_successful_video_job; +pub mod process_successful_audio_job; diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_one_order.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_one_order.rs --- a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_one_order.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_one_order.rs @@ -4,6 +4,7 @@ use errors::AnyhowResult; use kinovi_web_client::requests::poll_orders::poll_orders::{OrderMediaType, TaskStatus}; use crate::job_dependencies::JobDependencies; +use crate::jobs::order_processing_job::process_successful_audio_job::process_successful_audio_job; use crate::jobs::order_processing_job::process_failed_job::process_failed_job; use crate::jobs::order_processing_job::process_successful_image_job::process_successful_image_job; use crate::jobs::order_processing_job::process_successful_video_job::process_successful_video_job; @@ -25,6 +26,7 @@ pub async fn process_one_order( // response shapes that pre-date the `mediaType` field) and `Some(Video)` // both fall to the video handler for back-compat. match &order.media_type { + Some(OrderMediaType::Audio) => process_successful_audio_job(deps, job, order).await, Some(OrderMediaType::Image) => process_successful_image_job(deps, job, order).await, Some(OrderMediaType::Video) | None => process_successful_video_job(deps, job, order).await, Some(OrderMediaType::Unknown(other)) => { diff --git a/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_audio_job.rs b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_audio_job.rs new file mode 100644 --- /dev/null +++ b/crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_audio_job.rs @@ -0,0 +1,338 @@ +use anyhow::anyhow; +use log::{error, info, warn}; + +use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath; +use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType; +use enums::by_table::media_files::media_file_class::MediaFileClass; +use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory; +use enums::by_table::media_files::media_file_type::MediaFileType; +use enums::common::generation_provider::GenerationProvider; +use errors::AnyhowResult; +use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes; +use mimetypes::mimetype_for_bytes::get_mimetype_for_bytes; +use mysql_queries::queries::generic_inference::api_providers::kinovi_web::list_pending_kinovi_web_video_jobs::PendingKinoviWebJob; +use mysql_queries::queries::generic_inference::job::select_inference_job_status_for_update::select_inference_job_status_for_update; +use mysql_queries::queries::generic_inference::web::mark_generic_inference_job_successfully_done_by_token_with_executor::{mark_generic_inference_job_successfully_done_by_token_with_executor, MarkGenericInferenceJobSuccessfullyDoneByTokenWithExecutorArgs}; +use mysql_queries::queries::media_files::create::insert_builder::media_file_insert_builder::MediaFileInsertBuilder; +use kinovi_web_client::requests::poll_orders::poll_orders::{OrderStatus, MediaResult}; +use tokens::tokens::batch_generations::BatchGenerationToken; + +use crate::alert_on_error::alert_pager_and_return_err; +use crate::job_dependencies::JobDependencies; +use crate::jobs::order_processing_job::is_job_status_terminal::is_job_status_terminal; + +const PREFIX: &str = "artcraft_"; + +/// Import a completed Suno audio order from Kinovi. +/// +/// Preserve every returned clip in one batch and expose the first as the +/// primary result. Download and validate audio before marking the job complete. +pub async fn process_successful_audio_job( + deps: &JobDependencies, + job: &PendingKinoviWebJob, + order: &OrderStatus, +) -> AnyhowResult<()> { + let results = audio_results(order)?; + + info!( + "Processing completed audio order {}: {} audio(s) to download.", + order.order_id, + results.len(), + ); + + // Complete all network IO before acquiring a database connection. Media + // records and job completion are then committed together so a failed second + // clip or a retry cannot expose partial/duplicate batches in the library. + let maybe_batch_token = (results.len() > 1).then(BatchGenerationToken::generate); + let mut media_files = Vec::with_capacity(results.len()); + for (idx, result) in results.iter().enumerate() { + media_files.push( + download_and_upload_audio(deps, job, order, result, idx, maybe_batch_token.as_ref()).await?, + ); + } + + // Finalize inside a transaction: re-check the job is still pending under a row + // lock, then mark it complete. See `process_successful_video_job` for the rationale. + let mut transaction = deps.mysql_pool.begin().await.map_err(|err| { + anyhow!( + "error beginning finalize transaction for job {}: {:?}", + job.job_token.as_str(), + err + ) + })?; + + let maybe_status = select_inference_job_status_for_update(&mut *transaction, &job.job_token) + .await + .map_err(|err| { + anyhow!( + "error locking job {} for finalize: {:?}", + job.job_token.as_str(), + err + ) + })?; + + // A concurrent completion or cancellation must win over this import. + let status = match maybe_status { + Some(status) => status, + None => { + let _ = transaction.rollback().await; + return Err(anyhow!( + "Job {} vanished before finalize (order {})", + job.job_token.as_str(), + order.order_id, + )); + }, + }; + + if is_job_status_terminal(status) { + warn!( + "Job {} is already terminal ({:?}); skipping mark-done (order {}). \ + {} bucket object(s) may be orphaned.", + job.job_token.as_str(), + status, + order.order_id, + media_files.len(), + ); + let _ = transaction.rollback().await; + return Ok(()); + } + + let mut primary_token = None; + for media_file in &media_files { + let token = media_file.insert_acquire(&mut *transaction).await?; + if primary_token.is_none() { + primary_token = Some(token); + } + } + let primary_token = primary_token.ok_or_else(|| anyhow!("No audio results to finalize"))?; + + // Commit all clips and the job status within the same transaction. + + if let Err(err) = mark_generic_inference_job_successfully_done_by_token_with_executor( + MarkGenericInferenceJobSuccessfullyDoneByTokenWithExecutorArgs { + executor: &mut *transaction, + token: &job.job_token, + maybe_entity_type: Some(InferenceResultType::MediaFile), + maybe_entity_token: Some(primary_token.as_str()), + total_job_duration: None, + inference_duration: None, + }, + ) + .await + { + let _ = transaction.rollback().await; + error!( + "Error marking audio job {} done: {:?}", + job.job_token.as_str(), + err + ); + return alert_pager_and_return_err( + &deps.pager, + "Kinovi audio job completion update failed", + anyhow!("error marking job done: {:?}", err), + Some(job), + ); + } + + transaction.commit().await.map_err(|err| { + anyhow!( + "error committing finalize transaction for job {}: {:?}", + job.job_token.as_str(), + err + ) + })?; + + info!( + "Audio job {} completed successfully.", + job.job_token.as_str() + ); + + Ok(()) +} + +/// Upload a clip and prepare its media record without holding a DB connection. +async fn download_and_upload_audio( + deps: &JobDependencies, + job: &PendingKinoviWebJob, + order: &OrderStatus, + result: &MediaResult, + index: usize, + maybe_batch_token: Option<&BatchGenerationToken>, +) -> AnyhowResult { + let audio_url = result.url.as_str(); + + info!( + "Downloading audio {}/{} for order {} from: {}", + index + 1, + order.results.len().max(1), + order.order_id, + audio_url, + ); + + let audio_bytes: Vec = deps + .download_client + .get(audio_url) + .send() + .await + .map_err(|err| anyhow!("reqwest error downloading audio {}: {:?}", index, err))? + .error_for_status() + .map_err(|err| anyhow!("HTTP error downloading audio {}: {:?}", index, err))? + .bytes() + .await + .map_err(|err| anyhow!("error reading audio bytes for index {}: {:?}", index, err))? + .to_vec(); + + info!( + "Downloaded {} bytes for audio {}/{} of order {}", + audio_bytes.len(), + index + 1, + order.results.len().max(1), + order.order_id, + ); + + let (suffix, mime_type, file_type) = audio_metadata(&audio_bytes)?; + + let checksum = sha256_hash_bytes(&audio_bytes) + .map_err(|err| anyhow!("error hashing audio {}: {:?}", index, err))?; + + let bucket_path = MediaFileBucketPath::generate_new(Some(PREFIX), Some(suffix)); + let object_path = bucket_path.get_full_object_path_str(); + + info!( + "Uploading audio {}/{} to public bucket at path: {}", + index + 1, + order.results.len().max(1), + object_path, + ); + + deps + .public_bucket_client + .upload_file_with_content_type_process(object_path, &audio_bytes, mime_type) + .await + .map_err(|err| anyhow!("bucket upload failed for audio {}: {:?}", index, err))?; + + let media_file = MediaFileInsertBuilder::new() + .checksum_sha2(&checksum) + .creator_ip_address(&job.creator_ip_address) + .creator_set_visibility(job.creator_set_visibility) + .file_size_bytes(audio_bytes.len() as u64) + .maybe_batch_generation_token(maybe_batch_token) + .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref()) + .maybe_creator_user(job.maybe_creator_user_token.as_ref()) + .maybe_generation_provider(Some(GenerationProvider::Artcraft)) + .maybe_prompt_token(job.maybe_prompt_token.as_ref()) + .maybe_source_job_token(Some(&job.job_token)) + .maybe_platform_type(job.maybe_platform_type) + .media_file_class(MediaFileClass::Audio) + .media_file_origin_category(MediaFileOriginCategory::Inference) + .media_file_type(file_type) + .mime_type(mime_type) + .public_bucket_directory_hash(&bucket_path); + + Ok(media_file) +} + +/// Some provider responses expose only a top-level URL. Do not lose those +/// single-clip orders or import a duplicate of the primary clip in a batch. +fn audio_results(order: &OrderStatus) -> AnyhowResult> { + if !order.results.is_empty() { + return Ok(order.results.clone()); + } + let url = order + .result_url + .as_ref() + .filter(|url| !url.is_empty()) + .ok_or_else(|| anyhow!("Completed audio order {} has no results", order.order_id))?; + Ok(vec![MediaResult { + url: url.clone(), + maybe_width: None, + maybe_height: None, + }]) +} + +fn audio_metadata(bytes: &[u8]) -> AnyhowResult<(&'static str, &'static str, MediaFileType)> { + match get_mimetype_for_bytes(bytes) { + Some("audio/mpeg") => Ok((".mp3", "audio/mpeg", MediaFileType::Mp3)), + Some("audio/x-wav" | "audio/wav") => Ok((".wav", "audio/wav", MediaFileType::Wav)), + Some("audio/flac" | "audio/x-flac") => Ok((".flac", "audio/flac", MediaFileType::Flac)), + Some("audio/ogg") => Ok((".ogg", "audio/ogg", MediaFileType::Ogg)), + Some("audio/x-m4a" | "audio/m4a") => Ok((".m4a", "audio/mp4", MediaFileType::M4a)), + Some("audio/aac" | "audio/x-aac") => Ok((".aac", "audio/aac", MediaFileType::Aac)), + other => Err(anyhow!( + "Unsupported Kinovi audio result MIME type: {:?}", + other + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use kinovi_web_client::requests::poll_orders::poll_orders::{OrderMediaType, TaskStatus}; + + #[test] + fn accepts_mp3_and_wav_with_correct_file_metadata() { + assert_eq!( + audio_metadata(b"ID3\x04\x00\x00\x00\x00\x00\x00").unwrap(), + (".mp3", "audio/mpeg", MediaFileType::Mp3) + ); + assert_eq!( + audio_metadata(b"RIFF\x24\x00\x00\x00WAVEfmt ").unwrap(), + (".wav", "audio/wav", MediaFileType::Wav) + ); + } + + #[test] + fn rejects_video_and_error_pages_instead_of_importing_them_as_audio() { + assert!(audio_metadata(b"\x00\x00\x00\x20ftypisom\x00\x00\x02\x00isomiso2avc1mp41").is_err()); + assert!(audio_metadata(b"Access denied").is_err()); + } + + #[test] + fn preserves_both_clips_without_duplicating_the_primary_url() { + let mut order = completed_order(); + order.result_url = Some("https://example.test/one.mp3".into()); + order.results = vec![ + clip("https://example.test/one.mp3"), + clip("https://example.test/two.mp3"), + ]; + let results = audio_results(&order).unwrap(); + assert_eq!(results.len(), 2); + assert_eq!(results[0].url, "https://example.test/one.mp3"); + assert_eq!(results[1].url, "https://example.test/two.mp3"); + } + + #[test] + fn supports_top_level_only_results_and_rejects_missing_results() { + let mut order = completed_order(); + assert!(audio_results(&order).is_err()); + order.result_url = Some("https://example.test/song.mp3".into()); + let results = audio_results(&order).unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].url, "https://example.test/song.mp3"); + order.result_url = Some(String::new()); + assert!(audio_results(&order).is_err()); + } + + fn clip(url: &str) -> MediaResult { + MediaResult { + url: url.into(), + maybe_width: None, + maybe_height: None, + } + } + + fn completed_order() -> OrderStatus { + OrderStatus { + order_id: "audio-order".into(), + task_status: TaskStatus::Completed, + result_url: None, + results: vec![], + fail_reason: None, + created_at: "2026-10-18T00:00:00Z".into(), + created_at_utc: None, + media_type: Some(OrderMediaType::Audio), + total_credits: None, + } + } +}