job

crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/list_pending_gmicloud_jobs.rs

@@ -0,0 +1,98 @@
+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;
+use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
+
+/// A GmiCloud job that is waiting for a result from the external API.
+#[derive(Debug)]
+pub struct PendingGmiCloudJob {
+  pub job_token: InferenceJobToken,
+
+  /// The external request_id from GmiCloud.
+  pub request_id: String,
+
+  pub maybe_creator_user_token: Option<UserToken>,
+  pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
+  pub creator_ip_address: String,
+  pub creator_set_visibility: Visibility,
+
+  pub maybe_prompt_token: Option<PromptToken>,
+
+  pub maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
+}
+
+#[derive(Debug, Default)]
+struct RawRecord {
+  job_token: InferenceJobToken,
+  request_id: Option<String>,
+  maybe_creator_user_token: Option<UserToken>,
+  maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
+  creator_ip_address: String,
+  creator_set_visibility: Visibility,
+  maybe_prompt_token: Option<PromptToken>,
+  maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
+}
+
+/// Returns all non-terminal GmiCloud jobs that have an associated request_id.
+pub async fn list_pending_gmicloud_jobs(pool: &MySqlPool) -> AnyhowResult<Vec<PendingGmiCloudJob>> {
+  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 `request_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`,
+    jobs.maybe_wallet_ledger_entry_token as `maybe_wallet_ledger_entry_token: tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken`
+
+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::GmiCloud.to_str(),
+  )
+    .fetch_all(pool)
+    .await
+    .map_err(|err| anyhow!("error querying pending gmicloud jobs: {:?}", err))?;
+
+  let jobs = records
+    .into_iter()
+    .filter_map(|record| {
+      let request_id = match record.request_id {
+        Some(id) => id,
+        None => {
+          warn!("PendingGmiCloudJob has no request_id, skipping");
+          return None;
+        }
+      };
+
+      Some(PendingGmiCloudJob {
+        job_token: record.job_token,
+        request_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,
+        maybe_wallet_ledger_entry_token: record.maybe_wallet_ledger_entry_token,
+      })
+    })
+    .collect();
+
+  Ok(jobs)
+}

crates/schema/database/mysql_queries/src/queries/generic_inference/gmicloud/mod.rs

@@ -1 +1,2 @@
 pub mod insert_generic_inference_job_for_gmicloud;
+pub mod list_pending_gmicloud_jobs;

crates/service/job/gmicloud_job/Cargo.toml

@@ -0,0 +1,53 @@
+[package]
+name = "gmicloud-job"
+edition = "2021"
+version = "0.0.1"
+publish = false
+
+[[bin]]
+name = "gmicloud-job"
+path = "src/main.rs"
+
+[package.metadata.cargo-udeps.ignore]
+normal = ["workspace-build-acceleration"]
+
+[dependencies]
+# Workspace hack (faster compile times)
+workspace-build-acceleration = { workspace = true }
+
+# Internal
+actix_helpers.workspace = true
+bootstrap.workspace = true
+bucket_paths.workspace = true
+cloud_storage.workspace = true
+concurrency.workspace = true
+easyenv.workspace = true
+enums.workspace = true
+errors.workspace = true
+gmicloud_client.workspace = true
+hashing = { path = "../../../lib/files/hashing" }
+jobs_common.workspace = true
+mysql_queries.workspace = true
+server_environment = { path = "../../../lib/server_environment" }
+shared_env_var_config.workspace = true
+tokens.workspace = true
+
+# External
+anyhow.workspace = true
+log.workspace = true
+
+sqlx = { version = "0.7.4", features = [ "mysql", "runtime-tokio-rustls", "chrono" ] }
+
+# Runtime
+tokio = { version = "1.40.0", features = ["macros", "sync"] }
+
+# HTTP Client
+reqwest.workspace = true
+
+# HTTP Server
+actix-web.workspace = true
+
+# Serialization
+serde.workspace = true
+serde_derive.workspace = true
+serde_json.workspace = true

crates/service/job/gmicloud_job/config/gmicloud-job.development.env

@@ -0,0 +1 @@
+SERVER_ENVIRONMENT='development'

crates/service/job/gmicloud_job/src/http_server/endpoints/health_check_handler.rs

