From 062be6169420300b16bd0b45ffbd7bcce8134a74 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 29 Jun 2026 10:45:08 -0400 Subject: [PATCH] query macro --- .../backfill_user_spend_events.rs | 59 +++++++++++++++---- .../backfill_list_artcraft_customer_links.rs | 27 ++++----- .../backfill_list_artcraft_ledger_payments.rs | 45 +++++--------- .../insert_user_spend_event.rs | 42 +++++++------ 4 files changed, 95 insertions(+), 78 deletions(-) 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 index b54173534a..77e2b2bf5f 100644 --- 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 @@ -155,7 +155,10 @@ async fn backfill_payment_intents( 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); + let maybe_user = resolve_user( + pi.metadata.get("user_token").map(|s| s.as_str()), + &pi_id, customer_id.as_deref(), ledger_by_id, user_by_customer, + ); match maybe_user { Some(user) => { @@ -251,7 +254,10 @@ async fn backfill_invoices( 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); + let maybe_user = resolve_user( + invoice_metadata_user_token(inv), + &inv_id, customer_id.as_deref(), ledger_by_id, user_by_customer, + ); match maybe_user { Some(user) => { @@ -291,20 +297,23 @@ async fn backfill_invoices( // ---- 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. +/// Attribute a payment to a user, in order of reliability: +/// 1. Stripe object metadata `user_token` (set at checkout). +/// 2. The wallet ledger by Stripe id (`wallets.owner_user_token`) — this is +/// the "other half" that covers fulfilled subscriptions, which the customer +/// links table does NOT reliably have. +/// 3. `user_stripe_customer_links` by customer id — LAST resort (sparse for +/// subscriptions). fn resolve_user( - metadata: Option<&HashMap>, + metadata_user_token: Option<&str>, 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(user_token) = metadata_user_token { + if !user_token.is_empty() { + return Some(UserToken(user_token.to_string())); } } if let Some(ledger) = ledger_by_id.get(stripe_object_id) { @@ -318,6 +327,36 @@ fn resolve_user( None } +/// Extract our `user_token` from an invoice. Subscription invoices usually carry +/// it on the subscription/line-item metadata, NOT the top-level invoice metadata +/// (which is typically empty), so we check all three locations. +fn invoice_metadata_user_token(inv: &Invoice) -> Option<&str> { + if let Some(m) = inv.metadata.as_ref() { + if let Some(v) = m.get("user_token") { + if !v.is_empty() { + return Some(v.as_str()); + } + } + } + if let Some(sd) = inv.parent.as_ref().and_then(|p| p.subscription_details.as_ref()) { + if let Some(m) = sd.metadata.as_ref() { + if let Some(v) = m.get("user_token") { + if !v.is_empty() { + return Some(v.as_str()); + } + } + } + } + if let Some(line) = inv.lines.data.get(0) { + if let Some(v) = line.metadata.get("user_token") { + if !v.is_empty() { + return Some(v.as_str()); + } + } + } + None +} + struct UpsertInput<'a> { event_type: PaymentEventType, amount_usd_cents: i64, 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 index c90d03e972..ef8fa6b14f 100644 --- 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 @@ -1,12 +1,12 @@ use std::marker::PhantomData; -use sqlx::{Executor, MySql, Row}; +use sqlx::{Executor, MySql}; 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. +/// A (stripe_customer_id -> user_token) link in the ArtCraft payments namespace. +/// NB: this table is sparse for subscriptions, so it is the LAST-resort fallback +/// for attribution (Stripe metadata and the wallet ledger come first). pub struct ArtcraftCustomerLink { pub stripe_customer_id: String, pub user_token: UserToken, @@ -22,17 +22,18 @@ where /// 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( + let rows = sqlx::query_as!( + ArtcraftCustomerLink, r#" -SELECT stripe_customer_id, user_token +SELECT + stripe_customer_id AS `stripe_customer_id!`, + user_token AS `user_token!: UserToken` FROM user_stripe_customer_links WHERE payments_namespace = 'artcraft' "#, @@ -40,13 +41,5 @@ 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) + Ok(rows) } 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 index 64a906588f..5378b163f5 100644 --- 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 @@ -1,16 +1,15 @@ use std::marker::PhantomData; use chrono::{DateTime, Utc}; -use sqlx::{Executor, MySql, Row}; +use sqlx::{Executor, MySql}; 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`. +/// enumerated payment during backfill (Stripe is the authoritative payment list, +/// not this). 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_`). @@ -26,7 +25,7 @@ pub struct LedgerPaymentRef { pub owner_user_token: UserToken, /// Credits actually granted — becomes `user_spend_events.maybe_credits_granted`. - pub credits_delta: i64, + pub credits_delta: i32, /// When we credited the wallet (≈ settlement; spend events still bin on the /// Stripe object's own timestamp). @@ -47,22 +46,22 @@ where /// 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( + let rows = sqlx::query_as!( + LedgerPaymentRef, 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 +SELECT + wle.maybe_entity_ref AS `stripe_object_id!`, + wle.entry_type AS `entry_type!`, + wle.token AS `ledger_token!: WalletLedgerEntryToken`, + w.owner_user_token AS `owner_user_token!: UserToken`, + wle.credits_delta AS `credits_delta!`, + wle.created_at AS `created_at!: DateTime` FROM wallet_ledger_entries wle JOIN wallets w ON w.token = wle.wallet_token WHERE w.wallet_namespace = 'artcraft' @@ -70,22 +69,10 @@ WHERE w.wallet_namespace = 'artcraft' AND wle.created_at >= ? AND (wle.maybe_entity_ref LIKE 'pi\_%' OR wle.maybe_entity_ref LIKE 'in\_%') "#, + args.since, ) - .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) + Ok(rows) } diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_events/insert_user_spend_event.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_events/insert_user_spend_event.rs index d5004e5796..826cef1826 100644 --- a/crates/schema/database/mysql_queries/src/queries/user_spend_events/insert_user_spend_event.rs +++ b/crates/schema/database/mysql_queries/src/queries/user_spend_events/insert_user_spend_event.rs @@ -74,10 +74,8 @@ where /// with a `NULL` source_object_id are exempt and always insert, since MySQL /// treats each `NULL` in a unique index as distinct.) /// -/// NB: This uses the runtime `sqlx::query()` rather than the compile-time -/// `query!` macro because `user_spend_events` is a brand-new table that isn't in -/// the `.sqlx/` offline cache yet. Once the migration has been applied to a dev -/// DB and `cargo sqlx prepare` has been run, convert this to `query!`. +/// Compile-time-checked via `sqlx::query!` against the offline `.sqlx` cache +/// (regenerate with `cargo sqlx prepare` after schema changes). pub async fn insert_user_spend_event<'e, 'c: 'e, E>( args: InsertUserSpendEventArgs<'e, 'c, E>, ) -> Result<(), sqlx::Error> @@ -86,7 +84,7 @@ where { let token = UserSpendEventToken::generate(); - sqlx::query( + sqlx::query!( r#" INSERT INTO user_spend_events SET @@ -109,24 +107,24 @@ SET payment_occurred_at = ? ON DUPLICATE KEY UPDATE token = token "#, + token.as_str(), + args.payments_namespace.to_str(), + args.maybe_user_token.map(|t| t.as_str()), + args.event_type.to_str(), + args.amount_usd_cents, + args.maybe_credits_granted, + args.maybe_user_subscription_token.map(|t| t.as_str()), + args.maybe_wallet_ledger_entry_token.map(|t| t.as_str()), + args.payment_source.to_str(), + args.maybe_source_object_id, + args.maybe_stripe_customer_id, + args.maybe_stripe_invoice_id, + args.maybe_stripe_payment_intent_id, + args.maybe_stripe_charge_id, + args.maybe_stripe_event_id, + args.is_production, + args.payment_occurred_at, ) - .bind(token.as_str()) - .bind(args.payments_namespace.to_str()) - .bind(args.maybe_user_token.map(|t| t.as_str())) - .bind(args.event_type.to_str()) - .bind(args.amount_usd_cents) - .bind(args.maybe_credits_granted) - .bind(args.maybe_user_subscription_token.map(|t| t.as_str())) - .bind(args.maybe_wallet_ledger_entry_token.map(|t| t.as_str())) - .bind(args.payment_source.to_str()) - .bind(args.maybe_source_object_id) - .bind(args.maybe_stripe_customer_id) - .bind(args.maybe_stripe_invoice_id) - .bind(args.maybe_stripe_payment_intent_id) - .bind(args.maybe_stripe_charge_id) - .bind(args.maybe_stripe_event_id) - .bind(args.is_production) - .bind(args.payment_occurred_at) .execute(args.mysql_executor) .await?;