diff --git a/Cargo.lock b/Cargo.lock index 9c5b7fa284..a8593f469a 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/cli/production/db-backfill/Cargo.toml b/crates/cli/production/db-backfill/Cargo.toml index 4d4329f249..0e6eaa890e 100644 --- a/crates/cli/production/db-backfill/Cargo.toml +++ b/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" diff --git a/crates/cli/production/db-backfill/src/args.rs b/crates/cli/production/db-backfill/src/args.rs index 2830d6ccb7..c5d974e015 100644 --- a/crates/cli/production/db-backfill/src/args.rs +++ b/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, } diff --git a/crates/cli/production/db-backfill/src/main.rs b/crates/cli/production/db-backfill/src/main.rs index 1557de5352..e1d515e7a3 100644 --- a/crates/cli/production/db-backfill/src/main.rs +++ b/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(()) diff --git a/crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/README.md b/crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/README.md new file mode 100644 index 0000000000..5c885be612 --- /dev/null +++ b/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. diff --git a/crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/backfill_user_spend_events.rs b/crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/backfill_user_spend_events.rs new file mode 100644 index 0000000000..b54173534a --- /dev/null +++ b/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, + write_pool: &Pool, + 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 = + 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 = + links.into_iter().map(|l| (l.stripe_customer_id, l.user_token)).collect(); + + let mut stats = Stats::default(); + let mut unattributed: Vec = 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, + since: DateTime, + args: &BackfillUserSpendEventsArgs, + ledger_by_id: &HashMap, + user_by_customer: &HashMap, + stats: &mut Stats, + unattributed: &mut Vec, +) -> AnyhowResult<()> { + let mut starting_after: Option = 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 = 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, + since: DateTime, + args: &BackfillUserSpendEventsArgs, + ledger_by_id: &HashMap, + user_by_customer: &HashMap, + stats: &mut Stats, + unattributed: &mut Vec, +) -> AnyhowResult<()> { + let mut starting_after: Option = 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 = 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>, + stripe_object_id: &str, + customer_id: Option<&str>, + ledger_by_id: &HashMap, + user_by_customer: &HashMap, +) -> Option { + 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, +} + +async fn upsert_event(write_pool: &Pool, 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> { + 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, + occurred_at: DateTime, +} diff --git a/crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/mod.rs b/crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/mod.rs new file mode 100644 index 0000000000..677efd4e19 --- /dev/null +++ b/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; diff --git a/crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/sub_args.rs b/crates/cli/production/db-backfill/src/operations/backfill_user_spend_events/sub_args.rs new file mode 100644 index 0000000000..07ff885e19 --- /dev/null +++ b/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, +} + +/// 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)) +} diff --git a/crates/cli/production/db-backfill/src/operations/mod.rs b/crates/cli/production/db-backfill/src/operations/mod.rs index d978c206ac..089bf7fce5 100644 --- a/crates/cli/production/db-backfill/src/operations/mod.rs +++ b/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; \ No newline at end of file diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_events/backfill_list_artcraft_customer_links.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_events/backfill_list_artcraft_customer_links.rs new file mode 100644 index 0000000000..c90d03e972 --- /dev/null +++ b/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, 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) +} diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_events/backfill_list_artcraft_ledger_payments.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_events/backfill_list_artcraft_ledger_payments.rs new file mode 100644 index 0000000000..64a906588f --- /dev/null +++ b/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, +} + +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, + 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, 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::("credits_delta")? as i64, + created_at: row.try_get("created_at")?, + }); + } + + Ok(out) +} diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_events/mod.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_events/mod.rs index 43ffb30e01..0675bc647a 100644 --- a/crates/schema/database/mysql_queries/src/queries/user_spend_events/mod.rs +++ b/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;