@@ -0,0 +1,101 @@
+use std::sync::Arc;
+
+use actix_web::error::ResponseError;
+use actix_web::http::header::ContentType;
+use actix_web::http::StatusCode;
+use actix_web::{web, HttpRequest, HttpResponse};
+use log::error;
+
+use actix_helpers::response_serializers::error_to_json_http_response::error_to_json_http_response;
+
+use crate::http_server::http_server_shared_state::HttpServerSharedState;
+
+#[derive(Serialize)]
+pub struct HealthCheckResponse {
+  pub success: bool,
+  pub is_healthy: bool,
+  pub consecutive_failure_count: u64,
+  pub consecutive_success_count: u64,
+  pub total_failure_count: u64,
+  pub total_success_count: u64,
+  pub total_failure_ratio: f32,
+  pub total_success_ratio: f32,
+}
+
+#[derive(Debug, Serialize)]
+pub enum HealthCheckError {
+  ServerError,
+}
+
+impl ResponseError for HealthCheckError {
+  fn status_code(&self) -> StatusCode {
+    match *self {
+      HealthCheckError::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
+    }
+  }
+
+  fn error_response(&self) -> HttpResponse {
+    error_to_json_http_response(self)
+  }
+}
+
+impl std::fmt::Display for HealthCheckError {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    write!(f, "{:?}", self)
+  }
+}
+
+pub async fn get_health_check_handler(
+  _http_request: HttpRequest,
+  server_state: web::Data<Arc<HttpServerSharedState>>,
+) -> Result<HttpResponse, HealthCheckError> {
+  let job_stats = server_state
+    .job_stats
+    .get_status()
+    .map_err(|e| {
+      error!("Error serving health check status: {:?}", e);
+      HealthCheckError::ServerError
+    })?;
+
+  let total_tries = job_stats
+    .total_failure_count
+    .saturating_add(job_stats.total_success_count);
+
+  let total_success_ratio = if total_tries > 0 {
+    (job_stats.total_success_count as f32) / (total_tries as f32)
+  } else {
+    0.0
+  };
+
+  let total_failure_ratio = if total_tries > 0 {
+    1.0 - total_success_ratio
+  } else {
+    0.0
+  };
+
+  let is_healthy =
+    job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold;
+
+  let response = HealthCheckResponse {
+    success: true,
+    is_healthy,
+    consecutive_failure_count: job_stats.consecutive_failure_count,
+    consecutive_success_count: job_stats.consecutive_success_count,
+    total_failure_count: job_stats.total_failure_count,
+    total_success_count: job_stats.total_success_count,
+    total_failure_ratio,
+    total_success_ratio,
+  };
+
+  let body = serde_json::to_string(&response).map_err(|_e| HealthCheckError::ServerError)?;
+
+  if is_healthy {
+    Ok(HttpResponse::Ok()
+      .content_type(ContentType::json())
+      .body(body))
+  } else {
+    Ok(HttpResponse::InternalServerError()
+      .content_type(ContentType::json())
+      .body(body))
+  }
+}

crates/service/job/gmicloud_job/src/http_server/endpoints/mod.rs

@@ -0,0 +1 @@
+pub mod health_check_handler;

crates/service/job/gmicloud_job/src/http_server/http_server_shared_state.rs

@@ -0,0 +1,7 @@
+use jobs_common::job_stats::JobStats;
+
+#[derive(Clone)]
+pub struct HttpServerSharedState {
+  pub job_stats: JobStats,
+  pub consecutive_failure_unhealthy_threshold: u64,
+}

crates/service/job/gmicloud_job/src/http_server/mod.rs

@@ -0,0 +1,3 @@
+pub mod endpoints;
+pub mod http_server_shared_state;
+pub mod run_http_server;

crates/service/job/gmicloud_job/src/http_server/run_http_server.rs

@@ -0,0 +1,66 @@
+use std::sync::Arc;
+
+use actix_web::dev::Server;
+use actix_web::middleware::Logger;
+use actix_web::{web, App, HttpResponse, HttpServer};
+use log::info;
+
+use bootstrap::bootstrap::ContainerEnvironment;
+use errors::AnyhowResult;
+use jobs_common::job_stats::JobStats;
+
+use crate::http_server::endpoints::health_check_handler::get_health_check_handler;
+use crate::http_server::http_server_shared_state::HttpServerSharedState;
+
+const DEFAULT_BIND_ADDRESS: &str = "0.0.0.0:12345";
+const DEFAULT_NUM_WORKERS: usize = 2;
+
+pub struct CreateServerArgs {
+  pub container_environment: ContainerEnvironment,
+  pub job_stats: JobStats,
+}
+
+pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
+  let bind_address = easyenv::get_env_string_or_default("HTTP_BIND_ADDRESS", DEFAULT_BIND_ADDRESS);
+  let num_workers = easyenv::get_env_num("HTTP_NUM_WORKERS", DEFAULT_NUM_WORKERS)?;
+
+  let server_state = HttpServerSharedState {
+    job_stats: args.job_stats.clone(),
+    consecutive_failure_unhealthy_threshold: easyenv::get_env_num(
+      "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD",
+      3,
+    )?,
+  };
+
+  let server_state_arc = web::Data::new(Arc::new(server_state));
+
+  info!("Starting HTTP service (for k8s health checking).");
+
+  let log_format = "[%{HOSTNAME}e] IP=[%{X-Forwarded-For}i] \"%r\" %s %b \"%{Referer}i\" \"%{User-Agent}i\" %T";
+
+  let handle = HttpServer::new(move || {
+    App::new()
+      .app_data(server_state_arc.clone())
+      .wrap(Logger::new(&log_format).exclude("/_status"))
+      .service(
+        web::resource("/")
+          .route(web::get().to(|| HttpResponse::Ok()))
+          .route(web::head().to(|| HttpResponse::Ok())),
+      )
+      .service(
+        web::resource("/_status")
+          .route(web::get().to(get_health_check_handler))
+          .route(web::head().to(|| HttpResponse::Ok())),
+      )
+  })
+    .bind(bind_address)?
+    .workers(num_workers)
+    .run();
+
+  Ok(handle)
+}
+
+pub async fn launch_http_server(args: CreateServerArgs) -> AnyhowResult<()> {
+  run_http_server(args)?.await?;
+  Ok(())
+}

