diff --git a/.sqlx/query-f52c495a7ff5cafb357f79c4cdb503bbbef045fd193a5a37a66180fcac77f363.json b/.sqlx/query-f52c495a7ff5cafb357f79c4cdb503bbbef045fd193a5a37a66180fcac77f363.json new file mode 100644 index 0000000000..780eabe4e3 --- /dev/null +++ b/.sqlx/query-f52c495a7ff5cafb357f79c4cdb503bbbef045fd193a5a37a66180fcac77f363.json @@ -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" +} diff --git a/crates/schema/database/mysql_queries/src/queries/api_keys/get_api_key_user_by_value.rs b/crates/schema/database/mysql_queries/src/queries/api_keys/get_api_key_user_by_value.rs new file mode 100644 index 0000000000..59b8202ee2 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/api_keys/get_api_key_user_by_value.rs @@ -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, 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), + })) +} diff --git a/crates/schema/database/mysql_queries/src/queries/api_keys/mod.rs b/crates/schema/database/mysql_queries/src/queries/api_keys/mod.rs index 898cc84f80..deff03cc5b 100644 --- a/crates/schema/database/mysql_queries/src/queries/api_keys/mod.rs +++ b/crates/schema/database/mysql_queries/src/queries/api_keys/mod.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/docs/api_doc.rs b/crates/service/web/storyteller_web/src/docs/api_doc.rs index f15b9ca09d..2faf9fe2c9 100644 --- a/crates/service/web/storyteller_web/src/docs/api_doc.rs +++ b/crates/service/web/storyteller_web/src/docs/api_doc.rs @@ -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, diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs index 4d593ba4d7..bc28bdeafe 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/hydrate_to_router_request.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/hydrate_to_router_request.rs new file mode 100644 index 0000000000..566533f5b0 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/hydrate_to_router_request.rs @@ -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 { + 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 { + 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 { + 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 { + 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 { + let json = serde_json::to_string(quality)?; + serde_json::from_str(&json).map_err(|e| { + CommonWebError::BadInputWithSimpleMessage( + format!("Unsupported quality: {}", e), + ) + }) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/insert_fal_job.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/insert_fal_job.rs new file mode 100644 index 0000000000..cf27586134 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/insert_fal_job.rs @@ -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 { + 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)) + } + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/insert_seedance2pro_jobs.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/insert_seedance2pro_jobs.rs new file mode 100644 index 0000000000..05a87d6415 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/insert_seedance2pro_jobs.rs @@ -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, +} + +/// 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 { + 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 = 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, + }) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/mod.rs new file mode 100644 index 0000000000..751cbf87e7 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/mod.rs @@ -0,0 +1,3 @@ +pub mod insert_fal_job; +pub mod insert_seedance2pro_jobs; +pub mod shared_job_args; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/shared_job_args.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/shared_job_args.rs new file mode 100644 index 0000000000..e83ff66c7d --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/insert_db_job/shared_job_args.rs @@ -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, + pub maybe_platform_type: Option, + pub ip_address: &'a str, + pub transaction: &'a mut sqlx::Transaction<'tx, sqlx::MySql>, +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/mod.rs new file mode 100644 index 0000000000..2b127f5225 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/mod.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/omni_api_image_generate_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/omni_api_image_generate_handler.rs new file mode 100644 index 0000000000..1468e5aa9d --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/omni_api_image_generate_handler.rs @@ -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, + server_state: web::Data>, +) -> Result, 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 = 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, + })) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_result.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_result.rs new file mode 100644 index 0000000000..4d4bd9120d --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_result.rs @@ -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, +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_v2/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_v2/mod.rs new file mode 100644 index 0000000000..f207b21293 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_v2/mod.rs @@ -0,0 +1 @@ +pub mod run_pipeline_v2; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_v2/run_pipeline_v2.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_v2/run_pipeline_v2.rs new file mode 100644 index 0000000000..eb9dd59ed5 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/image/pipeline_v2/run_pipeline_v2.rs @@ -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 { + 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 { + 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 { + 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 { + 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 { + 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), + )) + }, + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/mod.rs new file mode 100644 index 0000000000..c10a6ca15e --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/mod.rs @@ -0,0 +1,2 @@ +pub mod image; +pub mod video; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/bill_wallet.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/bill_wallet.rs new file mode 100644 index 0000000000..4529d92d71 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/bill_wallet.rs @@ -0,0 +1,44 @@ +//! Shared wallet billing logic for both video generation pipelines. + +use log::info; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use tokens::tokens::users::UserToken; +use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken; + +use crate::billing::wallets::attempt_wallet_deduction::attempt_wallet_deduction_else_common_web_error; +use crate::http_server::common_responses::common_web_error::CommonWebError; + +pub struct BillWalletResult { + pub apriori_job_token: InferenceJobToken, + pub maybe_wallet_ledger_entry_token: Option, +} + +/// Generate an apriori job token and bill the user's wallet for the given cost. +/// +/// If cost is 0, no wallet deduction is made but the apriori token is still generated. +pub async fn bill_wallet( + user_token: &UserToken, + cost: u64, + mysql_connection: &mut sqlx::pool::PoolConnection, +) -> Result { + let apriori_job_token = InferenceJobToken::generate(); + + info!("Charging wallet: {} credits", cost); + + let maybe_wallet_ledger_entry_token = if cost > 0 { + let deduction_result = attempt_wallet_deduction_else_common_web_error( + user_token, + Some(apriori_job_token.as_str()), + cost, + mysql_connection, + ).await?; + Some(deduction_result.ledger_entry_token) + } else { + None + }; + + Ok(BillWalletResult { + apriori_job_token, + maybe_wallet_ledger_entry_token, + }) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/build_router_client.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/build_router_client.rs new file mode 100644 index 0000000000..0e153c7cb8 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/build_router_client.rs @@ -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 { + 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 { + 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))) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/hydrate_router_request.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/hydrate_router_request.rs new file mode 100644 index 0000000000..604697dc0e --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/hydrate_router_request.rs @@ -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 { + 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 { + 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 { + 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 { + let json = serde_json::to_string(res)?; + serde_json::from_str(&json).map_err(|e| { + CommonWebError::BadInputWithSimpleMessage( + format!("Unsupported resolution: {}", e), + ) + }) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/mod.rs new file mode 100644 index 0000000000..7108f22a3b --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/mod.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/pipeline_result.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/pipeline_result.rs new file mode 100644 index 0000000000..cb0648f34c --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/pipeline_result.rs @@ -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, +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/resolve_kinovi_character_ids.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/resolve_kinovi_character_ids.rs new file mode 100644 index 0000000000..25b0f8aeac --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/resolve_kinovi_character_ids.rs @@ -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, +) -> Result>, 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 = characters.iter() + .filter(|c| c.is_active) + .filter_map(|c| { + c.kinovi_character_id.as_ref().map(|kid| (c.token.clone(), kid.clone())) + }) + .collect(); + + if map.is_empty() { Ok(None) } else { Ok(Some(map)) } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/resolve_media_tokens_to_urls.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/resolve_media_tokens_to_urls.rs new file mode 100644 index 0000000000..8ee7d7f175 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/helpers/resolve_media_tokens_to_urls.rs @@ -0,0 +1,64 @@ +use std::collections::HashMap; + +use artcraft_router::api::audio_list_ref::AudioListRef; +use artcraft_router::api::image_list_ref::ImageListRef; +use artcraft_router::api::image_ref::ImageRef; +use artcraft_router::api::video_list_ref::VideoListRef; +use artcraft_router::generate::generate_video::generate_video_request_builder::GenerateVideoRequestBuilder; +use tokens::tokens::media_files::MediaFileToken; + +/// For providers that take URLs directly (GmiCloud, Fal), swap +/// `ImageRef::MediaFileToken` → `ImageRef::Url` using the resolved map. +pub fn resolve_media_tokens_to_urls( + builder: &mut GenerateVideoRequestBuilder, + url_map: Option<&HashMap>, +) { + let map = match url_map { + Some(m) => m, + None => return, + }; + + // start_frame + if let Some(ImageRef::MediaFileToken(ref token)) = builder.start_frame { + if let Some(url) = map.get(token) { + builder.start_frame = Some(ImageRef::Url(url.clone())); + } + } + + // end_frame + if let Some(ImageRef::MediaFileToken(ref token)) = builder.end_frame { + if let Some(url) = map.get(token) { + builder.end_frame = Some(ImageRef::Url(url.clone())); + } + } + + // reference_images + if let Some(ImageListRef::MediaFileTokens(ref tokens)) = builder.reference_images { + let urls: Vec = tokens.iter() + .filter_map(|t| map.get(t).cloned()) + .collect(); + if !urls.is_empty() { + builder.reference_images = Some(ImageListRef::Urls(urls)); + } + } + + // reference_videos + if let Some(VideoListRef::MediaFileTokens(ref tokens)) = builder.reference_videos { + let urls: Vec = tokens.iter() + .filter_map(|t| map.get(t).cloned()) + .collect(); + if !urls.is_empty() { + builder.reference_videos = Some(VideoListRef::Urls(urls)); + } + } + + // reference_audio + if let Some(AudioListRef::MediaFileTokens(ref tokens)) = builder.reference_audio { + let urls: Vec = tokens.iter() + .filter_map(|t| map.get(t).cloned()) + .collect(); + if !urls.is_empty() { + builder.reference_audio = Some(AudioListRef::Urls(urls)); + } + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_fal_job.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_fal_job.rs new file mode 100644 index 0000000000..b71ca9b184 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_fal_job.rs @@ -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 { + 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)) + } + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_gmicloud_job.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_gmicloud_job.rs new file mode 100644 index 0000000000..776318f419 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_gmicloud_job.rs @@ -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 { + 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)) + } + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_grok_api_job.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_grok_api_job.rs new file mode 100644 index 0000000000..76e4b70d00 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_grok_api_job.rs @@ -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 { + 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)) + } + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_seedance2pro_jobs.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_seedance2pro_jobs.rs new file mode 100644 index 0000000000..5da1d5e915 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/insert_seedance2pro_jobs.rs @@ -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, +} + +pub async fn insert_seedance2pro_jobs(args: InsertSeedance2proJobsArgs<'_, '_>) -> Result { + 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 = 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, + }) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/mod.rs new file mode 100644 index 0000000000..4591de2a46 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/mod.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/shared_job_args.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/shared_job_args.rs new file mode 100644 index 0000000000..69215d9ce0 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/insert_db_job/shared_job_args.rs @@ -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, + pub maybe_platform_type: Option, + pub maybe_cost_estimates: Option, + pub ip_address: &'a str, + pub transaction: &'a mut sqlx::Transaction<'tx, sqlx::MySql>, +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/kinovi_account.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/kinovi_account.rs new file mode 100644 index 0000000000..17277f6934 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/kinovi_account.rs @@ -0,0 +1,7 @@ + +#[derive(Debug, Clone, Copy)] +pub enum KinoviAccount { + Volcengine, + BytePlus, + BytePlusUltra, +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/mod.rs new file mode 100644 index 0000000000..95b669cdcd --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/mod.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/omni_api_video_generate_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/omni_api_video_generate_handler.rs new file mode 100644 index 0000000000..62c26f18b2 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/omni_api_video_generate_handler.rs @@ -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, + server_state: web::Data>, +) -> Result, 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 = 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 = 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> = 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> = + 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 +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/pipeline_v2/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/pipeline_v2/mod.rs new file mode 100644 index 0000000000..f207b21293 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/pipeline_v2/mod.rs @@ -0,0 +1 @@ +pub mod run_pipeline_v2; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/pipeline_v2/run_pipeline_v2.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/pipeline_v2/run_pipeline_v2.rs new file mode 100644 index 0000000000..19288349dd --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/generate/video/pipeline_v2/run_pipeline_v2.rs @@ -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>, + pub kinovi_character_id_map: &'a Option>, + 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 { + 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>, + kinovi_character_ids: Option<&HashMap>, + kinovi_account: KinoviAccount, +) -> Result { + + 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) + }) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/mod.rs new file mode 100644 index 0000000000..5cd6f81640 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/mod.rs @@ -0,0 +1,2 @@ +pub mod generate; +pub mod shared_utils; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/map_seedance2pro_router_error.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/map_seedance2pro_router_error.rs new file mode 100644 index 0000000000..6815fa3869 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/map_seedance2pro_router_error.rs @@ -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); + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/mod.rs new file mode 100644 index 0000000000..cf024715fc --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/mod.rs @@ -0,0 +1,2 @@ +pub mod map_seedance2pro_router_error; +pub mod video; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/mod.rs new file mode 100644 index 0000000000..bfdbcc2afa --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/mod.rs @@ -0,0 +1,2 @@ +pub mod validate_video_request; +mod validate_when_image_required; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/validate_video_request.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/validate_video_request.rs new file mode 100644 index 0000000000..e4ad72396e --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/validate_video_request.rs @@ -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(()) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/validate_when_image_required.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/validate_when_image_required.rs new file mode 100644 index 0000000000..c94faa6e48 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/omni_api/shared_utils/video/validate_when_image_required.rs @@ -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()); + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/mod.rs b/crates/service/web/storyteller_web/src/http_server/mod.rs index 82a70137ab..c4aeed0e92 100644 --- a/crates/service/web/storyteller_web/src/http_server/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/mod.rs @@ -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; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/http_server/routes/application_routes/add_application_routes.rs b/crates/service/web/storyteller_web/src/http_server/routes/application_routes/add_application_routes.rs index e3051809a7..3d0fddc7dc 100644 --- a/crates/service/web/storyteller_web/src/http_server/routes/application_routes/add_application_routes.rs +++ b/crates/service/web/storyteller_web/src/http_server/routes/application_routes/add_application_routes.rs @@ -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) diff --git a/crates/service/web/storyteller_web/src/http_server/routes/application_routes/mod.rs b/crates/service/web/storyteller_web/src/http_server/routes/application_routes/mod.rs index e7f8b9a836..80564ab5ae 100644 --- a/crates/service/web/storyteller_web/src/http_server/routes/application_routes/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/routes/application_routes/mod.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/http_server/routes/application_routes/omni_api_routes.rs b/crates/service/web/storyteller_web/src/http_server/routes/application_routes/omni_api_routes.rs new file mode 100644 index 0000000000..a889d7b66d --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/routes/application_routes/omni_api_routes.rs @@ -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(app: App) -> App +where + B: MessageBody, + T: ServiceFactory< + ServiceRequest, + Config = (), + Response = ServiceResponse, + 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())) + ) + ) + ) +} diff --git a/crates/service/web/storyteller_web/src/http_server/user_lookup/api_keys/mod.rs b/crates/service/web/storyteller_web/src/http_server/user_lookup/api_keys/mod.rs new file mode 100644 index 0000000000..c84538ff4f --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/user_lookup/api_keys/mod.rs @@ -0,0 +1 @@ +pub mod require_api_key_user; diff --git a/crates/service/web/storyteller_web/src/http_server/user_lookup/api_keys/require_api_key_user.rs b/crates/service/web/storyteller_web/src/http_server/user_lookup/api_keys/require_api_key_user.rs new file mode 100644 index 0000000000..e7a867bd34 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/user_lookup/api_keys/require_api_key_user.rs @@ -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 { + 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, + }) +} diff --git a/crates/service/web/storyteller_web/src/http_server/user_lookup/mod.rs b/crates/service/web/storyteller_web/src/http_server/user_lookup/mod.rs new file mode 100644 index 0000000000..02a4a96543 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/user_lookup/mod.rs @@ -0,0 +1 @@ +pub mod api_keys;