codegen db

.gitignore

@@ -68,6 +68,9 @@ __pycache__/
 # We use diesel for DB migrations only. We don't care about its codgen.
 _ignored_schema.rs
 
+# Diesel lock file
+.diesel_lock
+
 # AI Chat Sidecar has its own schema managed by diesel, but we don't care for diesel codegen.
 crates/programs/aichatbot-sidecar/_ignored_schema.rs
 

.sqlx/query-01f6369e0ab7b0fa5b95b4595827dcf15897f3b50cce5b1dc421101b157b7008.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "MySQL",
+  "query": "\nUPDATE wallets\nSET\n  banked_credits = ?,\n  version = version + 1\nWHERE token = ?\nLIMIT 1\n    ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "01f6369e0ab7b0fa5b95b4595827dcf15897f3b50cce5b1dc421101b157b7008"
+}

.sqlx/query-33aa8a1cf85bf1b8b68f4e2ab46541f25d7e8bb5eeb503c5ec14cb5e115f69ce.json

@@ -1,12 +0,0 @@
-{
-  "db_name": "MySQL",
-  "query": "\nINSERT INTO generic_inference_jobs\nSET\n  token = ?,\n  uuid_idempotency_token = ?,\n\n  job_type = ?,\n\n  maybe_external_third_party = ?,\n  maybe_external_third_party_id = ?,\n\n  product_category = ?,\n  inference_category = ?,\n\n  maybe_model_type = NULL,\n  maybe_model_token = NULL,\n\n  maybe_input_source_token = NULL,\n  maybe_input_source_token_type = NULL,\n\n  maybe_download_url = NULL,\n  maybe_cover_image_media_file_token = NULL,\n\n  maybe_prompt_token = ?,\n\n  maybe_raw_inference_text = NULL,\n\n  maybe_inference_args = ?,\n\n  maybe_creator_user_token = ?,\n  maybe_creator_anonymous_visitor_token = ?,\n  creator_ip_address = ?,\n  creator_set_visibility = ?,\n\n  priority_level = 0,\n  is_keepalive_required = FALSE,\n  max_duration_seconds = 0,\n\n  is_debug_request = FALSE,\n  maybe_routing_tag = NULL,\n\n  status = ?\n        ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Right": 14
-    },
-    "nullable": []
-  },
-  "hash": "33aa8a1cf85bf1b8b68f4e2ab46541f25d7e8bb5eeb503c5ec14cb5e115f69ce"
-}

.sqlx/query-6701a7adad65bc351e4600dad3ebf0f5dbdefa7dc9a0f890790cfaf368bf572d.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "MySQL",
+  "query": "\nUPDATE wallet_ledger_entries\nSET\n  is_refunded = TRUE,\n  maybe_linked_refund_ledger_token = ?\nWHERE token = ?\nLIMIT 1\n    ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "6701a7adad65bc351e4600dad3ebf0f5dbdefa7dc9a0f890790cfaf368bf572d"
+}

.sqlx/query-a7365998006f4d40ad26efb42e6958a7aed689562a4c5b8c91cbe0033d0270e7.json

@@ -0,0 +1,102 @@
+{
+  "db_name": "MySQL",
+  "query": "\nSELECT\n  wallet_token as `wallet_token: tokens::tokens::wallets::WalletToken`,\n  entry_type,\n  is_refunded as `is_refunded: bool`,\n  credits_delta,\n  banked_credits_before,\n  banked_credits_after,\n  monthly_credits_before,\n  monthly_credits_after\nFROM wallet_ledger_entries\nWHERE token = ?\nLIMIT 1\nFOR UPDATE\n    ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "wallet_token: tokens::tokens::wallets::WalletToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "NOT_NULL | MULTIPLE_KEY | BINARY | NO_DEFAULT_VALUE",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 1,
+        "name": "entry_type",
+        "type_info": {
+          "type": "VarString",
+          "flags": "NOT_NULL | BINARY | NO_DEFAULT_VALUE",
+          "char_set": 224,
+          "max_size": 128
+        }
+      },
+      {
+        "ordinal": 2,
+        "name": "is_refunded: bool",
+        "type_info": {
+          "type": "Tiny",
+          "flags": "NOT_NULL",
+          "char_set": 63,
+          "max_size": 1
+        }
+      },
+      {
+        "ordinal": 3,
+        "name": "credits_delta",
+        "type_info": {
+          "type": "Long",
+          "flags": "NOT_NULL",
+          "char_set": 63,
+          "max_size": 11
+        }
+      },
+      {
+        "ordinal": 4,
+        "name": "banked_credits_before",
+        "type_info": {
+          "type": "Long",
+          "flags": "NOT_NULL | UNSIGNED",
+          "char_set": 63,
+          "max_size": 10
+        }
+      },
+      {
+        "ordinal": 5,
+        "name": "banked_credits_after",
+        "type_info": {
+          "type": "Long",
+          "flags": "NOT_NULL | UNSIGNED",
+          "char_set": 63,
+          "max_size": 10
+        }
+      },
+      {
+        "ordinal": 6,
+        "name": "monthly_credits_before",
+        "type_info": {
+          "type": "Long",
+          "flags": "NOT_NULL | UNSIGNED",
+          "char_set": 63,
+          "max_size": 10
+        }
+      },
+      {
+        "ordinal": 7,
+        "name": "monthly_credits_after",
+        "type_info": {
+          "type": "Long",
+          "flags": "NOT_NULL | UNSIGNED",
+          "char_set": 63,
+          "max_size": 10
+        }
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "a7365998006f4d40ad26efb42e6958a7aed689562a4c5b8c91cbe0033d0270e7"
+}

.sqlx/query-cb33fd483f0fef7ec2a1f5950218f312a19e38c7971b88db949ccaa05bf7b7c4.json

@@ -1,6 +1,6 @@
 {
   "db_name": "MySQL",
-  "query": "\nSELECT\n    jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n    jobs.maybe_external_third_party_id as `order_id`,\n    jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n    jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n    jobs.creator_ip_address,\n    jobs.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`,\n    jobs.maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`\n\nFROM generic_inference_jobs as jobs\n\nWHERE jobs.maybe_external_third_party = ?\n  AND jobs.status NOT IN ('complete_success', 'complete_failure')\n  AND jobs.maybe_external_third_party_id IS NOT NULL\n\nLIMIT 25000\n    ",
+  "query": "\nSELECT\n    jobs.token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n    jobs.maybe_external_third_party_id as `order_id`,\n    jobs.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n    jobs.maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n    jobs.creator_ip_address,\n    jobs.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`,\n    jobs.maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,\n    jobs.maybe_wallet_ledger_entry_token as `maybe_wallet_ledger_entry_token: tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken`\n\nFROM generic_inference_jobs as jobs\n\nWHERE jobs.maybe_external_third_party = ?\n  AND jobs.status NOT IN ('complete_success', 'complete_failure')\n  AND jobs.maybe_external_third_party_id IS NOT NULL\n\nLIMIT 25000\n    ",
   "describe": {
     "columns": [
       {
@@ -72,6 +72,16 @@
           "char_set": 224,
           "max_size": 128
         }
+      },
+      {
+        "ordinal": 7,
+        "name": "maybe_wallet_ledger_entry_token: tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
       }
     ],
     "parameters": {
@@ -84,8 +94,9 @@
       true,
       false,
       false,
+      true,
       true
     ]
   },