crates/service/job/gmicloud_job/src/job_dependencies.rs

@@ -0,0 +1,26 @@
+use cloud_storage::bucket_client::BucketClient;
+use concurrency::relaxed_atomic_bool::RelaxedAtomicBool;
+use gmicloud_client::creds::gmicloud_api_key::GmiCloudApiKey;
+use jobs_common::job_stats::JobStats;
+use server_environment::ServerEnvironment;
+use sqlx::MySqlPool;
+
+pub struct JobDependencies {
+  pub mysql_pool: MySqlPool,
+
+  /// Public GCS/S3 bucket for storing generated videos.
+  pub public_bucket_client: BucketClient,
+
+  /// API key for GmiCloud.
+  pub gmicloud_api_key: GmiCloudApiKey,
+
+  pub server_environment: ServerEnvironment,
+
+  pub job_stats: JobStats,
+
+  /// How long to sleep between poll iterations (milliseconds).
+  pub poll_interval_millis: u64,
+
+  /// Set to `true` from another thread to trigger graceful shutdown.
+  pub application_shutdown: RelaxedAtomicBool,
+}

crates/service/job/gmicloud_job/src/main.rs

@@ -0,0 +1,140 @@
+// Never allow these
+#![forbid(private_bounds)]
+#![forbid(private_interfaces)]
+#![forbid(unused_must_use)]
+
+// Always allow
+#![allow(dead_code)]
+#![allow(non_snake_case)]
+
+#[macro_use] extern crate serde_derive;
+
+use std::time::Duration;
+
+use anyhow::anyhow;
+use log::{info, warn};
+use sqlx::mysql::MySqlPoolOptions;
+
+use bootstrap::bootstrap::{bootstrap, BootstrapArgs};
+use cloud_storage::bucket_client::BucketClient;
+use concurrency::relaxed_atomic_bool::RelaxedAtomicBool;
+use shared_env_var_config::logging::DEFAULT_RUST_LOG;
+use errors::AnyhowResult;
+use gmicloud_client::creds::gmicloud_api_key::GmiCloudApiKey;
+use jobs_common::job_stats::JobStats;
+use server_environment::ServerEnvironment;
+use shared_env_var_config::mysql::env_get_mysql_connection_string_or_default;
+
+use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs};
+use crate::main_loop::main_loop;
+use crate::job_dependencies::JobDependencies;
+
+pub mod http_server;
+pub mod job_dependencies;
+pub mod main_loop;
+pub mod process_job;
+
+// Bucket config
+const ENV_ACCESS_KEY: &str = "ACCESS_KEY";
+const ENV_SECRET_KEY: &str = "SECRET_KEY";
+const ENV_REGION_NAME: &str = "REGION_NAME";
+const ENV_PUBLIC_BUCKET_NAME: &str = "PUBLIC_BUCKET_NAME";
+const ENV_S3_ENDPOINT: &str = "S3_COMPATIBLE_ENDPOINT_URL";
+const ENV_GMICLOUD_API_KEY: &str = "GMICLOUD_API_KEY";
+
+#[actix_web::main]
+async fn main() -> AnyhowResult<()> {
+
+  let container_environment = bootstrap(BootstrapArgs {
+    app_name: "gmicloud-job",
+    default_logging_override: Some(DEFAULT_RUST_LOG),
+    config_search_directories: &[".", "./config", "crates/service/job/gmicloud_job/config"],
+    ignore_legacy_dot_env_file: true,
+  })?;
+
+  info!("Hostname: {}", &container_environment.hostname);
+
+  let _k8s_node_name = easyenv::get_env_string_optional("K8S_NODE_NAME");
+  let _k8s_pod_name = easyenv::get_env_string_optional("K8S_POD_NAME");
+
+  let db_connection_string = env_get_mysql_connection_string_or_default();
+
+  info!("Connecting to database...");
+
+  let mysql_pool = MySqlPoolOptions::new()
+    .max_connections(2)
+    .connect(&db_connection_string)
+    .await?;
+
+  info!("Connected to MySQL.");
+
+  let server_environment = ServerEnvironment::from_str(
+    &easyenv::get_env_string_required("SERVER_ENVIRONMENT")?,
+  )
+    .ok_or(anyhow!("invalid server environment"))?;
+
+  // Bucket setup
+  let access_key = easyenv::get_env_string_required(ENV_ACCESS_KEY)?;
+  let secret_key = easyenv::get_env_string_required(ENV_SECRET_KEY)?;
+  let region_name = easyenv::get_env_string_required(ENV_REGION_NAME)?;
+  let public_bucket_name = easyenv::get_env_string_required(ENV_PUBLIC_BUCKET_NAME)?;
+  let s3_compatible_endpoint_url = easyenv::get_env_string_required(ENV_S3_ENDPOINT)?;
+
+  let bucket_timeout = easyenv::get_env_duration_seconds_or_default(
+    "BUCKET_TIMEOUT_SECONDS",
+    Duration::from_secs(60 * 5),
+  );
+
+  let public_bucket_client = BucketClient::create(
+    &access_key,
+    &secret_key,
+    &region_name,
+    &public_bucket_name,
+    &s3_compatible_endpoint_url,
+    None,
+    Some(bucket_timeout),
+  )?;
+
+  // GmiCloud API key
+  let gmicloud_api_key_str = easyenv::get_env_string_required(ENV_GMICLOUD_API_KEY)?;
+  let gmicloud_api_key = GmiCloudApiKey::new(gmicloud_api_key_str);
+
+  // How often to poll for results (default: 15 seconds)
+  let poll_interval_millis: u64 = easyenv::get_env_num(
+    "GMICLOUD_POLL_INTERVAL_MILLIS",
+    15_000,
+  )?;
+
+  let application_shutdown = RelaxedAtomicBool::new(false);
+  let job_stats = JobStats::new();
+
+  let create_server_args = CreateServerArgs {
+    container_environment: container_environment.clone(),
+    job_stats: job_stats.clone(),
+  };
+
+  let job_dependencies = JobDependencies {
+    mysql_pool,
+    public_bucket_client,
+    gmicloud_api_key,
+    server_environment,
+    job_stats,
+    poll_interval_millis,
+    application_shutdown: application_shutdown.clone(),
+  };
+
+  std::thread::spawn(move || {
+    let actix_runtime = actix_web::rt::System::new();
+    let http_server_handle = launch_http_server(create_server_args);
+
+    actix_runtime.block_on(http_server_handle)
+      .expect("HTTP server should not exit.");
+
+    warn!("Server thread is shut down.");
+    application_shutdown.set(true);
+  });
+
+  main_loop(job_dependencies).await;
+
+  Ok(())
+}

