diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/fal/insert_generic_inference_job_for_fal_queue_with_apriori_job_token.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/fal/insert_generic_inference_job_for_fal_queue_with_apriori_job_token.rs index ffbbee23d7..925e8eafed 100644 --- a/crates/schema/database/mysql_queries/src/queries/generic_inference/fal/insert_generic_inference_job_for_fal_queue_with_apriori_job_token.rs +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/fal/insert_generic_inference_job_for_fal_queue_with_apriori_job_token.rs @@ -86,7 +86,6 @@ pub async fn insert_generic_inference_job_for_fal_queue_with_apriori_job_token<' let maybe_external_third_party = InferenceJobExternalThirdParty::Fal; - let maybe_external_third_party_id = ""; const STATUS : JobStatusPlus = JobStatusPlus::Pending; let query = sqlx::query!( diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/job/mark_job_failed_by_token.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/job/mark_job_failed_by_token.rs new file mode 100644 index 0000000000..0e17d212e9 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/job/mark_job_failed_by_token.rs @@ -0,0 +1,56 @@ +use anyhow::anyhow; +use sqlx::MySqlPool; + +use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; +use enums::common::job_status_plus::JobStatusPlus; +use errors::AnyhowResult; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; + +pub struct MarkJobFailedByTokenArgs<'a> { + pub pool: &'a MySqlPool, + pub job_token: &'a InferenceJobToken, + pub maybe_public_failure_reason: Option<&'a str>, + pub internal_debugging_failure_reason: &'a str, + pub maybe_frontend_failure_category: Option, +} + +/// Permanently mark an inference job as failed, looked up by its token. +/// Unlike `mark_generic_inference_job_failure`, this does not allow retries — +/// the job will always land in `complete_failure`. +pub async fn mark_job_failed_by_token(args: MarkJobFailedByTokenArgs<'_>) -> AnyhowResult<()> { + let maybe_public_failure_reason = args.maybe_public_failure_reason.map(|reason| { + let mut reason = reason.trim().to_string(); + reason.truncate(512); + reason + }); + + let mut internal_debugging_failure_reason = args.internal_debugging_failure_reason.trim().to_string(); + internal_debugging_failure_reason.truncate(512); + + const FAILURE_STATUS: &str = JobStatusPlus::CompleteFailure.to_str(); + + let query_result = sqlx::query!( + r#" +UPDATE generic_inference_jobs +SET + status = ?, + failure_reason = ?, + internal_debugging_failure_reason = ?, + frontend_failure_category = ?, + retry_at = NULL +WHERE token = ? + "#, + FAILURE_STATUS, + maybe_public_failure_reason.as_deref(), + &internal_debugging_failure_reason, + args.maybe_frontend_failure_category, + args.job_token.as_str() + ) + .execute(args.pool) + .await; + + match query_result { + Err(err) => Err(anyhow!("error with query: {:?}", err)), + Ok(_r) => Ok(()), + } +} diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/job/mod.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/job/mod.rs index 98ddd9b7a5..affd5103bf 100644 --- a/crates/schema/database/mysql_queries/src/queries/generic_inference/job/mod.rs +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/job/mod.rs @@ -3,6 +3,7 @@ pub mod count_untried_jobs_of_type; pub mod list_available_generic_inference_jobs; pub mod mark_generic_inference_job_completely_failed; pub mod mark_generic_inference_job_failure; +pub mod mark_job_failed_by_token; pub mod mark_generic_inference_job_pending_and_grab_lock; pub mod mark_generic_inference_job_successfully_done; pub mod reap_stale_fakeyou_jobs; diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/mod.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/mod.rs index 833d554e7c..75c08abda9 100644 --- a/crates/schema/database/mysql_queries/src/queries/generic_inference/mod.rs +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/mod.rs @@ -1,3 +1,4 @@ pub mod fal; pub mod job; +pub mod seedance2pro; pub mod web; diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token.rs new file mode 100644 index 0000000000..70e99d75af --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token.rs @@ -0,0 +1,137 @@ +use anyhow::anyhow; +use log::info; +use sqlx::{Executor, MySql}; +use std::marker::PhantomData; + +use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; +use enums::by_table::generic_inference_jobs::inference_job_external_third_party::InferenceJobExternalThirdParty; +use enums::by_table::generic_inference_jobs::inference_job_product_category::InferenceJobProductCategory; +use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType; +use enums::common::job_status_plus::JobStatusPlus; +use enums::common::visibility::Visibility; +use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use tokens::tokens::prompts::PromptToken; +use tokens::tokens::users::UserToken; + +use crate::errors::database_query_error::DatabaseQueryError; +use crate::payloads::generic_inference_args::generic_inference_args::GenericInferenceArgs; + +pub struct InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs<'e, 'c, E> + where E: 'e + Executor<'c, Database = MySql> +{ + pub uuid_idempotency_token: &'e str, + + // NOTE: We'll generate this ahead of time so we can save it with billing info! + pub apriori_job_token: &'e InferenceJobToken, + + /// The external order_id returned by the seedance2pro generate_video call. + pub maybe_external_third_party_id: &'e str, + + pub maybe_inference_args: Option, + + pub maybe_prompt_token: Option<&'e PromptToken>, + + pub maybe_creator_user_token: Option<&'e UserToken>, + pub maybe_avt_token: Option<&'e AnonymousVisitorTrackingToken>, + pub creator_ip_address: &'e str, + pub creator_set_visibility: Visibility, + + pub mysql_executor: E, + + pub phantom: PhantomData<&'c E>, +} + +pub async fn insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token<'e, 'c : 'e, E>( + args: InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs<'e, 'c, E> +) -> Result + where E: 'e + Executor<'c, Database = MySql> +{ + let serialized_args_payload = serde_json::ser::to_string(&args.maybe_inference_args) + .map_err(|_e| anyhow!("could not encode inference args"))?; + + const JOB_TYPE: InferenceJobType = InferenceJobType::Seedance2ProQueue; + const INFERENCE_CATEGORY: InferenceCategory = InferenceCategory::VideoGeneration; + const PRODUCT_CATEGORY: InferenceJobProductCategory = InferenceJobProductCategory::Seedance2ProVideo; + const EXTERNAL_THIRD_PARTY: InferenceJobExternalThirdParty = InferenceJobExternalThirdParty::Seedance2Pro; + const STATUS: JobStatusPlus = JobStatusPlus::Pending; + + let query = sqlx::query!( + r#" +INSERT INTO generic_inference_jobs +SET + token = ?, + uuid_idempotency_token = ?, + + job_type = ?, + + maybe_external_third_party = ?, + maybe_external_third_party_id = ?, + + product_category = ?, + inference_category = ?, + + maybe_model_type = NULL, + maybe_model_token = NULL, + + maybe_input_source_token = NULL, + maybe_input_source_token_type = NULL, + + maybe_download_url = NULL, + maybe_cover_image_media_file_token = NULL, + + maybe_prompt_token = ?, + + maybe_raw_inference_text = NULL, + + maybe_inference_args = ?, + + maybe_creator_user_token = ?, + maybe_creator_anonymous_visitor_token = ?, + creator_ip_address = ?, + creator_set_visibility = ?, + + priority_level = 0, + is_keepalive_required = FALSE, + max_duration_seconds = 0, + + is_debug_request = FALSE, + maybe_routing_tag = NULL, + + status = ? + "#, + args.apriori_job_token.as_str(), + args.uuid_idempotency_token, + + JOB_TYPE.to_str(), + + EXTERNAL_THIRD_PARTY.to_str(), + args.maybe_external_third_party_id, + + PRODUCT_CATEGORY.to_str(), + INFERENCE_CATEGORY.to_str(), + + args.maybe_prompt_token.map(|t| t.to_string()), + + serialized_args_payload, + + args.maybe_creator_user_token.map(|t| t.to_string()), + args.maybe_avt_token.map(|t| t.to_string()), + args.creator_ip_address, + args.creator_set_visibility.to_str(), + + STATUS.to_str(), + ); + + let query_result = query.execute(args.mysql_executor) + .await; + + let record_id = match query_result { + Err(err) => return Err(DatabaseQueryError::from(err)), + Ok(res) => res.last_insert_id(), + }; + + info!("Insert generic inference job for Seedance 2 Pro queue: {} with record ID {}", args.apriori_job_token, record_id); + + Ok(args.apriori_job_token.clone()) +} diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/list_pending_seedance2pro_jobs.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/list_pending_seedance2pro_jobs.rs new file mode 100644 index 0000000000..6e5f50c15c --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/list_pending_seedance2pro_jobs.rs @@ -0,0 +1,92 @@ +use anyhow::anyhow; +use log::warn; +use sqlx::MySqlPool; + +use enums::by_table::generic_inference_jobs::inference_job_external_third_party::InferenceJobExternalThirdParty; +use enums::common::visibility::Visibility; +use errors::AnyhowResult; +use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use tokens::tokens::prompts::PromptToken; +use tokens::tokens::users::UserToken; + +/// A Seedance2Pro job that is waiting for a result from the external API. +#[derive(Debug)] +pub struct PendingSeedance2ProJob { + pub job_token: InferenceJobToken, + + /// The external order_id from seedance2-pro.com. + pub order_id: String, + + pub maybe_creator_user_token: Option, + pub maybe_creator_anonymous_visitor_token: Option, + pub creator_ip_address: String, + pub creator_set_visibility: Visibility, + + pub maybe_prompt_token: Option, +} + +#[derive(Debug, Default)] +struct RawRecord { + job_token: InferenceJobToken, + order_id: Option, + maybe_creator_user_token: Option, + maybe_creator_anonymous_visitor_token: Option, + creator_ip_address: String, + creator_set_visibility: Visibility, + maybe_prompt_token: Option, +} + +/// Returns all non-terminal Seedance2Pro jobs that have an associated order_id. +pub async fn list_pending_seedance2pro_jobs(pool: &MySqlPool) -> AnyhowResult> { + let records = sqlx::query_as!( + RawRecord, + r#" +SELECT + jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`, + jobs.maybe_external_third_party_id as `order_id`, + jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`, + jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`, + jobs.creator_ip_address, + jobs.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`, + jobs.maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken` + +FROM generic_inference_jobs as jobs + +WHERE jobs.maybe_external_third_party = ? + AND jobs.status NOT IN ('complete_success', 'complete_failure') + AND jobs.maybe_external_third_party_id IS NOT NULL + +LIMIT 25000 + "#, + InferenceJobExternalThirdParty::Seedance2Pro.to_str(), + ) + .fetch_all(pool) + .await + .map_err(|err| anyhow!("error querying pending seedance2pro jobs: {:?}", err))?; + + let jobs = records + .into_iter() + .filter_map(|record| { + let order_id = match record.order_id { + Some(id) => id, + None => { + warn!("PendingSeedance2ProJob has no order_id, skipping"); + return None; + } + }; + + Some(PendingSeedance2ProJob { + job_token: record.job_token, + order_id, + maybe_creator_user_token: record.maybe_creator_user_token, + maybe_creator_anonymous_visitor_token: record.maybe_creator_anonymous_visitor_token, + creator_ip_address: record.creator_ip_address, + creator_set_visibility: record.creator_set_visibility, + maybe_prompt_token: record.maybe_prompt_token, + }) + }) + .collect(); + + Ok(jobs) +} diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/mod.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/mod.rs new file mode 100644 index 0000000000..24c6ee0ef4 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/mod.rs @@ -0,0 +1,2 @@ +pub mod insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token; +pub mod list_pending_seedance2pro_jobs;