diff --git a/crates/api_clients/artcraft/artcraft_router/src/generate/generate_video_v2/mod.rs b/crates/api_clients/artcraft/artcraft_router/src/generate/generate_video_v2/mod.rs index dcd09c1643..d862928fbe 100644 --- a/crates/api_clients/artcraft/artcraft_router/src/generate/generate_video_v2/mod.rs +++ b/crates/api_clients/artcraft/artcraft_router/src/generate/generate_video_v2/mod.rs @@ -2,4 +2,4 @@ pub mod providers; pub mod video_generation_request; pub mod video_generation_draft; pub mod video_generation_draft_or_request; -mod video_generation_draft_context; \ No newline at end of file +pub mod video_generation_draft_context; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/bill_wallet.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/bill_wallet.rs new file mode 100644 index 0000000000..61946ec2a7 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/bill_wallet.rs @@ -0,0 +1,44 @@ +//! Shared wallet billing logic for both video generation pipelines. + +use log::info; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use tokens::tokens::users::UserToken; +use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken; + +use crate::billing::wallets::attempt_wallet_deduction::attempt_wallet_deduction_else_common_web_error; +use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError; + +pub struct BillWalletResult { + pub apriori_job_token: InferenceJobToken, + pub maybe_wallet_ledger_entry_token: Option, +} + +/// Generate an apriori job token and bill the user's wallet for the given cost. +/// +/// If cost is 0, no wallet deduction is made but the apriori token is still generated. +pub async fn bill_wallet( + user_token: &UserToken, + cost: u64, + mysql_connection: &mut sqlx::pool::PoolConnection, +) -> Result { + let apriori_job_token = InferenceJobToken::generate(); + + info!("Charging wallet: {} credits", cost); + + let maybe_wallet_ledger_entry_token = if cost > 0 { + let deduction_result = attempt_wallet_deduction_else_common_web_error( + user_token, + Some(apriori_job_token.as_str()), + cost, + mysql_connection, + ).await?; + Some(deduction_result.ledger_entry_token) + } else { + None + }; + + Ok(BillWalletResult { + apriori_job_token, + maybe_wallet_ledger_entry_token, + }) +} 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 new file mode 100644 index 0000000000..e343232fde --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/helpers/mod.rs @@ -0,0 +1 @@ +pub mod bill_wallet; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/mod.rs index 4cf243587a..b69ae9ed49 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/mod.rs @@ -1,7 +1,8 @@ pub mod omni_gen_video_generate_handler; +mod helpers; pub mod pipeline_v1; +pub mod pipeline_v2; mod request_helper; #[cfg(test)] mod tests; -pub mod pipeline_v2; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/omni_gen_video_generate_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/omni_gen_video_generate_handler.rs index 156ea77d47..d64f8393c5 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/omni_gen_video_generate_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/omni_gen_video_generate_handler.rs @@ -10,6 +10,7 @@ use url::Url; use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_video_cost_and_generate_request::OmniGenVideoCostAndGenerateRequest; use artcraft_api_defs::omni_gen::generate_response::omni_gen_video_generate_response::OmniGenVideoGenerateResponse; use artcraft_router::api::provider::Provider; +use artcraft_router::generate::generate_video::generate_video_response::GenerateVideoResponse; use enums::by_table::prompt_context_items::prompt_context_semantic_type::PromptContextSemanticType; use enums::by_table::prompts::prompt_type::PromptType; use enums::common::generation::common_model_type::CommonModelType; @@ -31,14 +32,17 @@ use mysql_queries::queries::prompt_context_items::insert_batch_prompt_context_it insert_batch_prompt_context_items, InsertBatchArgs, PromptContextItem, }; use mysql_queries::queries::prompts::insert_prompt::{insert_prompt, InsertPromptArgs}; +use tokens::tokens::characters::CharacterToken; use tokens::tokens::generic_inference_jobs::InferenceJobToken; use tokens::tokens::media_files::MediaFileToken; -use crate::billing::wallets::attempt_wallet_deduction::attempt_wallet_deduction_else_common_web_error; use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError; use crate::http_server::endpoints::generate::common::payments_error_test::payments_error_test; +use crate::http_server::endpoints::omni_gen::generate::video::helpers::bill_wallet::bill_wallet; +use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v1::distill_helper::hydrate_to_router_request::hydrate_to_router_request; use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v1::distill_video_request::distill_video_request; use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v1::execute::execute_generation::execute_generation; +use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v2::execute::execute_pipeline_v2; use crate::http_server::endpoints::omni_gen::generate::video::request_helper::resolve_kinovi_character_ids::resolve_kinovi_character_ids; use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format; use crate::state::server_state::ServerState; @@ -63,7 +67,7 @@ pub async fn omni_gen_video_generate_handler( request: Json, server_state: web::Data>, ) -> Result, AdvancedCommonWebError> { - + info!("request: {:?}", request); payments_error_test(&request.prompt.as_deref().unwrap_or(""))?; @@ -112,8 +116,6 @@ pub async fn omni_gen_video_generate_handler( })?; // ==================== RESOLVE MEDIA TOKENS ==================== // - // Look up media file tokens BEFORE distilling. distill_video_request takes - // a pre-computed `MediaFileToken -> Url` map and does no I/O of its own. let media_file_hydration_map: Option> = { let mut all_tokens: Vec = Vec::new(); @@ -157,236 +159,245 @@ pub async fn omni_gen_video_generate_handler( } }; - // ==================== RESOLVE CHARACTER TOKENS ==================== // + // ==================== PIPELINE DECISION ==================== // + let use_v2 = matches!( + request.model, + Some(CommonVideoModel::Seedance2p0) | Some(CommonVideoModel::Seedance2p0Fast) + ); + + if use_v2 { + execute_v2_pipeline( + &http_request, + &request, + &server_state, + &mut mysql_connection, + user_token, + maybe_avt_token.as_ref(), + &idempotency_token, + maybe_prompt_model_type, + media_file_hydration_map, + ).await + } else { + execute_v1_pipeline( + &http_request, + &request, + &server_state, + &mut mysql_connection, + user_token, + maybe_avt_token.as_ref(), + &idempotency_token, + maybe_prompt_model_type, + media_file_hydration_map, + ).await + } +} + +// ==================== PIPELINE V1 (existing) ==================== // + +#[allow(clippy::too_many_arguments)] +async fn execute_v1_pipeline( + http_request: &HttpRequest, + request: &OmniGenVideoCostAndGenerateRequest, + server_state: &ServerState, + mysql_connection: &mut sqlx::pool::PoolConnection, + user_token: &tokens::tokens::users::UserToken, + maybe_avt_token: Option<&tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken>, + idempotency_token: &str, + maybe_prompt_model_type: Option, + media_file_hydration_map: Option>, +) -> Result, AdvancedCommonWebError> { let kinovi_character_ids = resolve_kinovi_character_ids( request.reference_character_tokens.as_deref(), - &mut mysql_connection, + mysql_connection, ).await?; - // ==================== DETERMINE PROVIDER ==================== // - let execution_provider = match request.model { Some(CommonVideoModel::Seedance2p0) => Provider::Seedance2Pro, Some(CommonVideoModel::Seedance2p0Fast) => Provider::Seedance2Pro, _ => Provider::Fal, }; - // ==================== DISTILL ==================== // + let distilled = distill_video_request(request, media_file_hydration_map.as_ref(), execution_provider)?; - let distilled = distill_video_request(&request, media_file_hydration_map.as_ref(), execution_provider)?; - - info!("distilled plan: {:?}", distilled.plan); - - // ==================== BILLING ==================== // + info!("v1 distilled plan: {:?}", distilled.plan); let cost = distilled.cost.cost_in_credits.unwrap_or(0); - - info!("Charging wallet: {} credits", cost); - - let apriori_job_token = InferenceJobToken::generate(); - - let maybe_deduction_result = if cost > 0 { - Some(attempt_wallet_deduction_else_common_web_error( - user_token, - Some(apriori_job_token.as_str()), - cost, - &mut mysql_connection, - ).await?) - } else { - None - }; - - // ==================== EXECUTE GENERATION ==================== // + let billing = bill_wallet(user_token, cost, mysql_connection).await?; let gen_result = execute_generation( &distilled, - &request, - &server_state, + request, + server_state, media_file_hydration_map.as_ref(), kinovi_character_ids, - maybe_deduction_result.as_ref().map(|d| &d.ledger_entry_token), - &mut mysql_connection, + billing.maybe_wallet_ledger_entry_token.as_ref(), + mysql_connection, ).await?; - // ==================== DB TRANSACTION ==================== // + // -- DB writes -- + let ip_address = get_request_ip(http_request); + let mut transaction = mysql_connection.begin().await.map_err(|err| { + error!("Error starting MySQL transaction: {:?}", err); + AdvancedCommonWebError::from_error(err) + })?; - let ip_address = get_request_ip(&http_request); - - let mut transaction = mysql_connection - .begin() - .await - .map_err(|err| { - error!("Error starting MySQL transaction: {:?}", err); - AdvancedCommonWebError::from_error(err) - })?; - - // -- Prompt -- - - let prompt_result = insert_prompt(InsertPromptArgs { - maybe_apriori_prompt_token: None, - prompt_type: PromptType::ArtcraftApp, - maybe_creator_user_token: Some(user_token), - maybe_model_type: maybe_prompt_model_type, - maybe_generation_provider: Some(GenerationProvider::Artcraft), - maybe_positive_prompt: request.prompt.as_deref(), - maybe_negative_prompt: request.negative_prompt.as_deref(), - maybe_other_args: None, - maybe_generation_mode: Some(gen_result.generation_mode), - maybe_aspect_ratio: None, - maybe_resolution: None, - maybe_batch_count: request.video_batch_count.map(|c| c as u8), - maybe_generate_audio: request.generate_audio, - maybe_duration_seconds: request.duration_seconds.map(|d| d as u32), - creator_ip_address: &ip_address, - mysql_executor: &mut *transaction, - phantom: Default::default(), - }).await; - - let prompt_token = match prompt_result { - Ok(token) => Some(token), - Err(err) => { - warn!("Error inserting prompt: {:?}", err); - None - } - }; - - // -- Prompt context items -- - - if let Some(token) = prompt_token.as_ref() { - let mut context_items = Vec::new(); - - if let Some(media_token) = &request.start_frame_image_media_token { - context_items.push(PromptContextItem { - media_token: media_token.clone(), - context_semantic_type: PromptContextSemanticType::VidStartFrame, - }); - } - - if let Some(media_token) = &request.end_frame_image_media_token { - context_items.push(PromptContextItem { - media_token: media_token.clone(), - context_semantic_type: PromptContextSemanticType::VidEndFrame, - }); - } - - if let Some(ref_tokens) = &request.reference_image_media_tokens { - for media_token in ref_tokens { - context_items.push(PromptContextItem { - media_token: media_token.clone(), - context_semantic_type: PromptContextSemanticType::Imgref, - }); - } - } - - if let Some(ref_tokens) = &request.reference_video_media_tokens { - for media_token in ref_tokens { - context_items.push(PromptContextItem { - media_token: media_token.clone(), - context_semantic_type: PromptContextSemanticType::VidRef, - }); - } - } - - if !context_items.is_empty() { - if let Err(err) = insert_batch_prompt_context_items(InsertBatchArgs { - prompt_token: token.clone(), - items: context_items, - transaction: &mut transaction, - }).await { - warn!("Error inserting batch prompt context items: {:?}", err); - } - } - } - - // -- Inference job -- - - let maybe_ledger_entry_token = maybe_deduction_result.as_ref() - .map(|d| &d.ledger_entry_token); + let prompt_token = insert_prompt_and_context( + request, &mut transaction, user_token, maybe_prompt_model_type, + Some(gen_result.generation_mode), &ip_address, + ).await; let job_token = if gen_result.is_seedance2pro { - // Seedance2Pro path: insert one job per order_id (batch support). - let fallback_ids = vec![gen_result.external_job_id.clone()]; - let order_ids = gen_result.maybe_seedance_order_ids - .as_deref() - .unwrap_or(&fallback_ids); - - let mut all_job_tokens: Vec = Vec::with_capacity(order_ids.len()); - - for (i, order_id) in order_ids.iter().enumerate() { - let job_token = if i == 0 { - apriori_job_token.clone() - } else { - InferenceJobToken::generate() - }; - - let idempotency_str = if i == 0 { - idempotency_token.clone() - } else { - format!("{}-batch-{}", idempotency_token, i) - }; - - let db_result = insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token( - InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs { - apriori_job_token: &job_token, - uuid_idempotency_token: &idempotency_str, - maybe_external_third_party_id: order_id, - maybe_inference_args: None, - maybe_prompt_token: prompt_token.as_ref(), - maybe_wallet_ledger_entry_token: maybe_ledger_entry_token, - maybe_creator_user_token: Some(user_token), - maybe_avt_token: maybe_avt_token.as_ref(), - creator_ip_address: &ip_address, - creator_set_visibility: Visibility::Public, - mysql_executor: &mut *transaction, - phantom: Default::default(), - } - ).await; - - match db_result { - Ok(token) => all_job_tokens.push(token), - Err(err) => { - warn!("Error inserting seedance2pro inference job (order_id={}): {:?}", order_id, err); - if i == 0 { - return Err(AdvancedCommonWebError::from_error(err)); - } - } - } - } - - all_job_tokens.first().cloned().ok_or_else(|| { - error!("No inference job token was created"); - AdvancedCommonWebError::server_error_with_message("No inference job token was created") - })? + insert_seedance2pro_jobs( + &gen_result.external_job_id, + gen_result.maybe_seedance_order_ids.as_deref(), + &billing.apriori_job_token, + idempotency_token, + prompt_token.as_ref(), + billing.maybe_wallet_ledger_entry_token.as_ref(), + user_token, + maybe_avt_token, + &ip_address, + &mut transaction, + ).await? } else { - // Fal / other providers path. - // TODO: Pass maybe_wallet_ledger_entry_token to fal jobs once the fal insert supports it. - let db_result = insert_generic_inference_job_for_fal_queue_with_apriori_job_token( - InsertGenericInferenceForFalWithAprioriJobTokenArgs { - apriori_job_token: &apriori_job_token, - uuid_idempotency_token: &idempotency_token, - maybe_external_third_party_id: &gen_result.external_job_id, - fal_category: FalCategory::VideoGeneration, - maybe_inference_args: None, - maybe_prompt_token: prompt_token.as_ref(), - maybe_creator_user_token: Some(user_token), - maybe_avt_token: maybe_avt_token.as_ref(), - creator_ip_address: &ip_address, - creator_set_visibility: Visibility::Public, - mysql_executor: &mut *transaction, - starting_job_status_override: None, - maybe_frontend_failure_category: None, - maybe_failure_reason: None, - phantom: Default::default(), - } - ).await; + insert_fal_job( + &gen_result.external_job_id, + &billing.apriori_job_token, + idempotency_token, + prompt_token.as_ref(), + user_token, + maybe_avt_token, + &ip_address, + &mut transaction, + ).await? + }; - match db_result { - Ok(token) => token, - Err(err) => { - warn!("Error inserting inference job: {:?}", err); - return Err(AdvancedCommonWebError::from_error(err)); - } + transaction.commit().await.map_err(|err| { + error!("Error committing transaction: {:?}", err); + AdvancedCommonWebError::from_error(err) + })?; + + Ok(Json(OmniGenVideoGenerateResponse { + success: true, + inference_job_token: job_token, + })) +} + +// ==================== PIPELINE V2 (new) ==================== // + +#[allow(clippy::too_many_arguments)] +async fn execute_v2_pipeline( + http_request: &HttpRequest, + request: &OmniGenVideoCostAndGenerateRequest, + server_state: &ServerState, + mysql_connection: &mut sqlx::pool::PoolConnection, + user_token: &tokens::tokens::users::UserToken, + maybe_avt_token: Option<&tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken>, + idempotency_token: &str, + maybe_prompt_model_type: Option, + media_file_hydration_map: Option>, +) -> Result, AdvancedCommonWebError> { + // 1. Build execution request (provider = Seedance2Pro for Kinovi execution) + let mut exec_builder = hydrate_to_router_request(request)?; + exec_builder.provider = Provider::Seedance2Pro; + + let draft_or_request = exec_builder.build2().map_err(|e| { + warn!("Failed to build2 for v2 pipeline: {}", e); + AdvancedCommonWebError::from_error(e) + })?; + + // 2. Calculate cost (swap provider to Artcraft for billing) + let mut cost_builder = hydrate_to_router_request(request)?; + cost_builder.provider = Provider::Artcraft; + + let cost_estimate = cost_builder.build2() + .map_err(|e| { + warn!("Failed to build2 cost estimate for v2 pipeline: {}", e); + AdvancedCommonWebError::from_error(e) + })? + .estimate_cost() + .map_err(|e| { + warn!("Failed to estimate cost for v2 pipeline: {}", e); + AdvancedCommonWebError::from_error(e) + })?; + + let cost = cost_estimate.cost_in_credits.unwrap_or(0); + info!("v2 estimated cost: {} credits", cost); + + // 3. Bill wallet + let billing = bill_wallet(user_token, cost, mysql_connection).await?; + + // 4. Build context maps for draft finalization + let media_file_urls_as_strings: Option> = + media_file_hydration_map.as_ref().map(|map| { + map.iter().map(|(k, v)| (k.clone(), v.to_string())).collect() + }); + + let kinovi_character_id_map: Option> = + resolve_kinovi_character_id_map( + request.reference_character_tokens.as_deref(), + mysql_connection, + ).await?; + + // 5. Execute: finalize draft (upload to Kinovi if needed) → send + let gen_response = execute_pipeline_v2( + draft_or_request, + server_state, + media_file_urls_as_strings.as_ref(), + kinovi_character_id_map.as_ref(), + billing.maybe_wallet_ledger_entry_token.as_ref(), + mysql_connection, + ).await?; + + // 6. DB writes + let ip_address = get_request_ip(http_request); + let mut transaction = mysql_connection.begin().await.map_err(|err| { + error!("Error starting MySQL transaction: {:?}", err); + AdvancedCommonWebError::from_error(err) + })?; + + let prompt_token = insert_prompt_and_context( + request, &mut transaction, user_token, maybe_prompt_model_type, + None, // TODO: determine generation_mode from v2 response + &ip_address, + ).await; + + let job_token = match gen_response { + GenerateVideoResponse::Seedance2Pro(ref payload) => { + insert_seedance2pro_jobs( + &payload.order_id, + None, // v2 Kinovi returns single order_id per request + &billing.apriori_job_token, + idempotency_token, + prompt_token.as_ref(), + billing.maybe_wallet_ledger_entry_token.as_ref(), + user_token, + maybe_avt_token, + &ip_address, + &mut transaction, + ).await? + } + GenerateVideoResponse::Artcraft(ref payload) => { + // Artcraft recursive: job was already created by the inner omni call + payload.inference_job_token.clone() + } + GenerateVideoResponse::Fal(ref payload) => { + let external_id = payload.request_id.clone().unwrap_or_default(); + insert_fal_job( + &external_id, + &billing.apriori_job_token, + idempotency_token, + prompt_token.as_ref(), + user_token, + maybe_avt_token, + &ip_address, + &mut transaction, + ).await? + } + _ => { + error!("Unexpected v2 response variant: {:?}", gen_response); + return Err(AdvancedCommonWebError::server_error_with_message("Unexpected v2 response")); } }; @@ -400,3 +411,220 @@ pub async fn omni_gen_video_generate_handler( inference_job_token: job_token, })) } + +// ==================== SHARED HELPERS ==================== // + +/// Insert prompt and prompt context items. Returns the prompt token if successful. +async fn insert_prompt_and_context( + request: &OmniGenVideoCostAndGenerateRequest, + transaction: &mut sqlx::Transaction<'_, sqlx::MySql>, + user_token: &tokens::tokens::users::UserToken, + maybe_prompt_model_type: Option, + maybe_generation_mode: Option, + ip_address: &str, +) -> Option { + let prompt_result = insert_prompt(InsertPromptArgs { + maybe_apriori_prompt_token: None, + prompt_type: PromptType::ArtcraftApp, + maybe_creator_user_token: Some(user_token), + maybe_model_type: maybe_prompt_model_type, + maybe_generation_provider: Some(GenerationProvider::Artcraft), + maybe_positive_prompt: request.prompt.as_deref(), + maybe_negative_prompt: request.negative_prompt.as_deref(), + maybe_other_args: None, + maybe_generation_mode, + maybe_aspect_ratio: None, + maybe_resolution: None, + maybe_batch_count: request.video_batch_count.map(|c| c as u8), + maybe_generate_audio: request.generate_audio, + maybe_duration_seconds: request.duration_seconds.map(|d| d as u32), + creator_ip_address: ip_address, + mysql_executor: &mut **transaction, + phantom: Default::default(), + }).await; + + let prompt_token = match prompt_result { + Ok(token) => Some(token), + Err(err) => { + warn!("Error inserting prompt: {:?}", err); + None + } + }; + + if let Some(token) = prompt_token.as_ref() { + let mut context_items = Vec::new(); + + if let Some(media_token) = &request.start_frame_image_media_token { + context_items.push(PromptContextItem { + media_token: media_token.clone(), + context_semantic_type: PromptContextSemanticType::VidStartFrame, + }); + } + if let Some(media_token) = &request.end_frame_image_media_token { + context_items.push(PromptContextItem { + media_token: media_token.clone(), + context_semantic_type: PromptContextSemanticType::VidEndFrame, + }); + } + if let Some(ref_tokens) = &request.reference_image_media_tokens { + for media_token in ref_tokens { + context_items.push(PromptContextItem { + media_token: media_token.clone(), + context_semantic_type: PromptContextSemanticType::Imgref, + }); + } + } + if let Some(ref_tokens) = &request.reference_video_media_tokens { + for media_token in ref_tokens { + context_items.push(PromptContextItem { + media_token: media_token.clone(), + context_semantic_type: PromptContextSemanticType::VidRef, + }); + } + } + + if !context_items.is_empty() { + if let Err(err) = insert_batch_prompt_context_items(InsertBatchArgs { + prompt_token: token.clone(), + items: context_items, + transaction, + }).await { + warn!("Error inserting batch prompt context items: {:?}", err); + } + } + } + + prompt_token +} + +#[allow(clippy::too_many_arguments)] +async fn insert_seedance2pro_jobs( + primary_order_id: &str, + maybe_additional_order_ids: Option<&[String]>, + apriori_job_token: &InferenceJobToken, + idempotency_token: &str, + prompt_token: Option<&tokens::tokens::prompts::PromptToken>, + maybe_wallet_ledger_entry_token: Option<&tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken>, + user_token: &tokens::tokens::users::UserToken, + maybe_avt_token: Option<&tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken>, + ip_address: &str, + transaction: &mut sqlx::Transaction<'_, sqlx::MySql>, +) -> Result { + let fallback_ids = vec![primary_order_id.to_string()]; + let order_ids = maybe_additional_order_ids.unwrap_or(&fallback_ids); + + let mut all_job_tokens: Vec = Vec::with_capacity(order_ids.len()); + + for (i, order_id) in order_ids.iter().enumerate() { + let job_token = if i == 0 { + apriori_job_token.clone() + } else { + InferenceJobToken::generate() + }; + + let idempotency_str = if i == 0 { + idempotency_token.to_string() + } else { + format!("{}-batch-{}", idempotency_token, i) + }; + + let db_result = insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token( + InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs { + apriori_job_token: &job_token, + uuid_idempotency_token: &idempotency_str, + maybe_external_third_party_id: order_id, + maybe_inference_args: None, + maybe_prompt_token: prompt_token, + maybe_wallet_ledger_entry_token, + maybe_creator_user_token: Some(user_token), + maybe_avt_token, + creator_ip_address: ip_address, + creator_set_visibility: Visibility::Public, + mysql_executor: &mut **transaction, + phantom: Default::default(), + } + ).await; + + match db_result { + Ok(token) => all_job_tokens.push(token), + Err(err) => { + warn!("Error inserting seedance2pro inference job (order_id={}): {:?}", order_id, err); + if i == 0 { + return Err(AdvancedCommonWebError::from_error(err)); + } + } + } + } + + all_job_tokens.first().cloned().ok_or_else(|| { + error!("No inference job token was created"); + AdvancedCommonWebError::server_error_with_message("No inference job token was created") + }) +} + +#[allow(clippy::too_many_arguments)] +async fn insert_fal_job( + external_job_id: &str, + apriori_job_token: &InferenceJobToken, + idempotency_token: &str, + prompt_token: Option<&tokens::tokens::prompts::PromptToken>, + user_token: &tokens::tokens::users::UserToken, + maybe_avt_token: Option<&tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken>, + ip_address: &str, + transaction: &mut sqlx::Transaction<'_, sqlx::MySql>, +) -> Result { + // TODO: Pass maybe_wallet_ledger_entry_token to fal jobs once the fal insert supports it. + let db_result = insert_generic_inference_job_for_fal_queue_with_apriori_job_token( + InsertGenericInferenceForFalWithAprioriJobTokenArgs { + apriori_job_token, + uuid_idempotency_token: idempotency_token, + maybe_external_third_party_id: external_job_id, + fal_category: FalCategory::VideoGeneration, + maybe_inference_args: None, + maybe_prompt_token: prompt_token, + maybe_creator_user_token: Some(user_token), + maybe_avt_token, + creator_ip_address: ip_address, + creator_set_visibility: Visibility::Public, + mysql_executor: &mut **transaction, + starting_job_status_override: None, + maybe_frontend_failure_category: None, + maybe_failure_reason: None, + phantom: Default::default(), + } + ).await; + + match db_result { + Ok(token) => Ok(token), + Err(err) => { + warn!("Error inserting fal inference job: {:?}", err); + Err(AdvancedCommonWebError::from_error(err)) + } + } +} + +/// Resolve character tokens to a map of CharacterToken → Kinovi character ID. +/// Used by pipeline v2's draft context. +async fn resolve_kinovi_character_id_map( + maybe_tokens: Option<&[CharacterToken]>, + connection: &mut sqlx::pool::PoolConnection, +) -> Result>, AdvancedCommonWebError> { + use mysql_queries::queries::characters::batch_lookup_characters_by_token_for_prompting::batch_lookup_characters_by_token_for_prompting; + + let tokens = match maybe_tokens { + None => return Ok(None), + Some(tokens) if tokens.is_empty() => return Ok(None), + Some(tokens) => tokens, + }; + + let characters = batch_lookup_characters_by_token_for_prompting(tokens, connection).await?; + + let map: HashMap = characters.iter() + .filter(|c| c.is_active) + .filter_map(|c| { + c.kinovi_character_id.as_ref().map(|kid| (c.token.clone(), kid.clone())) + }) + .collect(); + + if map.is_empty() { Ok(None) } else { Ok(Some(map)) } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/execute.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/execute.rs new file mode 100644 index 0000000000..729fbad4f8 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/execute.rs @@ -0,0 +1,118 @@ +//! Pipeline v2 execution: build2 → (optional finalize) → send_request. +//! +//! For Kinovi providers, the build2 output is a Draft that needs finalization +//! (media upload to Kinovi CDN). For Artcraft providers, build2 returns a +//! Request directly — no finalization needed. + +use std::collections::HashMap; + +use log::{error, info, warn}; + +use artcraft_router::client::router_client::RouterClient; +use artcraft_router::client::router_seedance2pro_client::RouterSeedance2ProClient; +use artcraft_router::generate::generate_video::generate_video_response::GenerateVideoResponse; +use artcraft_router::generate::generate_video_v2::video_generation_draft_context::VideoGenerationDraftContext; +use artcraft_router::generate::generate_video_v2::video_generation_draft_or_request::VideoGenerationDraftOrRequest; +use artcraft_router::generate::generate_video_v2::video_generation_request::VideoGenerationRequest; +use seedance2pro_client::creds::seedance2pro_session::Seedance2ProSession; +use tokens::tokens::characters::CharacterToken; +use tokens::tokens::media_files::MediaFileToken; +use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken; + +use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError; +use crate::http_server::endpoint_helpers::refund_wallet_after_api_failure::refund_wallet_after_api_failure; +use crate::state::server_state::ServerState; + +/// Execute the v2 pipeline: finalize the draft (if needed), then send the request. +/// +/// If the generation fails and is a Kinovi request with a wallet ledger entry, +/// the wallet is refunded. +pub async fn execute_pipeline_v2( + draft_or_request: VideoGenerationDraftOrRequest, + server_state: &ServerState, + media_file_urls_by_token: Option<&HashMap>, + kinovi_character_ids: Option<&HashMap>, + maybe_wallet_ledger_entry_token: Option<&WalletLedgerEntryToken>, + mysql_connection: &mut sqlx::pool::PoolConnection, +) -> Result { + // 1. Finalize draft → request (or use the request directly). + let (video_request, is_kinovi) = finalize_to_request( + draft_or_request, + server_state, + media_file_urls_by_token, + kinovi_character_ids, + ).await?; + + // 2. Build the client for send_request. + let client = build_kinovi_client(server_state); + + // 3. Send the request. + let result = video_request.send_request(&client).await; + + // 4. On failure, refund wallet if this is a Kinovi request. + if let Err(ref err) = result { + if is_kinovi { + if let Some(ledger_entry_token) = maybe_wallet_ledger_entry_token { + warn!("Kinovi v2 generation failed, issuing refund for {}: {:?}", ledger_entry_token.as_str(), err); + if let Err(refund_err) = refund_wallet_after_api_failure(ledger_entry_token, mysql_connection).await { + error!("Failed to refund wallet after Kinovi v2 failure: {:?}", refund_err); + } + } + } + } + + let response = result.map_err(|err| { + warn!("v2 video generation failed: {:?}", err); + AdvancedCommonWebError::from_error(err) + })?; + + info!("v2 generation response: {:?}", response); + + Ok(response) +} + +/// Finalize the draft (uploading media to Kinovi if needed) and return the request + is_kinovi flag. +async fn finalize_to_request( + draft_or_request: VideoGenerationDraftOrRequest, + server_state: &ServerState, + media_file_urls_by_token: Option<&HashMap>, + kinovi_character_ids: Option<&HashMap>, +) -> Result<(VideoGenerationRequest, bool), AdvancedCommonWebError> { + match draft_or_request { + VideoGenerationDraftOrRequest::Request(request) => { + let is_kinovi = matches!( + request, + VideoGenerationRequest::KinoviSeedance2p0(_) | VideoGenerationRequest::KinoviSeedance2p0Fast(_) + ); + Ok((request, is_kinovi)) + } + VideoGenerationDraftOrRequest::Draft(draft) => { + let seedance2pro_session = Seedance2ProSession::from_cookies_string( + server_state.seedance2pro.cookies.clone() + ); + let seedance2pro_client = RouterSeedance2ProClient::new(seedance2pro_session); + let router_client = RouterClient::Seedance2Pro(seedance2pro_client); + + let draft_context = VideoGenerationDraftContext { + client: Some(&router_client), + media_file_to_artcraft_url_map: media_file_urls_by_token, + character_token_to_kinovi_id_map: kinovi_character_ids, + }; + + let request = draft.finalize(draft_context).await.map_err(|err| { + warn!("Failed to finalize v2 draft: {:?}", err); + AdvancedCommonWebError::from_error(err) + })?; + + // Drafts are always Kinovi (Artcraft never produces drafts). + Ok((request, true)) + } + } +} + +fn build_kinovi_client(server_state: &ServerState) -> RouterClient { + let session = Seedance2ProSession::from_cookies_string( + server_state.seedance2pro.cookies.clone() + ); + RouterClient::Seedance2Pro(RouterSeedance2ProClient::new(session)) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/mod.rs index e69de29bb2..2e8bdddf98 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/pipeline_v2/mod.rs @@ -0,0 +1 @@ +pub mod execute;