crates/service/job/gmicloud_job/src/main_loop.rs

@@ -0,0 +1,93 @@
+use std::time::Duration;
+
+use log::{error, info, warn};
+use mysql_queries::queries::generic_inference::gmicloud::list_pending_gmicloud_jobs::list_pending_gmicloud_jobs;
+use gmicloud_client::requests::poll_request_queue::poll_request_queue::poll_gmicloud_request;
+
+use crate::process_job::process_failed_job::process_failed_job;
+use crate::process_job::process_successful_job::process_successful_job;
+use crate::job_dependencies::JobDependencies;
+
+pub async fn main_loop(job_dependencies: JobDependencies) {
+  while !job_dependencies.application_shutdown.get() {
+    let result = run_poll_iteration(&job_dependencies).await;
+
+    if let Err(err) = result {
+      error!("Error in poll iteration: {:?}", err);
+      let _ = job_dependencies.job_stats.increment_failure_count();
+    }
+
+    tokio::time::sleep(Duration::from_millis(job_dependencies.poll_interval_millis)).await;
+  }
+
+  warn!("GmiCloud job runner main loop is shut down.");
+}
+
+async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
+  let pending_jobs = list_pending_gmicloud_jobs(&deps.mysql_pool).await?;
+
+  if pending_jobs.is_empty() {
+    return Ok(());
+  }
+
+  info!("Found {} pending GmiCloud job(s).", pending_jobs.len());
+
+  for job in &pending_jobs {
+    let poll_result = match poll_gmicloud_request(&deps.gmicloud_api_key, &job.request_id).await {
+      Ok(response) => response,
+      Err(err) => {
+        warn!(
+          "Error polling GmiCloud request {} for job {}: {:?}",
+          job.request_id, job.job_token.as_str(), err
+        );
+        continue;
+      }
+    };
+
+    if poll_result.is_failed() {
+      let reason = "GmiCloud generation failed";
+      info!(
+        "Request {} failed. Processing job {} as failed.",
+        job.request_id, job.job_token.as_str()
+      );
+      process_failed_job(deps, job, reason).await;
+      continue;
+    }
+
+    if poll_result.is_in_progress() {
+      continue;
+    }
+
+    if poll_result.is_success() {
+      let video_url = match poll_result.video_url() {
+        Some(url) => url.to_string(),
+        None => {
+          warn!(
+            "Request {} succeeded but has no video_url for job {}. Skipping.",
+            job.request_id, job.job_token.as_str()
+          );
+          continue;
+        }
+      };
+
+      let thumbnail_url = poll_result.thumbnail_url().map(|s| s.to_string());
+
+      info!(
+        "Request {} completed, processing job {}.",
+        job.request_id, job.job_token.as_str()
+      );
+
+      if let Err(err) = process_successful_job(deps, job, &video_url, thumbnail_url.as_deref()).await {
+        warn!(
+          "Error processing completed request {} for job {}: {:?}",
+          job.request_id, job.job_token.as_str(), err
+        );
+        let _ = deps.job_stats.increment_failure_count();
+      } else {
+        let _ = deps.job_stats.increment_success_count();
+      }
+    }
+  }
+
+  Ok(())
+}

