backfill tool

Cargo.lock

@@ -3039,6 +3039,11 @@ name = "db-backfill"
 version = "0.0.1"
 dependencies = [
  "anyhow",
+ "async-stripe 1.0.0-alpha.2",
+ "async-stripe-billing",
+ "async-stripe-core",
+ "async-stripe-shared",
+ "async-stripe-types",
  "chrono",
  "clap 4.5.60",
  "datetimes",

crates/cli/production/db-backfill/Cargo.toml

@@ -37,7 +37,14 @@ log = "0.4.14"
 # Bigger pieces
 futures = { version = "0.3.17", features = ["thread-pool"] }
 sqlx = { version = "0.7.4", features = [ "mysql", "runtime-tokio-rustls", "chrono" ] }
-tokio = { version = "1.38.0", features = ["macros"] }
+tokio = { version = "1.38.0", features = ["macros", "rt-multi-thread"] }
+
+# Stripe (authoritative payment enumeration for the user_spend_events backfill)
+async-stripe = { version = "1.0.0-alpha.2", default-features = false, features = ["rustls-tls-native"] }
+async-stripe-core = { version = "1.0.0-alpha.2", features = ["payment_intent"] }
+async-stripe-billing = { version = "1.0.0-alpha.2", features = ["invoice"] }
+async-stripe-shared = { version = "1.0.0-alpha.2", features = ["deserialize"] }
+async-stripe-types = { version = "1.0.0-alpha.2" }
 
 # Serialization
 serde = "1.0.125"

crates/cli/production/db-backfill/src/args.rs

@@ -20,6 +20,7 @@ pub struct Args {
 // #[derive(Debug, Deserialize, Copy, Clone)]
 #[derive(ValueEnum, Clone, Copy, Debug)]
 pub enum Command {
+  BackfillUserSpendEvents,
   CalculateLegacyTtsResultsUsages,
   CalculateModelWeightsUsages,
 }

crates/cli/production/db-backfill/src/main.rs

@@ -10,6 +10,8 @@ use easyenv::init_all_with_default_logging;
 use errors::AnyhowResult;
 
 use crate::args::{Command, parse_cli_args};
+use crate::operations::backfill_user_spend_events::backfill_user_spend_events::backfill_user_spend_events;
+use crate::operations::backfill_user_spend_events::sub_args::parse_backfill_user_spend_events_args;
 use crate::operations::calculate_legacy_tts_results_usages::calculate_legacy_tts_results_usages::calculate_legacy_tts_result_usages;
 use crate::operations::calculate_model_weights_usages::run_migration::run_migration;
 
@@ -41,13 +43,26 @@ async fn main() -> AnyhowResult<()> {
   // (Hopefully this isn't getting out of hand at this point.)
   easyenv::from_filename(".env-db-backfill-secrets")?;
 
-  let mysql = get_mysql("MYSQL_PRODUCTION_URL").await?;
-
   info!("dispatching command: {:?}", command);
 
   match command.sub_command {
-    Command::CalculateModelWeightsUsages => run_migration(mysql).await?,
-    Command::CalculateLegacyTtsResultsUsages => calculate_legacy_tts_result_usages(mysql).await?,
+    Command::CalculateModelWeightsUsages => {
+      let mysql = get_mysql("MYSQL_PRODUCTION_URL").await?;
+      run_migration(mysql).await?
+    }
+    Command::CalculateLegacyTtsResultsUsages => {
+      let mysql = get_mysql("MYSQL_PRODUCTION_URL").await?;
+      calculate_legacy_tts_result_usages(mysql).await?
+    }
+    Command::BackfillUserSpendEvents => {
+      // Read pool = the replica (source); write pool = your target DB (local in
+      // a dry run, prod when you go live). Stripe key = the ArtCraft account.
+      let sub_args = parse_backfill_user_spend_events_args();
+      let read_pool = get_mysql("MYSQL_READ_URL").await?;
+      let write_pool = get_mysql("MYSQL_WRITE_URL").await?;
+      let stripe_client = stripe::Client::new(easyenv::get_env_string_required("STRIPE_ARTCRAFT_SECRET_KEY")?);
+      backfill_user_spend_events(&read_pool, &write_pool, &stripe_client, sub_args).await?;
+    }
   }
 
   Ok(())

crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/README.md

@@ -0,0 +1,62 @@
+# backfill-user-spend-events
+
+Backfills the `user_spend_events` table from historical Stripe payments
+(subscription creations, subscription renewals, and credit-pack purchases),
+idempotently and safely.
+
+## How it works
+
+1. **Stripe is the authoritative payment list** (so we don't miss any payment,
+   even ones the live webhook path failed to fulfill):
+   - Paid **invoices** with `billing_reason` ∈ {`subscription_create`,
+     `subscription_cycle`} → `subscription_initial` / `subscription_renewal`.
+   - Succeeded one-off **PaymentIntents** (carry our `user_token` metadata, or
+     match a ledger `credit_banked` entry) → `credit_pack_purchase`.
+   Using the ArtCraft Stripe key means this is ArtCraft-only (FakeYou is a
+   separate Stripe account), so there is no FakeYou contamination.
+2. **The ArtCraft ledger (read pool) enriches** each payment with the granted
+   credits (`maybe_credits_granted`) and the ledger-entry FK
+   (`maybe_wallet_ledger_entry_token`), looked up by the Stripe object id.
+3. **User attribution**, in order: (a) Stripe object metadata `user_token`,
+   (b) ledger entry by Stripe id, (c) `user_stripe_customer_links` by customer.
+   Payments that resolve to no user are **written to a CSV and NOT inserted**.
+4. **Upsert** via `insert_user_spend_event`, keyed on
+   `(payment_source, source_object_id)` — the same dedup key the webhook write
+   path uses — so rows already written live are no-ops (no duplicates).
+
+Read and write use **separate connections/credentials**, so you can dry-run
+against the read replica → your local DB before running it live.
+
+## Required env (loaded from `.env-db-backfill-secrets`)
+
+```
+MYSQL_READ_URL=mysql://USER:PASS@READ_REPLICA_HOST:PORT/storyteller
+MYSQL_WRITE_URL=mysql://USER:PASS@TARGET_HOST:PORT/storyteller   # local for dry runs; prod when live
+STRIPE_ARTCRAFT_SECRET_KEY=sk_live_or_test_for_the_ArtCraft_account
+# optional: MYSQL_MAX_CONNECTIONS (default 20)
+```
+
+## Usage
+
+```bash
+# Safe dry run: enumerate + resolve + write the unattributed CSV, but DON'T write events.
+cargo run -p db-backfill -- backfill-user-spend-events --since 2026-02-01 --dry-run --limit 50
+
+# Real run into MYSQL_WRITE_URL (point it at LOCAL first to validate):
+cargo run -p db-backfill -- backfill-user-spend-events --since 2026-02-01
+
+# Go live by pointing MYSQL_WRITE_URL at production and re-running (idempotent).
+```
+
+Flags: `--since YYYY-MM-DD` (default `2026-02-01`), `--dry-run`,
+`--unattributed-report PATH` (default `unattributed_payments.csv`),
+`--limit N` (cap per type, for testing).
+
+## Notes
+
+- Prorations (`subscription_update` invoices) are intentionally skipped, matching
+  the webhook write-path scope. The `PaymentEventType` enum has
+  `subscription_proration_upgrade/downgrade` if you later want them.
+- `maybe_stripe_event_id` is `NULL` for backfilled rows (no originating webhook).
+- Because it's idempotent, it's safe to re-run after fixing user links to pick up
+  previously-unattributed payments.

crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/backfill_user_spend_events.rs

@@ -0,0 +1,405 @@
+use std::collections::HashMap;
+use std::convert::TryFrom;
+use std::fs::File;
+use std::io::Write;
+use std::marker::PhantomData;
+
+use chrono::{DateTime, NaiveDate, TimeZone, Utc};
+use log::{info, warn};
+use sqlx::{MySql, Pool};
+
+use enums::by_table::user_spend_events::payment_event_type::PaymentEventType;
+use enums::by_table::user_spend_events::payment_source::PaymentSource;
+use enums::common::payments_namespace::PaymentsNamespace;
+use errors::AnyhowResult;
+use mysql_queries::queries::user_spend_events::backfill_list_artcraft_customer_links::{
+  backfill_list_artcraft_customer_links, BackfillListCustomerLinksArgs,
+};
+use mysql_queries::queries::user_spend_events::backfill_list_artcraft_ledger_payments::{
+  backfill_list_artcraft_ledger_payments, BackfillListLedgerPaymentsArgs, LedgerPaymentRef,
+};
+use mysql_queries::queries::user_spend_events::insert_user_spend_event::{
+  insert_user_spend_event, InsertUserSpendEventArgs,
+};
+use tokens::tokens::users::UserToken;
+
+use stripe::Client;
+use stripe_billing::invoice::ListInvoice;
+use stripe_core::payment_intent::ListPaymentIntent;
+use stripe_shared::{Invoice, InvoiceBillingReason, InvoiceStatus, PaymentIntent, PaymentIntentStatus};
+use stripe_types::{List, RangeQueryTs};
+
+use crate::operations::backfill_user_spend_events::sub_args::BackfillUserSpendEventsArgs;
+
+const PAGE_SIZE: i64 = 100;
+
+/// Backfill `user_spend_events` from historical Stripe payments.
+///
+/// Stripe is the authoritative payment list (so we don't miss charged-but-
+/// unfulfilled payments); the ArtCraft ledger (read pool) only ENRICHES each
+/// payment with the granted credits + ledger-entry FK. Every event is upserted
+/// (write pool) on the same `(payment_source, source_object_id)` dedup key the
+/// webhook path uses, so rows already written live are no-ops. Payments that
+/// can't be linked to a user are written to a CSV instead of inserted.
+pub async fn backfill_user_spend_events(
+  read_pool: &Pool<MySql>,
+  write_pool: &Pool<MySql>,
+  stripe_client: &Client,
+  args: BackfillUserSpendEventsArgs,
+) -> AnyhowResult<()> {
+  let since = parse_since(&args.since)?;
+  info!("Backfill user_spend_events since {since} (UTC). dry_run={}", args.dry_run);
+
+  // ---- 1. Enrichment maps from the READ pool (replica) ----
+  let ledger_rows = backfill_list_artcraft_ledger_payments(BackfillListLedgerPaymentsArgs {
+    since,
+    mysql_executor: read_pool,
+    phantom: PhantomData,
+  })
+  .await?;
+  info!("Loaded {} ArtCraft ledger payment refs (enrichment)", ledger_rows.len());
+  let ledger_by_id: HashMap<String, LedgerPaymentRef> =
+    ledger_rows.into_iter().map(|r| (r.stripe_object_id.clone(), r)).collect();
+
+  let links = backfill_list_artcraft_customer_links(BackfillListCustomerLinksArgs {
+    mysql_executor: read_pool,
+    phantom: PhantomData,
+  })
+  .await?;
+  info!("Loaded {} ArtCraft customer links (fallback attribution)", links.len());
+  let user_by_customer: HashMap<String, UserToken> =
+    links.into_iter().map(|l| (l.stripe_customer_id, l.user_token)).collect();
+
+  let mut stats = Stats::default();
+  let mut unattributed: Vec<UnattributedRow> = Vec::new();
+
+  // ---- 2. Credit-pack purchases: succeeded one-off PaymentIntents ----
+  backfill_payment_intents(
+    stripe_client, write_pool, since, &args, &ledger_by_id, &user_by_customer, &mut stats, &mut unattributed,
+  )
+  .await?;
+
+  // ---- 3. Subscription payments: paid invoices (create/cycle) ----
+  backfill_invoices(
+    stripe_client, write_pool, since, &args, &ledger_by_id, &user_by_customer, &mut stats, &mut unattributed,
+  )
+  .await?;
+
+  // ---- 4. Report unattributable payments ----
+  write_unattributed_report(&args.unattributed_report, &unattributed)?;
+
+  info!("==== BACKFILL SUMMARY ====");
+  info!("  credit-pack purchases upserted: {}", stats.credit_packs);
+  info!("  subscription payments upserted: {}", stats.subscriptions);
+  info!("  skipped (not a tracked type):   {}", stats.skipped);
+  info!("  UNATTRIBUTED (no user account): {} -> {}", unattributed.len(), args.unattributed_report);
+  if args.dry_run {
+    info!("  (dry run — NOTHING was written to the database)");
+  }
+  Ok(())
+}
+
+// ---- Enumerators ----
+
+async fn backfill_payment_intents(
+  stripe_client: &Client,
+  write_pool: &Pool<MySql>,
+  since: DateTime<Utc>,
+  args: &BackfillUserSpendEventsArgs,
+  ledger_by_id: &HashMap<String, LedgerPaymentRef>,
+  user_by_customer: &HashMap<String, UserToken>,
+  stats: &mut Stats,
+  unattributed: &mut Vec<UnattributedRow>,
+) -> AnyhowResult<()> {
+  let mut starting_after: Option<String> = None;
+  let mut processed = 0usize;
+
+  loop {
+    let mut req = ListPaymentIntent::new()
+      .created(RangeQueryTs::gte(since.timestamp()))
+      .limit(PAGE_SIZE);
+    if let Some(cursor) = &starting_after {
+      req = req.starting_after(cursor.clone());
+    }
+    let page: List<PaymentIntent> = req
+      .send(stripe_client)
+      .await
+      .map_err(|err| anyhow::anyhow!("Stripe ListPaymentIntent error: {err:?}"))?;
+
+    if page.data.is_empty() {
+      break;
+    }
+    let last_id = page.data.last().map(|pi| pi.id.to_string());
+
+    for pi in &page.data {
+      if let Some(limit) = args.limit {
+        if processed >= limit {
+          return Ok(());
+        }
+      }
+      if !matches!(pi.status, PaymentIntentStatus::Succeeded) {
+        continue;
+      }
+      let pi_id = pi.id.to_string();
+      // A one-off credit-pack PI carries our `user_token` metadata; subscription
+      // PIs have empty metadata. The ledger `credit_banked` match is a backstop
+      // for any historical pack PI that lacks metadata.
+      let is_credit_pack = pi.metadata.contains_key("user_token")
+        || ledger_by_id.get(&pi_id).map(|l| l.entry_type == "credit_banked").unwrap_or(false);
+      if !is_credit_pack {
+        stats.skipped += 1;
+        continue;
+      }
+      processed += 1;
+
+      let customer_id = pi.customer.as_ref().map(|c| c.id().to_string());
+      let charge_id = pi.latest_charge.as_ref().map(|c| c.id().to_string());
+      let occurred_at = DateTime::from_timestamp(pi.created, 0).unwrap_or_else(Utc::now);
+      let maybe_user = resolve_user(Some(&pi.metadata), &pi_id, customer_id.as_deref(), ledger_by_id, user_by_customer);
+
+      match maybe_user {
+        Some(user) => {
+          upsert_event(write_pool, args.dry_run, UpsertInput {
+            event_type: PaymentEventType::CreditPackPurchase,
+            amount_usd_cents: pi.amount_received,
+            source_object_id: &pi_id,
+            invoice_id: None,
+            payment_intent_id: Some(&pi_id),
+            charge_id: charge_id.as_deref(),
+            customer_id: customer_id.as_deref(),
+            user: &user,
+            ledger: ledger_by_id.get(&pi_id),
+            is_production: pi.livemode,
+            occurred_at,
+          })
+          .await?;
+          stats.credit_packs += 1;
+        }
+        None => unattributed.push(UnattributedRow {
+          kind: "credit_pack",
+          stripe_id: pi_id.clone(),
+          amount_usd_cents: pi.amount_received,
+          customer_id: customer_id.clone(),
+          occurred_at,
+        }),
+      }
+    }
+
+    if !page.has_more {
+      break;
+    }
+    starting_after = last_id;
+  }
+  Ok(())
+}
+
+async fn backfill_invoices(
+  stripe_client: &Client,
+  write_pool: &Pool<MySql>,
+  since: DateTime<Utc>,
+  args: &BackfillUserSpendEventsArgs,
+  ledger_by_id: &HashMap<String, LedgerPaymentRef>,
+  user_by_customer: &HashMap<String, UserToken>,
+  stats: &mut Stats,
+  unattributed: &mut Vec<UnattributedRow>,
+) -> AnyhowResult<()> {
+  let mut starting_after: Option<String> = None;
+  let mut processed = 0usize;
+
+  loop {
+    let mut req = ListInvoice::new()
+      .created(RangeQueryTs::gte(since.timestamp()))
+      .status(InvoiceStatus::Paid)
+      .limit(PAGE_SIZE);
+    if let Some(cursor) = &starting_after {
+      req = req.starting_after(cursor.clone());
+    }
+    let page: List<Invoice> = req
+      .send(stripe_client)
+      .await
+      .map_err(|err| anyhow::anyhow!("Stripe ListInvoice error: {err:?}"))?;
+
+    if page.data.is_empty() {
+      break;
+    }
+    let last_id = page.data.last().and_then(|inv| inv.id.as_ref().map(|id| id.to_string()));
+
+    for inv in &page.data {
+      if let Some(limit) = args.limit {
+        if processed >= limit {
+          return Ok(());
+        }
+      }
+      // Only subscription creation + renewal are in scope (prorations / manual
+      // invoices are skipped, matching the webhook write-path).
+      let event_type = match inv.billing_reason {
+        Some(InvoiceBillingReason::SubscriptionCreate) => PaymentEventType::SubscriptionInitial,
+        Some(InvoiceBillingReason::SubscriptionCycle) => PaymentEventType::SubscriptionRenewal,
+        _ => {
+          stats.skipped += 1;
+          continue;
+        }
+      };
+      let inv_id = match inv.id.as_ref() {
+        Some(id) => id.to_string(),
+        None => {
+          stats.skipped += 1;
+          continue;
+        }
+      };
+      processed += 1;
+
+      let customer_id = inv.customer.as_ref().map(|c| c.id().to_string());
+      let occurred_at = DateTime::from_timestamp(inv.created, 0).unwrap_or_else(Utc::now);
+      let maybe_user = resolve_user(inv.metadata.as_ref(), &inv_id, customer_id.as_deref(), ledger_by_id, user_by_customer);
+
+      match maybe_user {
+        Some(user) => {
+          upsert_event(write_pool, args.dry_run, UpsertInput {
+            event_type,
+            amount_usd_cents: inv.amount_paid,
+            source_object_id: &inv_id,
+            invoice_id: Some(&inv_id),
+            payment_intent_id: None,
+            charge_id: None,
+            customer_id: customer_id.as_deref(),
+            user: &user,
+            ledger: ledger_by_id.get(&inv_id),
+            is_production: inv.livemode,
+            occurred_at,
+          })
+          .await?;
+          stats.subscriptions += 1;
+        }
+        None => unattributed.push(UnattributedRow {
+          kind: "subscription",
+          stripe_id: inv_id.clone(),
+          amount_usd_cents: inv.amount_paid,
+          customer_id: customer_id.clone(),
+          occurred_at,
+        }),
+      }
+    }
+
+    if !page.has_more {
+      break;
+    }
+    starting_after = last_id;
+  }
+  Ok(())
+}
+
+// ---- Helpers ----
+
+/// Attribute a payment to a user: (1) Stripe object metadata `user_token`,
+/// (2) ledger entry by Stripe id, (3) `user_stripe_customer_links` by customer.
+fn resolve_user(
+  metadata: Option<&HashMap<String, String>>,
+  stripe_object_id: &str,
+  customer_id: Option<&str>,
+  ledger_by_id: &HashMap<String, LedgerPaymentRef>,
+  user_by_customer: &HashMap<String, UserToken>,
+) -> Option<UserToken> {
+  if let Some(md) = metadata {
+    if let Some(user_token) = md.get("user_token") {
+      if !user_token.is_empty() {
+        return Some(UserToken(user_token.clone()));
+      }
+    }
+  }
+  if let Some(ledger) = ledger_by_id.get(stripe_object_id) {
+    return Some(ledger.owner_user_token.clone());
+  }
+  if let Some(customer_id) = customer_id {
+    if let Some(user_token) = user_by_customer.get(customer_id) {
+      return Some(user_token.clone());
+    }
+  }
+  None
+}
+
+struct UpsertInput<'a> {
+  event_type: PaymentEventType,
+  amount_usd_cents: i64,
+  source_object_id: &'a str,
+  invoice_id: Option<&'a str>,
+  payment_intent_id: Option<&'a str>,
+  charge_id: Option<&'a str>,
+  customer_id: Option<&'a str>,
+  user: &'a UserToken,
+  ledger: Option<&'a LedgerPaymentRef>,
+  is_production: bool,
+  occurred_at: DateTime<Utc>,
+}
+
+async fn upsert_event(write_pool: &Pool<MySql>, dry_run: bool, input: UpsertInput<'_>) -> AnyhowResult<()> {
+  if dry_run {
+    return Ok(());
+  }
+  insert_user_spend_event(InsertUserSpendEventArgs {
+    payments_namespace: PaymentsNamespace::Artcraft,
+    maybe_user_token: Some(input.user),
+    event_type: input.event_type,
+    amount_usd_cents: input.amount_usd_cents,
+    maybe_credits_granted: input.ledger.and_then(|l| u32::try_from(l.credits_delta).ok()),
+    maybe_user_subscription_token: None,
+    maybe_wallet_ledger_entry_token: input.ledger.map(|l| &l.ledger_token),
+    payment_source: PaymentSource::Stripe,
+    maybe_source_object_id: Some(input.source_object_id),
+    maybe_stripe_customer_id: input.customer_id,
+    maybe_stripe_invoice_id: input.invoice_id,
+    maybe_stripe_payment_intent_id: input.payment_intent_id,
+    maybe_stripe_charge_id: input.charge_id,
+    maybe_stripe_event_id: None, // backfill has no originating webhook event
+    is_production: input.is_production,
+    payment_occurred_at: input.occurred_at,
+    mysql_executor: write_pool,
+    phantom: PhantomData,
+  })
+  .await?;
+  Ok(())
+}
+
+fn write_unattributed_report(path: &str, rows: &[UnattributedRow]) -> AnyhowResult<()> {
+  let mut file = File::create(path)?;
+  writeln!(file, "kind,stripe_object_id,amount_usd_cents,stripe_customer_id,occurred_at,reason")?;
+  for row in rows {
+    writeln!(
+      file,
+      "{},{},{},{},{},no user_token in metadata/ledger/customer_link",
+      row.kind,
+      row.stripe_id,
+      row.amount_usd_cents,
+      row.customer_id.as_deref().unwrap_or(""),
+      row.occurred_at.to_rfc3339(),
+    )?;
+  }
+  if !rows.is_empty() {
+    warn!("{} payment(s) could not be linked to a user — see {}", rows.len(), path);
+  }
+  Ok(())
+}
+
+fn parse_since(s: &str) -> AnyhowResult<DateTime<Utc>> {
+  let date = NaiveDate::parse_from_str(s, "%Y-%m-%d")
+    .map_err(|err| anyhow::anyhow!("invalid --since {s:?} (want YYYY-MM-DD): {err}"))?;
+  let naive = date
+    .and_hms_opt(0, 0, 0)
+    .ok_or_else(|| anyhow::anyhow!("invalid time for --since {s:?}"))?;
+  Ok(Utc.from_utc_datetime(&naive))
+}
+
+#[derive(Default)]
+struct Stats {
+  credit_packs: usize,
+  subscriptions: usize,
+  skipped: usize,
+}
+
+struct UnattributedRow {
+  kind: &'static str,
+  stripe_id: String,
+  amount_usd_cents: i64,
+  customer_id: Option<String>,
+  occurred_at: DateTime<Utc>,
+}

crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/mod.rs

@@ -0,0 +1,2 @@
+pub mod backfill_user_spend_events;
+pub mod sub_args;

crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/sub_args.rs

@@ -0,0 +1,35 @@
+use clap::Parser;
+
+/// Flags for the `backfill-user-spend-events` sub-command.
+///
+/// Parsed from the args AFTER the sub-command token, e.g.:
+///   db-backfill backfill-user-spend-events --since 2026-02-01 --dry-run
+#[derive(Parser, Debug, Clone)]
+#[command(name = "backfill-user-spend-events")]
+pub struct BackfillUserSpendEventsArgs {
+  /// Only backfill payments whose Stripe object was created on/after this UTC
+  /// date (YYYY-MM-DD). Payments effectively start in February.
+  #[arg(long, default_value = "2026-02-01")]
+  pub since: String,
+
+  /// Plan only: enumerate + resolve + report, but do NOT upsert anything.
+  #[arg(long, default_value_t = false)]
+  pub dry_run: bool,
+
+  /// CSV path where payments that could not be linked to a user account are
+  /// written (one row each) for manual follow-up.
+  #[arg(long, default_value = "unattributed_payments.csv")]
+  pub unattributed_report: String,
+
+  /// Optional cap on the number of Stripe objects processed per type. For
+  /// testing against the read replica without walking the whole account.
+  #[arg(long)]
+  pub limit: Option<usize>,
+}
+
+/// Parse the sub-command flags. The first element of `args()` is the binary and
+/// the second is the sub-command token; skipping one leaves the sub-command
+/// token as clap's "binary name" (ignored) followed by the flags.
+pub fn parse_backfill_user_spend_events_args() -> BackfillUserSpendEventsArgs {
+  BackfillUserSpendEventsArgs::parse_from(std::env::args().skip(1))
+}

crates/cli/production/db-backfill/src/operations/mod.rs

@@ -1,2 +1,3 @@
+pub mod backfill_user_spend_events;
 pub mod calculate_legacy_tts_results_usages;
 pub mod calculate_model_weights_usages;

crates/schema/database/mysql_queries/src/queries/user_spend_events/backfill_list_artcraft_customer_links.rs

@@ -0,0 +1,52 @@
+use std::marker::PhantomData;
+
+use sqlx::{Executor, MySql, Row};
+
+use tokens::tokens::users::UserToken;
+
+/// A (stripe_customer_id -> user_token) link in the ArtCraft payments namespace,
+/// used as the final fallback to attribute a Stripe-enumerated payment to a user
+/// when neither the object metadata nor the ledger yields a user token.
+pub struct ArtcraftCustomerLink {
+  pub stripe_customer_id: String,
+  pub user_token: UserToken,
+}
+
+pub struct BackfillListCustomerLinksArgs<'c, E>
+where
+  E: Executor<'c, Database = MySql>,
+{
+  pub mysql_executor: E,
+  pub phantom: PhantomData<&'c E>,
+}
+
+/// Load all ArtCraft `user_stripe_customer_links` for the backfill's in-memory
+/// customer→user fallback map.
+///
+/// NB: runtime `sqlx::query()` (read replica; keeps the offline cache out of it).
+pub async fn backfill_list_artcraft_customer_links<'c, E>(
+  args: BackfillListCustomerLinksArgs<'c, E>,
+) -> Result<Vec<ArtcraftCustomerLink>, sqlx::Error>
+where
+  E: Executor<'c, Database = MySql>,
+{
+  let rows = sqlx::query(
+    r#"
+SELECT stripe_customer_id, user_token
+FROM user_stripe_customer_links
+WHERE payments_namespace = 'artcraft'
+    "#,
+  )
+    .fetch_all(args.mysql_executor)
+    .await?;
+
+  let mut out = Vec::with_capacity(rows.len());
+  for row in rows {
+    out.push(ArtcraftCustomerLink {
+      stripe_customer_id: row.try_get("stripe_customer_id")?,
+      user_token: UserToken(row.try_get("user_token")?),
+    });
+  }
+
+  Ok(out)
+}

crates/schema/database/mysql_queries/src/queries/user_spend_events/backfill_list_artcraft_ledger_payments.rs

@@ -0,0 +1,91 @@
+use std::marker::PhantomData;
+
+use chrono::{DateTime, Utc};
+use sqlx::{Executor, MySql, Row};
+
+use tokens::tokens::users::UserToken;
+use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
+
+/// One ArtCraft credit-from-payment ledger entry, used to ENRICH a Stripe-
+/// enumerated payment during backfill (it is NOT the authoritative payment
+/// list — Stripe is). Keyed by `stripe_object_id` (`pi_…` for credit packs,
+/// `in_…` for subscription invoices), which equals the spend-event dedup
+/// `source_object_id`.
+pub struct LedgerPaymentRef {
+  /// `wallet_ledger_entries.maybe_entity_ref` — the Stripe payment_intent /
+  /// invoice id. Always present (the query filters to `pi_`/`in_`).
+  pub stripe_object_id: String,
+
+  /// `credit_banked` (credit pack) or `credit_monthly` (subscription).
+  pub entry_type: String,
+
+  /// The ledger entry — becomes `user_spend_events.maybe_wallet_ledger_entry_token`.
+  pub ledger_token: WalletLedgerEntryToken,
+
+  /// Wallet owner — becomes `user_spend_events.maybe_user_token`.
+  pub owner_user_token: UserToken,
+
+  /// Credits actually granted — becomes `user_spend_events.maybe_credits_granted`.
+  pub credits_delta: i64,
+
+  /// When we credited the wallet (≈ settlement; spend events still bin on the
+  /// Stripe object's own timestamp).
+  pub created_at: DateTime<Utc>,
+}
+
+pub struct BackfillListLedgerPaymentsArgs<'c, E>
+where
+  E: Executor<'c, Database = MySql>,
+{
+  /// Only entries at/after this instant (the payments start date, ~Feb).
+  pub since: DateTime<Utc>,
+  pub mysql_executor: E,
+  pub phantom: PhantomData<&'c E>,
+}
+
+/// Load every ArtCraft payment-backed credit ledger entry since `since`, for the
+/// backfill's in-memory enrichment map. ArtCraft-only via `wallet_namespace`
+/// (the ledger has no FakeYou rows, but we filter defensively). Refund/staff/
+/// deduct/create entries and non-Stripe refs are excluded.
+///
+/// NB: runtime `sqlx::query()` (read replica; keeps the offline cache out of it).
+pub async fn backfill_list_artcraft_ledger_payments<'c, E>(
+  args: BackfillListLedgerPaymentsArgs<'c, E>,
+) -> Result<Vec<LedgerPaymentRef>, sqlx::Error>
+where
+  E: Executor<'c, Database = MySql>,
+{
+  let rows = sqlx::query(
+    r#"
+SELECT wle.maybe_entity_ref AS stripe_object_id,
+       wle.entry_type        AS entry_type,
+       wle.token             AS ledger_token,
+       w.owner_user_token    AS owner_user_token,
+       wle.credits_delta     AS credits_delta,
+       wle.created_at         AS created_at
+FROM wallet_ledger_entries wle
+JOIN wallets w ON w.token = wle.wallet_token
+WHERE w.wallet_namespace = 'artcraft'
+  AND wle.entry_type IN ('credit_banked', 'credit_monthly')
+  AND wle.created_at >= ?
+  AND (wle.maybe_entity_ref LIKE 'pi\_%' OR wle.maybe_entity_ref LIKE 'in\_%')
+    "#,
+  )
+    .bind(args.since)
+    .fetch_all(args.mysql_executor)
+    .await?;
+
+  let mut out = Vec::with_capacity(rows.len());
+  for row in rows {
+    out.push(LedgerPaymentRef {
+      stripe_object_id: row.try_get("stripe_object_id")?,
+      entry_type: row.try_get("entry_type")?,
+      ledger_token: WalletLedgerEntryToken(row.try_get("ledger_token")?),
+      owner_user_token: UserToken(row.try_get("owner_user_token")?),
+      credits_delta: row.try_get::<i32, _>("credits_delta")? as i64,
+      created_at: row.try_get("created_at")?,
+    });
+  }
+
+  Ok(out)
+}

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

@@ -1 +1,3 @@
+pub mod backfill_list_artcraft_customer_links;
+pub mod backfill_list_artcraft_ledger_payments;
 pub mod insert_user_spend_event;