omni api image and video (#1652)

This commit is contained in:
Brandon Thomas authored and GitHub committed 2026-06-15 22:08:31 -04:00
1 parent e30b4366ce
commit 1e20f2bfff
47 files changed
+2680

No files matched your search

@@ -0,0 +1,47 @@
{
"db_name": "MySQL",
"query": "\nSELECT\n api_keys.token as `api_key_token: ApiKeyToken`,\n users.token as `user_token: UserToken`,\n users.is_banned\nFROM api_keys\nINNER JOIN users\n ON users.token = api_keys.owner_user_token\nWHERE api_keys.api_key = ?\n AND api_keys.maybe_deleted_at IS NULL\nLIMIT 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "api_key_token: ApiKeyToken",
"type_info": {
"type": "VarString",
"flags": "NOT_NULL | UNIQUE_KEY | BINARY | NO_DEFAULT_VALUE",
"char_set": 224,
"max_size": 128
}
},
{
"ordinal": 1,
"name": "user_token: UserToken",
"type_info": {
"type": "VarString",
"flags": "NOT_NULL | UNIQUE_KEY | BINARY | NO_DEFAULT_VALUE",
"char_set": 224,
"max_size": 128
}
},
{
"ordinal": 2,
"name": "is_banned",
"type_info": {
"type": "Tiny",
"flags": "NOT_NULL",
"char_set": 63,
"max_size": 1
}
}
],
"parameters": {
"Right": 1
},
"nullable": [
false,
false,
false
]
},
"hash": "f52c495a7ff5cafb357f79c4cdb503bbbef045fd193a5a37a66180fcac77f363"
}
@@ -0,0 +1,63 @@
use std::marker::PhantomData;
use sqlx::{Executor, MySql};
use artcraft_api_keys::ArtcraftApiKey;
use tokens::tokens::api_keys::ApiKeyToken;
use tokens::tokens::users::UserToken;
use crate::helpers::boolean_converters::i8_to_bool;
pub struct GetApiKeyUserByValueArgs<'e, 'c, E>
where
E: 'e + Executor<'c, Database = MySql>,
{
/// The actual API key secret presented by the client.
pub api_key: &'e ArtcraftApiKey,
pub mysql_executor: E,
pub phantom: PhantomData<&'c E>,
}
/// A live API key joined to its owning user. This is the minimal identity needed to authenticate
/// an API-key request — far smaller than a cookie `SessionUserRecord`.
pub struct ApiKeyUserRecord {
pub api_key_token: ApiKeyToken,
pub user_token: UserToken,
pub is_banned: bool,
}
/// Look up the user behind an API key secret, for authenticating an incoming request.
///
/// Returns `Ok(None)` when no row matches, when the matching API key is soft-deleted
/// (`maybe_deleted_at IS NOT NULL`), or when the owning user row does not exist (the `INNER JOIN`
/// requires both records to be present). The full secret is never echoed back.
pub async fn get_api_key_user_by_value<'e, 'c: 'e, E>(
args: GetApiKeyUserByValueArgs<'e, 'c, E>,
) -> Result<Option<ApiKeyUserRecord>, sqlx::Error>
where
E: 'e + Executor<'c, Database = MySql>,
{
let result = sqlx::query!(
r#"
SELECT
api_keys.token as `api_key_token: ApiKeyToken`,
users.token as `user_token: UserToken`,
users.is_banned
FROM api_keys
INNER JOIN users
ON users.token = api_keys.owner_user_token
WHERE api_keys.api_key = ?
AND api_keys.maybe_deleted_at IS NULL
LIMIT 1
"#,
args.api_key.as_str_be_careful(),
)
.fetch_optional(args.mysql_executor)
.await?;
Ok(result.map(|r| ApiKeyUserRecord {
api_key_token: r.api_key_token,
user_token: r.user_token,
is_banned: i8_to_bool(r.is_banned),
}))
}
@@ -1,6 +1,7 @@
pub mod api_key_row;
pub mod delete_api_key;
pub mod get_api_key_by_token;
pub mod get_api_key_user_by_value;
pub mod get_live_api_key_by_value;
pub mod insert_api_key;
pub mod list_api_keys_for_user;
@@ -550,6 +550,9 @@ use crate::http_server::endpoints::media_files::list::list_batch_generated_redux
crate::http_server::endpoints::omni_gen::generate::video::omni_gen_video_generate_handler::omni_gen_video_generate_handler,
crate::http_server::endpoints::omni_gen::models::image::omni_gen_image_models_handler::omni_gen_image_models_handler,
crate::http_server::endpoints::omni_gen::models::video::omni_gen_video_models_handler::omni_gen_video_models_handler,
// Omni API (API-key authenticated generate)
crate::http_server::endpoints::omni_api::generate::image::omni_api_image_generate_handler::omni_api_image_generate_handler,
crate::http_server::endpoints::omni_api::generate::video::omni_api_video_generate_handler::omni_api_video_generate_handler,
// Moderation
crate::http_server::endpoints::moderation::alerts::moderation_send_alert_handler::moderation_send_alert_handler,
crate::http_server::endpoints::moderation::info::moderator_token_info_handler::moderator_get_token_info_handler,
@@ -17,6 +17,7 @@ pub mod media_files;
pub mod misc;
pub mod model_download;
pub mod moderation;
pub mod omni_api;
pub mod omni_gen;
pub mod prompts;
pub mod service;
@@ -0,0 +1,100 @@
use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_image_cost_and_generate_request::OmniGenImageCostAndGenerateRequest;
use artcraft_router::api::router_aspect_ratio::RouterAspectRatio;
use artcraft_router::api::router_image_model::RouterImageModel;
use artcraft_router::api::router_quality::RouterQuality;
use artcraft_router::api::router_resolution::RouterResolution;
use artcraft_router::api::image_list_ref::ImageListRef;
use artcraft_router::api::router_provider::RouterProvider;
use artcraft_router::client::request_mismatch_mitigation_strategy::RequestMismatchMitigationStrategy;
use artcraft_router::generate::generate_image::generate_image_request_builder::GenerateImageRequestBuilder;
use enums::common::generation::common_aspect_ratio::CommonAspectRatio as CommonAspectRatioEnum;
use enums::common::generation::common_image_model::CommonImageModel as CommonImageModelEnum;
use enums::common::generation::common_quality::CommonQuality as CommonQualityEnum;
use enums::common::generation::common_resolution::CommonResolution as CommonResolutionEnum;
use crate::http_server::common_responses::common_web_error::CommonWebError;
pub fn hydrate_to_router_request(
request: &OmniGenImageCostAndGenerateRequest,
) -> Result<GenerateImageRequestBuilder, 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 quality = request.quality.as_ref()
.map(convert_quality)
.transpose()?;
Ok(GenerateImageRequestBuilder {
model,
provider: RouterProvider::Artcraft,
prompt: request.prompt.clone(),
image_inputs: request.image_media_tokens.clone()
.map(ImageListRef::MediaFileTokens),
resolution,
aspect_ratio,
quality,
image_batch_count: request.image_batch_count,
horizontal_angle: request.adjust_horizontal_angle,
vertical_angle: request.adjust_vertical_angle,
zoom: request.adjust_zoom,
request_mismatch_mitigation_strategy: RequestMismatchMitigationStrategy::PayMoreUpgrade,
generation_mode_mismatch_strategy: None,
idempotency_token: request.idempotency_token.clone(),
})
}
fn convert_model(
model: &CommonImageModelEnum,
) -> Result<RouterImageModel, CommonWebError> {
let json = serde_json::to_string(model)?;
serde_json::from_str(&json).map_err(|e| {
CommonWebError::BadInputWithSimpleMessage(
format!("Unsupported image 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_quality(
quality: &CommonQualityEnum,
) -> Result<RouterQuality, CommonWebError> {
let json = serde_json::to_string(quality)?;
serde_json::from_str(&json).map_err(|e| {
CommonWebError::BadInputWithSimpleMessage(
format!("Unsupported quality: {}", e),
)
})
}
@@ -0,0 +1,61 @@
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>,
}
/// Insert a single Fal-queue inference job row for an image generation.
/// Mirrors `omni_gen::generate::video::insert_db_job::insert_fal_job::insert_fal_job`
/// but tags the row with `FalCategory::ImageGeneration` so the worker
/// routes it to the image-generation queue.
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::ImageGeneration,
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: None,
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 (image): {:?}", err);
Err(CommonWebError::from_error(err))
}
}
}
@@ -0,0 +1,117 @@
use log::{error, warn};
use enums::common::visibility::Visibility;
use mysql_queries::queries::generic_inference::api_providers::seedance2pro::insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token::{
insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token,
InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs, KinoviVersion,
};
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
use uuid_utils::uuid::generate_random_uuid;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use super::shared_job_args::SharedJobArgs;
pub struct InsertSeedance2proJobsArgs<'a, 'tx> {
/// The order_id for the primary job (always inserted first).
pub primary_order_id: &'a str,
/// Additional order_ids returned by Kinovi when batch_count > 1. Each
/// becomes its own inference job row.
pub maybe_additional_order_ids: Option<&'a [String]>,
/// Wallet ledger entry from the up-front deduction. Stamped on each
/// inserted row so the worker can issue a refund on failure.
pub maybe_wallet_ledger_entry_token: Option<&'a WalletLedgerEntryToken>,
/// Which Kinovi account fulfilled the request — Volcengine, BytePlus,
/// or BytePlusUltra. For Midjourney image generation today this is
/// always Volcengine, but the helper stays version-agnostic so a future
/// account-picking layer can use it unchanged.
pub kinovi_version: KinoviVersion,
pub shared: SharedJobArgs<'a, 'tx>,
}
pub struct InsertSeedance2proJobsResult {
pub primary_job_token: InferenceJobToken,
pub all_job_tokens: Vec<InferenceJobToken>,
}
/// Insert one Seedance2Pro/Kinovi inference job per returned order_id.
/// Mirrors the video-side helper at
/// `omni_gen::generate::video::insert_db_job::insert_seedance2pro_jobs`.
pub async fn insert_seedance2pro_jobs(args: InsertSeedance2proJobsArgs<'_, '_>) -> Result<InsertSeedance2proJobsResult, CommonWebError> {
let InsertSeedance2proJobsArgs {
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() };
// The idempotency_token column has a max length too tight for the
// `{idempotency}-batch-{i}` suffix pattern used elsewhere; mint a
// fresh UUID for each additional row instead. The primary row still
// wants a deterministic value, but historic behavior on the video
// side has been to mint UUIDs for all rows, which we mirror here.
let idempotency_str = generate_random_uuid();
let db_result = insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token(
InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs {
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: None,
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 seedance2pro inference job (image, 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(InsertSeedance2proJobsResult {
primary_job_token,
all_job_tokens,
})
}
@@ -0,0 +1,3 @@
pub mod insert_fal_job;
pub mod insert_seedance2pro_jobs;
pub mod shared_job_args;
@@ -0,0 +1,22 @@
use enums::common::generation::common_model_type::CommonModelType;
use enums::common::platform_type::PlatformType;
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;
/// Fields every image-side `insert_*_job` helper needs.
/// Parallels `omni_gen::generate::video::insert_db_job::shared_job_args::SharedJobArgs`.
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 ip_address: &'a str,
pub transaction: &'a mut sqlx::Transaction<'tx, sqlx::MySql>,
}
@@ -0,0 +1,5 @@
pub mod omni_api_image_generate_handler;
pub(crate) mod hydrate_to_router_request;
pub(crate) mod insert_db_job;
pub(crate) mod pipeline_result;
pub mod pipeline_v2;
@@ -0,0 +1,313 @@
use std::sync::Arc;
use actix_web::web::Json;
use actix_web::{web, HttpRequest};
use log::{error, info, warn};
use sqlx::Acquire;
use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_image_cost_and_generate_request::OmniGenImageCostAndGenerateRequest;
use artcraft_api_defs::omni_gen::generate_response::omni_gen_image_generate_response::OmniGenImageGenerateResponse;
use artcraft_router::generate::generate_image::generate_image_response::GenerateImageResponse;
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_provider::GenerationProvider;
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::seedance2pro::insert_generic_inference_job_for_seedance2pro_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::generic_inference_jobs::InferenceJobToken;
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::payments_error_test::payments_error_test;
use crate::http_server::endpoints::omni_api::generate::image::hydrate_to_router_request::hydrate_to_router_request;
use crate::http_server::endpoints::omni_api::generate::image::insert_db_job::insert_fal_job::{insert_fal_job, InsertFalJobArgs};
use crate::http_server::endpoints::omni_api::generate::image::insert_db_job::insert_seedance2pro_jobs::{insert_seedance2pro_jobs, InsertSeedance2proJobsArgs};
use crate::http_server::endpoints::omni_api::generate::image::insert_db_job::shared_job_args::SharedJobArgs;
use crate::http_server::endpoints::omni_api::generate::image::pipeline_v2::run_pipeline_v2::{run_pipeline_v2, RunPipelineV2Args};
use crate::http_server::session::lookup::user_session_feature_flags::UserSessionFeatureFlags;
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::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 an image using the omni-api unified endpoint (API-key authenticated).
#[utoipa::path(
post,
tag = "Omni API",
path = "/v1/omni_api/generate/image",
request_body = OmniGenImageCostAndGenerateRequest,
responses(
(status = 200, description = "Success", body = OmniGenImageGenerateResponse),
(status = 400, description = "Bad input"),
(status = 401, description = "Unauthorized"),
(status = 402, description = "Payment required"),
(status = 500, description = "Server error"),
),
)]
pub async fn omni_api_image_generate_handler(
http_request: HttpRequest,
request: Json<OmniGenImageCostAndGenerateRequest>,
server_state: web::Data<Arc<ServerState>>,
) -> Result<Json<OmniGenImageGenerateResponse>, CommonWebError> {
info!("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.
let api_session = require_api_key_user(&http_request, &mut *mysql_connection).await?;
let user_token = &api_session.user_token;
let maybe_avt_token = server_state
.avt_cookie_manager
.get_avt_token_from_request(&http_request);
// ==================== 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())
})?;
// ==================== RESOLVE MEDIA TOKENS ==================== //
// Look up media file tokens BEFORE distilling. Pipeline execution should not do I/O.
let resolved_media = lookup_media_files_as_cdn_url_list_and_map(
&http_request,
&mut mysql_connection,
server_state.server_environment,
request.image_media_tokens.as_deref().unwrap_or(&[]),
).await?;
// ==================== HYDRATE ROUTER REQUEST ==================== //
let router_builder = hydrate_to_router_request(&request)?;
// ==================== 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,
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);
}
// ==================== PIPELINE ==================== //
// NB: Release the pooled DB connection before the (slow, external) generation call so we don't
// hold a pool slot idle while waiting on the provider — that's what starves the pool and causes
// PoolTimedOut on unrelated endpoints. We re-acquire below to write the result.
drop(mysql_connection);
let pipeline_result = run_pipeline_v2(RunPipelineV2Args {
router_builder: &router_builder,
server_state: &server_state,
user_token,
resolved_media: &resolved_media,
}).await?;
let mut mysql_connection = server_state.mysql_pool.acquire().await?;
// ==================== DEBUG LOG: FAL REQUEST ==================== //
if let GenerateImageResponse::Fal(ref fal_payload) = pipeline_result.response {
if let Some(ref outbound_request) = fal_payload.maybe_outbound_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::FalRequest,
message: &format!("{:#?}", outbound_request),
mysql_executor: &mut *mysql_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert Fal request debug log: {:?}", err);
}
}
}
// ==================== WRITE RESULT ==================== //
let ip_address = get_request_ip(&http_request);
let maybe_platform_type = 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 --
let generation_mode = if request.image_media_tokens.is_some() {
CommonGenerationMode::Edit
} else {
CommonGenerationMode::Text
};
let prompt_result = insert_prompt(InsertPromptArgs {
maybe_apriori_prompt_token: None,
prompt_type: PromptType::ArtcraftApp,
maybe_creator_user_token: Some(user_token),
maybe_model_type: maybe_prompt_model_type,
maybe_generation_provider: Some(GenerationProvider::Artcraft),
maybe_positive_prompt: request.prompt.as_deref(),
maybe_negative_prompt: None,
maybe_other_args: None,
maybe_generation_mode: Some(generation_mode),
maybe_aspect_ratio: request.aspect_ratio, // TODO: should be saved from router's decision as it could have changed
maybe_resolution: request.resolution,// TODO: should be saved from router's decision as it could have changed
maybe_batch_count: request.image_batch_count.map(|c| c as u8),
maybe_generate_audio: None, // NB: Images, not video
maybe_duration_seconds: None, // NB: Images, not video
creator_ip_address: &ip_address,
mysql_executor: &mut *transaction,
phantom: Default::default(),
}).await;
let prompt_token = match prompt_result {
Ok(token) => Some(token),
Err(err) => {
warn!("Error inserting prompt: {:?}", err);
None
}
};
// -- Prompt context items --
if let Some(token) = prompt_token.as_ref() {
let mut context_items = Vec::new();
if let Some(ref_tokens) = &request.image_media_tokens {
for media_token in ref_tokens {
context_items.push(PromptContextItem {
media_token: media_token.clone(),
context_semantic_type: PromptContextSemanticType::Imgref,
});
}
}
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 --
//
// Each provider has its own queue / worker. Branch by response variant
// so the row lands on the correct queue (the Seedance2Pro/Kinovi worker
// for Midjourney; the Fal worker for everything else).
let job_token: InferenceJobToken = match &pipeline_result.response {
GenerateImageResponse::Seedance2Pro(payload) => {
info!("Inserting seedance2pro image job(s) with token: {:?}", pipeline_result.apriori_job_token);
// The image-side omni pipeline always dispatches Midjourney via the
// Volcengine Kinovi account today. If we ever route to BytePlus /
// BytePlus Ultra here, mirror the video-side `kinovi_account` knob.
let kinovi_version = KinoviVersion::Volcengine;
let result = insert_seedance2pro_jobs(InsertSeedance2proJobsArgs {
primary_order_id: &payload.order_id,
maybe_additional_order_ids: payload.maybe_order_ids.as_deref(),
maybe_wallet_ledger_entry_token: pipeline_result.maybe_wallet_ledger_entry_token.as_ref(),
kinovi_version,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.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,
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?;
result.primary_job_token
}
GenerateImageResponse::Fal(payload) => {
info!("Inserting fal image job with token: {:?}", pipeline_result.apriori_job_token);
let external_job_id = payload.request_id.clone().unwrap_or_default();
insert_fal_job(InsertFalJobArgs {
external_job_id: &external_job_id,
shared: SharedJobArgs {
apriori_job_token: &pipeline_result.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,
ip_address: &ip_address,
transaction: &mut transaction,
},
}).await?
}
GenerateImageResponse::Artcraft(payload) => {
// The omni image pipeline never dispatches via the Artcraft provider
// itself today (everything routes to Fal or Kinovi), but the response
// variant exists so we cover it defensively — Artcraft jobs come back
// already inserted server-side, so just propagate the token.
payload.inference_job_token.clone()
}
};
transaction.commit().await.map_err(|err| {
error!("Error committing transaction: {:?}", err);
CommonWebError::from_error(err)
})?;
Ok(Json(OmniGenImageGenerateResponse {
success: true,
inference_job_token: job_token,
}))
}
@@ -0,0 +1,15 @@
use artcraft_router::generate::generate_image::generate_image_response::GenerateImageResponse;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
/// Both image pipelines produce this, then the shared handler suffix writes DB rows.
pub struct ImagePipelineResult {
pub apriori_job_token: InferenceJobToken,
pub response: GenerateImageResponse,
/// The wallet ledger entry created when the user was charged for this
/// generation, if any (None when the generation is free). Used as a
/// foreign key on the inference job row so the worker can refund on
/// failure.
pub maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
}
@@ -0,0 +1,208 @@
use log::{info, warn};
use artcraft_router::api::image_list_ref::ImageListRef;
use artcraft_router::api::router_image_model::RouterImageModel;
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_seedance2pro_client::RouterSeedance2ProClient;
use artcraft_router::generate::generate_image::generate_image_request_builder::GenerateImageRequestBuilder;
use artcraft_router::generate::generate_image::generate_image_response::GenerateImageResponse;
use artcraft_router::generate::generate_image::image_generation_draft_context::ImageGenerationDraftContext;
use artcraft_router::generate::generate_image::image_generation_draft_or_request::ImageGenerationDraftOrRequest;
use artcraft_router::generate::generate_image::image_generation_request::ImageGenerationRequest;
use seedance2pro_client::creds::seedance2pro_session::Seedance2ProSession;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::users::UserToken;
use crate::billing::wallets::attempt_wallet_deduction::attempt_wallet_deduction_else_common_web_error;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::omni_api::generate::image::pipeline_result::ImagePipelineResult;
use crate::http_server::endpoints::omni_api::shared_utils::map_seedance2pro_router_error::map_router_error_to_web_error;
use crate::state::server_state::ServerState;
use crate::util::lookup::lookup_media_files_as_cdn_url_list_and_map::MediaFilesAsCdnUrlListAndMap;
pub struct RunPipelineV2Args<'a> {
pub router_builder: &'a GenerateImageRequestBuilder,
pub server_state: &'a ServerState,
pub user_token: &'a UserToken,
pub resolved_media: &'a MediaFilesAsCdnUrlListAndMap,
}
// NB: This pipeline does an external generation call (`finalize_and_generate`) that can take many
// seconds. It deliberately does NOT hold a pooled DB connection across that call — it acquires a
// short-lived connection only for the wallet deduction. 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<ImagePipelineResult, CommonWebError> {
let RunPipelineV2Args {
router_builder,
server_state,
user_token,
resolved_media,
} = args;
let hydrated_builder = apply_hydrated_media_inputs(
router_builder,
resolved_media,
);
let cost = estimate_cost_in_credits(&router_builder)?;
let draft_or_request = build_execution_request(&hydrated_builder)?;
info!("Charging wallet: {} credits", cost);
let apriori_job_token = InferenceJobToken::generate();
let maybe_wallet_ledger_entry_token = if cost > 0 {
// NB: Short-lived connection — released before the external call below.
let mut billing_connection = server_state.mysql_pool.acquire().await
.map_err(|err| {
warn!("Failed to acquire MySQL connection for image billing: {:?}", err);
CommonWebError::from_error(err)
})?;
let deduction = attempt_wallet_deduction_else_common_web_error(
user_token,
Some(apriori_job_token.as_str()),
cost,
&mut billing_connection,
).await?;
Some(deduction.ledger_entry_token)
} else {
None
};
// NB: No pooled DB connection is held across this external generation call.
let response = finalize_and_generate(draft_or_request, server_state, resolved_media).await?;
Ok(ImagePipelineResult {
apriori_job_token,
response,
maybe_wallet_ledger_entry_token,
})
}
fn build_execution_request(
router_builder: &GenerateImageRequestBuilder,
) -> Result<ImageGenerationDraftOrRequest, CommonWebError> {
let mut execution_builder = router_builder.clone();
execution_builder.provider = provider_for_model(execution_builder.model);
execution_builder.build2().map_err(|e| {
warn!("Failed to build2 for image v2 pipeline: {}", e);
CommonWebError::from_error(e)
})
}
/// Route each image model to the provider that fulfils it. Midjourney is
/// served via the Seedance2Pro/Kinovi (Volcengine) backend; everything else
/// flows through Fal.
fn provider_for_model(model: RouterImageModel) -> RouterProvider {
match model {
RouterImageModel::Midjourney7
| RouterImageModel::Midjourney7Niji
| RouterImageModel::Midjourney8 => RouterProvider::Seedance2Pro,
_ => RouterProvider::Fal,
}
}
fn apply_hydrated_media_inputs(
router_builder: &GenerateImageRequestBuilder,
resolved_media: &MediaFilesAsCdnUrlListAndMap,
) -> GenerateImageRequestBuilder {
let mut hydrated_builder = router_builder.clone();
match hydrated_builder.image_inputs.as_ref() {
Some(ImageListRef::MediaFileTokens(tokens)) if !tokens.is_empty() => {
hydrated_builder.image_inputs = Some(ImageListRef::Urls(
resolved_media.ordered_url_list.clone(),
));
},
_ => {},
}
hydrated_builder
}
fn estimate_cost_in_credits(
router_builder: &GenerateImageRequestBuilder,
) -> Result<u64, CommonWebError> {
let mut cost_builder = router_builder.clone();
cost_builder.provider = RouterProvider::Artcraft;
let request = cost_builder.build2().map_err(|e| {
warn!("Failed to build image cost request: {}", e);
CommonWebError::from_error(e)
})?;
let cost = request.estimate_cost().map_err(|e| {
warn!("Failed to estimate image cost: {}", e);
CommonWebError::from_error(e)
})?;
Ok(cost.cost_in_credits.unwrap_or(0))
}
async fn finalize_and_generate(
draft_or_request: ImageGenerationDraftOrRequest,
server_state: &ServerState,
resolved_media: &MediaFilesAsCdnUrlListAndMap,
) -> Result<GenerateImageResponse, CommonWebError> {
let provider = draft_or_request.get_provider();
let client = build_router_client(provider, server_state)?;
// Inline the draft finalization so the draft context can borrow `client`.
let request: ImageGenerationRequest = match draft_or_request {
ImageGenerationDraftOrRequest::Request(request) => request,
ImageGenerationDraftOrRequest::Draft(draft) => {
let draft_context = ImageGenerationDraftContext {
client: Some(&client),
media_file_to_artcraft_url_map: Some(&resolved_media.token_to_url_map),
};
draft.finalize(draft_context)
.await
.map_err(|err| {
warn!("Failed to finalize image v2 draft: {:?}", err);
map_router_error_to_web_error(err)
})?
}
};
request.send_request(&client)
.await
.map_err(|err| {
warn!("v2 image generation failed: {:?}", err);
map_router_error_to_web_error(err)
})
}
fn build_router_client(
provider: RouterProvider,
server_state: &ServerState,
) -> Result<RouterClient, CommonWebError> {
match provider {
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::Seedance2Pro => {
// Midjourney image generation always uses the Volcengine account.
// Per-request account selection (like the video flow's
// `KinoviAccount` enum) isn't needed here yet — if it ever is, plumb
// a parallel knob through `RunPipelineV2Args`.
let cookies = server_state.inference_providers.seedance2pro.cookies_volcengine.clone();
let session = Seedance2ProSession::from_cookies_string(cookies);
Ok(RouterClient::Seedance2Pro(RouterSeedance2ProClient::new(session)))
},
other => {
Err(CommonWebError::server_error_with_message(
&format!("Unsupported provider for image v2 generation: {:?}", other),
))
},
}
}
@@ -0,0 +1,2 @@
pub mod image;
pub mod video;
@@ -0,0 +1,44 @@
//! Shared wallet billing logic for both video generation pipelines.
use log::info;
use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::users::UserToken;
use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
use crate::billing::wallets::attempt_wallet_deduction::attempt_wallet_deduction_else_common_web_error;
use crate::http_server::common_responses::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,
})
}
@@ -0,0 +1,61 @@
//! 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_seedance2pro_client::RouterSeedance2ProClient;
use seedance2pro_client::creds::seedance2pro_session::Seedance2ProSession;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::omni_api::generate::video::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::Seedance2Pro => {
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 seedance2pro = &server_state.inference_providers.seedance2pro;
let cookies = match kinovi_account {
KinoviAccount::Volcengine => seedance2pro.cookies_volcengine.clone(),
KinoviAccount::BytePlus => seedance2pro.cookies_byteplus.clone(),
KinoviAccount::BytePlusUltra => seedance2pro.cookies_byteplus_ultra.clone(),
};
let session = Seedance2ProSession::from_cookies_string(cookies);
Ok(RouterClient::Seedance2Pro(RouterSeedance2ProClient::new(session)))
}
@@ -0,0 +1,97 @@
use crate::http_server::common_responses::common_web_error::CommonWebError;
use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_video_cost_and_generate_request::OmniGenVideoCostAndGenerateRequest;
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_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_resolution::CommonResolution as CommonResolutionEnum;
use enums::common::generation::common_video_model::CommonVideoModel as CommonVideoModelEnum;
pub fn hydrate_to_router_request(
request: &OmniGenVideoCostAndGenerateRequest,
) -> 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()?;
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,
duration_seconds: request.duration_seconds,
video_batch_count: request.video_batch_count,
generate_audio: request.generate_audio,
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),
)
})
}
@@ -0,0 +1,6 @@
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;
@@ -0,0 +1,14 @@
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,
}
@@ -0,0 +1,51 @@
//! 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)) }
}
@@ -0,0 +1,64 @@
use std::collections::HashMap;
use artcraft_router::api::audio_list_ref::AudioListRef;
use artcraft_router::api::image_list_ref::ImageListRef;
use artcraft_router::api::image_ref::ImageRef;
use artcraft_router::api::video_list_ref::VideoListRef;
use artcraft_router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder;
use tokens::tokens::media_files::MediaFileToken;
/// For providers that take URLs directly (GmiCloud, Fal), swap
/// `ImageRef::MediaFileToken` → `ImageRef::Url` using the resolved map.
pub fn resolve_media_tokens_to_urls(
builder: &mut GenerateVideoRequestBuilder,
url_map: Option<&HashMap<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));
}
}
}
@@ -0,0 +1,57 @@
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))
}
}
}
@@ -0,0 +1,48 @@
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))
}
}
}
@@ -0,0 +1,48 @@
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))
}
}
}
@@ -0,0 +1,95 @@
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::seedance2pro::insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token::{insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token, InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs, 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 InsertSeedance2proJobsArgs<'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 InsertSeedance2proJobsResult {
pub primary_job_token: InferenceJobToken,
pub all_job_tokens: Vec<InferenceJobToken>,
}
pub async fn insert_seedance2pro_jobs(args: InsertSeedance2proJobsArgs<'_, '_>) -> Result<InsertSeedance2proJobsResult, CommonWebError> {
let InsertSeedance2proJobsArgs {
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_seedance2pro_queue_with_apriori_job_token(
InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs {
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 seedance2pro 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(InsertSeedance2proJobsResult {
primary_job_token,
all_job_tokens,
})
}
@@ -0,0 +1,5 @@
pub mod insert_fal_job;
pub mod insert_gmicloud_job;
pub mod insert_grok_api_job;
pub mod insert_seedance2pro_jobs;
pub mod shared_job_args;
@@ -0,0 +1,22 @@
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>,
}
@@ -0,0 +1,7 @@
#[derive(Debug, Clone, Copy)]
pub enum KinoviAccount {
Volcengine,
BytePlus,
BytePlusUltra,
}
@@ -0,0 +1,5 @@
pub mod omni_api_video_generate_handler;
pub mod helpers;
pub mod pipeline_v2;
pub mod insert_db_job;
pub mod kinovi_account;
@@ -0,0 +1,477 @@
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 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 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::seedance2pro::insert_generic_inference_job_for_seedance2pro_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::payments_error_test::payments_error_test;
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_seedance2pro_jobs::{insert_seedance2pro_jobs, InsertSeedance2proJobsArgs};
use crate::http_server::endpoints::omni_api::generate::video::insert_db_job::shared_job_args::SharedJobArgs;
use crate::http_server::endpoints::omni_api::generate::video::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::session::lookup::user_session_feature_flags::UserSessionFeatureFlags;
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::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-api unified endpoint (API-key authenticated).
#[utoipa::path(
post,
tag = "Omni API",
path = "/v1/omni_api/generate/video",
request_body = OmniGenVideoCostAndGenerateRequest,
responses(
(status = 200, description = "Success", body = OmniGenVideoGenerateResponse),
(status = 400, description = "Bad input"),
(status = 401, description = "Unauthorized"),
(status = 402, description = "Payment required"),
(status = 500, description = "Server error"),
),
)]
pub async fn omni_api_video_generate_handler(
http_request: HttpRequest,
request: Json<OmniGenVideoCostAndGenerateRequest>,
server_state: web::Data<Arc<ServerState>>,
) -> Result<Json<OmniGenVideoGenerateResponse>, CommonWebError> {
info!("request: {:?}", request);
// Reject doomed combos (e.g. grok_imagine_video_1p5 without an image)
// before any billable or DB-mutating work — see helper for the rules.
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.
let api_session = require_api_key_user(&http_request, &mut *mysql_connection).await?;
let user_token = &api_session.user_token;
let maybe_avt_token = server_state
.avt_cookie_manager
.get_avt_token_from_request(&http_request);
// ==================== 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())
})?;
// ==================== 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,
&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?;
// ==================== HYDRATE ROUTER REQUEST ==================== //
let router_builder = hydrate_to_router_request(&request)?;
// ==================== PIPELINE DISPATCH ==================== //
let kinovi_account = match request.model {
// BytePlus Ultra
Some(CommonVideoModel::Seedance2p0BytePlusUltra) => KinoviAccount::BytePlusUltra,
Some(CommonVideoModel::Seedance2p0BytePlusUltraFast) => KinoviAccount::BytePlusUltra,
// BytePlus
Some(CommonVideoModel::Seedance2p0BytePlus) => KinoviAccount::BytePlus,
Some(CommonVideoModel::Seedance2p0BytePlusFast) => 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,
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: Release the pooled DB connection before the (slow, external) generation call so we don't
// hold a pool slot idle while waiting on the provider — that's what starves the pool and causes
// PoolTimedOut on unrelated endpoints. We re-acquire below to write the result.
drop(mysql_connection);
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,
}).await?;
let mut mysql_connection = server_state.mysql_pool.acquire().await?;
// ==================== DEBUG LOG: FAL REQUEST ==================== //
if let GenerateVideoResponse::Fal(ref fal_payload) = pipeline_result.response {
if let Some(ref outbound_request) = fal_payload.maybe_outbound_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::FalRequest,
message: &format!("{:#?}", outbound_request),
mysql_executor: &mut *mysql_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert Fal request debug log: {:?}", err);
}
}
}
// ==================== DEBUG LOG: GROK API REQUEST ==================== //
if let GenerateVideoResponse::Grok(ref grok_payload) = pipeline_result.response {
if let Some(ref outbound_request) = grok_payload.maybe_outbound_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::GrokApiRequest,
message: &format!("{:#?}", outbound_request),
mysql_executor: &mut *mysql_connection,
phantom: Default::default(),
}).await {
warn!("Failed to insert Grok API request debug log: {:?}", err);
}
}
}
// ==================== WRITE RESULT ==================== //
let ip_address = get_request_ip(&http_request);
let maybe_platform_type = 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 --
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_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::Seedance2Pro(payload) => {
info!("Inserting seedance2pro 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_seedance2pro_jobs(InsertSeedance2proJobsArgs {
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,
}))
}
fn determine_generation_mode(request: &OmniGenVideoCostAndGenerateRequest) -> 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;
}
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
}
@@ -0,0 +1,227 @@
use std::convert::TryFrom;
use std::collections::HashMap;
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::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::generate::video::kinovi_account::KinoviAccount;
use crate::http_server::endpoints::omni_api::shared_utils::map_seedance2pro_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,
}
// 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,
} = args;
let mut router_builder = router_builder.clone();
match router_builder.model {
RouterVideoModel::PreviewModel |
RouterVideoModel::Seedance2p0BytePlus |
RouterVideoModel::Seedance2p0BytePlusUltra => {
router_builder.model = RouterVideoModel::Seedance2p0;
},
RouterVideoModel::PreviewModelFast |
RouterVideoModel::Seedance2p0BytePlusFast |
RouterVideoModel::Seedance2p0BytePlusUltraFast => {
router_builder.model = RouterVideoModel::Seedance2p0Fast;
},
_ => {}, // Fall-through
}
let provider = match router_builder.model {
RouterVideoModel::HappyHorse1p0 => RouterProvider::Seedance2Pro,
RouterVideoModel::Seedance2p0 => RouterProvider::Seedance2Pro,
RouterVideoModel::Seedance2p0Fast => RouterProvider::Seedance2Pro,
//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;
// 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 (seedance2pro_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 (short-lived connection — released before the external call below).
let billing = {
let mut billing_connection = server_state.mysql_pool.acquire().await
.map_err(|err| {
error!("Failed to acquire MySQL connection for billing: {:?}", err);
CommonWebError::from_error(err)
})?;
bill_wallet(user_token, cost, &mut billing_connection).await?
};
// 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,
).await;
// 5. On failure, refund wallet for Kinovi requests.
if let Err(ref err) = result {
if matches!(provider, RouterProvider::Seedance2Pro) {
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,
) -> 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,
};
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)
})
}
@@ -0,0 +1,2 @@
pub mod generate;
pub mod shared_utils;
@@ -0,0 +1,90 @@
//! Maps artcraft router errors to web errors for the omni_gen pipelines.
//!
//! Known Kinovi (Seedance2Pro) bad requests (content violations, over-long
//! prompts, too many urls) become 400s with a user-facing message. Everything
//! else is unanticipated and stays a 500.
use artcraft_router::errors::artcraft_router_error::ArtcraftRouterError;
use artcraft_router::errors::provider_error::ProviderError;
use seedance2pro_client::error::seedance2pro_bad_request_api_error::Seedance2ProBadRequestApiError;
use seedance2pro_client::error::seedance2pro_error::Seedance2ProError;
use crate::http_server::common_responses::common_web_error::CommonWebError;
/// Map an artcraft router error to a web error, unwrapping Kinovi bad requests.
pub fn map_router_error_to_web_error(error: ArtcraftRouterError) -> CommonWebError {
match error {
ArtcraftRouterError::Provider(
ProviderError::Seedance2Pro(Seedance2ProError::ApiBadRequest(bad_request))
) => {
CommonWebError::BadInputWithSimpleMessage(user_facing_bad_request_message(&bad_request))
}
other => CommonWebError::from_error(other),
}
}
fn user_facing_bad_request_message(error: &Seedance2ProBadRequestApiError) -> String {
match error {
Seedance2ProBadRequestApiError::VideoGenerationViolation { .. } => {
"The generation request was flagged as a content violation. \
Please adjust your prompt or input media and try again.".to_string()
}
Seedance2ProBadRequestApiError::PromptIsTooLong { .. } => {
"The prompt is too long. Please shorten it below 10,000 characters and try again.".to_string()
}
Seedance2ProBadRequestApiError::TooManyUrls { .. } => {
"Too many input files were attached. \
Please remove some and try again.".to_string()
}
}
}
#[cfg(test)]
mod tests {
use actix_web::ResponseError;
use super::*;
#[test]
fn kinovi_bad_requests_become_400s_with_user_facing_messages() {
let cases = [
(
Seedance2ProBadRequestApiError::VideoGenerationViolation { raw_body: "{}".to_string() },
"content violation",
),
(
Seedance2ProBadRequestApiError::PromptIsTooLong { raw_body: "{}".to_string() },
"prompt is too long",
),
(
Seedance2ProBadRequestApiError::TooManyUrls { raw_body: "{}".to_string() },
"Too many input files",
),
];
for (bad_request, expected_message_fragment) in cases {
let error = map_router_error_to_web_error(
ArtcraftRouterError::Provider(
ProviderError::Seedance2Pro(
Seedance2ProError::ApiBadRequest(bad_request))));
assert_eq!(error.status_code().as_u16(), 400);
match error {
CommonWebError::BadInputWithSimpleMessage(message) => {
assert!(
message.contains(expected_message_fragment),
"message {:?} should contain {:?}", message, expected_message_fragment);
}
other => panic!("Expected BadInputWithSimpleMessage, got: {:?}", other),
}
}
}
#[test]
fn other_router_errors_stay_500s() {
let error = map_router_error_to_web_error(
ArtcraftRouterError::UnsupportedModel("some_model".to_string()));
assert_eq!(error.status_code().as_u16(), 500);
}
}
@@ -0,0 +1,2 @@
pub mod map_seedance2pro_router_error;
pub mod video;
@@ -0,0 +1,2 @@
pub mod validate_video_request;
mod validate_when_image_required;
@@ -0,0 +1,11 @@
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::endpoints::omni_api::shared_utils::video::validate_when_image_required::validate_when_image_required;
use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_video_cost_and_generate_request::OmniGenVideoCostAndGenerateRequest;
/// Validate requests before they incur user costs or send API requests
pub fn validate_video_request(
request: &OmniGenVideoCostAndGenerateRequest,
) -> Result<(), CommonWebError> {
validate_when_image_required(request)?;
Ok(())
}
@@ -0,0 +1,157 @@
use artcraft_api_defs::omni_gen::cost_and_generate_requests::omni_gen_video_cost_and_generate_request::OmniGenVideoCostAndGenerateRequest;
use enums::common::generation::common_video_model::CommonVideoModel;
use crate::http_server::common_responses::common_web_error::CommonWebError;
/// xAI's `grok-imagine-video-1.5-preview` model rejects text-to-video at
/// the upstream API. Reject the request at the handler layer so callers see
/// a clean 400 with an actionable message and we don't charge their wallet
/// for an inference job we know will fail.
pub (super) fn validate_when_image_required(
request: &OmniGenVideoCostAndGenerateRequest,
) -> Result<(), CommonWebError> {
match request.model {
Some(CommonVideoModel::GrokImagineVideo1p5) => {}, // Fall-through
_ => return Ok(()),
}
let requires_image = matches!(request.model, Some(CommonVideoModel::GrokImagineVideo1p5));
if !requires_image {
return Ok(());
}
let has_start_frame = request.start_frame_image_media_token.is_some();
let has_reference_images = request
.reference_image_media_tokens
.as_ref()
.is_some_and(|v| !v.is_empty());
if has_start_frame || has_reference_images {
return Ok(());
}
Err(CommonWebError::BadInputWithSimpleMessage(
"Image is required".to_string(),
))
}
#[cfg(test)]
mod tests {
use super::*;
use tokens::tokens::media_files::MediaFileToken;
fn base_request() -> OmniGenVideoCostAndGenerateRequest {
OmniGenVideoCostAndGenerateRequest {
idempotency_token: Some("a1b2c3d4-e5f6-7890-abcd-ef1234567890".to_string()),
model: None,
prompt: Some("test".to_string()),
negative_prompt: None,
start_frame_image_media_token: None,
end_frame_image_media_token: None,
reference_image_media_tokens: None,
reference_video_media_tokens: None,
reference_audio_media_tokens: None,
reference_character_tokens: None,
resolution: None,
aspect_ratio: None,
quality: None,
duration_seconds: Some(5),
video_batch_count: Some(1),
generate_audio: None,
}
}
fn token(id: &str) -> MediaFileToken {
MediaFileToken::new(id.to_string())
}
#[test]
fn passes_when_model_is_none() {
let req = base_request();
assert!(validate_when_image_required(&req).is_ok());
}
#[test]
fn passes_for_other_models_without_image() {
// Only v1.5 needs the gate; other models can do text-to-video.
let req = OmniGenVideoCostAndGenerateRequest {
model: Some(CommonVideoModel::Seedance2p0),
..base_request()
};
assert!(validate_when_image_required(&req).is_ok());
}
#[test]
fn rejects_v1p5_with_no_image_inputs() {
let req = OmniGenVideoCostAndGenerateRequest {
model: Some(CommonVideoModel::GrokImagineVideo1p5),
..base_request()
};
let err = validate_when_image_required(&req).expect_err("should reject");
match err {
CommonWebError::BadInputWithSimpleMessage(msg) => {
assert_eq!(msg, "Image is required");
}
other => panic!("expected BadInputWithSimpleMessage, got {:?}", other),
}
}
#[test]
fn rejects_v1p5_with_empty_reference_list_and_no_start_frame() {
let req = OmniGenVideoCostAndGenerateRequest {
model: Some(CommonVideoModel::GrokImagineVideo1p5),
reference_image_media_tokens: Some(vec![]),
..base_request()
};
assert!(matches!(
validate_when_image_required(&req),
Err(CommonWebError::BadInputWithSimpleMessage(_))
));
}
#[test]
fn accepts_v1p5_with_start_frame_only() {
let req = OmniGenVideoCostAndGenerateRequest {
model: Some(CommonVideoModel::GrokImagineVideo1p5),
start_frame_image_media_token: Some(token("mf_start")),
..base_request()
};
assert!(validate_when_image_required(&req).is_ok());
}
#[test]
fn accepts_v1p5_with_reference_images_only() {
let req = OmniGenVideoCostAndGenerateRequest {
model: Some(CommonVideoModel::GrokImagineVideo1p5),
reference_image_media_tokens: Some(vec![token("mf_a"), token("mf_b")]),
..base_request()
};
assert!(validate_when_image_required(&req).is_ok());
}
#[test]
fn accepts_v1p5_with_both_start_frame_and_reference_images() {
let req = OmniGenVideoCostAndGenerateRequest {
model: Some(CommonVideoModel::GrokImagineVideo1p5),
start_frame_image_media_token: Some(token("mf_start")),
reference_image_media_tokens: Some(vec![token("mf_a")]),
..base_request()
};
assert!(validate_when_image_required(&req).is_ok());
}
#[test]
fn end_frame_alone_does_not_count_as_image_input() {
// end_frame_image_media_token is technically an image, but xAI's v1.5
// wire shape only uses start_frame / reference_images. end_frame
// shouldn't satisfy the gate.
let req = OmniGenVideoCostAndGenerateRequest {
model: Some(CommonVideoModel::GrokImagineVideo1p5),
end_frame_image_media_token: Some(token("mf_end")),
..base_request()
};
assert!(validate_when_image_required(&req).is_err());
}
}
@@ -7,5 +7,6 @@ pub mod middleware;
pub mod requests;
pub mod routes;
pub mod session;
pub mod user_lookup;
pub mod validations;
pub mod web_utils;
@@ -11,6 +11,7 @@ use crate::http_server::routes::application_routes::generate_routes::add_generat
use crate::http_server::routes::application_routes::job_routes::add_job_routes;
use crate::http_server::routes::application_routes::media_files_routes::add_media_file_routes;
use crate::http_server::routes::application_routes::moderation_routes::add_moderator_routes;
use crate::http_server::routes::application_routes::omni_api_routes::add_omni_api_routes;
use crate::http_server::routes::application_routes::omni_gen_routes::add_omni_gen_routes;
use crate::http_server::routes::application_routes::prompts_routes::add_prompts_routes;
use crate::http_server::routes::application_routes::stripe_artcraft_routes::add_stripe_artcraft_routes;
@@ -48,6 +49,7 @@ where
// Artcraft surface area
app = add_generate_routes(app); // /v1/generate/...
app = add_omni_gen_routes(app); // /v1/omni_gen/...
app = add_omni_api_routes(app); // /v1/omni_api/... (API-key authenticated generate)
app = add_character_routes(app); // /v1/character/... and /v1/characters/...
app = add_webhook_routes(app); // /v1/webhooks/... (fal)
@@ -11,6 +11,7 @@ mod generate_routes;
mod job_routes;
mod media_files_routes;
mod moderation_routes;
mod omni_api_routes;
mod omni_gen_routes;
mod prompts_routes;
mod stripe_artcraft_routes;
@@ -0,0 +1,36 @@
use actix_http::body::MessageBody;
use actix_service::ServiceFactory;
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_web::{web, App, Error, HttpResponse};
use crate::http_server::endpoints::omni_api::generate::image::omni_api_image_generate_handler::omni_api_image_generate_handler;
use crate::http_server::endpoints::omni_api::generate::video::omni_api_video_generate_handler::omni_api_video_generate_handler;
/// API-key authenticated generation endpoints. These mirror `/v1/omni_gen/generate/*` but read the
/// caller's identity from the `Authorization` header API key instead of a session cookie. The
/// `cost` and `models` endpoints are not duplicated here — they need no user and the existing
/// `/v1/omni_gen/{cost,models}/*` routes suffice.
pub fn add_omni_api_routes<T, B>(app: App<T>) -> App<T>
where
B: MessageBody,
T: ServiceFactory<
ServiceRequest,
Config = (),
Response = ServiceResponse<B>,
Error = Error,
InitError = (),
>,
{
app.service(web::scope("/v1/omni_api")
.service(web::scope("/generate")
.service(web::resource("/video")
.route(web::post().to(omni_api_video_generate_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(web::resource("/image")
.route(web::post().to(omni_api_image_generate_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
)
)
}
@@ -0,0 +1 @@
pub mod require_api_key_user;
@@ -0,0 +1,83 @@
use std::marker::PhantomData;
use actix_web::HttpRequest;
use log::{info, warn};
use sqlx::MySqlConnection;
use mysql_queries::queries::api_keys::get_api_key_user_by_value::{
get_api_key_user_by_value, GetApiKeyUserByValueArgs,
};
use tokens::tokens::api_keys::ApiKeyToken;
use tokens::tokens::users::UserToken;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::web_utils::get_authorization_header_api_key::get_authorization_header_api_key;
/// The authenticated identity behind an API-key request — the API-key analogue of a cookie
/// `SessionUserRecord`, but deliberately minimal (just what the omni-api handlers need).
///
/// Unlike cookie sessions, this is **never** cached in Redis: every request performs a fresh MySQL
/// lookup.
pub struct UserApiSession {
pub api_key_token: ApiKeyToken,
pub user_token: UserToken,
pub is_banned: bool,
}
/// Authenticate a request using the API key in its `Authorization` header (see
/// `get_authorization_header_api_key` for the accepted formats).
///
/// Returns `CommonWebError::NotAuthorized` (401) when the header is missing/unparseable, when the
/// key does not match a live (non-soft-deleted) API key whose owning user still exists, or when
/// that owner is banned. `mysql_executor` can be any sqlx executor — pass `&mut *connection` to
/// reuse a connection the handler already holds.
pub async fn require_api_key_user(
http_request: &HttpRequest,
mysql_connection: &mut MySqlConnection,
) -> Result<UserApiSession, CommonWebError> {
let api_key = match get_authorization_header_api_key(http_request) {
Some(api_key) => api_key,
None => {
warn!("Missing or malformed Authorization header API key");
return Err(CommonWebError::NotAuthorized);
}
};
let maybe_record = get_api_key_user_by_value(GetApiKeyUserByValueArgs {
api_key: &api_key,
mysql_executor: &mut *mysql_connection,
phantom: PhantomData,
})
.await
.map_err(|err| {
warn!("API key user lookup error: {:?}", err);
CommonWebError::from_error(err)
})?;
// A missing or soft-deleted key (or a key whose owner no longer exists) is a 401, not a leak of
// which case occurred.
let record = match maybe_record {
Some(record) => record,
None => {
warn!("No live API key user for presented key: {:?}", api_key);
return Err(CommonWebError::NotAuthorized);
}
};
if record.is_banned {
warn!("API key owner is banned: {:?}", record.user_token.as_str());
return Err(CommonWebError::NotAuthorized);
}
info!(
"API key user authenticated: api_key_token={:?} user_token={:?}",
record.api_key_token.as_str(),
record.user_token.as_str(),
);
Ok(UserApiSession {
api_key_token: record.api_key_token,
user_token: record.user_token,
is_banned: record.is_banned,
})
}
@@ -0,0 +1 @@
pub mod api_keys;