crates/service/job/gmicloud_job/src/process_job/mod.rs

@@ -0,0 +1,2 @@
+pub mod process_failed_job;
+pub mod process_successful_job;

crates/service/job/gmicloud_job/src/process_job/process_failed_job.rs

@@ -0,0 +1,113 @@
+use log::{error, info, warn};
+
+use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
+use mysql_queries::queries::generic_inference::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs};
+use mysql_queries::queries::generic_inference::gmicloud::list_pending_gmicloud_jobs::PendingGmiCloudJob;
+use mysql_queries::queries::wallets::refund::try_to_refund_ledger_entry::{try_to_refund_ledger_entry, WalletRefundOutcome};
+
+use crate::job_dependencies::JobDependencies;
+
+pub async fn process_failed_job(
+  deps: &JobDependencies,
+  job: &PendingGmiCloudJob,
+  reason: &str,
+) {
+  // --- Step 1: Attempt the refund before touching the job status. ---
+  //
+  // Refund first so a crash between refund and status update never leaves
+  // a failed job without a refund. If the refund fails, bail out early:
+  // the job stays pending and will be retried on the next poll cycle.
+
+  match &job.maybe_wallet_ledger_entry_token {
+    None => {
+      warn!(
+        "Job {} has no wallet ledger entry token; skipping refund.",
+        job.job_token.as_str()
+      );
+    }
+    Some(ledger_token) => {
+      let mut transaction = match deps.mysql_pool.begin().await {
+        Ok(tx) => tx,
+        Err(err) => {
+          error!(
+            "Failed to begin refund transaction for job {} (ledger {}): {:?}. \
+             Job will NOT be marked failed yet and will be retried next poll.",
+            job.job_token.as_str(), ledger_token.as_str(), err
+          );
+          return;
+        }
+      };
+
+      match try_to_refund_ledger_entry(ledger_token, &mut transaction).await {
+        Ok(WalletRefundOutcome::Refunded(summary)) => {
+          info!(
+            "Refunded {} credits for failed job {} (ledger {} -> refund ledger {}).",
+            summary.refund_amount,
+            job.job_token.as_str(),
+            ledger_token.as_str(),
+            summary.refund_ledger_entry_token.as_str(),
+          );
+          if let Err(err) = transaction.commit().await {
+            error!(
+              "Failed to commit refund transaction for job {} (ledger {}): {:?}. \
+               Job will NOT be marked failed yet and will be retried next poll.",
+              job.job_token.as_str(), ledger_token.as_str(), err
+            );
+            return;
+          }
+        }
+        Ok(WalletRefundOutcome::AlreadyRefunded) => {
+          info!(
+            "Ledger entry {} for job {} was already refunded; proceeding to mark job failed.",
+            ledger_token.as_str(), job.job_token.as_str(),
+          );
+          let _ = transaction.rollback().await;
+        }
+        Err(err) => {
+          error!(
+            "Failed to refund ledger entry {} for job {}: {:?}. \
+             Job will NOT be marked failed yet and will be retried next poll.",
+            ledger_token.as_str(), job.job_token.as_str(), err,
+          );
+          let _ = transaction.rollback().await;
+          return;
+        }
+      }
+    }
+  }
+
+  // --- Step 2: Mark the job record as failed. ---
+
+  let reason_lower = reason.to_lowercase();
+
+  let platform_rules_violation = reason_lower.contains("violates")
+    || reason_lower.contains("platform rules")
+    || reason_lower.contains("please modify")
+    || reason_lower.contains("content policy");
+
+  let frontend_failure_category = if platform_rules_violation {
+    Some(FrontendFailureCategory::ModelRulesViolation)
+  } else {
+    None
+  };
+
+  warn!(
+    "Request for job {} failed: {}. Marking job failed.",
+    job.job_token.as_str(), reason
+  );
+
+  let mark_failed_result = mark_job_failed_by_token(MarkJobFailedByTokenArgs {
+    pool: &deps.mysql_pool,
+    job_token: &job.job_token,
+    maybe_public_failure_reason: Some(reason),
+    internal_debugging_failure_reason: reason,
+    maybe_frontend_failure_category: frontend_failure_category,
+  }).await;
+
+  if let Err(err) = mark_failed_result {
+    error!(
+      "Error marking job {} as failed: {:?}",
+      job.job_token.as_str(), err
+    );
+  }
+}

crates/service/job/gmicloud_job/src/process_job/process_successful_job.rs