-  "hash": "ca96942d8214edb84f163cf40131ceed873209e86718054e1502219fc81d2dc6"
+  "hash": "cb33fd483f0fef7ec2a1f5950218f312a19e38c7971b88db949ccaa05bf7b7c4"
 }

.sqlx/query-ee228ab4a079fde6b3683b6cde2a2384d3def27a5d09baa5caf00a521d5d2bcd.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "MySQL",
+  "query": "\nINSERT INTO generic_inference_jobs\nSET\n  token = ?,\n  uuid_idempotency_token = ?,\n\n  job_type = ?,\n\n  maybe_external_third_party = ?,\n  maybe_external_third_party_id = ?,\n\n  product_category = ?,\n  inference_category = ?,\n\n  maybe_model_type = NULL,\n  maybe_model_token = NULL,\n\n  maybe_input_source_token = NULL,\n  maybe_input_source_token_type = NULL,\n\n  maybe_download_url = NULL,\n  maybe_cover_image_media_file_token = NULL,\n\n  maybe_prompt_token = ?,\n\n  maybe_wallet_ledger_entry_token = ?,\n\n  maybe_raw_inference_text = NULL,\n\n  maybe_inference_args = ?,\n\n  maybe_creator_user_token = ?,\n  maybe_creator_anonymous_visitor_token = ?,\n  creator_ip_address = ?,\n  creator_set_visibility = ?,\n\n  priority_level = 0,\n  is_keepalive_required = FALSE,\n  max_duration_seconds = 0,\n\n  is_debug_request = FALSE,\n  maybe_routing_tag = NULL,\n\n  status = ?\n        ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 15
+    },
+    "nullable": []
+  },
+  "hash": "ee228ab4a079fde6b3683b6cde2a2384d3def27a5d09baa5caf00a521d5d2bcd"
+}

_database/sql/migrations/2026-02-26-083102-0000_alter_jobs_add_wallet_ledger_entry_token/down.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+ALTER TABLE generic_inference_jobs
+DROP COLUMN
+maybe_wallet_ledger_entry_token;

_database/sql/migrations/2026-02-26-083102-0000_alter_jobs_add_wallet_ledger_entry_token/up.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+ALTER TABLE generic_inference_jobs
+ADD COLUMN
+maybe_wallet_ledger_entry_token VARCHAR(32) DEFAULT NULL;

_database/sql/migrations/2026-02-26-083139-0000_alter_wallet_ledger_entries_add_is_refunded/down.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+ALTER TABLE wallet_ledger_entries
+DROP COLUMN
+is_refunded;

_database/sql/migrations/2026-02-26-083139-0000_alter_wallet_ledger_entries_add_is_refunded/up.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+ALTER TABLE wallet_ledger_entries
+ADD COLUMN
+is_refunded BOOLEAN NOT NULL DEFAULT FALSE;

_database/sql/migrations/2026-02-26-083221-0000_alter_wallet_ledger_entries_add_linked_refund_token/down.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+ALTER TABLE wallet_ledger_entries
+DROP COLUMN
+maybe_linked_refund_ledger_token;

_database/sql/migrations/2026-02-26-083221-0000_alter_wallet_ledger_entries_add_linked_refund_token/up.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+ALTER TABLE wallet_ledger_entries
+ADD COLUMN
+maybe_linked_refund_ledger_token VARCHAR(32) DEFAULT NULL;

_database/sql/migrations_squashed/generic_inference_jobs.sql

