server/artcraft
1
0
Fork 0
mirror of https://github.com/storytold/artcraft.git synced 2026-10-20 17:02:53 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge omni api with omni gen (#1876)

Browse Source
This commit is contained in:
hanashi
2026-08-09 23:54:01 -04:00
committed by GitHub
parent c62a269638
commit 91762e2f6c
23 changed files with 882 additions and 1863 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/bill_wallet.rs
-44
View File
@@ -1,44 +0,0 @@
//! 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::common_web_error::CommonWebError;
pub struct BillWalletResult {
pub apriori_job_token: InferenceJobToken,
pub maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
}
/// 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<sqlx::MySql>,
) -> Result<BillWalletResult, CommonWebError> {
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,
})
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/build_router_client.rs
-61
View File
@@ -1,61 +0,0 @@
//! Build a RouterClient for the given provider from server state.
use artcraft_router::api::router_provider::RouterProvider;
use artcraft_router::client::router_client::RouterClient;
use artcraft_router::client::router_fal_client::RouterFalClient;
use artcraft_router::client::router_gmicloud_client::RouterGmiCloudClient;
use artcraft_router::client::router_grok_api_client::RouterGrokApiClient;
use artcraft_router::client::router_kinovi_web_client::RouterKinoviWebClient;
use kinovi_web_client::creds::kinovi_web_session::KinoviWebSession;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::omni_api::shared_utils::kinovi_account::KinoviAccount;
use crate::state::server_state::ServerState;
pub fn build_router_client(
provider: RouterProvider,
server_state: &ServerState,
kinovi_account: KinoviAccount,
) -> Result<RouterClient, CommonWebError> {
match provider {
RouterProvider::KinoviWeb => {
kinovi_provider(server_state, kinovi_account)
}
RouterProvider::Fal => {
let fal_client = RouterFalClient::new_with_webhook(
server_state.inference_providers.fal.api_key.clone(),
server_state.inference_providers.fal.webhook_url.clone(),
);
Ok(RouterClient::Fal(fal_client))
}
RouterProvider::GmiCloud => {
Ok(RouterClient::GmiCloud(RouterGmiCloudClient::new(
server_state.inference_providers.gmicloud.api_key.clone(),
)))
}
RouterProvider::GrokApi => {
Ok(RouterClient::GrokApi(RouterGrokApiClient::new(
server_state.inference_providers.grok_api.api_key.clone(),
)))
}
other => {
Err(CommonWebError::server_error_with_message(
&format!("Unsupported provider for video generation: {:?}", other),
))
}
}
}
fn kinovi_provider(server_state: &ServerState, kinovi_account: KinoviAccount) -> Result<RouterClient, CommonWebError> {
let kinovi_web = &server_state.inference_providers.kinovi_web;
let cookies = match kinovi_account {
KinoviAccount::Volcengine => kinovi_web.cookies_volcengine.clone(),
KinoviAccount::BytePlus => kinovi_web.cookies_byteplus.clone(),
KinoviAccount::BytePlusUltra => kinovi_web.cookies_byteplus_ultra.clone(),
};
let session = KinoviWebSession::from_cookies_string(cookies);
Ok(RouterClient::KinoviWeb(RouterKinoviWebClient::new(session)))
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/hydrate_router_request.rs
-181
View File
@@ -1,181 +0,0 @@
use crate::http_server::common_responses::common_web_error::CommonWebError;
use artcraft_api_defs::omni_api::generate_requests::omni_api_video_generate_request::OmniApiVideoGenerateRequest;
use artcraft_router::api::audio_list_ref::AudioListRef;
use artcraft_router::api::character_list_ref::CharacterListRef;
use artcraft_router::api::router_aspect_ratio::RouterAspectRatio;
use artcraft_router::api::router_bitrate::RouterBitrate;
use artcraft_router::api::router_resolution::RouterResolution;
use artcraft_router::api::router_video_model::RouterVideoModel;
use artcraft_router::api::image_list_ref::ImageListRef;
use artcraft_router::api::image_ref::ImageRef;
use artcraft_router::api::router_provider::RouterProvider;
use artcraft_router::api::video_list_ref::VideoListRef;
use artcraft_router::client::request_mismatch_mitigation_strategy::RequestMismatchMitigationStrategy;
use artcraft_router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder;
use enums::common::generation::common_aspect_ratio::CommonAspectRatio as CommonAspectRatioEnum;
use enums::common::generation::common_bitrate::CommonBitrate as CommonBitrateEnum;
use enums::common::generation::common_resolution::CommonResolution as CommonResolutionEnum;
use enums::common::generation::common_video_model::CommonVideoModel as CommonVideoModelEnum;
pub fn hydrate_to_router_request(
request: &OmniApiVideoGenerateRequest,
) -> Result<GenerateVideoRequestBuilder, CommonWebError> {
let api_model = request.model
.as_ref()
.ok_or_else(|| CommonWebError::BadInputWithSimpleMessage(
"model is required".to_string(),
))?;
let model = convert_model(api_model)?;
let aspect_ratio = request.aspect_ratio
.as_ref()
.map(convert_aspect_ratio)
.transpose()?;
let resolution = request.resolution
.as_ref()
.map(convert_resolution)
.transpose()?;
let bitrate = request.bitrate
.as_ref()
.map(convert_bitrate)
.transpose()?;
Ok(GenerateVideoRequestBuilder {
model,
provider: RouterProvider::Artcraft,
prompt: request.prompt.clone(),
negative_prompt: request.negative_prompt.clone(),
start_frame: request.start_frame_image_media_token.clone()
.map(ImageRef::MediaFileToken),
end_frame: request.end_frame_image_media_token.clone()
.map(ImageRef::MediaFileToken),
reference_images: request.reference_image_media_tokens.clone()
.map(ImageListRef::MediaFileTokens),
reference_videos: request.reference_video_media_tokens.clone()
.map(VideoListRef::MediaFileTokens),
reference_audio: request.reference_audio_media_tokens.clone()
.map(AudioListRef::MediaFileTokens),
reference_character_tokens: request.reference_character_tokens.clone()
.map(CharacterListRef::CharacterTokens),
resolution,
aspect_ratio,
bitrate,
duration_seconds: request.duration_seconds,
video_batch_count: request.video_batch_count,
generate_audio: request.generate_audio,
// Set by the handler after probing reference video durations (only
// models that bill input seconds use it).
total_reference_video_input_seconds: None,
request_mismatch_mitigation_strategy: RequestMismatchMitigationStrategy::PayMoreUpgrade,
idempotency_token: request.idempotency_token.clone(),
})
}
fn convert_model(
model: &CommonVideoModelEnum,
) -> Result<RouterVideoModel, CommonWebError> {
let json = serde_json::to_string(model)?;
serde_json::from_str(&json).map_err(|e| {
CommonWebError::BadInputWithSimpleMessage(
format!("Unsupported video model: {}", e),
)
})
}
fn convert_aspect_ratio(
ar: &CommonAspectRatioEnum,
) -> Result<RouterAspectRatio, CommonWebError> {
let json = serde_json::to_string(ar)?;
serde_json::from_str(&json).map_err(|e| {
CommonWebError::BadInputWithSimpleMessage(
format!("Unsupported aspect ratio: {}", e),
)
})
}
fn convert_resolution(
res: &CommonResolutionEnum,
) -> Result<RouterResolution, CommonWebError> {
let json = serde_json::to_string(res)?;
serde_json::from_str(&json).map_err(|e| {
CommonWebError::BadInputWithSimpleMessage(
format!("Unsupported resolution: {}", e),
)
})
}
fn convert_bitrate(
bitrate: &CommonBitrateEnum,
) -> Result<RouterBitrate, CommonWebError> {
let json = serde_json::to_string(bitrate)?;
serde_json::from_str(&json).map_err(|e| {
CommonWebError::BadInputWithSimpleMessage(
format!("Unsupported bitrate: {}", e),
)
})
}
#[cfg(test)]
mod tests {
use super::*;
mod bitrate_hydration {
use super::*;
#[test]
fn high_is_hydrated() {
let request = OmniApiVideoGenerateRequest {
bitrate: Some(CommonBitrateEnum::High),
..base_request()
};
let builder = hydrate_to_router_request(&request).expect("hydrate should succeed");
assert_eq!(builder.bitrate, Some(RouterBitrate::High));
}
#[test]
fn normal_is_hydrated() {
let request = OmniApiVideoGenerateRequest {
bitrate: Some(CommonBitrateEnum::Normal),
..base_request()
};
let builder = hydrate_to_router_request(&request).expect("hydrate should succeed");
assert_eq!(builder.bitrate, Some(RouterBitrate::Normal));
}
#[test]
fn none_stays_none() {
let builder = hydrate_to_router_request(&base_request()).expect("hydrate should succeed");
assert!(builder.bitrate.is_none());
}
}
fn base_request() -> OmniApiVideoGenerateRequest {
OmniApiVideoGenerateRequest {
idempotency_token: None,
model: Some(CommonVideoModelEnum::Seedance2p0),
prompt: None,
negative_prompt: None,
start_frame_image_media_token: None,
start_frame_image_url: None,
end_frame_image_media_token: None,
end_frame_image_url: None,
reference_image_media_tokens: None,
reference_image_urls: None,
reference_video_media_tokens: None,
reference_video_urls: None,
reference_audio_media_tokens: None,
reference_audio_urls: None,
reference_character_tokens: None,
resolution: None,
aspect_ratio: None,
bitrate: None,
quality: None,
duration_seconds: None,
video_batch_count: None,
generate_audio: None,
}
}
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/mod.rs
-6
View File
@@ -1,6 +0,0 @@
pub mod bill_wallet;
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;
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/pipeline_result.rs
-14
View File
@@ -1,14 +0,0 @@
use artcraft_router::generate::generate_video::generate_video_response::GenerateVideoResponse;
use mysql_queries::queries::generic_inference::common::job_cost_estimates::JobCostEstimates;
use crate::http_server::endpoints::omni_api::generate::video::helpers::bill_wallet::BillWalletResult;
/// Both pipelines produce this, then the shared suffix handles DB writes.
pub struct PipelineResult {
pub billing: BillWalletResult,
pub response: GenerateVideoResponse,
/// System (user-facing) and provider-side cost estimates, written onto the
/// generic_inference_jobs rows.
pub cost_estimates: JobCostEstimates,
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/resolve_kinovi_character_ids.rs
-51
View File
@@ -1,51 +0,0 @@
//! Resolve character tokens to Kinovi character IDs for prompting.
use std::collections::HashMap;
use log::warn;
use sqlx::MySql;
use mysql_queries::queries::characters::batch_lookup_characters_by_token_for_prompting::batch_lookup_characters_by_token_for_prompting;
use tokens::tokens::characters::CharacterToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
/// Resolve character tokens to a map of CharacterToken → Kinovi character ID.
///
/// Looks up the characters, filters to active ones with kinovi IDs, and warns about
/// any that are missing or inactive (but doesn't fail the request).
pub async fn resolve_kinovi_character_ids(
maybe_tokens: Option<&[CharacterToken]>,
connection: &mut sqlx::pool::PoolConnection<MySql>,
) -> Result<Option<HashMap<CharacterToken, String>>, CommonWebError> {
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?;
if characters.len() != tokens.len() {
warn!(
"Not all character tokens were found: requested {}, found {}",
tokens.len(), characters.len(),
);
}
for character in &characters {
if !character.is_active {
warn!("Character {} is not yet active, skipping", character.token);
}
}
let map: HashMap<CharacterToken, String> = 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)) }
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/resolve_media_tokens_to_urls.rs
-64
View File
@@ -1,64 +0,0 @@
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<MediaFileToken, String>>,
) {
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<String> = 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<String> = 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<String> = tokens.iter()
.filter_map(|t| map.get(t).cloned())
.collect();
if !urls.is_empty() {
builder.reference_audio = Some(AudioListRef::Urls(urls));
}
}
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_fal_job.rs
-57
View File
@@ -1,57 +0,0 @@
use log::warn;
use enums::common::visibility::Visibility;
use mysql_queries::queries::generic_inference::api_providers::fal::insert_generic_inference_job_for_fal_queue::FalCategory;
use mysql_queries::queries::generic_inference::api_providers::fal::insert_generic_inference_job_for_fal_queue_with_apriori_job_token::{
insert_generic_inference_job_for_fal_queue_with_apriori_job_token,
InsertGenericInferenceForFalWithAprioriJobTokenArgs,
};
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use super::shared_job_args::SharedJobArgs;
pub struct InsertFalJobArgs<'a, 'tx> {
pub external_job_id: &'a str,
pub shared: SharedJobArgs<'a, 'tx>,
}
pub async fn insert_fal_job(args: InsertFalJobArgs<'_, '_>) -> Result<InferenceJobToken, CommonWebError> {
let InsertFalJobArgs {
external_job_id,
shared,
} = args;
let db_result = insert_generic_inference_job_for_fal_queue_with_apriori_job_token(
InsertGenericInferenceForFalWithAprioriJobTokenArgs {
apriori_job_token: shared.apriori_job_token,
uuid_idempotency_token: shared.idempotency_token,
maybe_external_third_party_id: external_job_id,
fal_category: FalCategory::VideoGeneration,
maybe_model_type: shared.maybe_model_type,
maybe_prompt_token: shared.maybe_prompt_token,
maybe_creator_user_token: Some(shared.user_token),
maybe_avt_token: shared.maybe_avt_token,
creator_ip_address: shared.ip_address,
creator_set_visibility: Visibility::Public,
maybe_platform_type: shared.maybe_platform_type,
maybe_cost_estimates: shared.maybe_cost_estimates,
maybe_debug_log_event_token: shared.maybe_debug_log_event_token,
mysql_executor: &mut **shared.transaction,
maybe_inference_args: None,
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(CommonWebError::from_error(err))
}
}
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_gmicloud_job.rs
-48
View File
@@ -1,48 +0,0 @@
use log::warn;
use mysql_queries::queries::generic_inference::api_providers::gmicloud::insert_generic_inference_job_for_gmicloud::{
insert_generic_inference_job_for_gmicloud, InsertGmiCloudInferenceJobArgs,
};
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use super::shared_job_args::SharedJobArgs;
pub struct InsertGmiCloudJobArgs<'a, 'tx> {
pub external_request_id: &'a str,
pub shared: SharedJobArgs<'a, 'tx>,
}
pub async fn insert_gmicloud_job(args: InsertGmiCloudJobArgs<'_, '_>) -> Result<InferenceJobToken, CommonWebError> {
let InsertGmiCloudJobArgs {
external_request_id,
shared,
} = args;
let db_result = insert_generic_inference_job_for_gmicloud(
InsertGmiCloudInferenceJobArgs {
apriori_job_token: shared.apriori_job_token,
uuid_idempotency_token: shared.idempotency_token,
external_request_id,
maybe_model_type: shared.maybe_model_type,
maybe_prompt_token: shared.maybe_prompt_token,
maybe_creator_user_token: Some(shared.user_token),
maybe_avt_token: shared.maybe_avt_token,
creator_ip_address: shared.ip_address,
maybe_platform_type: shared.maybe_platform_type,
maybe_cost_estimates: shared.maybe_cost_estimates,
maybe_debug_log_event_token: shared.maybe_debug_log_event_token,
mysql_executor: &mut **shared.transaction,
phantom: Default::default(),
}
).await;
match db_result {
Ok(token) => Ok(token),
Err(err) => {
warn!("Error inserting GmiCloud inference job: {:?}", err);
Err(CommonWebError::from_error(err))
}
}
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_grok_api_job.rs
-48
View File
@@ -1,48 +0,0 @@
use log::warn;
use mysql_queries::queries::generic_inference::api_providers::grok_api::insert_generic_inference_job_for_grok_api::{
insert_generic_inference_job_for_grok_api, InsertGrokApiInferenceJobArgs,
};
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use super::shared_job_args::SharedJobArgs;
pub struct InsertGrokApiJobArgs<'a, 'tx> {
pub external_request_id: &'a str,
pub shared: SharedJobArgs<'a, 'tx>,
}
pub async fn insert_grok_api_job(args: InsertGrokApiJobArgs<'_, '_>) -> Result<InferenceJobToken, CommonWebError> {
let InsertGrokApiJobArgs {
external_request_id,
shared,
} = args;
let db_result = insert_generic_inference_job_for_grok_api(
InsertGrokApiInferenceJobArgs {
apriori_job_token: shared.apriori_job_token,
uuid_idempotency_token: shared.idempotency_token,
external_request_id,
maybe_model_type: shared.maybe_model_type,
maybe_prompt_token: shared.maybe_prompt_token,
maybe_creator_user_token: Some(shared.user_token),
maybe_avt_token: shared.maybe_avt_token,
creator_ip_address: shared.ip_address,
maybe_platform_type: shared.maybe_platform_type,
maybe_cost_estimates: shared.maybe_cost_estimates,
maybe_debug_log_event_token: shared.maybe_debug_log_event_token,
mysql_executor: &mut **shared.transaction,
phantom: Default::default(),
}
).await;
match db_result {
Ok(token) => Ok(token),
Err(err) => {
warn!("Error inserting Grok (xAI) API inference job: {:?}", err);
Err(CommonWebError::from_error(err))
}
}
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_kinovi_web_jobs.rs
-95
View File
@@ -1,95 +0,0 @@
use log::{error, warn};
use crate::http_server::common_responses::common_web_error::CommonWebError;
use enums::common::visibility::Visibility;
use mysql_queries::queries::generic_inference::api_providers::kinovi_web::insert_generic_inference_job_for_kinovi_web_queue_with_apriori_job_token::{insert_generic_inference_job_for_kinovi_web_queue_with_apriori_job_token, InsertGenericInferenceForKinoviWebWithAprioriJobTokenArgs, KinoviVersion};
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
use uuid_utils::uuid::generate_random_uuid;
use super::shared_job_args::SharedJobArgs;
pub struct InsertKinoviWebJobsArgs<'a, 'tx> {
pub primary_order_id: &'a str,
pub maybe_additional_order_ids: Option<&'a [String]>,
pub maybe_wallet_ledger_entry_token: Option<&'a WalletLedgerEntryToken>,
pub kinovi_version: KinoviVersion,
pub shared: SharedJobArgs<'a, 'tx>,
}
pub struct InsertKinoviWebJobsResult {
pub primary_job_token: InferenceJobToken,
pub all_job_tokens: Vec<InferenceJobToken>,
}
pub async fn insert_kinovi_web_jobs(args: InsertKinoviWebJobsArgs<'_, '_>) -> Result<InsertKinoviWebJobsResult, CommonWebError> {
let InsertKinoviWebJobsArgs {
primary_order_id,
maybe_additional_order_ids,
maybe_wallet_ledger_entry_token,
kinovi_version,
mut shared,
} = args;
// Build a deduplicated list with primary_order_id first.
let mut all_order_ids = vec![primary_order_id.to_string()];
if let Some(additional) = maybe_additional_order_ids {
for id in additional {
if id != primary_order_id {
all_order_ids.push(id.clone());
}
}
}
let mut all_job_tokens: Vec<InferenceJobToken> = Vec::with_capacity(all_order_ids.len());
for (i, order_id) in all_order_ids.iter().enumerate() {
let job_token = if i == 0 { shared.apriori_job_token.clone() } else { InferenceJobToken::generate() };
// TODO: These strings are too long for the database:
//let idempotency_str = if i == 0 { shared.idempotency_token.to_string() } else { format!("{}-batch-{}", shared.idempotency_token, i) };
let idempotency_str = generate_random_uuid();
let db_result = insert_generic_inference_job_for_kinovi_web_queue_with_apriori_job_token(
InsertGenericInferenceForKinoviWebWithAprioriJobTokenArgs {
kinovi_version,
apriori_job_token: &job_token,
uuid_idempotency_token: &idempotency_str,
maybe_external_third_party_id: order_id,
maybe_model_type: shared.maybe_model_type,
maybe_prompt_token: shared.maybe_prompt_token,
maybe_wallet_ledger_entry_token,
maybe_creator_user_token: Some(shared.user_token),
maybe_avt_token: shared.maybe_avt_token,
creator_ip_address: shared.ip_address,
creator_set_visibility: Visibility::Public,
maybe_platform_type: shared.maybe_platform_type,
maybe_cost_estimates: shared.maybe_cost_estimates,
maybe_debug_log_event_token: shared.maybe_debug_log_event_token,
mysql_executor: &mut **shared.transaction,
maybe_inference_args: None,
phantom: Default::default(),
}
).await;
match db_result {
Ok(token) => all_job_tokens.push(token),
Err(err) => {
warn!("Error inserting kinovi_web inference job (order_id={}): {:?}", order_id, err);
if i == 0 { return Err(CommonWebError::from_error(err)); }
}
}
}
let primary_job_token = all_job_tokens.first().cloned().ok_or_else(|| {
error!("No inference job token was created");
CommonWebError::server_error_with_message("No inference job token was created")
})?;
Ok(InsertKinoviWebJobsResult {
primary_job_token,
all_job_tokens,
})
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/mod.rs
-5
View File
@@ -1,5 +0,0 @@
pub mod insert_fal_job;
pub mod insert_gmicloud_job;
pub mod insert_grok_api_job;
pub mod insert_kinovi_web_jobs;
pub mod shared_job_args;
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/shared_job_args.rs
-22
View File
@@ -1,22 +0,0 @@
use enums::common::generation::common_model_type::CommonModelType;
use enums::common::platform_type::PlatformType;
use mysql_queries::queries::generic_inference::common::job_cost_estimates::JobCostEstimates;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken;
use tokens::tokens::prompts::PromptToken;
use tokens::tokens::users::UserToken;
pub struct SharedJobArgs<'a, 'tx> {
pub apriori_job_token: &'a InferenceJobToken,
pub idempotency_token: &'a str,
pub user_token: &'a UserToken,
pub maybe_avt_token: Option<&'a AnonymousVisitorTrackingToken>,
pub maybe_prompt_token: Option<&'a PromptToken>,
pub maybe_debug_log_event_token: Option<&'a DebugLogEventToken>,
pub maybe_model_type: Option<CommonModelType>,
pub maybe_platform_type: Option<PlatformType>,
pub maybe_cost_estimates: Option<JobCostEstimates>,
pub ip_address: &'a str,
pub transaction: &'a mut sqlx::Transaction<'tx, sqlx::MySql>,
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/mod.rs
-3
View File
@@ -1,6 +1,3 @@
pub mod omni_api_video_generate_handler;
pub mod check_request;
pub mod helpers;
pub mod ingest_url_inputs;
pub mod pipeline_v2;
pub mod insert_db_job;
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/omni_api_video_generate_handler.rs
+58 -467
View File
@@ -1,61 +1,33 @@
use std::collections::HashMap;
use std::sync::Arc;
use actix_web::web::Json;
use actix_web::{web, HttpRequest};
use log::{error, info, warn};
use sqlx::Acquire;
use log::info;
use artcraft_api_defs::omni_api::generate_requests::omni_api_video_generate_request::OmniApiVideoGenerateRequest;
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::generate::generate_video::generate_video_response::GenerateVideoResponse;
use enums::by_table::debug_logs::debug_log_type::DebugLogType;
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_generation_mode::CommonGenerationMode;
use enums::common::generation::common_model_type::CommonModelType;
use enums::common::generation::common_video_model::CommonVideoModel;
use enums::common::generation_provider::GenerationProvider;
use enums::common::platform_type::PlatformType;
use http_server_common::request::get_request_ip::get_request_ip;
use enums::by_table::debug_logs::debug_log_level::DebugLogLevel;
use mysql_queries::queries::debug_logs::insert_debug_log::{insert_debug_log, InsertDebugLogArgs};
use mysql_queries::queries::generic_inference::api_providers::kinovi_web::insert_generic_inference_job_for_kinovi_web_queue_with_apriori_job_token::KinoviVersion;
use mysql_queries::queries::idepotency_tokens::insert_idempotency_token::insert_idempotency_token;
use mysql_queries::queries::prompt_context_items::insert_batch_prompt_context_items::{
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 tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::generate::common::generation_debug_logs::GenerationDebugLogContext;
use crate::http_server::endpoints::generate::common::payments_error_test::payments_error_test;
use crate::http_server::endpoints::generate::common::probed_reference_videos::{
download_and_probe_reference_videos, fetch_reference_video_sources, ProbedReferenceVideos,
};
use crate::http_server::endpoints::omni_api::generate::video::check_request::check_request;
use crate::http_server::endpoints::omni_api::generate::video::ingest_url_inputs::ingest_url_inputs;
use crate::http_server::endpoints::omni_api::generate::video::helpers::hydrate_router_request::hydrate_to_router_request;
use crate::http_server::endpoints::omni_api::generate::video::helpers::resolve_kinovi_character_ids::resolve_kinovi_character_ids;
use crate::http_server::endpoints::omni_api::generate::video::insert_db_job::insert_fal_job::{insert_fal_job, InsertFalJobArgs};
use crate::http_server::endpoints::omni_api::generate::video::insert_db_job::insert_gmicloud_job::{insert_gmicloud_job, InsertGmiCloudJobArgs};
use crate::http_server::endpoints::omni_api::generate::video::insert_db_job::insert_grok_api_job::{insert_grok_api_job, InsertGrokApiJobArgs};
use crate::http_server::endpoints::omni_api::generate::video::insert_db_job::insert_kinovi_web_jobs::{insert_kinovi_web_jobs, InsertKinoviWebJobsArgs};
use crate::http_server::endpoints::omni_api::generate::video::insert_db_job::shared_job_args::SharedJobArgs;
use crate::http_server::endpoints::omni_api::shared_utils::kinovi_account::KinoviAccount;
use crate::http_server::endpoints::omni_api::generate::video::pipeline_v2::run_pipeline_v2::{run_pipeline_v2, RunPipelineV2Args};
use crate::http_server::endpoints::omni_api::shared_utils::video::validate_video_request::validate_video_request;
use crate::http_server::user_lookup::user_session::session_utils::lookup::user_session_feature_flags::UserSessionFeatureFlags;
use crate::http_server::endpoints::omni_gen::generate::video::shared_video_generation::{
run_authenticated_video_generation, VideoGenerationAuth,
};
use crate::http_server::user_lookup::api_keys::require_api_key_user::require_api_key_user;
use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format;
use crate::state::server_state::ServerState;
use crate::util::lookup::lookup_media_files_as_cdn_url_list_and_map::lookup_media_files_as_cdn_url_list_and_map;
/// Generate a video using the omni-api unified endpoint (API-key authenticated).
///
/// Razor-thin handler: validate the omni-api extras (URL-based media inputs),
/// authenticate the API key, ingest any URL inputs into media files, then
/// delegate to the shared generation core in the omni_gen module — which is
/// authoritative for generation behavior. This endpoint is the API-key-only
/// twin of `/v1/omni_gen/generate/video`.
#[utoipa::path(
post,
tag = "Omni API",
@@ -82,55 +54,29 @@ pub async fn omni_api_video_generate_handler(
// Reject doomed combos (e.g. grok_imagine_video_1p5 without an image)
// before any billable or DB-mutating work — see helper for the rules.
// (This is the URL-aware twin of the omni_gen validator: URL inputs count
// toward the same limits before they are ingested below.)
validate_video_request(&request)?;
payments_error_test(&request.prompt.as_deref().unwrap_or(""))?;
let debug_log_event_token = DebugLogEventToken::generate();
let maybe_prompt_model_type: Option<CommonModelType> = request.model
.as_ref()
.map(|m| m.to_common_model_type());
// ==================== API KEY USER ==================== //
let mut mysql_connection = server_state.mysql_pool.acquire().await?;
// API-key authentication (Authorization header) instead of a session cookie. Never cached.
// API-key authentication (Authorization header) ONLY — never a web
// session. Web/session callers use /v1/omni_gen/generate/video.
let api_session = require_api_key_user(&http_request, &mut *mysql_connection).await?;
let user_token = &api_session.user_token;
// ==================== MODEL ACCESS CHECK ==================== //
// API-key users have no per-user feature flags today.
let user_feature_flags = UserSessionFeatureFlags::new(None);
// ==================== IDEMPOTENCY ==================== //
let idempotency_token = request.idempotency_token.as_deref()
.unwrap_or("")
.to_string();
if let Err(reason) = validate_idempotency_token_format(&idempotency_token) {
return Err(CommonWebError::BadInputWithSimpleMessage(reason));
}
insert_idempotency_token(&idempotency_token, &mut *mysql_connection)
.await
.map_err(|err| {
error!("Error inserting idempotency token: {:?}", err);
CommonWebError::BadInputWithSimpleMessage("repeated idempotency token".to_string())
})?;
let ip_address = get_request_ip(&http_request);
let request_url = http_request.uri().to_string();
// ==================== INGEST URL INPUTS ==================== //
// Download any URL media inputs into media files owned by this user, then
// treat them as media tokens. Release the pooled connection first so we don't
// hold a pool slot during the (network) downloads.
// treat them as media tokens. Release the pooled connection first so we
// don't hold a pool slot during the (network) downloads. This happens
// BEFORE the shared core consumes the idempotency token, so a failed
// ingest never burns the caller's token.
let has_url_inputs = request.start_frame_image_url.is_some()
|| request.end_frame_image_url.is_some()
|| request.reference_image_urls.is_some()
@@ -139,409 +85,54 @@ pub async fn omni_api_video_generate_handler(
if has_url_inputs {
drop(mysql_connection);
ingest_url_inputs(&mut request, &server_state, user_token, &ip_address).await?;
ingest_url_inputs(&mut request, &server_state, &api_session.user_token, &ip_address).await?;
mysql_connection = server_state.mysql_pool.acquire().await?;
}
// ==================== RESOLVE MEDIA TOKENS ==================== //
// ==================== DELEGATE TO THE SHARED CORE ==================== //
let mut all_tokens: Vec<MediaFileToken> = Vec::new();
let omni_gen_request = to_omni_gen_request(request.into_inner());
if let Some(token) = &request.start_frame_image_media_token {
all_tokens.push(token.clone());
}
if let Some(token) = &request.end_frame_image_media_token {
all_tokens.push(token.clone());
}
if let Some(tokens) = &request.reference_image_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
if let Some(tokens) = &request.reference_video_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
if let Some(tokens) = &request.reference_audio_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
let media_file_to_url_map: Option<HashMap<MediaFileToken, String>> = if all_tokens.is_empty() {
None
} else {
info!("Resolving {} media file tokens to CDN URLs", all_tokens.len());
let resolved = lookup_media_files_as_cdn_url_list_and_map(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&all_tokens,
).await?;
Some(resolved.token_to_url_map)
let auth = VideoGenerationAuth {
user_token: &api_session.user_token,
// AVT tokens are web-session only; API-key callers have none.
maybe_avt_token: None,
// Omni API requests are always API-key authenticated.
maybe_platform_type: Some(PlatformType::ApiKey),
};
// ==================== RESOLVE CHARACTERS ==================== //
let kinovi_character_id_map: Option<HashMap<CharacterToken, String>> =
resolve_kinovi_character_ids(
request.reference_character_tokens.as_deref(),
&mut mysql_connection,
).await?;
// ==================== REFERENCE VIDEO INPUT SECONDS ==================== //
// Seedance 2.5 bills reference-video input seconds on top of the output
// duration. Stored durations can't be trusted for billing, so every
// reference video is downloaded and ffprobed; the downloaded files are
// kept and handed to the pipeline so the Kinovi upload reuses them
// instead of downloading the same bytes twice.
let mut maybe_probed_reference_videos: Option<ProbedReferenceVideos> = None;
if matches!(request.model, Some(CommonVideoModel::Seedance2p5 | CommonVideoModel::Seedance2p5Ultra)) {
if let Some(video_tokens) = request.reference_video_media_tokens.as_deref().filter(|tokens| !tokens.is_empty()) {
let video_sources = fetch_reference_video_sources(
video_tokens,
&http_request,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&mut mysql_connection,
).await?;
// Downloads + ffprobe are slow — release the pool slot across them,
// then re-acquire for billing. Probing never fails the generation:
// unmeasurable files bill at the 30-second worst case.
drop(mysql_connection);
let probed = download_and_probe_reference_videos(&video_sources).await;
mysql_connection = server_state.mysql_pool.acquire().await?;
maybe_probed_reference_videos = Some(probed);
}
}
// ==================== HYDRATE ROUTER REQUEST ==================== //
let mut router_builder = hydrate_to_router_request(&request)?;
router_builder.total_reference_video_input_seconds =
maybe_probed_reference_videos.as_ref().map(|probed| probed.total_input_seconds);
// ==================== PIPELINE DISPATCH ==================== //
let kinovi_account = match request.model {
// BytePlus Ultra
Some(CommonVideoModel::Seedance2p0BytePlusUltra) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p0BytePlusUltraFast) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p0BytePlusUltraMini) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p5Ultra) => KinoviAccount::BytePlusUltra,
// BytePlus
Some(CommonVideoModel::Seedance2p0BytePlus) => KinoviAccount::BytePlus,
Some(CommonVideoModel::Seedance2p0BytePlusFast) => KinoviAccount::BytePlus,
Some(CommonVideoModel::Seedance2p0BytePlusMini) => KinoviAccount::BytePlus,
Some(CommonVideoModel::PreviewModel) => KinoviAccount::BytePlus,
Some(CommonVideoModel::PreviewModelFast) => KinoviAccount::BytePlus,
// Everything else goes through Volcengine
_ => KinoviAccount::Volcengine,
};
// ==================== DEBUG LOG: HTTP REQUEST ==================== //
if let Err(err) = insert_debug_log(InsertDebugLogArgs {
apriori_debug_log_event_token: Some(&debug_log_event_token),
maybe_creator_user_token: Some(user_token),
debug_log_type: DebugLogType::HttpRequest,
maybe_log_level: Some(DebugLogLevel::Info),
maybe_ip_address: Some(&ip_address),
maybe_url: Some(&request_url),
message: &serde_json::to_string(&*request).unwrap_or_default(),
mysql_executor: &mut *mysql_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert HTTP request debug log: {:?}", err);
}
// NB: The pipeline takes over the connection for its remaining pre-request DB writes (billing,
// outbound provider request debug log) and releases it before the (slow, external) generation
// call — holding a pool slot across that call is what starves the pool and causes PoolTimedOut
// on unrelated endpoints. We re-acquire below to write the result.
let debug_log_context = GenerationDebugLogContext {
event_token: &debug_log_event_token,
user_token,
ip_address: &ip_address,
request_url: &request_url,
};
let pipeline_result = run_pipeline_v2(RunPipelineV2Args {
router_builder: &router_builder,
server_state: &server_state,
user_token,
media_file_to_url_map: &media_file_to_url_map,
kinovi_character_id_map: &kinovi_character_id_map,
kinovi_account,
debug_log_context: &debug_log_context,
predownloaded_media_paths: maybe_probed_reference_videos.as_ref().map(|probed| probed.local_paths_by_url()),
run_authenticated_video_generation(
&http_request,
&omni_gen_request,
&server_state,
auth,
mysql_connection,
}).await;
// ==================== DEBUG LOG: PIPELINE ERROR ==================== //
let pipeline_result = match pipeline_result {
Ok(result) => result,
Err(err) => {
// Best-effort error log; never mask the original error.
if let Ok(mut error_log_connection) = server_state.mysql_pool.acquire().await {
if let Err(log_err) = insert_debug_log(InsertDebugLogArgs {
apriori_debug_log_event_token: Some(&debug_log_event_token),
maybe_creator_user_token: Some(user_token),
debug_log_type: DebugLogType::BackendFailure,
maybe_log_level: Some(DebugLogLevel::Error),
maybe_ip_address: Some(&ip_address),
maybe_url: Some(&request_url),
message: &format!("Video generation pipeline failed: {:?}", err),
mysql_executor: &mut *error_log_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert pipeline error debug log: {:?}", log_err);
}
}
return Err(err);
}
};
let mut mysql_connection = server_state.mysql_pool.acquire().await?;
// NB: Outbound provider requests (Fal/Grok/Kinovi) are debug-logged inside
// the pipeline BEFORE the send, so the payload is captured even on failure.
// ==================== WRITE RESULT ==================== //
// Omni API requests are always API-key authenticated; hardcode the platform type.
let maybe_platform_type = Some(PlatformType::ApiKey);
let mut transaction = mysql_connection.begin().await.map_err(|err| {
error!("Error starting MySQL transaction: {:?}", err);
CommonWebError::from_error(err)
})?;
// -- Prompt --
let prompt_token = match 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(determine_generation_mode(&request)),
maybe_aspect_ratio: request.aspect_ratio,
maybe_resolution: request.resolution,
maybe_bitrate: request.bitrate,
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 {
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 (primary_job_token, all_job_tokens) = match &pipeline_result.response {
GenerateVideoResponse::KinoviWeb(payload) => {
info!("Inserting kinovi_web job(s) with token: {:?}", pipeline_result.billing.apriori_job_token);
let kinovi_version = match kinovi_account {
KinoviAccount::Volcengine => KinoviVersion::Volcengine,
KinoviAccount::BytePlus => KinoviVersion::BytePlus,
KinoviAccount::BytePlusUltra => KinoviVersion::BytePlusUltra,
};
let result = insert_kinovi_web_jobs(InsertKinoviWebJobsArgs {
primary_order_id: &payload.order_id,
maybe_additional_order_ids: payload.maybe_order_ids.as_deref(),
maybe_wallet_ledger_entry_token: pipeline_result.billing.maybe_wallet_ledger_entry_token.as_ref(),
kinovi_version,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: None, // AVT tokens are web-session only; API-key callers have none.
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(result.primary_job_token, result.all_job_tokens)
}
GenerateVideoResponse::Fal(payload) => {
let external_id = payload.request_id.as_deref().ok_or_else(|| {
error!("Fal generation response missing request_id");
CommonWebError::server_error_with_message("Fal generation response missing request_id")
})?;
info!("Inserting fal job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_fal_job(InsertFalJobArgs {
external_job_id: external_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: None, // AVT tokens are web-session only; API-key callers have none.
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
GenerateVideoResponse::Artcraft(payload) => {
(
payload.inference_job_token.clone(),
vec![payload.inference_job_token.clone()],
)
}
GenerateVideoResponse::GmiCloud(payload) => {
info!("Inserting GmiCloud job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_gmicloud_job(InsertGmiCloudJobArgs {
external_request_id: &payload.request_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: None, // AVT tokens are web-session only; API-key callers have none.
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
GenerateVideoResponse::Grok(payload) => {
info!("Inserting Grok (xAI) API job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_grok_api_job(InsertGrokApiJobArgs {
external_request_id: &payload.request_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: None, // AVT tokens are web-session only; API-key callers have none.
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
other => {
error!("Unexpected generation response variant: {:?}", other);
return Err(CommonWebError::server_error_with_message("Unexpected generation response"));
}
};
transaction.commit().await.map_err(|err| {
error!("Error committing transaction: {:?}", err);
CommonWebError::from_error(err)
})?;
Ok(Json(OmniGenVideoGenerateResponse {
success: true,
inference_job_token: primary_job_token,
all_job_tokens,
}))
).await
}
fn determine_generation_mode(request: &OmniApiVideoGenerateRequest) -> CommonGenerationMode {
let has_keyframe = request.start_frame_image_media_token.is_some()
|| request.end_frame_image_media_token.is_some();
if has_keyframe {
return CommonGenerationMode::Keyframe;
/// The omni-api request is the omni-gen request plus URL-based media inputs.
/// By the time this runs, `ingest_url_inputs` has folded every URL into its
/// media-token twin, so the conversion is a plain field map.
fn to_omni_gen_request(request: OmniApiVideoGenerateRequest) -> OmniGenVideoCostAndGenerateRequest {
OmniGenVideoCostAndGenerateRequest {
idempotency_token: request.idempotency_token,
model: request.model,
prompt: request.prompt,
negative_prompt: request.negative_prompt,
start_frame_image_media_token: request.start_frame_image_media_token,
end_frame_image_media_token: request.end_frame_image_media_token,
reference_image_media_tokens: request.reference_image_media_tokens,
reference_video_media_tokens: request.reference_video_media_tokens,
reference_audio_media_tokens: request.reference_audio_media_tokens,
reference_character_tokens: request.reference_character_tokens,
resolution: request.resolution,
aspect_ratio: request.aspect_ratio,
bitrate: request.bitrate,
quality: request.quality,
duration_seconds: request.duration_seconds,
video_batch_count: request.video_batch_count,
generate_audio: request.generate_audio,
// Cost-estimation hints are an omni_gen-only field; generation ignores it.
estimate_only: None,
}
let has_reference = request.reference_image_media_tokens.as_ref().is_some_and(|t| !t.is_empty())
|| request.reference_video_media_tokens.as_ref().is_some_and(|t| !t.is_empty())
|| request.reference_audio_media_tokens.as_ref().is_some_and(|t| !t.is_empty())
|| request.reference_character_tokens.as_ref().is_some_and(|t| !t.is_empty());
if has_reference {
return CommonGenerationMode::Reference;
}
CommonGenerationMode::Text
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/pipeline_v2/mod.rs
-1
View File
@@ -1 +0,0 @@
pub mod run_pipeline_v2;
crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/pipeline_v2/run_pipeline_v2.rs
-282
View File
@@ -1,282 +0,0 @@
use std::convert::TryFrom;
use std::collections::HashMap;
use std::path::PathBuf;
use log::{error, info, warn};
use artcraft_router::api::router_video_model::RouterVideoModel;
use artcraft_router::api::router_provider::RouterProvider;
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::video_generation_draft_context::VideoGenerationDraftContext;
use artcraft_router::generate::generate_video::video_generation_draft_or_request::VideoGenerationDraftOrRequest;
use tokens::tokens::characters::CharacterToken;
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::users::UserToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoint_helpers::refund_wallet_after_api_failure::refund_wallet_after_api_failure;
use crate::http_server::endpoints::generate::common::generation_debug_logs::{
insert_provider_request_debug_log, provider_request_debug_log_type, GenerationDebugLogContext,
};
use crate::http_server::endpoints::omni_api::generate::video::helpers::bill_wallet::bill_wallet;
use crate::http_server::endpoints::omni_api::generate::video::helpers::build_router_client::build_router_client;
use crate::http_server::endpoints::omni_api::generate::video::helpers::pipeline_result::PipelineResult;
use crate::http_server::endpoints::omni_api::generate::video::helpers::resolve_media_tokens_to_urls::resolve_media_tokens_to_urls;
use mysql_queries::queries::generic_inference::common::job_cost_estimates::JobCostEstimates;
use crate::http_server::endpoints::omni_api::shared_utils::kinovi_account::KinoviAccount;
use crate::http_server::endpoints::omni_api::shared_utils::map_kinovi_web_router_error::map_router_error_to_web_error;
use crate::state::server_state::ServerState;
pub struct RunPipelineV2Args<'a> {
pub router_builder: &'a GenerateVideoRequestBuilder,
pub server_state: &'a ServerState,
pub user_token: &'a UserToken,
pub media_file_to_url_map: &'a Option<HashMap<MediaFileToken, String>>,
pub kinovi_character_id_map: &'a Option<HashMap<CharacterToken, String>>,
pub kinovi_account: KinoviAccount,
pub debug_log_context: &'a GenerationDebugLogContext<'a>,
/// Source URL → local file path for reference videos the handler already
/// downloaded (for input-seconds billing). The Kinovi upload reads these
/// instead of downloading the same bytes again.
pub predownloaded_media_paths: Option<&'a HashMap<String, PathBuf>>,
/// The handler's open connection. The pipeline uses it for its remaining
/// pre-request DB writes (billing, outbound-request debug log) and releases
/// it BEFORE the external provider call.
pub mysql_connection: sqlx::pool::PoolConnection<sqlx::MySql>,
}
// NB: This pipeline does an external generation call (`upload_and_generate`) that can take many
// seconds. It deliberately does NOT hold a pooled DB connection across that call — it acquires
// short-lived connections only for the billing and (on failure) refund writes. Holding a pooled
// connection across the external call is what starves the pool and causes `PoolTimedOut`.
pub async fn run_pipeline_v2(args: RunPipelineV2Args<'_>) -> Result<PipelineResult, CommonWebError> {
let RunPipelineV2Args {
router_builder,
server_state,
user_token,
media_file_to_url_map,
kinovi_character_id_map,
kinovi_account,
debug_log_context,
predownloaded_media_paths,
mut mysql_connection,
} = args;
let router_builder = router_builder.clone();
let provider = match router_builder.model {
RouterVideoModel::HappyHorse1p0 => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0 => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0Fast => RouterProvider::KinoviWeb,
RouterVideoModel::PreviewModel => RouterProvider::KinoviWeb,
RouterVideoModel::PreviewModelFast => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0BytePlus => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0BytePlusFast => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0BytePlusUltra => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0BytePlusUltraFast => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0Mini => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0BytePlusMini => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p0BytePlusUltraMini => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p5Preview => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p5 => RouterProvider::KinoviWeb,
RouterVideoModel::Seedance2p5Ultra => RouterProvider::KinoviWeb,
//RouterVideoModel::Seedance2p0Ultra => RouterProvider::GmiCloud,
//RouterVideoModel::Seedance2p0UltraFast => RouterProvider::GmiCloud,
RouterVideoModel::GrokImagineVideo => RouterProvider::GrokApi,
RouterVideoModel::GrokImagineVideo1p5 => RouterProvider::GrokApi,
_ => RouterProvider::Fal,
};
// 1. Build execution request
let mut exec_builder = router_builder.clone();
exec_builder.provider = provider;
// These variants are FULFILLED by the base model (they build the same
// execution request), but they are PRICED as themselves — each has its own
// cost entry under the Artcraft provider.
//
// IMPORTANT: the model rewrite must apply to the EXECUTION builder only.
// The cost estimate below must see the ORIGINAL model. Rewriting before
// the cost estimate makes every variant price as its base model — a
// pricing bug we actually shipped, where these variants charged the base
// rate while the cost endpoint quoted their real prices. Do not "hoist"
// or "deduplicate" this match to the top of the function.
match exec_builder.model {
RouterVideoModel::PreviewModel |
RouterVideoModel::Seedance2p0BytePlus |
RouterVideoModel::Seedance2p0BytePlusUltra => {
exec_builder.model = RouterVideoModel::Seedance2p0;
},
RouterVideoModel::PreviewModelFast |
RouterVideoModel::Seedance2p0BytePlusFast |
RouterVideoModel::Seedance2p0BytePlusUltraFast => {
exec_builder.model = RouterVideoModel::Seedance2p0Fast;
},
RouterVideoModel::Seedance2p5Ultra => {
exec_builder.model = RouterVideoModel::Seedance2p5;
},
_ => {}, // Fall-through
}
// Fal, GmiCloud, and Grok (xAI) take image URLs directly, not media file tokens.
// Resolve tokens to URLs before building.
if matches!(provider, RouterProvider::Fal | RouterProvider::GmiCloud | RouterProvider::GrokApi) {
resolve_media_tokens_to_urls(&mut exec_builder, media_file_to_url_map.as_ref());
}
let draft_or_request = exec_builder.build2()
.map_err(|e| {
warn!("Failed to build2 for v2 pipeline: {}", e);
CommonWebError::from_error(e)
})?;
// 2. Calculate cost.
// For Artcraft-billable models, swap provider to Artcraft so credits = cents.
// For GmiCloud, use the execution request's cost directly (no Artcraft equivalent).
let system_cost_estimate = {
let mut cost_builder = router_builder.clone();
cost_builder.provider = RouterProvider::Artcraft;
cost_builder.build2()
.map_err(|e| {
warn!("Failed to build2 cost estimate for v2: {}", e);
CommonWebError::from_error(e)
})?
.estimate_cost()
.map_err(|e| {
warn!("Failed to estimate cost for v2: {}", e);
CommonWebError::from_error(e)
})?
};
let cost = system_cost_estimate.cost_in_credits.unwrap_or(0);
// Provider-side estimate (what the fulfilling provider charges us). The
// router defers to the underlying provider crates (kinovi_web_client,
// gmicloud_client, grok_api_client, the fal pricing modules, etc.) per
// request variant. Bookkeeping only — failures must not block generation.
let maybe_provider_cost_estimate = match draft_or_request.estimate_cost() {
Ok(estimate) => Some(estimate),
Err(err) => {
warn!("Failed to estimate provider cost for v2 video: {}", err);
None
}
};
let cost_estimates = JobCostEstimates {
maybe_external_third_party_cost_credits: maybe_provider_cost_estimate.as_ref()
.and_then(|e| e.cost_in_credits)
.and_then(|v| u32::try_from(v).ok()),
maybe_external_third_party_cost_usd_cents: maybe_provider_cost_estimate.as_ref()
.and_then(|e| e.cost_in_usd_cents)
.and_then(|v| u32::try_from(v).ok()),
maybe_system_cost_credits: system_cost_estimate.cost_in_credits
.and_then(|v| u32::try_from(v).ok()),
maybe_system_cost_usd_cents: system_cost_estimate.cost_in_usd_cents
.and_then(|v| u32::try_from(v).ok()),
};
info!("v2 estimated cost: {} credits (estimates: {:?})", cost, cost_estimates);
// 3. Bill wallet on the handler's connection (same pre-request DB phase).
let billing = bill_wallet(user_token, cost, &mut mysql_connection).await?;
// Debug-log the outbound provider request BEFORE the send — still on the
// handler's connection — so the payload is captured even when the
// upload/enqueue fails.
if let Some(debug_log_type) = provider_request_debug_log_type(provider) {
insert_provider_request_debug_log(
debug_log_context,
debug_log_type,
&format!("{:#?}", draft_or_request),
&mut *mysql_connection,
).await;
}
// NB: Done with pre-request DB writes. Release the pooled connection before
// the (slow, external) provider call — holding it across that call is what
// starves the pool and causes PoolTimedOut. Post-send writes re-acquire.
drop(mysql_connection);
// 4. Upload media (if draft) and generate video.
// The entire block is wrapped so Kinovi failures trigger a refund.
// NB: No pooled DB connection is held across this call.
let result = upload_and_generate(
draft_or_request,
server_state,
media_file_to_url_map.as_ref(),
kinovi_character_id_map.as_ref(),
kinovi_account,
predownloaded_media_paths,
).await;
// 5. On failure, refund wallet for Kinovi requests.
if let Err(ref err) = result {
if matches!(provider, RouterProvider::KinoviWeb) {
if let Some(ledger_entry_token) = billing.maybe_wallet_ledger_entry_token.as_ref() {
warn!("Kinovi v2 generation failed, issuing refund for {}: {:?}", ledger_entry_token.as_str(), err);
match server_state.mysql_pool.acquire().await {
Ok(mut refund_connection) => {
if let Err(refund_err) = refund_wallet_after_api_failure(ledger_entry_token, &mut refund_connection).await {
error!("Failed to refund wallet after Kinovi v2 failure: {:?}", refund_err);
}
}
Err(acquire_err) => {
error!("Failed to acquire MySQL connection to refund wallet after Kinovi v2 failure: {:?}", acquire_err);
}
}
}
}
}
let response = result?;
info!("v2 generation response: {:?}", response);
Ok(PipelineResult { billing, response, cost_estimates })
}
/// Finalize the draft (uploading media if needed), then send the generation request.
///
/// This is the block that gets refunded on failure for Kinovi providers.
async fn upload_and_generate(
draft_or_request: VideoGenerationDraftOrRequest,
server_state: &ServerState,
media_file_urls_by_token: Option<&HashMap<MediaFileToken, String>>,
kinovi_character_ids: Option<&HashMap<CharacterToken, String>>,
kinovi_account: KinoviAccount,
predownloaded_media_paths: Option<&HashMap<String, PathBuf>>,
) -> Result<GenerateVideoResponse, CommonWebError> {
let provider = draft_or_request.get_provider();
let client = build_router_client(provider, server_state, kinovi_account)?;
let video_request = match draft_or_request {
VideoGenerationDraftOrRequest::Request(request) => request,
VideoGenerationDraftOrRequest::Draft(draft) => {
let draft_context = VideoGenerationDraftContext {
client: Some(&client),
media_file_to_artcraft_url_map: media_file_urls_by_token,
character_token_to_kinovi_id_map: kinovi_character_ids,
predownloaded_media_paths,
};
draft.finalize(draft_context)
.await
.map_err(|err| {
warn!("Failed to finalize v2 draft: {:?}", err);
map_router_error_to_web_error(err)
})?
}
};
video_request.send_request(&client)
.await
.map_err(|err| {
warn!("v2 video generation failed: {:?}", err);
map_router_error_to_web_error(err)
})
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/mod.rs
+1
View File
@@ -1,4 +1,5 @@
pub mod omni_gen_video_generate_handler;
pub mod shared_video_generation;
pub mod first_party_minimax_h3;
pub mod helpers;
pub mod pipeline_v2;
crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/omni_gen_video_generate_handler.rs
+19 -410
View File
@@ -1,59 +1,29 @@
use std::collections::HashMap;
use std::sync::Arc;
use actix_web::web::Json;
use actix_web::{web, HttpRequest};
use log::{error, info, warn};
use sqlx::Acquire;
use log::info;
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::generate::generate_video::generate_video_response::GenerateVideoResponse;
use enums::by_table::debug_logs::debug_log_type::DebugLogType;
use enums::common::generation::common_model_type::CommonModelType;
use enums::common::generation::common_video_model::CommonVideoModel;
use enums::common::platform_type::PlatformType;
use http_server_common::request::get_request_ip::get_request_ip;
use enums::by_table::debug_logs::debug_log_level::DebugLogLevel;
use mysql_queries::queries::debug_logs::insert_debug_log::{insert_debug_log, InsertDebugLogArgs};
use mysql_queries::queries::generic_inference::api_providers::kinovi_web::insert_generic_inference_job_for_kinovi_web_queue_with_apriori_job_token::KinoviVersion;
use mysql_queries::queries::idepotency_tokens::insert_idempotency_token::insert_idempotency_token;
use tokens::tokens::characters::CharacterToken;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::generate::common::generation_debug_logs::GenerationDebugLogContext;
use crate::http_server::endpoints::generate::common::payments_error_test::payments_error_test;
use crate::http_server::endpoints::generate::common::probed_reference_videos::{
download_and_probe_reference_videos, fetch_reference_video_sources, ProbedReferenceVideos,
use crate::http_server::endpoints::omni_gen::generate::video::shared_video_generation::{
run_authenticated_video_generation, VideoGenerationAuth,
};
use crate::http_server::endpoints::omni_gen::generate::video::first_party_minimax_h3::enqueue_first_party_minimax_h3_job::{
enqueue_first_party_minimax_h3_job, first_party_minimax_h3_model, EnqueueFirstPartyMinimaxH3JobArgs,
};
use crate::http_server::endpoints::omni_gen::generate::video::helpers::hydrate_router_request::hydrate_to_router_request;
use crate::http_server::endpoints::omni_gen::generate::video::helpers::resolve_kinovi_character_ids::resolve_kinovi_character_ids;
use crate::http_server::endpoints::omni_gen::generate::video::helpers::write_prompt_records::{
write_prompt_records, WritePromptRecordsArgs,
};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::insert_fal_job::{insert_fal_job, InsertFalJobArgs};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::insert_gmicloud_job::{insert_gmicloud_job, InsertGmiCloudJobArgs};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::insert_grok_api_job::{insert_grok_api_job, InsertGrokApiJobArgs};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::insert_kinovi_web_jobs::{insert_kinovi_web_jobs, InsertKinoviWebJobsArgs};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::shared_job_args::SharedJobArgs;
use crate::http_server::endpoints::omni_gen::shared_utils::kinovi_account::KinoviAccount;
use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v2::run_pipeline_v2::{run_pipeline_v2, RunPipelineV2Args};
use crate::http_server::endpoints::omni_gen::shared_utils::video::validate_video_request::validate_video_request;
use crate::http_server::user_lookup::api_or_web_session::require_api_or_web_session::{require_api_or_web_session, SessionType};
use crate::http_server::user_lookup::user_session::session_utils::lookup::user_session_feature_flags::UserSessionFeatureFlags;
use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format;
use crate::http_server::web_utils::get_request_platform_type::get_request_platform_type;
use crate::state::server_state::ServerState;
use crate::util::lookup::lookup_media_files_as_cdn_url_list_and_map::lookup_media_files_as_cdn_url_list_and_map;
/// Generate a video using the omni-gen unified endpoint.
/// Authenticates as either a web-session (cookie) user or an API-key (`Authorization` header) user.
///
/// Razor-thin handler: authenticate, then delegate to the shared
/// generation core in `shared_video_generation` (which is authoritative for
/// behavior — the omni_api endpoint delegates to the same core).
#[utoipa::path(
post,
tag = "Omni Gen",
@@ -81,12 +51,6 @@ pub async fn omni_gen_video_generate_handler(
payments_error_test(&request.prompt.as_deref().unwrap_or(""))?;
let debug_log_event_token = DebugLogEventToken::generate();
let maybe_prompt_model_type: Option<CommonModelType> = request.model
.as_ref()
.map(|m| m.to_common_model_type());
// ==================== SESSION ==================== //
let mut mysql_connection = server_state.mysql_pool.acquire().await?;
@@ -99,378 +63,23 @@ pub async fn omni_gen_video_generate_handler(
&mut *mysql_connection,
).await?;
let user_token = &session.user_token;
// AVT tokens are web-session only; API-key sessions never carry one.
let maybe_avt_token = session.maybe_avt_token.clone();
// ==================== MODEL ACCESS CHECK ==================== //
let user_feature_flags =
UserSessionFeatureFlags::new(session.maybe_feature_flags.as_deref());
// ==================== IDEMPOTENCY ==================== //
let idempotency_token = request.idempotency_token.as_deref()
.unwrap_or("")
.to_string();
if let Err(reason) = validate_idempotency_token_format(&idempotency_token) {
return Err(CommonWebError::BadInputWithSimpleMessage(reason));
}
insert_idempotency_token(&idempotency_token, &mut *mysql_connection)
.await
.map_err(|err| {
error!("Error inserting idempotency token: {:?}", err);
CommonWebError::BadInputWithSimpleMessage("repeated idempotency token".to_string())
})?;
// ==================== RESOLVE MEDIA TOKENS ==================== //
let mut all_tokens: Vec<MediaFileToken> = Vec::new();
if let Some(token) = &request.start_frame_image_media_token {
all_tokens.push(token.clone());
}
if let Some(token) = &request.end_frame_image_media_token {
all_tokens.push(token.clone());
}
if let Some(tokens) = &request.reference_image_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
if let Some(tokens) = &request.reference_video_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
if let Some(tokens) = &request.reference_audio_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
let media_file_to_url_map: Option<HashMap<MediaFileToken, String>> = if all_tokens.is_empty() {
None
} else {
info!("Resolving {} media file tokens to CDN URLs", all_tokens.len());
let resolved = lookup_media_files_as_cdn_url_list_and_map(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&all_tokens,
).await?;
Some(resolved.token_to_url_map)
};
// ==================== RESOLVE CHARACTERS ==================== //
let kinovi_character_id_map: Option<HashMap<CharacterToken, String>> =
resolve_kinovi_character_ids(
request.reference_character_tokens.as_deref(),
&mut mysql_connection,
).await?;
// ==================== REFERENCE VIDEO INPUT SECONDS ==================== //
// Seedance 2.5 bills reference-video input seconds on top of the output
// duration. Stored durations can't be trusted for billing, so every
// reference video is downloaded and ffprobed; the downloaded files are
// kept and handed to the pipeline so the Kinovi upload reuses them
// instead of downloading the same bytes twice.
let mut maybe_probed_reference_videos: Option<ProbedReferenceVideos> = None;
if matches!(request.model, Some(CommonVideoModel::Seedance2p5 | CommonVideoModel::Seedance2p5Ultra)) {
if let Some(video_tokens) = request.reference_video_media_tokens.as_deref().filter(|tokens| !tokens.is_empty()) {
let video_sources = fetch_reference_video_sources(
video_tokens,
&http_request,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&mut mysql_connection,
).await?;
// Downloads + ffprobe are slow — release the pool slot across them,
// then re-acquire for billing. Probing never fails the generation:
// unmeasurable files bill at the 30-second worst case.
drop(mysql_connection);
let probed = download_and_probe_reference_videos(&video_sources).await;
mysql_connection = server_state.mysql_pool.acquire().await?;
maybe_probed_reference_videos = Some(probed);
}
}
// ==================== HYDRATE ROUTER REQUEST ==================== //
let mut router_builder = hydrate_to_router_request(&request)?;
router_builder.total_reference_video_input_seconds =
maybe_probed_reference_videos.as_ref().map(|probed| probed.total_input_seconds);
// ==================== PIPELINE DISPATCH ==================== //
let kinovi_account = match request.model {
// BytePlus Ultra
Some(CommonVideoModel::Seedance2p0BytePlusUltra) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p0BytePlusUltraFast) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p0BytePlusUltraMini) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p5Ultra) => KinoviAccount::BytePlusUltra,
// BytePlus
Some(CommonVideoModel::Seedance2p0BytePlus) => KinoviAccount::BytePlus,
Some(CommonVideoModel::Seedance2p0BytePlusFast) => KinoviAccount::BytePlus,
Some(CommonVideoModel::Seedance2p0BytePlusMini) => KinoviAccount::BytePlus,
Some(CommonVideoModel::PreviewModel) => KinoviAccount::BytePlus,
Some(CommonVideoModel::PreviewModelFast) => KinoviAccount::BytePlus,
// Everything else goes through Volcengine
_ => KinoviAccount::Volcengine,
};
// ==================== DEBUG LOG: HTTP REQUEST ==================== //
let ip_address = get_request_ip(&http_request);
let request_url = http_request.uri().to_string();
if let Err(err) = insert_debug_log(InsertDebugLogArgs {
apriori_debug_log_event_token: Some(&debug_log_event_token),
maybe_creator_user_token: Some(user_token),
debug_log_type: DebugLogType::HttpRequest,
maybe_log_level: Some(DebugLogLevel::Info),
maybe_ip_address: Some(&ip_address),
maybe_url: Some(&request_url),
message: &serde_json::to_string(&*request).unwrap_or_default(),
mysql_executor: &mut *mysql_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert HTTP request debug log: {:?}", err);
}
// ==================== FIRST-PARTY MINIMAX H3 (TURBO / ULTRA) ==================== //
// These models run on our own GPU inference — no provider call is made.
// The module bills the wallet (Ultra only), writes the prompt + job
// records, and returns; a scheduler picks the pending jobs up later.
if let Some(minimax_model) = first_party_minimax_h3_model(request.model) {
let maybe_platform_type = match session.session_type {
SessionType::Api => Some(PlatformType::ApiKey),
SessionType::WebSession => get_request_platform_type(&http_request),
};
return enqueue_first_party_minimax_h3_job(EnqueueFirstPartyMinimaxH3JobArgs {
minimax_model,
request: &request,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_prompt_model_type,
maybe_platform_type,
idempotency_token: &idempotency_token,
ip_address: &ip_address,
debug_log_event_token: &debug_log_event_token,
mysql_connection,
}).await;
}
// NB: The pipeline takes over the connection for its remaining pre-request DB writes (billing,
// outbound provider request debug log) and releases it before the (slow, external) generation
// call — holding a pool slot across that call is what starves the pool and causes PoolTimedOut
// on unrelated endpoints. We re-acquire below to write the result.
let debug_log_context = GenerationDebugLogContext {
event_token: &debug_log_event_token,
user_token,
ip_address: &ip_address,
request_url: &request_url,
};
let pipeline_result = run_pipeline_v2(RunPipelineV2Args {
router_builder: &router_builder,
server_state: &server_state,
user_token,
media_file_to_url_map: &media_file_to_url_map,
kinovi_character_id_map: &kinovi_character_id_map,
kinovi_account,
debug_log_context: &debug_log_context,
predownloaded_media_paths: maybe_probed_reference_videos.as_ref().map(|probed| probed.local_paths_by_url()),
mysql_connection,
}).await;
// ==================== DEBUG LOG: PIPELINE ERROR ==================== //
let pipeline_result = match pipeline_result {
Ok(result) => result,
Err(err) => {
// Best-effort error log; never mask the original error.
if let Ok(mut error_log_connection) = server_state.mysql_pool.acquire().await {
if let Err(log_err) = insert_debug_log(InsertDebugLogArgs {
apriori_debug_log_event_token: Some(&debug_log_event_token),
maybe_creator_user_token: Some(user_token),
debug_log_type: DebugLogType::BackendFailure,
maybe_log_level: Some(DebugLogLevel::Error),
maybe_ip_address: Some(&ip_address),
maybe_url: Some(&request_url),
message: &format!("Video generation pipeline failed: {:?}", err),
mysql_executor: &mut *error_log_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert pipeline error debug log: {:?}", log_err);
}
}
return Err(err);
}
};
let mut mysql_connection = server_state.mysql_pool.acquire().await?;
// NB: Outbound provider requests (Fal/Grok/Kinovi) are debug-logged inside
// the pipeline BEFORE the send, so the payload is captured even on failure.
// ==================== WRITE RESULT ==================== //
let maybe_platform_type = match session.session_type {
SessionType::Api => Some(PlatformType::ApiKey),
SessionType::WebSession => get_request_platform_type(&http_request),
};
let mut transaction = mysql_connection.begin().await.map_err(|err| {
error!("Error starting MySQL transaction: {:?}", err);
CommonWebError::from_error(err)
})?;
// -- Prompt + context items --
let prompt_token = write_prompt_records(WritePromptRecordsArgs {
request: &request,
user_token,
maybe_prompt_model_type,
ip_address: &ip_address,
transaction: &mut transaction,
}).await;
// -- Inference job --
let (primary_job_token, all_job_tokens) = match &pipeline_result.response {
GenerateVideoResponse::KinoviWeb(payload) => {
info!("Inserting kinovi_web job(s) with token: {:?}", pipeline_result.billing.apriori_job_token);
let kinovi_version = match kinovi_account {
KinoviAccount::Volcengine => KinoviVersion::Volcengine,
KinoviAccount::BytePlus => KinoviVersion::BytePlus,
KinoviAccount::BytePlusUltra => KinoviVersion::BytePlusUltra,
};
let result = insert_kinovi_web_jobs(InsertKinoviWebJobsArgs {
primary_order_id: &payload.order_id,
maybe_additional_order_ids: payload.maybe_order_ids.as_deref(),
maybe_wallet_ledger_entry_token: pipeline_result.billing.maybe_wallet_ledger_entry_token.as_ref(),
kinovi_version,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(result.primary_job_token, result.all_job_tokens)
}
GenerateVideoResponse::Fal(payload) => {
let external_id = payload.request_id.as_deref().ok_or_else(|| {
error!("Fal generation response missing request_id");
CommonWebError::server_error_with_message("Fal generation response missing request_id")
})?;
info!("Inserting fal job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_fal_job(InsertFalJobArgs {
external_job_id: external_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
GenerateVideoResponse::Artcraft(payload) => {
(
payload.inference_job_token.clone(),
vec![payload.inference_job_token.clone()],
)
}
GenerateVideoResponse::GmiCloud(payload) => {
info!("Inserting GmiCloud job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_gmicloud_job(InsertGmiCloudJobArgs {
external_request_id: &payload.request_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
GenerateVideoResponse::Grok(payload) => {
info!("Inserting Grok (xAI) API job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_grok_api_job(InsertGrokApiJobArgs {
external_request_id: &payload.request_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
other => {
error!("Unexpected generation response variant: {:?}", other);
return Err(CommonWebError::server_error_with_message("Unexpected generation response"));
}
let auth = VideoGenerationAuth {
user_token: &session.user_token,
// AVT tokens are web-session only; API-key sessions never carry one.
maybe_avt_token: session.maybe_avt_token.clone(),
maybe_platform_type,
};
transaction.commit().await.map_err(|err| {
error!("Error committing transaction: {:?}", err);
CommonWebError::from_error(err)
})?;
Ok(Json(OmniGenVideoGenerateResponse {
success: true,
inference_job_token: primary_job_token,
all_job_tokens,
}))
run_authenticated_video_generation(
&http_request,
&request,
&server_state,
auth,
mysql_connection,
).await
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/shared_video_generation.rs
+452
View File
@@ -0,0 +1,452 @@
//! The shared video-generation core behind BOTH generate endpoints:
//!
//! - `/v1/omni_gen/generate/video` (web session or API key)
//! - `/v1/omni_api/generate/video` (API key only; URL inputs pre-ingested)
//!
//! The handlers are razor thin: they authenticate, build a
//! [`VideoGenerationAuth`], and delegate here. THIS module is authoritative
//! for generation behavior — validation of the omni_gen request shape,
//! idempotency, media/character resolution, reference-video probing,
//! billing, provider dispatch, and job/prompt record writing.
use std::collections::HashMap;
use std::sync::Arc;
use actix_web::web::Json;
use actix_web::{web, HttpRequest};
use log::{error, info, warn};
use sqlx::pool::PoolConnection;
use sqlx::{Acquire, MySql};
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::generate::generate_video::generate_video_response::GenerateVideoResponse;
use enums::by_table::debug_logs::debug_log_level::DebugLogLevel;
use enums::by_table::debug_logs::debug_log_type::DebugLogType;
use enums::common::generation::common_model_type::CommonModelType;
use enums::common::generation::common_video_model::CommonVideoModel;
use enums::common::platform_type::PlatformType;
use http_server_common::request::get_request_ip::get_request_ip;
use mysql_queries::queries::debug_logs::insert_debug_log::{insert_debug_log, InsertDebugLogArgs};
use mysql_queries::queries::generic_inference::api_providers::kinovi_web::insert_generic_inference_job_for_kinovi_web_queue_with_apriori_job_token::KinoviVersion;
use mysql_queries::queries::idepotency_tokens::insert_idempotency_token::insert_idempotency_token;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
use tokens::tokens::characters::CharacterToken;
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken;
use tokens::tokens::users::UserToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::generate::common::generation_debug_logs::GenerationDebugLogContext;
use crate::http_server::endpoints::generate::common::probed_reference_videos::{
download_and_probe_reference_videos, fetch_reference_video_sources, ProbedReferenceVideos,
};
use crate::http_server::endpoints::omni_gen::generate::video::first_party_minimax_h3::enqueue_first_party_minimax_h3_job::{
enqueue_first_party_minimax_h3_job, first_party_minimax_h3_model, EnqueueFirstPartyMinimaxH3JobArgs,
};
use crate::http_server::endpoints::omni_gen::generate::video::helpers::hydrate_router_request::hydrate_to_router_request;
use crate::http_server::endpoints::omni_gen::generate::video::helpers::resolve_kinovi_character_ids::resolve_kinovi_character_ids;
use crate::http_server::endpoints::omni_gen::generate::video::helpers::write_prompt_records::{
write_prompt_records, WritePromptRecordsArgs,
};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::insert_fal_job::{insert_fal_job, InsertFalJobArgs};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::insert_gmicloud_job::{insert_gmicloud_job, InsertGmiCloudJobArgs};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::insert_grok_api_job::{insert_grok_api_job, InsertGrokApiJobArgs};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::insert_kinovi_web_jobs::{insert_kinovi_web_jobs, InsertKinoviWebJobsArgs};
use crate::http_server::endpoints::omni_gen::generate::video::insert_db_job::shared_job_args::SharedJobArgs;
use crate::http_server::endpoints::omni_gen::generate::video::pipeline_v2::run_pipeline_v2::{run_pipeline_v2, RunPipelineV2Args};
use crate::http_server::endpoints::omni_gen::shared_utils::kinovi_account::KinoviAccount;
use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format;
use crate::state::server_state::ServerState;
use crate::util::lookup::lookup_media_files_as_cdn_url_list_and_map::lookup_media_files_as_cdn_url_list_and_map;
/// Who is generating, as established by the calling handler's authentication.
pub struct VideoGenerationAuth<'a> {
pub user_token: &'a UserToken,
/// Web sessions only; API-key callers pass None.
pub maybe_avt_token: Option<AnonymousVisitorTrackingToken>,
/// API-key callers pass `Some(PlatformType::ApiKey)`; web sessions infer
/// the platform from the request.
pub maybe_platform_type: Option<PlatformType>,
}
/// Run one authenticated video generation end to end and return the HTTP
/// response body. `mysql_connection` is the handler's already-open
/// connection (it authenticated on it); the pipeline manages its lifetime
/// from here.
pub async fn run_authenticated_video_generation(
http_request: &HttpRequest,
request: &OmniGenVideoCostAndGenerateRequest,
server_state: &web::Data<Arc<ServerState>>,
auth: VideoGenerationAuth<'_>,
mut mysql_connection: PoolConnection<MySql>,
) -> Result<Json<OmniGenVideoGenerateResponse>, CommonWebError> {
let debug_log_event_token = DebugLogEventToken::generate();
let maybe_prompt_model_type: Option<CommonModelType> = request.model
.as_ref()
.map(|m| m.to_common_model_type());
let VideoGenerationAuth {
user_token,
maybe_avt_token,
maybe_platform_type,
} = auth;
// ==================== IDEMPOTENCY ==================== //
let idempotency_token = request.idempotency_token.as_deref()
.unwrap_or("")
.to_string();
if let Err(reason) = validate_idempotency_token_format(&idempotency_token) {
return Err(CommonWebError::BadInputWithSimpleMessage(reason));
}
insert_idempotency_token(&idempotency_token, &mut *mysql_connection)
.await
.map_err(|err| {
error!("Error inserting idempotency token: {:?}", err);
CommonWebError::BadInputWithSimpleMessage("repeated idempotency token".to_string())
})?;
// ==================== RESOLVE MEDIA TOKENS ==================== //
let mut all_tokens: Vec<MediaFileToken> = Vec::new();
if let Some(token) = &request.start_frame_image_media_token {
all_tokens.push(token.clone());
}
if let Some(token) = &request.end_frame_image_media_token {
all_tokens.push(token.clone());
}
if let Some(tokens) = &request.reference_image_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
if let Some(tokens) = &request.reference_video_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
if let Some(tokens) = &request.reference_audio_media_tokens {
all_tokens.extend(tokens.iter().cloned());
}
let media_file_to_url_map: Option<HashMap<MediaFileToken, String>> = if all_tokens.is_empty() {
None
} else {
info!("Resolving {} media file tokens to CDN URLs", all_tokens.len());
let resolved = lookup_media_files_as_cdn_url_list_and_map(
&http_request,
&mut mysql_connection,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&all_tokens,
).await?;
Some(resolved.token_to_url_map)
};
// ==================== RESOLVE CHARACTERS ==================== //
let kinovi_character_id_map: Option<HashMap<CharacterToken, String>> =
resolve_kinovi_character_ids(
request.reference_character_tokens.as_deref(),
&mut mysql_connection,
).await?;
// ==================== REFERENCE VIDEO INPUT SECONDS ==================== //
// Seedance 2.5 bills reference-video input seconds on top of the output
// duration. Stored durations can't be trusted for billing, so every
// reference video is downloaded and ffprobed; the downloaded files are
// kept and handed to the pipeline so the Kinovi upload reuses them
// instead of downloading the same bytes twice.
let mut maybe_probed_reference_videos: Option<ProbedReferenceVideos> = None;
if matches!(request.model, Some(CommonVideoModel::Seedance2p5 | CommonVideoModel::Seedance2p5Ultra)) {
if let Some(video_tokens) = request.reference_video_media_tokens.as_deref().filter(|tokens| !tokens.is_empty()) {
let video_sources = fetch_reference_video_sources(
video_tokens,
&http_request,
server_state.server_environment,
server_state.maybe_media_cdn_override_url.as_deref(),
&mut mysql_connection,
).await?;
// Downloads + ffprobe are slow — release the pool slot across them,
// then re-acquire for billing. Probing never fails the generation:
// unmeasurable files bill at the 30-second worst case.
drop(mysql_connection);
let probed = download_and_probe_reference_videos(&video_sources).await;
mysql_connection = server_state.mysql_pool.acquire().await?;
maybe_probed_reference_videos = Some(probed);
}
}
// ==================== HYDRATE ROUTER REQUEST ==================== //
let mut router_builder = hydrate_to_router_request(&request)?;
router_builder.total_reference_video_input_seconds =
maybe_probed_reference_videos.as_ref().map(|probed| probed.total_input_seconds);
// ==================== PIPELINE DISPATCH ==================== //
let kinovi_account = match request.model {
// BytePlus Ultra
Some(CommonVideoModel::Seedance2p0BytePlusUltra) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p0BytePlusUltraFast) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p0BytePlusUltraMini) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p5Ultra) => KinoviAccount::BytePlusUltra,
// BytePlus
Some(CommonVideoModel::Seedance2p0BytePlus) => KinoviAccount::BytePlus,
Some(CommonVideoModel::Seedance2p0BytePlusFast) => KinoviAccount::BytePlus,
Some(CommonVideoModel::Seedance2p0BytePlusMini) => KinoviAccount::BytePlus,
Some(CommonVideoModel::PreviewModel) => KinoviAccount::BytePlus,
Some(CommonVideoModel::PreviewModelFast) => KinoviAccount::BytePlus,
// Everything else goes through Volcengine
_ => KinoviAccount::Volcengine,
};
// ==================== DEBUG LOG: HTTP REQUEST ==================== //
let ip_address = get_request_ip(&http_request);
let request_url = http_request.uri().to_string();
if let Err(err) = insert_debug_log(InsertDebugLogArgs {
apriori_debug_log_event_token: Some(&debug_log_event_token),
maybe_creator_user_token: Some(user_token),
debug_log_type: DebugLogType::HttpRequest,
maybe_log_level: Some(DebugLogLevel::Info),
maybe_ip_address: Some(&ip_address),
maybe_url: Some(&request_url),
message: &serde_json::to_string(&*request).unwrap_or_default(),
mysql_executor: &mut *mysql_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert HTTP request debug log: {:?}", err);
}
// ==================== FIRST-PARTY MINIMAX H3 (TURBO / ULTRA) ==================== //
// These models run on our own GPU inference — no provider call is made.
// The module bills the wallet (Ultra only), writes the prompt + job
// records, and returns; a scheduler picks the pending jobs up later.
if let Some(minimax_model) = first_party_minimax_h3_model(request.model) {
return enqueue_first_party_minimax_h3_job(EnqueueFirstPartyMinimaxH3JobArgs {
minimax_model,
request: &request,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_prompt_model_type,
maybe_platform_type,
idempotency_token: &idempotency_token,
ip_address: &ip_address,
debug_log_event_token: &debug_log_event_token,
mysql_connection,
}).await;
}
// NB: The pipeline takes over the connection for its remaining pre-request DB writes (billing,
// outbound provider request debug log) and releases it before the (slow, external) generation
// call — holding a pool slot across that call is what starves the pool and causes PoolTimedOut
// on unrelated endpoints. We re-acquire below to write the result.
let debug_log_context = GenerationDebugLogContext {
event_token: &debug_log_event_token,
user_token,
ip_address: &ip_address,
request_url: &request_url,
};
let pipeline_result = run_pipeline_v2(RunPipelineV2Args {
router_builder: &router_builder,
server_state: &server_state,
user_token,
media_file_to_url_map: &media_file_to_url_map,
kinovi_character_id_map: &kinovi_character_id_map,
kinovi_account,
debug_log_context: &debug_log_context,
predownloaded_media_paths: maybe_probed_reference_videos.as_ref().map(|probed| probed.local_paths_by_url()),
mysql_connection,
}).await;
// ==================== DEBUG LOG: PIPELINE ERROR ==================== //
let pipeline_result = match pipeline_result {
Ok(result) => result,
Err(err) => {
// Best-effort error log; never mask the original error.
if let Ok(mut error_log_connection) = server_state.mysql_pool.acquire().await {
if let Err(log_err) = insert_debug_log(InsertDebugLogArgs {
apriori_debug_log_event_token: Some(&debug_log_event_token),
maybe_creator_user_token: Some(user_token),
debug_log_type: DebugLogType::BackendFailure,
maybe_log_level: Some(DebugLogLevel::Error),
maybe_ip_address: Some(&ip_address),
maybe_url: Some(&request_url),
message: &format!("Video generation pipeline failed: {:?}", err),
mysql_executor: &mut *error_log_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert pipeline error debug log: {:?}", log_err);
}
}
return Err(err);
}
};
let mut mysql_connection = server_state.mysql_pool.acquire().await?;
// NB: Outbound provider requests (Fal/Grok/Kinovi) are debug-logged inside
// the pipeline BEFORE the send, so the payload is captured even on failure.
// ==================== WRITE RESULT ==================== //
let mut transaction = mysql_connection.begin().await.map_err(|err| {
error!("Error starting MySQL transaction: {:?}", err);
CommonWebError::from_error(err)
})?;
// -- Prompt + context items --
let prompt_token = write_prompt_records(WritePromptRecordsArgs {
request: &request,
user_token,
maybe_prompt_model_type,
ip_address: &ip_address,
transaction: &mut transaction,
}).await;
// -- Inference job --
let (primary_job_token, all_job_tokens) = match &pipeline_result.response {
GenerateVideoResponse::KinoviWeb(payload) => {
info!("Inserting kinovi_web job(s) with token: {:?}", pipeline_result.billing.apriori_job_token);
let kinovi_version = match kinovi_account {
KinoviAccount::Volcengine => KinoviVersion::Volcengine,
KinoviAccount::BytePlus => KinoviVersion::BytePlus,
KinoviAccount::BytePlusUltra => KinoviVersion::BytePlusUltra,
};
let result = insert_kinovi_web_jobs(InsertKinoviWebJobsArgs {
primary_order_id: &payload.order_id,
maybe_additional_order_ids: payload.maybe_order_ids.as_deref(),
maybe_wallet_ledger_entry_token: pipeline_result.billing.maybe_wallet_ledger_entry_token.as_ref(),
kinovi_version,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(result.primary_job_token, result.all_job_tokens)
}
GenerateVideoResponse::Fal(payload) => {
let external_id = payload.request_id.as_deref().ok_or_else(|| {
error!("Fal generation response missing request_id");
CommonWebError::server_error_with_message("Fal generation response missing request_id")
})?;
info!("Inserting fal job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_fal_job(InsertFalJobArgs {
external_job_id: external_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
GenerateVideoResponse::Artcraft(payload) => {
(
payload.inference_job_token.clone(),
vec![payload.inference_job_token.clone()],
)
}
GenerateVideoResponse::GmiCloud(payload) => {
info!("Inserting GmiCloud job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_gmicloud_job(InsertGmiCloudJobArgs {
external_request_id: &payload.request_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
GenerateVideoResponse::Grok(payload) => {
info!("Inserting Grok (xAI) API job with token: {:?}", pipeline_result.billing.apriori_job_token);
let token = insert_grok_api_job(InsertGrokApiJobArgs {
external_request_id: &payload.request_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.billing.apriori_job_token,
idempotency_token: &idempotency_token,
user_token,
maybe_avt_token: maybe_avt_token.as_ref(),
maybe_model_type: request.model.map(|v| v.to_common_model_type()),
maybe_prompt_token: prompt_token.as_ref(),
maybe_debug_log_event_token: Some(&debug_log_event_token),
maybe_platform_type,
maybe_cost_estimates: Some(pipeline_result.cost_estimates),
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
(
token.clone(),
vec![token],
)
}
other => {
error!("Unexpected generation response variant: {:?}", other);
return Err(CommonWebError::server_error_with_message("Unexpected generation response"));
}
};
transaction.commit().await.map_err(|err| {
error!("Error committing transaction: {:?}", err);
CommonWebError::from_error(err)
})?;
Ok(Json(OmniGenVideoGenerateResponse {
success: true,
inference_job_token: primary_job_token,
all_job_tokens,
}))
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/tests/mod.rs
+1
View File
@@ -25,5 +25,6 @@
pub mod support;
mod omni_api_parity_tests;
mod seedance_2p0;
mod seedance_2p5;
crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/tests/omni_api_parity_tests.rs
+197
View File
@@ -0,0 +1,197 @@
//! Parity tests for the omni_api (API-key only) generate endpoint.
//!
//! The omni_api handler is a razor-thin wrapper that delegates to the same
//! shared generation core as omni_gen. These tests pin that: the same
//! request bills the SAME credits through either endpoint, and the omni_api
//! endpoint only accepts API-key authentication (session cookies are
//! rejected before any billable work).
use enums::common::generation::common_resolution::CommonResolution;
use enums::common::generation::common_video_model::CommonVideoModel;
use artcraft_api_keys::ArtcraftApiKey;
use crate::http_server::endpoints::omni_gen::generate::video::tests::support::{
base_generate_request, to_omni_api_request, ExpectedCredits, Seconds, TestHarness,
STARTING_CREDITS,
};
mod billing_parity {
use super::*;
/// Base Seedance 2.0 at 720p 5s — same request, both endpoints, same debit.
#[tokio::test]
#[cfg_attr(feature = "skip_database_tests", ignore)]
async fn seedance_2p0_bills_identically_on_both_endpoints() {
assert_endpoints_bill_identically(
CommonVideoModel::Seedance2p0,
Some(CommonResolution::SevenTwentyP),
Seconds(5),
ExpectedCredits(80),
).await;
}
/// BytePlus Ultra at 720p 5s — the canonical collapse-bug shape (125, NOT
/// the base model's 80). If the omni_api path ever bills the collapsed
/// model again, this fails.
#[tokio::test]
#[cfg_attr(feature = "skip_database_tests", ignore)]
async fn seedance_2p0_byteplus_ultra_bills_identically_on_both_endpoints() {
assert_endpoints_bill_identically(
CommonVideoModel::Seedance2p0BytePlusUltra,
Some(CommonResolution::SevenTwentyP),
Seconds(5),
ExpectedCredits(125),
).await;
}
}
mod url_ingestion {
use super::*;
/// The omni_api-only URL-input flow, end to end: a stub-hosted video URL
/// is ingested (download → bucket upload → media_files row owned by the
/// API user), then billed exactly like the same request made with a media
/// TOKEN through omni_gen. Seedance 2.0 at 720p 5s with one reference
/// video: 111 credits on both paths, fully successful, no refund.
#[tokio::test]
#[cfg_attr(feature = "skip_database_tests", ignore)]
async fn reference_video_url_bills_like_a_media_token() {
const EXPECTED_CREDITS: u64 = 111;
let harness = TestHarness::create().await;
// omni_gen with a probeable media TOKEN pointing at the 5s stub fixture.
let web_user = harness.create_funded_user(STARTING_CREDITS).await;
let video_token = mysql_testing::fixtures::media_files::create_probeable_test_video_media_file(
&harness.pool,
&web_user.user_token,
5_000,
)
.await
.expect("probeable video fixture");
let mut request = base_generate_request(CommonVideoModel::Seedance2p0);
request.resolution = Some(CommonResolution::SevenTwentyP);
request.duration_seconds = Some(5);
request.reference_video_media_tokens = Some(vec![video_token]);
let response = harness
.post_generate(&web_user, request)
.await
.expect("omni_gen generation with reference token");
assert!(response.success);
// omni_api with a URL to the same 5s fixture, no token.
let api_user = harness.create_funded_user(STARTING_CREDITS).await;
let api_key = harness.create_api_key(&api_user).await;
let mut base = base_generate_request(CommonVideoModel::Seedance2p0);
base.resolution = Some(CommonResolution::SevenTwentyP);
base.duration_seconds = Some(5);
let mut api_request = to_omni_api_request(base);
api_request.reference_video_urls =
Some(vec![format!("{}/media/dur5000ms/source.mp4", harness.stub_kinovi_base_url)]);
let response = harness
.post_omni_api_request(&api_key, api_request)
.await
.expect("omni_api generation with reference URL");
assert!(response.success);
let web_debit = STARTING_CREDITS - harness.wallet_balance(&web_user).await;
let api_debit = STARTING_CREDITS - harness.wallet_balance(&api_user).await;
assert_eq!(web_debit, EXPECTED_CREDITS, "omni_gen token reference debited the wrong amount");
assert_eq!(api_debit, EXPECTED_CREDITS, "omni_api URL reference debited the wrong amount");
let entries = harness.ledger_entries(&api_user).await;
let debit = entries
.iter()
.find(|entry| entry.credits_delta < 0)
.expect("omni_api URL generation must have a debit ledger entry");
assert!(!debit.is_refunded, "successful URL-input generation must not be refunded");
}
}
mod api_key_only_auth {
use super::*;
/// A key that was never inserted is a 401, and nothing is charged.
#[tokio::test]
#[cfg_attr(feature = "skip_database_tests", ignore)]
async fn unknown_api_key_is_rejected() {
let harness = TestHarness::create().await;
let bogus_key = ArtcraftApiKey::new_from_str("artcraft_api_test_never_inserted_key");
let request = base_generate_request(CommonVideoModel::Seedance2p0);
let result = harness.post_generate_via_api_key(&bogus_key, request).await;
assert!(result.is_err(), "unknown API key must be rejected");
}
/// The omni_api endpoint is API-key ONLY: a valid session cookie (which
/// omni_gen accepts) must be rejected, and nothing charged.
#[tokio::test]
#[cfg_attr(feature = "skip_database_tests", ignore)]
async fn session_cookie_is_rejected() {
let harness = TestHarness::create().await;
let user = harness.create_funded_user(STARTING_CREDITS).await;
let request = base_generate_request(CommonVideoModel::Seedance2p0);
let result = harness
.post_generate_via_session_cookie_on_omni_api(&user, request)
.await;
assert!(result.is_err(), "session cookies must not authenticate omni_api");
assert_eq!(
harness.wallet_balance(&user).await,
STARTING_CREDITS,
"rejected session-cookie request must not charge",
);
}
}
/// Run the same generation through omni_gen (session cookie) and omni_api
/// (API key) as two independent fixture users; assert both succeed and both
/// debit exactly `expected_credits`.
async fn assert_endpoints_bill_identically(
model: CommonVideoModel,
resolution: Option<CommonResolution>,
Seconds(duration_seconds): Seconds,
ExpectedCredits(expected_credits): ExpectedCredits,
) {
let harness = TestHarness::create().await;
// omni_gen, session-cookie user.
let web_user = harness.create_funded_user(STARTING_CREDITS).await;
let mut request = base_generate_request(model);
request.resolution = resolution;
request.duration_seconds = Some(duration_seconds);
let response = harness
.post_generate(&web_user, request)
.await
.unwrap_or_else(|err| panic!("{:?}: omni_gen generation failed: {:?}", model, err));
assert!(response.success);
// omni_api, API-key user.
let api_user = harness.create_funded_user(STARTING_CREDITS).await;
let api_key = harness.create_api_key(&api_user).await;
let mut request = base_generate_request(model);
request.resolution = resolution;
request.duration_seconds = Some(duration_seconds);
let response = harness
.post_generate_via_api_key(&api_key, request)
.await
.unwrap_or_else(|err| panic!("{:?}: omni_api generation failed: {:?}", model, err));
assert!(response.success);
let web_debit = STARTING_CREDITS - harness.wallet_balance(&web_user).await;
let api_debit = STARTING_CREDITS - harness.wallet_balance(&api_user).await;
assert_eq!(
web_debit, expected_credits,
"{:?}: omni_gen debited the wrong amount", model,
);
assert_eq!(
api_debit, expected_credits,
"{:?}: omni_api debited the wrong amount", model,
);
}
crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/tests/support.rs
+154 -4
View File
@@ -4,6 +4,7 @@
//! with dummy Actix HTTP requests as a fixture user.
use std::collections::HashSet;
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;
@@ -18,7 +19,9 @@ use actix_artcraft::sessions::anonymous_visitor_tracking::avt_cookie_manager::Av
use actix_artcraft::sessions::user_sessions::http_user_session_manager::HttpUserSessionManager;
use actix_helpers::middleware::banned_cidr_filter::banned_cidr_set::BannedCidrSet;
use actix_helpers::middleware::banned_ip_filter::ip_ban_list::ip_ban_list::IpBanList;
use artcraft_api_defs::omni_api::generate_requests::omni_api_video_generate_request::OmniApiVideoGenerateRequest;
use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_video_cost_and_generate_request::OmniGenVideoCostAndGenerateRequest;
use artcraft_api_keys::ArtcraftApiKey;
use artcraft_api_defs::omni_gen::generate_response::omni_gen_video_generate_response::OmniGenVideoGenerateResponse;
use billing_artcraft_component::utils::artcraft_stripe_config::ArtcraftStripeConfig;
use billing_component::stripe::stripe_config::{
@@ -34,6 +37,7 @@ use kinovi_web_client::requests::kinovi_host::{
use memory_caching::arc_ttl_sieve::ArcTtlSieve;
use memory_caching::single_item_ttl_cache::SingleItemTtlCache;
use mysql_queries::mediators::badge_granter::BadgeGranter;
use mysql_queries::queries::api_keys::insert_api_key::{insert_api_key, InsertApiKeyArgs};
use mysql_queries::mediators::firehose_publisher::FirehosePublisher;
use mysql_testing::fixtures::users::TestUser;
use redis_caching::redis_ttl_cache::RedisTtlCache;
@@ -43,6 +47,7 @@ use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
use crate::configs::app_startup::redis_rate_limiters::configure_redis_rate_limiters;
use crate::configs::static_api_tokens::StaticApiTokenSet;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::omni_api::generate::video::omni_api_video_generate_handler::omni_api_video_generate_handler;
use crate::http_server::endpoints::omni_gen::generate::video::omni_gen_video_generate_handler::omni_gen_video_generate_handler;
use crate::http_server::user_lookup::user_session::session_utils::session_checker::SessionChecker;
use crate::http_server::web_utils::scoped_temp_dir_creator::ScopedTempDirCreator;
@@ -147,6 +152,92 @@ impl TestHarness {
.map(Json::into_inner)
}
/// Mint and insert an API key owned by `user`.
pub async fn create_api_key(&self, user: &TestUser) -> ArtcraftApiKey {
// Derive the key value from the (unique) user token so parallel tests
// never collide on the api_key column.
let entropy = user.user_token.as_str().trim_start_matches("user_").to_string();
let api_key = ArtcraftApiKey::new_from_str(&format!("artcraft_api_test{entropy}"));
let mut connection = self.pool.acquire().await.expect("acquire for api key");
insert_api_key(InsertApiKeyArgs {
owner_user_token: &user.user_token,
ip_address: "127.0.0.1",
name: "test key",
maybe_description: None,
api_key: &api_key,
mysql_executor: &mut *connection,
phantom: PhantomData,
})
.await
.expect("insert api key");
api_key
}
/// POST the request through the omni_api (API-key only) generate handler,
/// authenticating with `api_key` in the Authorization header. Token-based
/// fields are mapped onto the omni_api request shape; URL fields stay None.
pub async fn post_generate_via_api_key(
&self,
api_key: &ArtcraftApiKey,
request: OmniGenVideoCostAndGenerateRequest,
) -> Result<OmniGenVideoGenerateResponse, CommonWebError> {
self.post_omni_api_request(api_key, to_omni_api_request(request)).await
}
/// POST a raw omni_api request — URL input fields allowed — with API-key
/// authentication. Use this to exercise the URL-ingestion path.
pub async fn post_omni_api_request(
&self,
api_key: &ArtcraftApiKey,
api_request: OmniApiVideoGenerateRequest,
) -> Result<OmniGenVideoGenerateResponse, CommonWebError> {
let http_request = TestRequest::post()
.uri("/v1/omni_api/generate/video")
.insert_header(("Authorization", format!("Bearer {}", api_key.as_str_be_careful())))
.peer_addr("127.0.0.1:9999".parse().expect("peer addr"))
.to_http_request();
omni_api_video_generate_handler(
http_request,
Json(api_request),
Data::new(self.server_state.clone()),
)
.await
.map(Json::into_inner)
}
/// POST through the omni_api handler with a SESSION COOKIE and no API key.
/// The endpoint is API-key only, so this must fail — the test asserting
/// that lives in `omni_api_parity_tests`.
pub async fn post_generate_via_session_cookie_on_omni_api(
&self,
user: &TestUser,
request: OmniGenVideoCostAndGenerateRequest,
) -> Result<OmniGenVideoGenerateResponse, CommonWebError> {
let cookie = self
.server_state
.session_cookie_manager
.create_cookie(&user.session_token, &user.user_token)
.expect("create session cookie");
let cookie = Cookie::new("session", cookie.value().to_string());
let http_request = TestRequest::post()
.uri("/v1/omni_api/generate/video")
.cookie(cookie)
.peer_addr("127.0.0.1:9999".parse().expect("peer addr"))
.to_http_request();
let api_request = to_omni_api_request(request);
omni_api_video_generate_handler(
http_request,
Json(api_request),
Data::new(self.server_state.clone()),
)
.await
.map(Json::into_inner)
}
/// A dummy HTTP request carrying the user's session cookie.
fn authed_request(&self, user: &TestUser) -> HttpRequest {
let cookie = self
@@ -242,8 +333,12 @@ fn process_stub_kinovi_base_url() -> &'static String {
/// - `GET` paths containing `dur{millis}ms` serve a REAL generated video of
/// that duration (ffmpeg, cached per duration) — the reference-video
/// billing probe downloads and ffprobes these.
/// - `GET` paths under the `video_` bucket prefix (URL-ingested uploads)
/// serve a default 5s fixture video; other GETs are 404 so the
/// unreachable-token fixtures keep failing as intended.
/// - `POST …uploads.signedUploadUrl` answers with an upload URL back at this
/// stub; the subsequent `PUT` is accepted with 200.
/// stub; any `PUT` (that upload, or an S3 PutObject from the test bucket
/// client) is accepted with 200.
/// - Any other `POST` answers as a successful workflow.runTask with
/// process-unique order ids.
fn spawn_stub_server() -> String {
@@ -291,6 +386,13 @@ fn spawn_stub_server() -> String {
if request_line.starts_with("GET") {
match extract_fixture_duration_millis(&request_line) {
Some(millis) => ("200 OK", "video/mp4", fixture_video_bytes(millis)),
// URL-ingested uploads land under the `video_` bucket prefix
// (see MediaUploadKind::bucket_prefix); serve those a default
// 5s fixture. Token-fixture paths without a prefix stay 404 —
// the charge-then-refund tests depend on that.
None if request_line.contains("/video_") => {
("200 OK", "video/mp4", fixture_video_bytes(5_000))
}
None => ("404 Not Found", "text/plain", b"not found".to_vec()),
}
} else if request_line.starts_with("PUT") {
@@ -413,14 +515,17 @@ fn build_test_server_state(pool: MySqlPool, media_cdn_override_url: String) -> S
}
use std::sync::OnceLock;
static DUMMIES: OnceLock<ExpensiveDummies> = OnceLock::new();
let dummies = DUMMIES.get_or_init(|| {
let dummies = DUMMIES.get_or_init(|| retry_keychain_flake(|| {
// Bucket uploads (e.g. URL-input ingestion) PUT to the stub, which
// accepts anything; downloads then come back through the media CDN
// override, also the stub.
let dummy_bucket = || {
LegacyBucketClient::create(
"test-access-key",
"test-secret-key",
"auto",
"test-bucket",
"http://127.0.0.1:1",
&media_cdn_override_url,
None,
Some(Duration::from_secs(5)),
)
@@ -463,7 +568,7 @@ fn build_test_server_state(pool: MySqlPool, media_cdn_override_url: String) -> S
auto_gc_bucket_client: dummy_bucket(),
redis_rate_limiters: configure_redis_rate_limiters().expect("rate limiters"),
}
});
}));
ServerState {
env_config: EnvConfig {
@@ -775,3 +880,48 @@ pub async fn assert_real_input_videos_charge(
.unwrap_or_else(|| panic!("{:?}: no debit ledger entry found", model));
assert!(!debit.is_refunded, "{:?}: successful generation must not be refunded", model);
}
/// Map an omni_gen request onto the omni_api request shape (token fields
/// only; the URL fields start None — tests exercising URL ingestion set
/// them afterwards).
pub fn to_omni_api_request(request: OmniGenVideoCostAndGenerateRequest) -> OmniApiVideoGenerateRequest {
OmniApiVideoGenerateRequest {
idempotency_token: request.idempotency_token,
model: request.model,
prompt: request.prompt,
negative_prompt: request.negative_prompt,
start_frame_image_media_token: request.start_frame_image_media_token,
start_frame_image_url: None,
end_frame_image_media_token: request.end_frame_image_media_token,
end_frame_image_url: None,
reference_image_media_tokens: request.reference_image_media_tokens,
reference_image_urls: None,
reference_video_media_tokens: request.reference_video_media_tokens,
reference_video_urls: None,
reference_audio_media_tokens: request.reference_audio_media_tokens,
reference_audio_urls: None,
reference_character_tokens: request.reference_character_tokens,
resolution: request.resolution,
aspect_ratio: request.aspect_ratio,
bitrate: request.bitrate,
quality: request.quality,
duration_seconds: request.duration_seconds,
video_batch_count: request.video_batch_count,
generate_audio: request.generate_audio,
}
}
/// Construct a value whose builder loads the macOS keychain (async-stripe's
/// TLS connector does), retrying on the intermittent keychain I/O failure
/// (error -36) that occurs while dozens of parallel tests hit it at once.
/// The builders panic internally on that failure, so retry via catch_unwind;
/// the final attempt runs unguarded so a real error still fails the test.
fn retry_keychain_flake<T>(build: impl Fn() -> T) -> T {
for _ in 0..20 {
match std::panic::catch_unwind(std::panic::AssertUnwindSafe(&build)) {
Ok(value) => return value,
Err(_) => std::thread::sleep(Duration::from_millis(50)),
}
}
build()
}