@@ -0,0 +1,189 @@
+use anyhow::anyhow;
+use log::{error, info, warn};
+
+use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath;
+use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
+use enums::by_table::media_files::media_file_class::MediaFileClass;
+use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory;
+use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory;
+use enums::by_table::media_files::media_file_type::MediaFileType;
+use errors::AnyhowResult;
+use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes;
+use mysql_queries::queries::generic_inference::gmicloud::list_pending_gmicloud_jobs::PendingGmiCloudJob;
+use mysql_queries::queries::media_files::create::insert_builder::media_file_insert_builder::MediaFileInsertBuilder;
+use mysql_queries::queries::generic_inference::web::mark_generic_inference_job_successfully_done_by_token::mark_generic_inference_job_successfully_done_by_token;
+use tokens::tokens::media_files::MediaFileToken;
+
+use crate::job_dependencies::JobDependencies;
+
+const VIDEO_PREFIX: &str = "artcraft_";
+const VIDEO_SUFFIX: &str = ".mp4";
+
+const THUMB_PREFIX: &str = "artcraft_";
+
+/// Download the completed video, upload to bucket, create media file record, and mark job done.
+pub async fn process_successful_job(
+  deps: &JobDependencies,
+  job: &PendingGmiCloudJob,
+  video_url: &str,
+  maybe_thumbnail_url: Option<&str>,
+) -> AnyhowResult<()> {
+
+  // --- Step 1: Download and upload thumbnail as cover image (if available). ---
+
+  let maybe_cover_token = match maybe_thumbnail_url {
+    Some(thumbnail_url) => {
+      match download_and_upload_thumbnail(deps, job, thumbnail_url).await {
+        Ok(token) => Some(token),
+        Err(err) => {
+          warn!(
+            "Failed to create thumbnail for job {}: {:?}. Continuing without cover.",
+            job.job_token.as_str(), err
+          );
+          None
+        }
+      }
+    }
+    None => None,
+  };
+
+  // --- Step 2: Download and upload the video. ---
+
+  info!("Downloading video for job {} from: {}", job.job_token.as_str(), video_url);
+
+  let video_bytes: Vec<u8> = reqwest::get(video_url)
+    .await
+    .map_err(|err| anyhow!("reqwest error downloading video: {:?}", err))?
+    .bytes()
+    .await
+    .map_err(|err| anyhow!("error reading video bytes: {:?}", err))?
+    .to_vec();
+
+  info!("Downloaded {} bytes for job {}", video_bytes.len(), job.job_token.as_str());
+
+  let checksum = sha256_hash_bytes(&video_bytes)
+    .map_err(|err| anyhow!("error hashing video: {:?}", err))?;
+
+  let bucket_path = MediaFileBucketPath::generate_new(Some(VIDEO_PREFIX), Some(VIDEO_SUFFIX));
+  let object_path = bucket_path.get_full_object_path_str();
+
+  info!("Uploading video to public bucket at path: {}", object_path);
+
+  deps
+    .public_bucket_client
+    .upload_file_with_content_type_process(object_path, &video_bytes, "video/mp4")
+    .await
+    .map_err(|err| anyhow!("error uploading video to bucket: {:?}", err))?;
+
+  info!("Uploaded video for job {}. Creating media file record.", job.job_token.as_str());
+
+  let media_file_token = MediaFileInsertBuilder::new()
+    .maybe_creator_user(job.maybe_creator_user_token.as_ref())
+    .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref())
+    .creator_ip_address(&job.creator_ip_address)
+    .creator_set_visibility(job.creator_set_visibility)
+    .media_file_class(MediaFileClass::Video)
+    .media_file_type(MediaFileType::Mp4)
+    .media_file_origin_category(MediaFileOriginCategory::Inference)
+    .media_file_origin_product_category(MediaFileOriginProductCategory::VideoGeneration)
+    .mime_type("video/mp4")
+    .file_size_bytes(video_bytes.len() as u64)
+    .checksum_sha2(&checksum)
+    .maybe_prompt_token(job.maybe_prompt_token.as_ref())
+    .maybe_cover_image_media_file_token(maybe_cover_token.as_ref())
+    .public_bucket_directory_hash(&bucket_path)
+    .insert_pool(&deps.mysql_pool)
+    .await
+    .map_err(|err| anyhow!("error inserting media file record: {:?}", err))?;
+
+  info!(
+    "Created media file {} for job {}. Marking job complete.",
+    media_file_token.as_str(), job.job_token.as_str()
+  );
+
+  mark_generic_inference_job_successfully_done_by_token(
+    &deps.mysql_pool,
+    &job.job_token,
+    Some(InferenceResultType::MediaFile),
+    Some(media_file_token.as_str()),
+    None,
+    None,
+  )
+    .await
+    .map_err(|err| {
+      error!("Error marking job {} done: {:?}", job.job_token.as_str(), err);
+      anyhow!("error marking job done: {:?}", err)
+    })?;
+
+  info!("Job {} completed successfully.", job.job_token.as_str());
+
+  Ok(())
+}
+
+fn guess_image_format_from_url(url: &str) -> (&'static str, &'static str, MediaFileType) {
+  let path = url.split('?').next().unwrap_or(url);
+  let path = path.split('#').next().unwrap_or(path);
+
+  if let Some(dot) = path.rfind('.') {
+    match &path[dot..] {
+      ".jpg" | ".jpeg" => (".jpg", "image/jpeg", MediaFileType::Jpg),
+      ".png" => (".png", "image/png", MediaFileType::Png),
+      _ => (".png", "image/png", MediaFileType::Png),
+    }
+  } else {
+    (".png", "image/png", MediaFileType::Png)
+  }
+}
+
+async fn download_and_upload_thumbnail(
+  deps: &JobDependencies,
+  job: &PendingGmiCloudJob,
+  thumbnail_url: &str,
+) -> AnyhowResult<MediaFileToken> {
+  info!("Downloading thumbnail for job {} from: {}", job.job_token.as_str(), thumbnail_url);
+
+  let (suffix, mime_type, media_file_type) = guess_image_format_from_url(thumbnail_url);
+
+  let thumb_bytes: Vec<u8> = reqwest::get(thumbnail_url)
+    .await
+    .map_err(|err| anyhow!("reqwest error downloading thumbnail: {:?}", err))?
+    .bytes()
+    .await
+    .map_err(|err| anyhow!("error reading thumbnail bytes: {:?}", err))?
+    .to_vec();
+
+  let checksum = sha256_hash_bytes(&thumb_bytes)
+    .map_err(|err| anyhow!("error hashing thumbnail: {:?}", err))?;
+
+  let bucket_path = MediaFileBucketPath::generate_new(Some(THUMB_PREFIX), Some(suffix));
+  let object_path = bucket_path.get_full_object_path_str();
+
+  deps
+    .public_bucket_client
+    .upload_file_with_content_type_process(object_path, &thumb_bytes, mime_type)
+    .await
+    .map_err(|err| anyhow!("error uploading thumbnail to bucket: {:?}", err))?;
+
+  let thumb_token = MediaFileInsertBuilder::new()
+    .maybe_creator_user(job.maybe_creator_user_token.as_ref())
+    .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref())
+    .creator_ip_address(&job.creator_ip_address)
+    .creator_set_visibility(job.creator_set_visibility)
+    .media_file_class(MediaFileClass::Image)
+    .media_file_type(media_file_type)
+    .is_intermediate_system_file(true)
+    .media_file_origin_category(MediaFileOriginCategory::Inference)
+    .media_file_origin_product_category(MediaFileOriginProductCategory::VideoGeneration)
+    .mime_type(mime_type)
+    .file_size_bytes(thumb_bytes.len() as u64)
+    .checksum_sha2(&checksum)
+    .maybe_prompt_token(job.maybe_prompt_token.as_ref())
+    .public_bucket_directory_hash(&bucket_path)
+    .insert_pool(&deps.mysql_pool)
+    .await
+    .map_err(|err| anyhow!("error inserting thumbnail media file record: {:?}", err))?;
+
+  info!("Created thumbnail media file {} for job {}.", thumb_token.as_str(), job.job_token.as_str());
+
+  Ok(thumb_token)
+}