@@ -182,6 +182,9 @@ CREATE TABLE generic_inference_jobs (
   is_from_api_user BOOLEAN NOT NULL DEFAULT FALSE,
   is_for_twitch BOOLEAN NOT NULL DEFAULT FALSE,
 
+  -- If credits were spent for this job, this references the corresponding wallet ledger entry.
+  maybe_wallet_ledger_entry_token VARCHAR(32) DEFAULT NULL,
+
   -- ========== DEVELOPMENT AND DEBUGGING METADATA ==========
 
   -- If true, the request gets a "debug" flag, which may do different

_database/sql/migrations_squashed/wallets/wallet_ledger_entries.sql

@@ -40,6 +40,12 @@ CREATE TABLE wallet_ledger_entries (
   -- Balance of monthly credits after the event.
   monthly_credits_after INTEGER UNSIGNED NOT NULL DEFAULT 0,
 
+  -- Whether this ledger entry has been refunded.
+  is_refunded BOOLEAN NOT NULL DEFAULT FALSE,
+
+  -- If this entry was refunded, this points to the refund ledger entry.
+  maybe_linked_refund_ledger_token VARCHAR(32) DEFAULT NULL,
+
   -- ========== RECORD TIMESTAMPS ==========
 
   created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,

crates/api_clients/seedance2pro/src/requests/generate_video/generate_video.rs

@@ -160,10 +160,10 @@ pub async fn generate_video(args: GenerateVideoArgs<'_>) -> Result<GenerateVideo
         api_params: ApiParams {
           prompt: args.prompt,
           resolution: args.resolution.as_str().to_string(),
-          mode: "normal",
+          content_mode: "normal",
           model: "seedance-20",
           duration,
-          video_input_mode,
+          mode: video_input_mode,
           uploaded_urls,
           batch_count,
         },
@@ -344,7 +344,7 @@ mod tests {
       session: &session,
       prompt: "A dog shakes the glasses off its head. The camera pans out as the shiba shakes. The shiba barks.".to_string(),
       resolution: Resolution::Landscape16x9,
-      duration_seconds: 15,
+      duration_seconds: 5,
       batch_count: BatchCount::One,
       start_frame_url: Some("https://static.seedance2-pro.com/materials/20260219/1771496300184-fb32e08c.jpg".to_string()),
       end_frame_url: None,

crates/api_clients/seedance2pro/src/requests/generate_video/request_types.rs

@@ -23,11 +23,11 @@ pub (super) struct BatchRequestJson {
 pub (super) struct ApiParams {
   pub prompt: String,
   pub resolution: String,
-  pub mode: &'static str,
+  #[serde(rename = "contentMode")]
+  pub content_mode: &'static str,
   pub model: &'static str,
   pub duration: String,
-  #[serde(rename = "videoInputMode")]
-  pub video_input_mode: &'static str,
+  pub mode: &'static str,
   #[serde(rename = "uploadedUrls", skip_serializing_if = "Option::is_none")]
   pub uploaded_urls: Option<Vec<String>>,
   #[serde(rename = "batchCount", skip_serializing_if = "Option::is_none")]

crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token.rs

@@ -13,6 +13,7 @@ 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;
 
 use crate::errors::database_query_error::DatabaseQueryError;
 use crate::payloads::generic_inference_args::generic_inference_args::GenericInferenceArgs;
@@ -32,6 +33,8 @@ pub struct InsertGenericInferenceForSeedance2ProWithAprioriJobTokenArgs<'e, 'c,
 
   pub maybe_prompt_token: Option<&'e PromptToken>,
 
+  pub maybe_wallet_ledger_entry_token: Option<&'e WalletLedgerEntryToken>,
+
   pub maybe_creator_user_token: Option<&'e UserToken>,
   pub maybe_avt_token: Option<&'e AnonymousVisitorTrackingToken>,
   pub creator_ip_address: &'e str,
@@ -82,6 +85,8 @@ SET
 
   maybe_prompt_token = ?,
 
+  maybe_wallet_ledger_entry_token = ?,
+
   maybe_raw_inference_text = NULL,
 
   maybe_inference_args = ?,
@@ -113,6 +118,8 @@ SET
 
         args.maybe_prompt_token.map(|t| t.to_string()),
 
+        args.maybe_wallet_ledger_entry_token.map(|t| t.to_string()),
+
         serialized_args_payload,
 
         args.maybe_creator_user_token.map(|t| t.to_string()),

crates/schema/database/mysql_queries/src/queries/generic_inference/seedance2pro/list_pending_seedance2pro_jobs.rs

@@ -9,6 +9,7 @@ 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 Seedance2Pro job that is waiting for a result from the external API.
 #[derive(Debug)]
@@ -24,6 +25,8 @@ pub struct PendingSeedance2ProJob {
   pub creator_set_visibility: Visibility,
 
   pub maybe_prompt_token: Option<PromptToken>,
+
+  pub maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
 }
 
 #[derive(Debug, Default)]
@@ -35,6 +38,7 @@ struct RawRecord {
   creator_ip_address: String,
   creator_set_visibility: Visibility,
   maybe_prompt_token: Option<PromptToken>,
+  maybe_wallet_ledger_entry_token: Option<WalletLedgerEntryToken>,
 }
 
 /// Returns all non-terminal Seedance2Pro jobs that have an associated order_id.
@@ -49,7 +53,8 @@ SELECT
     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_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
 
@@ -84,6 +89,7 @@ LIMIT 25000
         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();

crates/schema/database/mysql_queries/src/queries/wallet_ledger_entries/internal_insert_wallet_ledger_entry.rs

@@ -21,7 +21,7 @@ pub (crate) struct InsertWalletLedgerEntry<'a> {
   /// An optional reference for this entry. Could be an internal ID (eg. job) or an
   /// external ID (eg. Stripe payment intent, invoice, etc.)
   pub maybe_entity_ref: Option<String>,
-  
+
   /// Change in credits (positive or negative) across banked and/or monthly credits.
   pub credits_delta: i64,
 
@@ -40,41 +40,35 @@ pub (crate) struct InsertWalletLedgerEntry<'a> {
 
 impl <'a> InsertWalletLedgerEntry<'a> {
 
-  pub async fn upsert(&'a self, mysql_pool: &MySqlPool) -> AnyhowResult<()> {
+  pub async fn upsert_with_pool(&'a self, mysql_pool: &MySqlPool) -> AnyhowResult<WalletLedgerEntryToken> {
     let mut conn = mysql_pool.acquire().await?;
     self.upsert_with_connection(&mut conn).await
   }
 
-  pub async fn upsert_with_connection(&'a self, mysql_connection: &mut PoolConnection<MySql>) -> AnyhowResult<()> {
-    let query = self.query();
-    let query_result = query.execute(&mut **mysql_connection).await;
+  pub async fn upsert_with_connection(&'a self, mysql_connection: &mut PoolConnection<MySql>) -> AnyhowResult<WalletLedgerEntryToken> {
+    let ledger_token = WalletLedgerEntryToken::generate();
+    let query_result = self.query_with_token(&ledger_token).execute(&mut **mysql_connection).await;
 
-    let _record_id = match query_result {
-      Ok(res) => res.last_insert_id(),
-      Err(err) => return Err(anyhow!("Error upserting wallet ledger entry record: {:?}", err)),
-    };
-
-    Ok(())
+    match query_result {
+      Ok(_) => Ok(ledger_token),
+      Err(err) => Err(anyhow!("Error upserting wallet ledger entry record: {:?}", err)),
+    }
   }
 
-  pub async fn upsert_with_transaction(&'a self, transaction: &mut Transaction<'_, MySql>) -> Result<(), sqlx::Error> {
-    let query = self.query();
-    let query_result = query.execute(&mut **transaction).await;
+  pub async fn upsert_with_transaction(&'a self, transaction: &mut Transaction<'_, MySql>) -> Result<WalletLedgerEntryToken, sqlx::Error> {
+    let ledger_token = WalletLedgerEntryToken::generate();
+    let query_result = self.query_with_token(&ledger_token).execute(&mut **transaction).await;
 
-    let _record_id = match query_result {
-      Ok(res) => res.last_insert_id(),
+    match query_result {
+      Ok(_) => Ok(ledger_token),
       Err(err) => {
         error!("Error upserting wallet ledger entry record: {:?}", err);
-        return Err(err);
+        Err(err)
       },
-    };
-
-    Ok(())
+    }
   }
 
-  fn query(&self) -> Query<MySql, MySqlArguments> {
-    let token = WalletLedgerEntryToken::generate().to_string();
-
+  fn query_with_token(&self, ledger_token: &WalletLedgerEntryToken) -> Query<MySql, MySqlArguments> {
     sqlx::query!(
         r#"
 INSERT INTO wallet_ledger_entries
@@ -89,7 +83,7 @@ SET
   monthly_credits_before = ?,
   monthly_credits_after = ?
         "#,
-      token.as_str(),
+      ledger_token.as_str(),
       self.wallet_token.as_str(),
       self.entry_type.to_str(),
       self.maybe_entity_ref,

crates/schema/database/mysql_queries/src/queries/wallets/add_durable_banked_balance_to_wallet.rs

@@ -52,11 +52,12 @@ pub async fn add_durable_banked_balance_to_wallet(
     monthly_credits_after: wallet.monthly_credits,
   };
 
-  record.upsert_with_transaction(transaction).await?;
+  let wallet_ledger_entry_token = record.upsert_with_transaction(transaction).await?;
 
 
   Ok(WalletUpdateSummary {
-    token: wallet.token,
+    wallet_token: wallet.token,
+    wallet_ledger_entry_token,
     namespace: wallet.namespace,
     owner_user_token: wallet.owner_user_token,
 

crates/schema/database/mysql_queries/src/queries/wallets/mod.rs

@@ -4,5 +4,6 @@ pub mod create_new_artcraft_wallet_for_owner_user;
 pub mod find_primary_wallet_for_owner;
 pub mod find_primary_wallet_token_for_owner;
 pub mod refill_monthly_credits_balance_on_wallet;
+pub mod refund;
 pub mod spend;
 pub mod wallet_update_summary;

crates/schema/database/mysql_queries/src/queries/wallets/refill_monthly_credits_balance_on_wallet.rs

@@ -51,11 +51,12 @@ pub async fn refill_monthly_credits_balance_on_wallet(
     banked_credits_after: wallet.banked_credits,
   };
 
-  record.upsert_with_transaction(transaction).await?;
+  let wallet_ledger_entry_token = record.upsert_with_transaction(transaction).await?;
 
 
   Ok(WalletUpdateSummary {
-    token: wallet.token,
+    wallet_token: wallet.token,
+    wallet_ledger_entry_token,
     namespace: wallet.namespace,
     owner_user_token: wallet.owner_user_token,
 

crates/schema/database/mysql_queries/src/queries/wallets/refund/mod.rs

@@ -0,0 +1,2 @@
+pub mod try_to_refund_ledger_entry;
+pub mod wallet_refund_error;

crates/schema/database/mysql_queries/src/queries/wallets/refund/try_to_refund_ledger_entry.rs

@@ -0,0 +1,219 @@
+use log::info;
+use sqlx::MySql;
+
+use enums::by_table::wallet_ledger_entries::wallet_ledger_entry_type::WalletLedgerEntryType;
+use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
+use tokens::tokens::wallets::WalletToken;
+
+use crate::queries::wallet_ledger_entries::internal_insert_wallet_ledger_entry::InsertWalletLedgerEntry;
+use crate::queries::wallets::internal_select_wallet_balance_for_update::internal_select_wallet_balance_for_update;
+use crate::queries::wallets::refund::wallet_refund_error::WalletRefundError;
+
+pub enum WalletRefundOutcome {
+  /// The refund was successfully applied.
+  Refunded(WalletRefundSummary),
+
+  /// The ledger entry was already marked as refunded; no action was taken.
+  /// This is not an error — it is safe to call refund multiple times.
+  AlreadyRefunded,
+}
+
+pub struct WalletRefundSummary {
+  pub wallet_token: WalletToken,
+  pub original_ledger_entry_token: WalletLedgerEntryToken,
+  pub refund_ledger_entry_token: WalletLedgerEntryToken,
+  pub refund_amount: u64,
+  pub banked_credits_before: u64,
+  pub banked_credits_after: u64,
+}
+
+/// Refund a wallet ledger entry, crediting the amount back as banked (durable) credits.
+///
+/// NB: BE VERY CAREFUL WITH THIS FUNCTION!
+/// It locks both the ledger entry row and the wallet row for the duration of the transaction.
+///
+/// We always refund into banked credits rather than monthly credits. Refunding into monthly
+/// credits near a billing cycle cutoff could create race conditions with the monthly refill job,
+/// so banked credits are the safer, permanent choice.
+pub async fn try_to_refund_ledger_entry(
+  ledger_entry_token: &WalletLedgerEntryToken,
+  transaction: &mut sqlx::Transaction<'_, MySql>,
+) -> Result<WalletRefundOutcome, WalletRefundError> {
+
+  // Step 1: Lock the original ledger entry for the duration of this transaction.
+  let original_entry = select_ledger_entry_for_update(ledger_entry_token, transaction).await?;
+
+  // Step 2: Guard — bail out gracefully if it was already refunded.
+  if original_entry.is_refunded {
+    return Ok(WalletRefundOutcome::AlreadyRefunded);
+  }
+
+  // Step 3: Guard — only deduct-type entries may be refunded.
+  let is_deduct_type = matches!(
+    original_entry.entry_type,
+    WalletLedgerEntryType::DeductMixed
+      | WalletLedgerEntryType::DeductBanked
+      | WalletLedgerEntryType::DeductMonthly,
+  );
+
+  if !is_deduct_type {
+    return Err(WalletRefundError::NotADeductEntry(original_entry.entry_type));
+  }
+
+  // Step 4: Lock the wallet row so our balance arithmetic is atomic.
+  let wallet = internal_select_wallet_balance_for_update(
+    &original_entry.wallet_token,
+    transaction,
+  ).await.map_err(|_| WalletRefundError::WalletNotFound)?;
+
+  // credits_delta was negative for a deduction (e.g. -100). The refund amount is its magnitude.
+  let refund_amount = original_entry.credits_delta.unsigned_abs();
+
+  let banked_credits_before = wallet.banked_credits;
+  let banked_credits_after = banked_credits_before.saturating_add(refund_amount);
+
+  // Step 5: Insert the refund ledger entry.
+  //
+  // We record the original ledger entry token as the entity_ref so there is a clear audit trail
+  // linking the refund back to the original spend. The new entry itself has is_refunded = false
+  // because it is the refund, not the thing being refunded.
+  let refund_ledger_record = InsertWalletLedgerEntry {
+    wallet_token: &original_entry.wallet_token,
+    entry_type: WalletLedgerEntryType::RefundBanked,
+    maybe_entity_ref: Some(ledger_entry_token.to_string()),
+
+    credits_delta: refund_amount as i64,
+
+    banked_credits_before,
+    banked_credits_after,
+
+    // Monthly credits are untouched by a banked refund.
+    monthly_credits_before: wallet.monthly_credits,
+    monthly_credits_after: wallet.monthly_credits,
+  };
+
+  let refund_ledger_entry_token = refund_ledger_record
+    .upsert_with_transaction(transaction)
+    .await?;
+
+  // Step 6: Credit the wallet's banked balance.
+  sqlx::query!(
+    r#"
+UPDATE wallets
+SET
+  banked_credits = ?,
+  version = version + 1
+WHERE token = ?
+LIMIT 1
+    "#,
+    banked_credits_after,
+    original_entry.wallet_token.as_str(),
+  ).execute(&mut **transaction).await?;
+
+  // Step 7: Mark the original entry as refunded and link it to the new refund record.
+  sqlx::query!(
+    r#"
+UPDATE wallet_ledger_entries
+SET
+  is_refunded = TRUE,
+  maybe_linked_refund_ledger_token = ?
+WHERE token = ?
+LIMIT 1
+    "#,
+    refund_ledger_entry_token.as_str(),
+    ledger_entry_token.as_str(),
+  ).execute(&mut **transaction).await?;
+
+  info!(
+    "Refunded ledger entry {} → new refund entry {}; wallet {} banked credits: {} → {}",
+    ledger_entry_token.as_str(),
+    refund_ledger_entry_token.as_str(),
+    original_entry.wallet_token.as_str(),
+    banked_credits_before,
+    banked_credits_after,
+  );
+
+  Ok(WalletRefundOutcome::Refunded(WalletRefundSummary {
+    wallet_token: original_entry.wallet_token,
+    original_ledger_entry_token: ledger_entry_token.clone(),
+    refund_ledger_entry_token,
+    refund_amount,
+    banked_credits_before,
+    banked_credits_after,
+  }))
+}
+
+// ===== Internal helpers =====
+
+struct LedgerEntryForUpdate {
+  wallet_token: WalletToken,
+  entry_type: WalletLedgerEntryType,
+  is_refunded: bool,
+  /// Negative for deductions, positive for credits.
+  credits_delta: i64,
+  banked_credits_before: u64,
+  banked_credits_after: u64,
+  monthly_credits_before: u64,
+  monthly_credits_after: u64,
+}
+
+/// SELECT ... FOR UPDATE on wallet_ledger_entries.
+/// Locks the row for the duration of the enclosing transaction.
+async fn select_ledger_entry_for_update(
+  ledger_entry_token: &WalletLedgerEntryToken,
+  transaction: &mut sqlx::Transaction<'_, MySql>,
+) -> Result<LedgerEntryForUpdate, WalletRefundError> {
+
+  struct RawLedgerEntryForUpdate {
+    wallet_token: WalletToken,
+    entry_type: String,
+    is_refunded: bool,
+    credits_delta: i32,
+    banked_credits_before: u32,
+    banked_credits_after: u32,
+    monthly_credits_before: u32,
+    monthly_credits_after: u32,
+  }
+
+  let raw = sqlx::query_as!(
+    RawLedgerEntryForUpdate,
+    r#"
+SELECT
+  wallet_token as `wallet_token: tokens::tokens::wallets::WalletToken`,
+  entry_type,
+  is_refunded as `is_refunded: bool`,
+  credits_delta,
+  banked_credits_before,
+  banked_credits_after,
+  monthly_credits_before,
+  monthly_credits_after
+FROM wallet_ledger_entries
+WHERE token = ?
+LIMIT 1
+FOR UPDATE
+    "#,
+    ledger_entry_token.as_str(),
+  )
+    .fetch_one(&mut **transaction)
+    .await
+    .map_err(|e| match e {
+      sqlx::Error::RowNotFound => WalletRefundError::LedgerEntryNotFound,
+      err => WalletRefundError::SqlxError(err),
+    })?;
+
+  let entry_type = WalletLedgerEntryType::from_str(&raw.entry_type)
+    .map_err(|_| WalletRefundError::SqlxError(sqlx::Error::Decode(
+      format!("unknown wallet_ledger_entry entry_type: {:?}", raw.entry_type).into()
+    )))?;
+
+  Ok(LedgerEntryForUpdate {
+    wallet_token: raw.wallet_token,
+    entry_type,
+    is_refunded: raw.is_refunded,
+    credits_delta: raw.credits_delta as i64,
+    banked_credits_before: raw.banked_credits_before as u64,
+    banked_credits_after: raw.banked_credits_after as u64,
+    monthly_credits_before: raw.monthly_credits_before as u64,
+    monthly_credits_after: raw.monthly_credits_after as u64,
+  })
+}

crates/schema/database/mysql_queries/src/queries/wallets/refund/wallet_refund_error.rs

@@ -0,0 +1,51 @@
+use crate::errors::select_exactly_one_error::SelectExactlyOneError;
+use enums::by_table::wallet_ledger_entries::wallet_ledger_entry_type::WalletLedgerEntryType;
+use std::error::Error;
+use std::fmt::{Display, Formatter};
+
+#[derive(Debug)]
+pub enum WalletRefundError {
+  /// The wallet ledger entry to refund was not found.
+  LedgerEntryNotFound,
+
+  /// The wallet referenced by the ledger entry was not found.
+  WalletNotFound,
+
+  /// The ledger entry is not a deduct-type entry; only DeductMixed, DeductBanked,
+  /// and DeductMonthly entries can be refunded.
+  NotADeductEntry(WalletLedgerEntryType),
+
+  /// Error selecting a record.
+  SelectError(SelectExactlyOneError),
+
+  /// Underlying database error.
+  SqlxError(sqlx::Error),
+}
+
+impl Error for WalletRefundError {}
+
+impl Display for WalletRefundError {
+  fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+    match self {
+      WalletRefundError::LedgerEntryNotFound => write!(f, "Ledger entry not found"),
+      WalletRefundError::WalletNotFound => write!(f, "Wallet not found"),
+      WalletRefundError::NotADeductEntry(entry_type) => {
+        write!(f, "Cannot refund a non-deduct ledger entry: {:?}", entry_type)
+      }
+      WalletRefundError::SelectError(err) => write!(f, "Database select error: {}", err),
+      WalletRefundError::SqlxError(err) => write!(f, "Database error: {}", err),
+    }
+  }
+}
+
+impl From<SelectExactlyOneError> for WalletRefundError {
+  fn from(err: SelectExactlyOneError) -> Self {
+    WalletRefundError::SelectError(err)
+  }
+}
+
+impl From<sqlx::Error> for WalletRefundError {
+  fn from(err: sqlx::Error) -> Self {
+    WalletRefundError::SqlxError(err)
+  }
+}

crates/schema/database/mysql_queries/src/queries/wallets/spend/try_to_spend_wallet_balance.rs

@@ -115,10 +115,11 @@ pub async fn try_to_spend_wallet_balance(
     banked_credits_after: updated_banked_balance
   };
 
-  record.upsert_with_transaction(transaction).await?;
+  let wallet_ledger_entry_token = record.upsert_with_transaction(transaction).await?;
 
   Ok(WalletUpdateSummary {
-    token: wallet.token,
+    wallet_token: wallet.token,
+    wallet_ledger_entry_token,
     namespace: wallet.namespace,
     owner_user_token: wallet.owner_user_token,
 

crates/schema/database/mysql_queries/src/queries/wallets/wallet_update_summary.rs

@@ -1,9 +1,12 @@
 use enums::common::payments_namespace::PaymentsNamespace;
 use tokens::tokens::users::UserToken;
+use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
 use tokens::tokens::wallets::WalletToken;
 
 pub struct WalletUpdateSummary {
-  pub token: WalletToken,
+  pub wallet_token: WalletToken,
+
+  pub wallet_ledger_entry_token: WalletLedgerEntryToken,
 
   pub namespace: PaymentsNamespace,
 

crates/schema/public/enums/src/by_table/wallet_ledger_entries/wallet_ledger_entry_type.rs

@@ -36,6 +36,18 @@ pub enum WalletLedgerEntryType {
   /// Deduct monthly credits
   #[serde(rename = "deduct_monthly")]
   DeductMonthly,
+
+  /// Refund banked credits
+  #[serde(rename = "refund_banked")]
+  RefundBanked,
+
+  // TODO: No clean way to do "mixed" refunds yet, and if we
+  //  refund close to the cutoff it might be unfair. Let's
+  //  just not do monthly refunds yet and instead credit our
+  //  users with banked credits.
+  // /// Refund monthly credits
+  // #[serde(rename = "refund_monthly")]
+  // RefundMonthly,
 }
 
 // TODO(bt, 2022-12-21): This desperately needs MySQL integration tests!
@@ -53,6 +65,7 @@ impl WalletLedgerEntryType {
       Self::DeductMixed => "deduct_mixed",
       Self::DeductBanked => "deduct_banked",
       Self::DeductMonthly => "deduct_monthly",
+      Self::RefundBanked => "refund_banked",
     }
   }
 
@@ -64,6 +77,7 @@ impl WalletLedgerEntryType {
       "deduct_mixed" => Ok(Self::DeductMixed),
       "deduct_banked" => Ok(Self::DeductBanked),
       "deduct_monthly" => Ok(Self::DeductMonthly),
+      "refund_banked" => Ok(Self::RefundBanked),
       _ => Err(format!("invalid value: {:?}", value)),
     }
   }
@@ -78,6 +92,7 @@ impl WalletLedgerEntryType {
       Self::DeductMixed,
       Self::DeductBanked,
       Self::DeductMonthly,
+      Self::RefundBanked,
     ])
   }
 }
@@ -98,6 +113,7 @@ mod tests {
       assert_serialization(WalletLedgerEntryType::DeductMixed, "deduct_mixed");
       assert_serialization(WalletLedgerEntryType::DeductBanked, "deduct_banked");
       assert_serialization(WalletLedgerEntryType::DeductMonthly, "deduct_monthly");
+      assert_serialization(WalletLedgerEntryType::RefundBanked, "refund_banked");
     }
   }
 
@@ -112,6 +128,7 @@ mod tests {
       assert_eq!(WalletLedgerEntryType::DeductMixed.to_str(), "deduct_mixed");
       assert_eq!(WalletLedgerEntryType::DeductBanked.to_str(), "deduct_banked");
       assert_eq!(WalletLedgerEntryType::DeductMonthly.to_str(), "deduct_monthly");
+      assert_eq!(WalletLedgerEntryType::RefundBanked.to_str(), "refund_banked");
     }
 
     #[test]
@@ -122,6 +139,7 @@ mod tests {
       assert_eq!(WalletLedgerEntryType::from_str("deduct_mixed").unwrap(), WalletLedgerEntryType::DeductMixed);
       assert_eq!(WalletLedgerEntryType::from_str("deduct_banked").unwrap(), WalletLedgerEntryType::DeductBanked);
       assert_eq!(WalletLedgerEntryType::from_str("deduct_monthly").unwrap(), WalletLedgerEntryType::DeductMonthly);
+      assert_eq!(WalletLedgerEntryType::from_str("refund_banked").unwrap(), WalletLedgerEntryType::RefundBanked);
       assert!(WalletLedgerEntryType::from_str("foo").is_err());
     }
   }
@@ -132,13 +150,14 @@ mod tests {
     #[test]
     fn all_variants() {
       let mut variants = WalletLedgerEntryType::all_variants();
-      assert_eq!(variants.len(), 6);
+      assert_eq!(variants.len(), 7);
       assert_eq!(variants.pop_first(), Some(WalletLedgerEntryType::Create));
       assert_eq!(variants.pop_first(), Some(WalletLedgerEntryType::CreditBanked));
       assert_eq!(variants.pop_first(), Some(WalletLedgerEntryType::CreditMonthly));
       assert_eq!(variants.pop_first(), Some(WalletLedgerEntryType::DeductMixed));
       assert_eq!(variants.pop_first(), Some(WalletLedgerEntryType::DeductBanked));
       assert_eq!(variants.pop_first(), Some(WalletLedgerEntryType::DeductMonthly));
+      assert_eq!(variants.pop_first(), Some(WalletLedgerEntryType::RefundBanked));
       assert_eq!(variants.pop_first(), None);
     }
   }

crates/service/job/seedance2_pro_job/src/job/job_loop/mod.rs

@@ -1,2 +0,0 @@
-pub mod main_loop;
-pub mod process_completed_order;

crates/service/job/seedance2_pro_job/src/job/mod.rs

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

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

@@ -26,12 +26,13 @@ use seedance2pro::creds::seedance2pro_session::Seedance2ProSession;
 use server_environment::ServerEnvironment;
 
 use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs};
-use crate::job::job_loop::main_loop::main_loop;
+use crate::main_loop::main_loop;
 use crate::job_dependencies::JobDependencies;
 
 pub mod http_server;
-pub mod job;
 pub mod job_dependencies;
+pub mod main_loop;
+pub mod process_job;
 
 // Bucket config
 const ENV_ACCESS_KEY: &str = "ACCESS_KEY";

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

@@ -2,12 +2,11 @@ use std::collections::HashMap;
 use std::time::Duration;
 
 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::seedance2pro::list_pending_seedance2pro_jobs::list_pending_seedance2pro_jobs;
 use seedance2pro::requests::poll_orders::poll_orders::{poll_orders, PollOrdersArgs, TaskStatus};
 
-use crate::job::job_loop::process_completed_order::{mark_job_failed, process_completed_order};
+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) {
@@ -80,7 +79,7 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
           order.order_id,
           job.job_token.as_str()
         );
-        if let Err(err) = process_completed_order(deps, &job, order).await {
+        if let Err(err) = process_successful_job(deps, &job, order).await {
           warn!(
             "Error processing completed order {}: {:?}",
             order.order_id, err
@@ -91,43 +90,7 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
         }
       }
       TaskStatus::Failed => {
-        let reason = order
-          .fail_reason
-          .as_deref()
-          .unwrap_or("unknown failure reason");
-
-        let reason_lower = reason.to_lowercase();
-
-        let platform_rules_violation = reason_lower.contains("violates") ||
-            reason_lower.contains("platform rules") ||
-            reason_lower.contains("please modify");
-
-        let frontend_failure_category = if platform_rules_violation {
-          Some(FrontendFailureCategory::ModelRulesViolation)
-        } else {
-          None
-        };
-
-        warn!(
-          "Order {} failed: {}. Marking job {} failed.",
-          order.order_id, reason, job.job_token.as_str()
-        );
-
-        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
-          );
-        }
+        process_failed_job(deps, &job, order).await;
       }
       TaskStatus::Pending | TaskStatus::Processing => {
         // Still in progress — check again next poll.

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

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

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

@@ -0,0 +1,124 @@
+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::seedance2pro::list_pending_seedance2pro_jobs::PendingSeedance2ProJob;
+use mysql_queries::queries::wallets::refund::try_to_refund_ledger_entry::{try_to_refund_ledger_entry, WalletRefundOutcome};
+use seedance2pro::requests::poll_orders::poll_orders::OrderStatus;
+
+use crate::job_dependencies::JobDependencies;
+
+pub async fn process_failed_job(
+  deps: &JobDependencies,
+  job: &PendingSeedance2ProJob,
+  order: &OrderStatus,
+) {
+  let reason = order
+    .fail_reason
+    .as_deref()
+    .unwrap_or("unknown failure reason");
+
+  // --- Step 1: Attempt the refund before touching the job status. ---
+  //
+  // We do this first so that a crash or error between the refund and the status update
+  // can never result in a failed job with no refund. If the refund itself fails we bail
+  // out early: the job stays pending and will be retried on the next poll cycle.
+
+  match &job.maybe_wallet_ledger_entry_token {
+    None => {
+      // No ledger token recorded — job was likely submitted before billing was wired up.
+      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) => {
+          // Idempotent — nothing to do, safe to proceed.
+          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");
+
+  let frontend_failure_category = if platform_rules_violation {
+    Some(FrontendFailureCategory::ModelRulesViolation)
+  } else {
+    None
+  };
+
+  warn!(
+    "Order {} failed: {}. Marking job {} failed.",
+    order.order_id, reason, job.job_token.as_str()
+  );
+
+  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/seedance2_pro_job/src/process_job/process_successful_job.rs

@@ -9,18 +9,17 @@ use enums::by_table::media_files::media_file_origin_product_category::MediaFileO
 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::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs};
 use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedance2pro_jobs::PendingSeedance2ProJob;
 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 seedance2pro::requests::poll_orders::poll_orders::OrderStatus;
 use crate::job_dependencies::JobDependencies;
 
-const PREFIX : &str = "artcraft_";
-const SUFFIX : &str = ".mp4";
+const PREFIX: &str = "artcraft_";
+const SUFFIX: &str = ".mp4";
 
 /// Download the completed video, upload to bucket, create media file record, and mark job done.
-pub async fn process_completed_order(
+pub async fn process_successful_job(
   deps: &JobDependencies,
   job: &PendingSeedance2ProJob,
   order: &OrderStatus,
@@ -143,18 +142,3 @@ pub async fn process_completed_order(
 
   Ok(())
 }
-
-/// Mark a job as permanently failed.
-pub async fn mark_job_failed(
-  deps: &JobDependencies,
-  job: &PendingSeedance2ProJob,
-  fail_reason: &str,
-) -> AnyhowResult<()> {
-  mark_job_failed_by_token(MarkJobFailedByTokenArgs {
-    pool: &deps.mysql_pool,
-    job_token: &job.job_token,
-    maybe_public_failure_reason: Some(fail_reason),
-    internal_debugging_failure_reason: fail_reason,
-    maybe_frontend_failure_category: None,
-  }).await
-}

crates/service/web/storyteller_web/src/billing/wallets/attempt_wallet_deduction.rs

@@ -1,25 +1,28 @@
 use crate::http_server::common_responses::common_web_error::CommonWebError;
-use anyhow::anyhow;
 use enums::common::payments_namespace::PaymentsNamespace;
 use errors::AnyhowResult;
-use log::{error, info, warn};
+use log::{error, info};
 use mysql_queries::queries::wallets::create_new_artcraft_wallet_for_owner_user::create_new_artcraft_wallet_for_owner_user;
-use mysql_queries::queries::wallets::find_primary_wallet_token_for_owner::{find_primary_wallet_token_for_owner_using_connection, find_primary_wallet_token_for_owner_using_transaction};
+use mysql_queries::queries::wallets::find_primary_wallet_token_for_owner::find_primary_wallet_token_for_owner_using_connection;
 use mysql_queries::queries::wallets::spend::try_to_spend_wallet_balance::try_to_spend_wallet_balance;
 use mysql_queries::queries::wallets::spend::wallet_spend_error::WalletSpendError;
 use sqlx::pool::PoolConnection;
 use sqlx::{Acquire, MySql};
 use tokens::tokens::users::UserToken;
+use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
 use tokens::tokens::wallets::WalletToken;
 
-// TODO: This hasn't really been tested yet...
+pub struct WalletDeductionResult {
+  pub wallet_token: WalletToken,
+  pub ledger_entry_token: WalletLedgerEntryToken,
+}
 
 pub async fn attempt_wallet_deduction_else_common_web_error(
   user_token: &UserToken,
   maybe_reference_token: Option<&str>,
   amount_to_deduct: u64,
   connection: &mut PoolConnection<MySql>
-) -> Result<(), CommonWebError> {
+) -> Result<WalletDeductionResult, CommonWebError> {
 
   let result = try_wallet_deduction(
     user_token,
@@ -29,32 +32,31 @@ pub async fn attempt_wallet_deduction_else_common_web_error(
   ).await;
 
   // Infallible for now.
-  if let Err(err) = result {
-    return match err {
+  match result {
+    Ok(deduction_result) => Ok(deduction_result),
+    Err(err) => Err(match err {
       WalletSpendError::InvalidAmountToSpend => {
         log::error!("invalid spend amount charged");
-        Err(CommonWebError::PaymentRequired)
+        CommonWebError::PaymentRequired
       }
       WalletSpendError::InsufficientBalance { requested_to_spend_amount, available_amount } => {
         log::error!("payment is required - requested: {}, available: {}", requested_to_spend_amount, available_amount);
-        Err(CommonWebError::PaymentRequired)
+        CommonWebError::PaymentRequired
       }
       WalletSpendError::SelectError(err) => {
         log::error!("SQL error (select) in attempt_wallet_deduction: {:?}", err);
-        Err(CommonWebError::ServerError)
+        CommonWebError::ServerError
       }
       WalletSpendError::SelectOptionalError(err) => {
         log::error!("SQL error (select optional) in attempt_wallet_deduction: {:?}", err);
-        Err(CommonWebError::ServerError)
+        CommonWebError::ServerError
       }
       WalletSpendError::SqlxError(err) => {
         log::error!("SQL error (sqlx) in attempt_wallet_deduction: {:?}", err);
-        Err(CommonWebError::ServerError)
+        CommonWebError::ServerError
       }
-    }
+    }),
   }
-
-  Ok(())
 }
 
 async fn try_wallet_deduction(
@@ -62,7 +64,7 @@ async fn try_wallet_deduction(
   maybe_reference_token: Option<&str>,
   amount_to_deduct: u64,
   connection: &mut PoolConnection<MySql>
-) -> Result<(), WalletSpendError>
+) -> Result<WalletDeductionResult, WalletSpendError>
 {
   let maybe_wallet_token = find_primary_wallet_token_for_owner_using_connection(
     owner_user_token,
@@ -81,8 +83,9 @@ async fn try_wallet_deduction(
   ).await;
 
   match result {
-    Ok(()) => {
+    Ok(deduction_result) => {
       transaction.commit().await?;
+      Ok(deduction_result)
     },
     Err(err) => {
       error!("Error handling temporary wallet deduction for user {:?} : {:?}",
@@ -90,11 +93,9 @@ async fn try_wallet_deduction(
 
       transaction.rollback().await?;
 
-      return Err(err);
+      Err(err)
     }
   }
-
-  Ok(())
 }
 
 async fn try_wallet_deduction_with_transaction(
@@ -103,7 +104,7 @@ async fn try_wallet_deduction_with_transaction(
   maybe_reference_token: Option<&str>,
   amount_to_deduct: u64,
   transaction: &mut sqlx::Transaction<'_, MySql>,
-) -> Result<(), WalletSpendError>
+) -> Result<WalletDeductionResult, WalletSpendError>
 {
   let wallet_token = match maybe_wallet_token {
     Some(token) => token,
@@ -113,22 +114,23 @@ async fn try_wallet_deduction_with_transaction(
     }
   };
 
-  let result = try_to_spend_wallet_balance(
-    &wallet_token, 
-    amount_to_deduct, 
-    maybe_reference_token, 
+  let summary = try_to_spend_wallet_balance(
+    &wallet_token,
+    amount_to_deduct,
+    maybe_reference_token,
     transaction
-  ).await;
-  
-  if let Err(err) = result {
-    error!("Failed to deduct {} credits from wallet {} for user {} : {:?}",
-      amount_to_deduct,
-      wallet_token.as_str(),
-      owner_user_token.as_str(),
-      err);
-    
-    return Err(err);
-  }
+  ).await
+    .map_err(|err| {
+      error!("Failed to deduct {} credits from wallet {} for user {} : {:?}",
+        amount_to_deduct,
+        wallet_token.as_str(),
+        owner_user_token.as_str(),
+        err);
+      err
+    })?;
 
-  Ok(())
+  Ok(WalletDeductionResult {
+    wallet_token: summary.wallet_token,
+    ledger_entry_token: summary.wallet_ledger_entry_token,
+  })
 }

crates/service/web/storyteller_web/src/http_server/endpoints/generate/video/multi_function/seedance_2p0_multi_function_video_gen_handler.rs

@@ -29,6 +29,7 @@ use mysql_queries::queries::prompt_context_items::insert_batch_prompt_context_it
   insert_batch_prompt_context_items, InsertBatchArgs, PromptContextItem,
 };
 use mysql_queries::queries::prompts::insert_prompt::{insert_prompt, InsertPromptArgs};
+use mysql_queries::queries::wallets::refund::try_to_refund_ledger_entry::{try_to_refund_ledger_entry, WalletRefundOutcome};
 use seedance2pro::creds::seedance2pro_session::Seedance2ProSession;
 use seedance2pro::requests::generate_video::generate_video::{
   generate_video, BatchCount, GenerateVideoArgs, Resolution,
@@ -203,7 +204,7 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
 
   info!("Charging wallet: {} cents ({} credits)", cost_in_cents, cost_in_cents);
 
-  attempt_wallet_deduction_else_common_web_error(
+  let deduction_result = attempt_wallet_deduction_else_common_web_error(
     user_token,
     Some(apriori_job_token.as_str()),
     cost_in_cents,
@@ -212,12 +213,20 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
 
   // --- Call seedance2pro API ---
 
-  let gen_response = generate_video(video_gen_args)
-      .await
-      .map_err(|err| {
-        warn!("Error calling seedance2pro generate_video: {:?}", err);
-        CommonWebError::ServerError
-      })?;
+  // TODO: The refund logic here could use work. We shouldn't hold MySQL transactions open during the HTTP
+  //  transaction (as that could last forever), but this is slightly risky in that a tiny percentage of errors
+  //  may leak. We need to engineer two-phase transactions (a lot of work) to make every step atomic.
+
+  let gen_response = match generate_video(video_gen_args).await {
+    Ok(resp) => resp,
+    Err(err) => {
+      warn!("Error calling seedance2pro generate_video: {:?}", err);
+      refund_after_api_failure(&deduction_result.ledger_entry_token, &mut mysql_connection).await;
+
+      // TODO: We need to start returning robust failure messages to the caller.
+      return Err(CommonWebError::ServerError);
+    }
+  };
 
   info!(
     "Seedance2pro task_id={}, order_id={}",
@@ -315,6 +324,7 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
         maybe_external_third_party_id: order_id,
         maybe_inference_args: None,
         maybe_prompt_token: prompt_token.as_ref(),
+        maybe_wallet_ledger_entry_token: Some(&deduction_result.ledger_entry_token),
         maybe_creator_user_token: maybe_user_session.as_ref().map(|s| &s.user_token),
         maybe_avt_token: maybe_avt_token.as_ref(),
         creator_ip_address: &ip_address,
@@ -357,6 +367,53 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
   }))
 }
 
+async fn refund_after_api_failure(
+  ledger_entry_token: &tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken,
+  connection: &mut sqlx::pool::PoolConnection<sqlx::MySql>,
+) {
+  let mut transaction = match connection.begin().await {
+    Ok(tx) => tx,
+    Err(err) => {
+      error!(
+        "Failed to begin refund transaction after API failure (ledger {}): {:?}",
+        ledger_entry_token.as_str(), err
+      );
+      return;
+    }
+  };
+
+  match try_to_refund_ledger_entry(ledger_entry_token, &mut transaction).await {
+    Ok(WalletRefundOutcome::Refunded(summary)) => {
+      info!(
+        "Refunded {} credits after API failure (ledger {} → refund ledger {}).",
+        summary.refund_amount,
+        ledger_entry_token.as_str(),
+        summary.refund_ledger_entry_token.as_str(),
+      );
+      if let Err(err) = transaction.commit().await {
+        error!(
+          "Failed to commit refund after API failure (ledger {}): {:?}",
+          ledger_entry_token.as_str(), err
+        );
+      }
+    }
+    Ok(WalletRefundOutcome::AlreadyRefunded) => {
+      info!(
+        "Ledger entry {} was already refunded; no action needed.",
+        ledger_entry_token.as_str()
+      );
+      let _ = transaction.rollback().await;
+    }
+    Err(err) => {
+      error!(
+        "Failed to refund ledger entry {} after API failure: {:?}",
+        ledger_entry_token.as_str(), err
+      );
+      let _ = transaction.rollback().await;
+    }
+  }
+}
+
 async fn upload_to_seedance2pro(
   session: &Seedance2ProSession,
   our_cdn_url: &str,