diff --git a/Cargo.lock b/Cargo.lock index 334f2afccf..12d54217d0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4710,7 +4710,10 @@ dependencies = [ "jobs_common", "log", "mysql_queries", + "pager", "reqwest 0.12.23", + "rootly_client", + "rootly_config", "serde", "serde_derive", "serde_json", diff --git a/crates/service/job/gmicloud_job/Cargo.toml b/crates/service/job/gmicloud_job/Cargo.toml index 4253d843d8..2820c5e613 100644 --- a/crates/service/job/gmicloud_job/Cargo.toml +++ b/crates/service/job/gmicloud_job/Cargo.toml @@ -28,6 +28,9 @@ gmicloud_client.workspace = true hashing = { path = "../../../lib/files/hashing" } jobs_common.workspace = true mysql_queries.workspace = true +pager = { path = "../../../lib/pager" } +rootly_client.workspace = true +rootly_config.workspace = true server_environment = { path = "../../../lib/server_environment" } shared_env_var_config.workspace = true tokens.workspace = true diff --git a/crates/service/job/gmicloud_job/src/alert_on_error.rs b/crates/service/job/gmicloud_job/src/alert_on_error.rs new file mode 100644 index 0000000000..9c7e8288a5 --- /dev/null +++ b/crates/service/job/gmicloud_job/src/alert_on_error.rs @@ -0,0 +1,35 @@ +use std::sync::Arc; +use log::error; +use mysql_queries::queries::generic_inference::gmicloud::list_pending_gmicloud_jobs::PendingGmiCloudJob; +use pager::client::pager::Pager; +use pager::notification::notification_details_builder::NotificationDetailsBuilder; +use pager::notification::notification_urgency::NotificationUrgency; + +/// Enqueue a pager alert for the error, then return it as `Err`. +pub fn alert_pager_and_return_err( + pager: &Pager, + title: &str, + error: Box, + job: Option<&PendingGmiCloudJob>, +) -> anyhow::Result { + let err_message = format!("{:#}", error); + + let mut builder = NotificationDetailsBuilder::from_boxed_error(error) + .set_title(title.to_string()) + .set_urgency(Some(NotificationUrgency::Medium)); + + if let Some(job) = job { + builder = builder + .set_inference_job_token(Some(job.job_token.to_string())) + .set_third_party_id(Some(job.request_id.clone())) + .set_user_token(job.maybe_creator_user_token.as_ref().map(|t| t.to_string())); + } + + let notification = builder.build(); + + if let Err(pager_err) = pager.enqueue_page(notification) { + error!("Failed to enqueue pager alert: {:?}", pager_err); + } + + Err(anyhow::anyhow!(err_message)) +} diff --git a/crates/service/job/gmicloud_job/src/job_dependencies.rs b/crates/service/job/gmicloud_job/src/job_dependencies.rs index f6d8979ac0..1f3cdfc656 100644 --- a/crates/service/job/gmicloud_job/src/job_dependencies.rs +++ b/crates/service/job/gmicloud_job/src/job_dependencies.rs @@ -2,6 +2,7 @@ use cloud_storage::bucket_client::BucketClient; use concurrency::relaxed_atomic_bool::RelaxedAtomicBool; use gmicloud_client::creds::gmicloud_api_key::GmiCloudApiKey; use jobs_common::job_stats::JobStats; +use pager::client::pager::Pager; use server_environment::ServerEnvironment; use sqlx::MySqlPool; @@ -16,10 +17,15 @@ pub struct JobDependencies { pub server_environment: ServerEnvironment, + pub pager: Pager, + pub job_stats: JobStats, - /// How long to sleep between poll iterations (milliseconds). - pub poll_interval_millis: u64, + /// How long to sleep after a successful poll iteration (milliseconds). + pub poll_interval_success_millis: u64, + + /// How long to sleep after a failed poll iteration (milliseconds). + pub poll_interval_failure_millis: u64, /// Set to `true` from another thread to trigger graceful shutdown. pub application_shutdown: RelaxedAtomicBool, diff --git a/crates/service/job/gmicloud_job/src/main.rs b/crates/service/job/gmicloud_job/src/main.rs index 3a22a6e64e..7b2a6b0123 100644 --- a/crates/service/job/gmicloud_job/src/main.rs +++ b/crates/service/job/gmicloud_job/src/main.rs @@ -28,11 +28,14 @@ use shared_env_var_config::mysql::env_get_mysql_connection_string_or_default; use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs}; use crate::main_loop::main_loop; use crate::job_dependencies::JobDependencies; +use crate::startup::build_pager::build_pager; +pub mod alert_on_error; pub mod http_server; pub mod job_dependencies; pub mod main_loop; pub mod process_job; +pub mod startup; // Bucket config const ENV_ACCESS_KEY: &str = "ACCESS_KEY"; @@ -99,15 +102,31 @@ async fn main() -> AnyhowResult<()> { let gmicloud_api_key_str = easyenv::get_env_string_required(ENV_GMICLOUD_API_KEY)?; let gmicloud_api_key = GmiCloudApiKey::new(gmicloud_api_key_str); - // How often to poll for results (default: 15 seconds) - let poll_interval_millis: u64 = easyenv::get_env_num( - "GMICLOUD_POLL_INTERVAL_MILLIS", + // How often to poll after a successful iteration (default: 3 seconds) + let poll_interval_success_millis: u64 = easyenv::get_env_num( + "GMICLOUD_POLL_INTERVAL_SUCCESS_MILLIS", + 3_000, + )?; + + // How often to poll after a failed iteration (default: 15 seconds) + let poll_interval_failure_millis: u64 = easyenv::get_env_num( + "GMICLOUD_POLL_INTERVAL_FAILURE_MILLIS", 15_000, )?; let application_shutdown = RelaxedAtomicBool::new(false); let job_stats = JobStats::new(); + // Pager setup + let (pager, pager_worker) = build_pager(server_environment, &container_environment.hostname); + + info!("Spawning pager worker."); + + std::thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().expect("pager worker tokio runtime"); + rt.block_on(pager_worker.run()); + }); + let create_server_args = CreateServerArgs { container_environment: container_environment.clone(), job_stats: job_stats.clone(), @@ -118,8 +137,10 @@ async fn main() -> AnyhowResult<()> { public_bucket_client, gmicloud_api_key, server_environment, + pager, job_stats, - poll_interval_millis, + poll_interval_success_millis, + poll_interval_failure_millis, application_shutdown: application_shutdown.clone(), }; diff --git a/crates/service/job/gmicloud_job/src/main_loop.rs b/crates/service/job/gmicloud_job/src/main_loop.rs index fb6616303b..e96621f5ad 100644 --- a/crates/service/job/gmicloud_job/src/main_loop.rs +++ b/crates/service/job/gmicloud_job/src/main_loop.rs @@ -12,12 +12,16 @@ pub async fn main_loop(job_dependencies: JobDependencies) { while !job_dependencies.application_shutdown.get() { let result = run_poll_iteration(&job_dependencies).await; - if let Err(err) = result { - error!("Error in poll iteration: {:?}", err); - let _ = job_dependencies.job_stats.increment_failure_count(); - } + let sleep_millis = match result { + Ok(_) => job_dependencies.poll_interval_success_millis, + Err(err) => { + error!("Error in poll iteration: {:?}", err); + let _ = job_dependencies.job_stats.increment_failure_count(); + job_dependencies.poll_interval_failure_millis + } + }; - tokio::time::sleep(Duration::from_millis(job_dependencies.poll_interval_millis)).await; + tokio::time::sleep(Duration::from_millis(sleep_millis)).await; } warn!("GmiCloud job runner main loop is shut down."); diff --git a/crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs b/crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs index 42a5547258..4f92a35245 100644 --- a/crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs +++ b/crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs @@ -1,9 +1,8 @@ -use log::{error, info, warn}; +use log::{error, warn}; use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; use mysql_queries::queries::generic_inference::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs}; use mysql_queries::queries::generic_inference::gmicloud::list_pending_gmicloud_jobs::PendingGmiCloudJob; -use mysql_queries::queries::wallets::refund::try_to_refund_ledger_entry::{try_to_refund_ledger_entry, WalletRefundOutcome}; use crate::job_dependencies::JobDependencies; @@ -12,72 +11,6 @@ pub async fn process_failed_job( job: &PendingGmiCloudJob, reason: &str, ) { - // --- Step 1: Attempt the refund before touching the job status. --- - // - // Refund first so a crash between refund and status update never leaves - // a failed job without a refund. If the refund fails, bail out early: - // the job stays pending and will be retried on the next poll cycle. - - match &job.maybe_wallet_ledger_entry_token { - None => { - warn!( - "Job {} has no wallet ledger entry token; skipping refund.", - job.job_token.as_str() - ); - } - Some(ledger_token) => { - let mut transaction = match deps.mysql_pool.begin().await { - Ok(tx) => tx, - Err(err) => { - error!( - "Failed to begin refund transaction for job {} (ledger {}): {:?}. \ - Job will NOT be marked failed yet and will be retried next poll.", - job.job_token.as_str(), ledger_token.as_str(), err - ); - return; - } - }; - - match try_to_refund_ledger_entry(ledger_token, &mut transaction).await { - Ok(WalletRefundOutcome::Refunded(summary)) => { - info!( - "Refunded {} credits for failed job {} (ledger {} -> refund ledger {}).", - summary.refund_amount, - job.job_token.as_str(), - ledger_token.as_str(), - summary.refund_ledger_entry_token.as_str(), - ); - if let Err(err) = transaction.commit().await { - error!( - "Failed to commit refund transaction for job {} (ledger {}): {:?}. \ - Job will NOT be marked failed yet and will be retried next poll.", - job.job_token.as_str(), ledger_token.as_str(), err - ); - return; - } - } - Ok(WalletRefundOutcome::AlreadyRefunded) => { - info!( - "Ledger entry {} for job {} was already refunded; proceeding to mark job failed.", - ledger_token.as_str(), job.job_token.as_str(), - ); - let _ = transaction.rollback().await; - } - Err(err) => { - error!( - "Failed to refund ledger entry {} for job {}: {:?}. \ - Job will NOT be marked failed yet and will be retried next poll.", - ledger_token.as_str(), job.job_token.as_str(), err, - ); - let _ = transaction.rollback().await; - return; - } - } - } - } - - // --- Step 2: Mark the job record as failed. --- - let reason_lower = reason.to_lowercase(); let platform_rules_violation = reason_lower.contains("violates") diff --git a/crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs b/crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs index 76573dbf14..7c19e5f97a 100644 --- a/crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs +++ b/crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs @@ -1,3 +1,4 @@ +use std::sync::Arc; use anyhow::anyhow; use log::{error, info, warn}; @@ -14,6 +15,7 @@ use mysql_queries::queries::media_files::create::insert_builder::media_file_inse use mysql_queries::queries::generic_inference::web::mark_generic_inference_job_successfully_done_by_token::mark_generic_inference_job_successfully_done_by_token; use tokens::tokens::media_files::MediaFileToken; +use crate::alert_on_error::alert_pager_and_return_err; use crate::job_dependencies::JobDependencies; const VIDEO_PREFIX: &str = "artcraft_"; @@ -47,20 +49,38 @@ pub async fn process_successful_job( None => None, }; - // --- Step 2: Download and upload the video. --- + // --- Step 2: Download the video. --- info!("Downloading video for job {} from: {}", job.job_token.as_str(), video_url); - let video_bytes: Vec = reqwest::get(video_url) - .await - .map_err(|err| anyhow!("reqwest error downloading video: {:?}", err))? - .bytes() - .await - .map_err(|err| anyhow!("error reading video bytes: {:?}", err))? - .to_vec(); + let video_bytes: Vec = match reqwest::get(video_url).await { + Ok(resp) => match resp.bytes().await { + Ok(bytes) => bytes.to_vec(), + Err(err) => { + error!("Error reading video bytes for job {}: {:?}", job.job_token.as_str(), err); + return alert_pager_and_return_err( + &deps.pager, + "GmiCloud video download failed", + err.into(), + Some(job), + ); + } + }, + Err(err) => { + error!("Error downloading video for job {}: {:?}", job.job_token.as_str(), err); + return alert_pager_and_return_err( + &deps.pager, + "GmiCloud video download failed", + err.into(), + Some(job), + ); + } + }; info!("Downloaded {} bytes for job {}", video_bytes.len(), job.job_token.as_str()); + // --- Step 3: Hash and upload. --- + let checksum = sha256_hash_bytes(&video_bytes) .map_err(|err| anyhow!("error hashing video: {:?}", err))?; @@ -69,15 +89,26 @@ pub async fn process_successful_job( info!("Uploading video to public bucket at path: {}", object_path); - deps + let upload_result = deps .public_bucket_client .upload_file_with_content_type_process(object_path, &video_bytes, "video/mp4") - .await - .map_err(|err| anyhow!("error uploading video to bucket: {:?}", err))?; + .await; + + if let Err(err) = upload_result { + error!("Error uploading video for job {}: {:?}", job.job_token.as_str(), err); + return alert_pager_and_return_err( + &deps.pager, + "GmiCloud bucket upload failed", + err.into(), + Some(job), + ); + } info!("Uploaded video for job {}. Creating media file record.", job.job_token.as_str()); - let media_file_token = MediaFileInsertBuilder::new() + // --- Step 4: Create media file record. --- + + let media_file_result = MediaFileInsertBuilder::new() .maybe_creator_user(job.maybe_creator_user_token.as_ref()) .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref()) .creator_ip_address(&job.creator_ip_address) @@ -93,27 +124,44 @@ pub async fn process_successful_job( .maybe_cover_image_media_file_token(maybe_cover_token.as_ref()) .public_bucket_directory_hash(&bucket_path) .insert_pool(&deps.mysql_pool) - .await - .map_err(|err| anyhow!("error inserting media file record: {:?}", err))?; + .await; + + let media_file_token = match media_file_result { + Ok(token) => token, + Err(err) => { + error!("Error inserting media file record for job {}: {:?}", job.job_token.as_str(), err); + return alert_pager_and_return_err( + &deps.pager, + "GmiCloud media file insert failed", + err.into(), + Some(job), + ); + } + }; info!( "Created media file {} for job {}. Marking job complete.", media_file_token.as_str(), job.job_token.as_str() ); - mark_generic_inference_job_successfully_done_by_token( + // --- Step 5: Mark job as done. --- + + if let Err(err) = mark_generic_inference_job_successfully_done_by_token( &deps.mysql_pool, &job.job_token, Some(InferenceResultType::MediaFile), Some(media_file_token.as_str()), None, None, - ) - .await - .map_err(|err| { - error!("Error marking job {} done: {:?}", job.job_token.as_str(), err); - anyhow!("error marking job done: {:?}", err) - })?; + ).await { + error!("Error marking job {} done: {:?}", job.job_token.as_str(), err); + return alert_pager_and_return_err( + &deps.pager, + "GmiCloud job completion update failed", + err.into(), + Some(job), + ); + } info!("Job {} completed successfully.", job.job_token.as_str()); diff --git a/crates/service/job/gmicloud_job/src/startup/build_pager.rs b/crates/service/job/gmicloud_job/src/startup/build_pager.rs new file mode 100644 index 0000000000..5a8d1865cf --- /dev/null +++ b/crates/service/job/gmicloud_job/src/startup/build_pager.rs @@ -0,0 +1,64 @@ +use log::{info, warn}; +use pager::client::pager::Pager; +use pager::client::pager_builder::PagerBuilder; +use pager::worker::pager_worker::PagerWorker; +use rootly_client::creds::rootly_api_key::RootlyApiKey; +use rootly_config::services::ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB; +use rootly_config::urgencies::{ROOTLY_URGENCY_ID_HIGH, ROOTLY_URGENCY_ID_LOW, ROOTLY_URGENCY_ID_MEDIUM}; +use shared_env_var_config::paging::{env_enable_paging_default_false, env_optional_rootly_api_key, env_optional_rootly_notification_target_id, env_optional_rootly_notification_target_type}; + +pub fn build_pager( + server_environment: server_environment::ServerEnvironment, + hostname: &str, +) -> (Pager, PagerWorker) { + let is_paging_enabled = env_enable_paging_default_false(); + + info!("Paging enabled: {}", is_paging_enabled); + + let environment = if server_environment.is_deployed_in_production() { + "production" + } else { + "development" + }; + + let builder = PagerBuilder::new() + .application_name("gmicloud-job".to_string()) + .environment(environment.to_string()) + .hostname(hostname.to_string()) + .service_id(ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB.to_string()); + + if !is_paging_enabled { + warn!("ENABLE_PAGING is false. Pager will be NoOp."); + return builder.build_with_worker(); + } + + let maybe_api_key = env_optional_rootly_api_key(); + + match maybe_api_key { + Some(api_key) => { + info!("Rootly API key found. Configuring pager with Rootly backend."); + build_rootly_pager(builder, api_key) + } + None => { + warn!("ROOTLY_API_KEY not set. Pager will not send real pages."); + builder.build_with_worker() + } + } +} + +fn build_rootly_pager(builder: PagerBuilder, api_key: String) -> (Pager, PagerWorker) { + let mut rootly_builder = builder + .rootly(RootlyApiKey::new(api_key)) + .urgency_id_high(ROOTLY_URGENCY_ID_HIGH.to_string()) + .urgency_id_medium(ROOTLY_URGENCY_ID_MEDIUM.to_string()) + .urgency_id_low(ROOTLY_URGENCY_ID_LOW.to_string()); + + let target_type = env_optional_rootly_notification_target_type(); + let target_id = env_optional_rootly_notification_target_id(); + + if let (Some(t_type), Some(t_id)) = (target_type, target_id) { + rootly_builder = rootly_builder.notification_target(t_type, t_id); + } + + rootly_builder.build_with_worker() +} diff --git a/crates/service/job/gmicloud_job/src/startup/mod.rs b/crates/service/job/gmicloud_job/src/startup/mod.rs new file mode 100644 index 0000000000..d074c6c0ce --- /dev/null +++ b/crates/service/job/gmicloud_job/src/startup/mod.rs @@ -0,0 +1 @@ +pub mod build_pager; diff --git a/crates/service/job/seedance2_pro_job/src/jobs/video_polling_job/process_job/process_successful_job.rs b/crates/service/job/seedance2_pro_job/src/jobs/video_polling_job/process_job/process_successful_job.rs index ec7edc70e0..516579b26c 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/video_polling_job/process_job/process_successful_job.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/video_polling_job/process_job/process_successful_job.rs @@ -94,11 +94,12 @@ pub async fn process_successful_job( ); // Upload to public bucket. - if let Err(err) = deps + let upload_result = deps .public_bucket_client .upload_file_with_content_type_process(object_path, &video_bytes, "video/mp4") - .await - { + .await; + + if let Err(err) = upload_result { error!("Error uploading video for order {}: {:?}", order.order_id, err); return alert_pager_and_return_err( &deps.pager, @@ -118,7 +119,7 @@ pub async fn process_successful_job( let maybe_frame_height = order.results.first().map(|r| r.height); // Insert media file record. - let media_file_token = match MediaFileInsertBuilder::new() + let media_file_result = MediaFileInsertBuilder::new() .maybe_creator_user(job.maybe_creator_user_token.as_ref()) .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref()) .creator_ip_address(&job.creator_ip_address) @@ -135,8 +136,9 @@ pub async fn process_successful_job( .maybe_prompt_token(job.maybe_prompt_token.as_ref()) .public_bucket_directory_hash(&bucket_path) .insert_pool(&deps.mysql_pool) - .await - { + .await; + + let media_file_token = match media_file_result { Ok(token) => token, Err(err) => { error!("Error inserting media file record for order {}: {:?}", order.order_id, err); diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/cost/video/omni_gen_video_cost_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/cost/video/omni_gen_video_cost_handler.rs index 9b065b2cf2..0221211912 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/cost/video/omni_gen_video_cost_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/cost/video/omni_gen_video_cost_handler.rs @@ -8,7 +8,7 @@ use artcraft_router::api::provider::Provider; use artcraft_router::errors::artcraft_router_error::ArtcraftRouterError; use artcraft_router::generate::generate_video::video_generation_cost_estimate::VideoGenerationCostEstimate; use log::warn; - +use artcraft_router::api::common_video_model::CommonVideoModel; use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError; use crate::http_server::endpoints::omni_gen::generate::video::helpers::hydrate_router_request::hydrate_to_router_request; use crate::state::server_state::ServerState; @@ -31,8 +31,16 @@ pub async fn omni_gen_video_cost_handler( _server_state: web::Data>, ) -> Result, AdvancedCommonWebError> { let mut builder = hydrate_to_router_request(&request)?; + builder.provider = Provider::Artcraft; + // We bill global Seedance 2.0 at the same rate (for now) as regular Seedance 2.0 + match builder.model { + CommonVideoModel::Seedance2p0Global => builder.model = CommonVideoModel::Seedance2p0, + CommonVideoModel::Seedance2p0FastGlobal => builder.model = CommonVideoModel::Seedance2p0Fast, + _ => {}, + } + let estimate = if builder.use_new_builder() { builder.build2() .map_err(|e| { diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/mod.rs index 903ea1091b..7108f22a3b 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/mod.rs @@ -3,3 +3,4 @@ pub mod build_router_client; pub(crate) mod hydrate_router_request; pub mod pipeline_result; pub mod resolve_kinovi_character_ids; +pub mod resolve_media_tokens_to_urls; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/resolve_media_tokens_to_urls.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/resolve_media_tokens_to_urls.rs new file mode 100644 index 0000000000..8ee7d7f175 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/resolve_media_tokens_to_urls.rs @@ -0,0 +1,64 @@ +use std::collections::HashMap; + +use artcraft_router::api::audio_list_ref::AudioListRef; +use artcraft_router::api::image_list_ref::ImageListRef; +use artcraft_router::api::image_ref::ImageRef; +use artcraft_router::api::video_list_ref::VideoListRef; +use artcraft_router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder; +use tokens::tokens::media_files::MediaFileToken; + +/// For providers that take URLs directly (GmiCloud, Fal), swap +/// `ImageRef::MediaFileToken` → `ImageRef::Url` using the resolved map. +pub fn resolve_media_tokens_to_urls( + builder: &mut GenerateVideoRequestBuilder, + url_map: Option<&HashMap>, +) { + let map = match url_map { + Some(m) => m, + None => return, + }; + + // start_frame + if let Some(ImageRef::MediaFileToken(ref token)) = builder.start_frame { + if let Some(url) = map.get(token) { + builder.start_frame = Some(ImageRef::Url(url.clone())); + } + } + + // end_frame + if let Some(ImageRef::MediaFileToken(ref token)) = builder.end_frame { + if let Some(url) = map.get(token) { + builder.end_frame = Some(ImageRef::Url(url.clone())); + } + } + + // reference_images + if let Some(ImageListRef::MediaFileTokens(ref tokens)) = builder.reference_images { + let urls: Vec = tokens.iter() + .filter_map(|t| map.get(t).cloned()) + .collect(); + if !urls.is_empty() { + builder.reference_images = Some(ImageListRef::Urls(urls)); + } + } + + // reference_videos + if let Some(VideoListRef::MediaFileTokens(ref tokens)) = builder.reference_videos { + let urls: Vec = tokens.iter() + .filter_map(|t| map.get(t).cloned()) + .collect(); + if !urls.is_empty() { + builder.reference_videos = Some(VideoListRef::Urls(urls)); + } + } + + // reference_audio + if let Some(AudioListRef::MediaFileTokens(ref tokens)) = builder.reference_audio { + let urls: Vec = tokens.iter() + .filter_map(|t| map.get(t).cloned()) + .collect(); + if !urls.is_empty() { + builder.reference_audio = Some(AudioListRef::Urls(urls)); + } + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/run_pipeline_v2.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/run_pipeline_v2.rs index fc70046617..63cae94d5e 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/run_pipeline_v2.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/run_pipeline_v2.rs @@ -2,12 +2,8 @@ use std::collections::HashMap; use log::{error, info, warn}; use sqlx::pool::PoolConnection; -use artcraft_router::api::audio_list_ref::AudioListRef; use artcraft_router::api::common_video_model::CommonVideoModel; -use artcraft_router::api::image_list_ref::ImageListRef; -use artcraft_router::api::image_ref::ImageRef; use artcraft_router::api::provider::Provider; -use artcraft_router::api::video_list_ref::VideoListRef; use artcraft_router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder; use artcraft_router::generate::generate_video::generate_video_response::GenerateVideoResponse; use artcraft_router::generate::generate_video_v2::video_generation_draft_context::VideoGenerationDraftContext; @@ -21,6 +17,7 @@ use crate::http_server::endpoint_helpers::refund_wallet_after_api_failure::refun use crate::http_server::endpoints::omni_gen::generate::video::helpers::bill_wallet::bill_wallet; use crate::http_server::endpoints::omni_gen::generate::video::helpers::build_router_client::build_router_client; use crate::http_server::endpoints::omni_gen::generate::video::helpers::pipeline_result::PipelineResult; +use crate::http_server::endpoints::omni_gen::generate::video::helpers::resolve_media_tokens_to_urls::resolve_media_tokens_to_urls; use crate::state::server_state::ServerState; pub struct RunPipelineV2Args<'a> { @@ -70,18 +67,19 @@ pub async fn run_pipeline_v2(args: RunPipelineV2Args<'_>) -> Result cost_builder.model = CommonVideoModel::Seedance2p0, + CommonVideoModel::Seedance2p0FastGlobal => cost_builder.model = CommonVideoModel::Seedance2p0Fast, + _ => {}, + } + } + cost_builder.build2() .map_err(|e| { warn!("Failed to build2 cost estimate for v2: {}", e); @@ -170,59 +168,3 @@ async fn upload_and_generate( AdvancedCommonWebError::from_error(err) }) } - -/// For providers that take URLs directly (GmiCloud, Fal), swap -/// `ImageRef::MediaFileToken` → `ImageRef::Url` using the resolved map. -fn resolve_media_tokens_to_urls( - builder: &mut GenerateVideoRequestBuilder, - url_map: Option<&HashMap>, -) { - let map = match url_map { - Some(m) => m, - None => return, - }; - - // start_frame - if let Some(ImageRef::MediaFileToken(ref token)) = builder.start_frame { - if let Some(url) = map.get(token) { - builder.start_frame = Some(ImageRef::Url(url.clone())); - } - } - - // end_frame - if let Some(ImageRef::MediaFileToken(ref token)) = builder.end_frame { - if let Some(url) = map.get(token) { - builder.end_frame = Some(ImageRef::Url(url.clone())); - } - } - - // reference_images - if let Some(ImageListRef::MediaFileTokens(ref tokens)) = builder.reference_images { - let urls: Vec = tokens.iter() - .filter_map(|t| map.get(t).cloned()) - .collect(); - if !urls.is_empty() { - builder.reference_images = Some(ImageListRef::Urls(urls)); - } - } - - // reference_videos - if let Some(VideoListRef::MediaFileTokens(ref tokens)) = builder.reference_videos { - let urls: Vec = tokens.iter() - .filter_map(|t| map.get(t).cloned()) - .collect(); - if !urls.is_empty() { - builder.reference_videos = Some(VideoListRef::Urls(urls)); - } - } - - // reference_audio - if let Some(AudioListRef::MediaFileTokens(ref tokens)) = builder.reference_audio { - let urls: Vec = tokens.iter() - .filter_map(|t| map.get(t).cloned()) - .collect(); - if !urls.is_empty() { - builder.reference_audio = Some(AudioListRef::Urls(urls)); - } - } -}