crates/service/web/storyteller_web/src/configs/omni_gen/video_models.rs

@@ -155,84 +155,84 @@ fn build_omni_gen_video_models() -> Vec<OmniGenVideoModelDetails> {
     ..Default::default()
   });
 
-//  models.push(OmniGenVideoModelDetails {
-//    model: CommonVideoModel::Seedance2p0Global,
-//    model_creator: Some(ModelCreator::Bytedance),
-//    full_name: Some("Seedance 2.0 (Global)".to_string()),
-//    text_prompt_supported: Some(true),
-//    starting_keyframe_supported: Some(true),
-//    ending_keyframe_supported: Some(true),
-//    image_references_supported: Some(true),
-//    image_references_max: Some(9),
-//    audio_references_supported: Some(true),
-//    audio_references_max: Some(3),
-//    audio_references_max_total_duration_seconds: Some(15),
-//    video_references_supported: Some(true),
-//    video_references_max: Some(3),
-//    video_references_max_total_duration_seconds: Some(15),
-//    character_references_supported: Some(true),
-//    character_references_max: Some(9),
-//    aspect_ratio_options: Some(vec![
-//      CommonAspectRatio::WideTwentyOneByNine,
-//      CommonAspectRatio::WideSixteenByNine,
-//      CommonAspectRatio::WideFourByThree,
-//      CommonAspectRatio::Square,
-//      CommonAspectRatio::TallThreeByFour,
-//      CommonAspectRatio::TallNineBySixteen,
-//    ]),
-//    aspect_ratio_default: Some(CommonAspectRatio::WideSixteenByNine),
-//    resolution_options: Some(vec![
-//      CommonResolution::FourEightyP,
-//      CommonResolution::SevenTwentyP,
-//      CommonResolution::TenEightyP,
-//    ]),
-//    resolution_default: Some(CommonResolution::SevenTwentyP),
-//    batch_size_options: Some(vec![1, 2, 4]),
-//    batch_size_default: Some(1),
-//    duration_seconds_min: Some(4),
-//    duration_seconds_max: Some(15),
-//    duration_seconds_default: Some(5),
-//    ..Default::default()
-//  });
-//
-//  models.push(OmniGenVideoModelDetails {
-//    model: CommonVideoModel::Seedance2p0FastGlobal,
-//    model_creator: Some(ModelCreator::Bytedance),
-//    full_name: Some("Seedance 2.0 Fast (Global)".to_string()),
-//    text_prompt_supported: Some(true),
-//    starting_keyframe_supported: Some(true),
-//    ending_keyframe_supported: Some(true),
-//    image_references_supported: Some(true),
-//    image_references_max: Some(9),
-//    audio_references_supported: Some(true),
-//    audio_references_max: Some(3),
-//    audio_references_max_total_duration_seconds: Some(15),
-//    video_references_supported: Some(true),
-//    video_references_max: Some(3),
-//    video_references_max_total_duration_seconds: Some(15),
-//    character_references_supported: Some(true),
-//    character_references_max: Some(9),
-//    aspect_ratio_options: Some(vec![
-//      CommonAspectRatio::WideTwentyOneByNine,
-//      CommonAspectRatio::WideSixteenByNine,
-//      CommonAspectRatio::WideFourByThree,
-//      CommonAspectRatio::Square,
-//      CommonAspectRatio::TallThreeByFour,
-//      CommonAspectRatio::TallNineBySixteen,
-//    ]),
-//    aspect_ratio_default: Some(CommonAspectRatio::WideSixteenByNine),
-//    resolution_options: Some(vec![
-//      CommonResolution::FourEightyP,
-//      CommonResolution::SevenTwentyP,
-//    ]),
-//    resolution_default: Some(CommonResolution::SevenTwentyP),
-//    batch_size_options: Some(vec![1, 2, 4]),
-//    batch_size_default: Some(1),
-//    duration_seconds_min: Some(4),
-//    duration_seconds_max: Some(15),
-//    duration_seconds_default: Some(5),
-//    ..Default::default()
-//  });
+  models.push(OmniGenVideoModelDetails {
+    model: CommonVideoModel::Seedance2p0Global,
+    model_creator: Some(ModelCreator::Bytedance),
+    full_name: Some("Seedance 2.0 (Global)".to_string()),
+    text_prompt_supported: Some(true),
+    starting_keyframe_supported: Some(true),
+    ending_keyframe_supported: Some(true),
+    image_references_supported: Some(true),
+    image_references_max: Some(9),
+    audio_references_supported: Some(true),
+    audio_references_max: Some(3),
+    audio_references_max_total_duration_seconds: Some(15),
+    video_references_supported: Some(true),
+    video_references_max: Some(3),
+    video_references_max_total_duration_seconds: Some(15),
+    character_references_supported: Some(true),
+    character_references_max: Some(9),
+    aspect_ratio_options: Some(vec![
+      CommonAspectRatio::WideTwentyOneByNine,
+      CommonAspectRatio::WideSixteenByNine,
+      CommonAspectRatio::WideFourByThree,
+      CommonAspectRatio::Square,
+      CommonAspectRatio::TallThreeByFour,
+      CommonAspectRatio::TallNineBySixteen,
+    ]),
+    aspect_ratio_default: Some(CommonAspectRatio::WideSixteenByNine),
+    resolution_options: Some(vec![
+      CommonResolution::FourEightyP,
+      CommonResolution::SevenTwentyP,
+      CommonResolution::TenEightyP,
+    ]),
+    resolution_default: Some(CommonResolution::SevenTwentyP),
+    batch_size_options: Some(vec![1, 2, 4]),
+    batch_size_default: Some(1),
+    duration_seconds_min: Some(4),
+    duration_seconds_max: Some(15),
+    duration_seconds_default: Some(5),
+    ..Default::default()
+  });
+
+  models.push(OmniGenVideoModelDetails {
+    model: CommonVideoModel::Seedance2p0FastGlobal,
+    model_creator: Some(ModelCreator::Bytedance),
+    full_name: Some("Seedance 2.0 Fast (Global)".to_string()),
+    text_prompt_supported: Some(true),
+    starting_keyframe_supported: Some(true),
+    ending_keyframe_supported: Some(true),
+    image_references_supported: Some(true),
+    image_references_max: Some(9),
+    audio_references_supported: Some(true),
+    audio_references_max: Some(3),
+    audio_references_max_total_duration_seconds: Some(15),
+    video_references_supported: Some(true),
+    video_references_max: Some(3),
+    video_references_max_total_duration_seconds: Some(15),
+    character_references_supported: Some(true),
+    character_references_max: Some(9),
+    aspect_ratio_options: Some(vec![
+      CommonAspectRatio::WideTwentyOneByNine,
+      CommonAspectRatio::WideSixteenByNine,
+      CommonAspectRatio::WideFourByThree,
+      CommonAspectRatio::Square,
+      CommonAspectRatio::TallThreeByFour,
+      CommonAspectRatio::TallNineBySixteen,
+    ]),
+    aspect_ratio_default: Some(CommonAspectRatio::WideSixteenByNine),
+    resolution_options: Some(vec![
+      CommonResolution::FourEightyP,
+      CommonResolution::SevenTwentyP,
+    ]),
+    resolution_default: Some(CommonResolution::SevenTwentyP),
+    batch_size_options: Some(vec![1, 2, 4]),
+    batch_size_default: Some(1),
+    duration_seconds_min: Some(4),
+    duration_seconds_max: Some(15),
+    duration_seconds_default: Some(5),
+    ..Default::default()
+  });
 
   // TODO(bt,2026-04-10): Veo 2 image-to-video doesn't support aspect ratio
   models.push(OmniGenVideoModelDetails {