query macro

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<String, String>>,
+  metadata_user_token: Option<&str>,
   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(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,

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<Vec<ArtcraftCustomerLink>, 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)
 }

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<Vec<LedgerPaymentRef>, 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<Utc>`
 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::<i32, _>("credits_delta")? as i64,
-      created_at: row.try_get("created_at")?,
-    });
-  }
-
-  Ok(out)
+  Ok(rows)
 }

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?;