payments informatics (part 1) (#1704)
This commit is contained in:
Brandon Thomas authored and 
GitHub committed
1 parent
3bccac5b2e
commit
3e4f998599
43 files changed
+2426
-12
No files matched your search
+25
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"db_name": "MySQL",
|
||||
"query": "\nSELECT token AS `token: UserToken`\nFROM users\nWHERE email_address = ?\nLIMIT 1\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "token: UserToken",
|
||||
"type_info": {
|
||||
"type": "VarString",
|
||||
"flags": "NOT_NULL | UNIQUE_KEY | BINARY | NO_DEFAULT_VALUE",
|
||||
"char_set": 224,
|
||||
"max_size": 128
|
||||
}
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 1
|
||||
},
|
||||
"nullable": [
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "0d2eadc64240b9ce5348b3b238b0ae6229df21697c952b08e930d03f97414cf4"
|
||||
}
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"db_name": "MySQL",
|
||||
"query": "\nSELECT\n stripe_customer_id AS `stripe_customer_id!`,\n user_token AS `user_token!: UserToken`\nFROM user_stripe_customer_links\nWHERE payments_namespace = 'artcraft'\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "stripe_customer_id!",
|
||||
"type_info": {
|
||||
"type": "VarString",
|
||||
"flags": "NOT_NULL | UNIQUE_KEY | BINARY | NO_DEFAULT_VALUE",
|
||||
"char_set": 224,
|
||||
"max_size": 1020
|
||||
}
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "user_token!: UserToken",
|
||||
"type_info": {
|
||||
"type": "VarString",
|
||||
"flags": "NOT_NULL | MULTIPLE_KEY | BINARY | NO_DEFAULT_VALUE",
|
||||
"char_set": 224,
|
||||
"max_size": 128
|
||||
}
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 0
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "2d49cfbafc29bce35f3e03135054c7521fb0182663aefa971245f91f33a6fc49"
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"db_name": "MySQL",
|
||||
"query": "\nINSERT INTO user_spend_events\nSET\n token = ?,\n payments_namespace = ?,\n maybe_user_token = ?,\n event_type = ?,\n amount_usd_cents = ?,\n maybe_credits_granted = ?,\n maybe_user_subscription_token = ?,\n maybe_wallet_ledger_entry_token = ?,\n payment_source = ?,\n source_object_id = ?,\n maybe_stripe_customer_id = ?,\n maybe_stripe_invoice_id = ?,\n maybe_stripe_payment_intent_id = ?,\n maybe_stripe_charge_id = ?,\n maybe_stripe_event_id = ?,\n is_production = ?,\n payment_occurred_at = ?\nON DUPLICATE KEY UPDATE token = token\n ",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Right": 17
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "3e5a71cedc31cc656177c8331c2de1d1d2d2eb310aabeb7e6238943b073f6998"
|
||||
}
|
||||
+80
@@ -0,0 +1,80 @@
|
||||
{
|
||||
"db_name": "MySQL",
|
||||
"query": "\nSELECT\n wle.maybe_entity_ref AS `stripe_object_id!`,\n wle.entry_type AS `entry_type!`,\n wle.token AS `ledger_token!: WalletLedgerEntryToken`,\n w.owner_user_token AS `owner_user_token!: UserToken`,\n wle.credits_delta AS `credits_delta!`,\n wle.created_at AS `created_at!: DateTime<Utc>`\nFROM wallet_ledger_entries wle\nJOIN wallets w ON w.token = wle.wallet_token\nWHERE w.wallet_namespace = 'artcraft'\n AND wle.entry_type IN ('credit_banked', 'credit_monthly')\n AND wle.created_at >= ?\n AND (wle.maybe_entity_ref LIKE 'pi\\_%' OR wle.maybe_entity_ref LIKE 'in\\_%')\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "stripe_object_id!",
|
||||
"type_info": {
|
||||
"type": "VarString",
|
||||
"flags": "BINARY",
|
||||
"char_set": 224,
|
||||
"max_size": 1020
|
||||
}
|
||||
},
|
||||
{
|
||||
"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": "ledger_token!: WalletLedgerEntryToken",
|
||||
"type_info": {
|
||||
"type": "VarString",
|
||||
"flags": "NOT_NULL | UNIQUE_KEY | BINARY | NO_DEFAULT_VALUE",
|
||||
"char_set": 224,
|
||||
"max_size": 128
|
||||
}
|
||||
},
|
||||
{
|
||||
"ordinal": 3,
|
||||
"name": "owner_user_token!: UserToken",
|
||||
"type_info": {
|
||||
"type": "VarString",
|
||||
"flags": "NOT_NULL | MULTIPLE_KEY | BINARY | NO_DEFAULT_VALUE",
|
||||
"char_set": 224,
|
||||
"max_size": 128
|
||||
}
|
||||
},
|
||||
{
|
||||
"ordinal": 4,
|
||||
"name": "credits_delta!",
|
||||
"type_info": {
|
||||
"type": "Long",
|
||||
"flags": "NOT_NULL",
|
||||
"char_set": 63,
|
||||
"max_size": 11
|
||||
}
|
||||
},
|
||||
{
|
||||
"ordinal": 5,
|
||||
"name": "created_at!: DateTime<Utc>",
|
||||
"type_info": {
|
||||
"type": "Timestamp",
|
||||
"flags": "NOT_NULL | BINARY | TIMESTAMP",
|
||||
"char_set": 63,
|
||||
"max_size": 19
|
||||
}
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 1
|
||||
},
|
||||
"nullable": [
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "de69f3d564830f8fff90f754d13982de623fcda50118d261466eeaf034de5ecb"
|
||||
}
|
||||
Generated
+2
@@ -3049,6 +3049,8 @@ dependencies = [
|
||||
"futures 0.3.32",
|
||||
"log",
|
||||
"mysql_queries",
|
||||
"reqwest 0.12.23",
|
||||
"rustls 0.23.36",
|
||||
"serde",
|
||||
"serde_derive",
|
||||
"serde_json",
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
DROP TABLE IF EXISTS user_spend_events;
|
||||
@@ -0,0 +1,103 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
-- Append-only ledger of real money movements (USD cents), one row per settled
|
||||
-- event. Derived from settled Stripe webhooks (invoice.paid, payment_intent.
|
||||
-- succeeded / checkout.session.completed, charge.refunded, charge.dispute.*),
|
||||
-- NOT from every webhook. Idempotent via uq_source_object so retries never
|
||||
-- double-count. We always work in USD via Stripe, so there are no currency or
|
||||
-- fee columns.
|
||||
CREATE TABLE user_spend_events (
|
||||
-- Replication only.
|
||||
id BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
|
||||
|
||||
-- Effective public primary key. Token prefix "spend_".
|
||||
token VARCHAR(32) NOT NULL,
|
||||
|
||||
-- Multi-site segregation. Reuses the existing PaymentsNamespace enum.
|
||||
payments_namespace VARCHAR(32) NOT NULL,
|
||||
|
||||
-- Who paid. Nullable so we can record money we can't yet attribute and
|
||||
-- backfill -- but expect this set ~always.
|
||||
maybe_user_token VARCHAR(32) DEFAULT NULL,
|
||||
|
||||
-- PaymentEventType enum. Splits subscription vs credit-pack revenue:
|
||||
-- 'subscription_initial' -- first paid invoice of a new subscription
|
||||
-- 'subscription_renewal' -- recurring paid invoice (monthly, or annual once/yr)
|
||||
-- 'subscription_proration_upgrade' -- mid-cycle upgrade proration charge (positive)
|
||||
-- 'subscription_proration_downgrade' -- mid-cycle downgrade proration credit (rare as a settled event)
|
||||
-- 'credit_pack_purchase' -- one-time credit pack (the ~10x-volume case)
|
||||
-- 'refund' -- refund (amount negative)
|
||||
-- 'chargeback' -- dispute/chargeback (amount negative)
|
||||
-- 'manual_adjustment' -- staff/console correction (payment_source = 'manual')
|
||||
-- 'subscription_monthly_refill' -- OPTIONAL non-revenue event for annual-plan
|
||||
-- monthly credit refills (amount 0)
|
||||
event_type VARCHAR(32) NOT NULL,
|
||||
|
||||
-- ========== MONEY ==========
|
||||
|
||||
-- USD cents. SIGNED: refunds/chargebacks negative; refills are 0. BIGINT so
|
||||
-- downstream lifetime sums never overflow.
|
||||
amount_usd_cents BIGINT NOT NULL,
|
||||
|
||||
-- For credit-pack purchases / subscription grants/refills: how many of OUR
|
||||
-- credits this granted. Lets us correlate dollars-in vs credits-granted vs
|
||||
-- credits-used. NULL for refunds/chargebacks.
|
||||
maybe_credits_granted INTEGER UNSIGNED DEFAULT NULL,
|
||||
|
||||
-- ========== INTERNAL LINKAGE ==========
|
||||
|
||||
-- The subscription this payment belongs to, if any (-> user_subscriptions.token).
|
||||
maybe_user_subscription_token VARCHAR(32) DEFAULT NULL,
|
||||
|
||||
-- The credit-grant ledger entry this produced, if any
|
||||
-- (-> wallet_ledger_entries.token). Ties a dollar event to its credit effect.
|
||||
maybe_wallet_ledger_entry_token VARCHAR(32) DEFAULT NULL,
|
||||
|
||||
-- ========== SOURCE / PROVIDER ==========
|
||||
|
||||
-- PaymentSource enum: 'stripe' or 'manual'. (USD/Stripe only for the
|
||||
-- foreseeable future; 'manual' covers staff/console adjustments.)
|
||||
payment_source VARCHAR(16) NOT NULL,
|
||||
|
||||
-- The authoritative object this row represents -- the dedup anchor. NULLABLE.
|
||||
-- For Stripe: invoice id (subscriptions), payment_intent id (one-time), or
|
||||
-- refund/dispute id (refunds). For payment_source='manual' (no provider
|
||||
-- object) leave it NULL. MySQL allows multiple NULLs in a unique index, so
|
||||
-- manual rows freely duplicate, while any present id is still deduped.
|
||||
source_object_id VARCHAR(255) DEFAULT NULL,
|
||||
|
||||
-- Convenience copies for joins/forensics (255 + utf8mb4_bin for Stripe IDs).
|
||||
maybe_stripe_customer_id VARCHAR(255) DEFAULT NULL,
|
||||
maybe_stripe_invoice_id VARCHAR(255) DEFAULT NULL,
|
||||
maybe_stripe_payment_intent_id VARCHAR(255) DEFAULT NULL,
|
||||
maybe_stripe_charge_id VARCHAR(255) DEFAULT NULL,
|
||||
-- The webhook event that caused this row (provenance only).
|
||||
maybe_stripe_event_id VARCHAR(255) DEFAULT NULL,
|
||||
|
||||
-- Stripe livemode. Keep test-mode money out of dashboards.
|
||||
is_production BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
|
||||
-- ========== TIMESTAMPS ==========
|
||||
|
||||
-- When the money actually moved (Stripe settlement time). Bin/aggregate on
|
||||
-- THIS, not created_at (which is when we ingested the webhook).
|
||||
payment_occurred_at TIMESTAMP NOT NULL,
|
||||
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
|
||||
-- ========== INDICES ==========
|
||||
PRIMARY KEY (id),
|
||||
UNIQUE KEY (token),
|
||||
-- Idempotency: one row per settled object. NULL source_object_id rows (manual)
|
||||
-- are exempt -- MySQL treats each NULL in a unique index as distinct.
|
||||
UNIQUE KEY uq_source_object (payment_source, source_object_id),
|
||||
KEY idx_user_occurred (maybe_user_token, payment_occurred_at),
|
||||
KEY idx_occurred (payment_occurred_at),
|
||||
KEY idx_namespace_occurred (payments_namespace, payment_occurred_at),
|
||||
KEY idx_event_type (event_type),
|
||||
KEY idx_stripe_customer (maybe_stripe_customer_id),
|
||||
KEY idx_stripe_invoice (maybe_stripe_invoice_id)
|
||||
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
|
||||
@@ -0,0 +1,5 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
DROP TABLE IF EXISTS user_daily_spends;
|
||||
@@ -0,0 +1,44 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
-- Per-user, per-day spend rollup (L1). Net = gross - refunds. Subscription vs
|
||||
-- credit-pack splits expose the revenue mix per day without touching the event
|
||||
-- table. Sparsely populated -- no row for days with zero spend and zero refunds.
|
||||
-- Derived from (and fully rebuildable from) user_spend_events.
|
||||
CREATE TABLE user_daily_spends (
|
||||
-- Replication only.
|
||||
id BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
|
||||
|
||||
user_token VARCHAR(32) NOT NULL,
|
||||
payments_namespace VARCHAR(32) NOT NULL,
|
||||
|
||||
-- Reporting day. Bins user_spend_events.payment_occurred_at in a FIXED tz
|
||||
-- (UTC). DATE, not TIMESTAMP.
|
||||
spend_date DATE NOT NULL,
|
||||
|
||||
-- Revenue mix (subset of gross). Category totals can never go negative, so
|
||||
-- they're UNSIGNED.
|
||||
subscription_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
credits_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
|
||||
gross_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- sum of positive amounts (UNSIGNED)
|
||||
refund_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- refund/chargeback magnitude (UNSIGNED)
|
||||
-- NET is the ONE field here that must be SIGNED: a single day can net negative
|
||||
-- when refunds (often of an earlier day's purchase) exceed that day's purchases.
|
||||
net_spend_usd_cents BIGINT NOT NULL DEFAULT 0, -- gross - refund (SIGNED)
|
||||
|
||||
payment_count INT UNSIGNED NOT NULL DEFAULT 0, -- # positive money events
|
||||
credits_granted BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
|
||||
-- ========== INDICES ==========
|
||||
PRIMARY KEY (id),
|
||||
UNIQUE KEY uq_user_day (user_token, payments_namespace, spend_date),
|
||||
-- "Top spenders / spend in a date range": filter by date, then group by user.
|
||||
KEY idx_day_user_net (user_token, payments_namespace, spend_date, net_spend_usd_cents),
|
||||
KEY idx_user_day (user_token, spend_date)
|
||||
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
|
||||
@@ -0,0 +1,5 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
DROP TABLE IF EXISTS user_spend_summaries;
|
||||
@@ -0,0 +1,95 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
-- Per-user spend summary (L2). Lifetime/recency facts are event-maintained (they
|
||||
-- only grow); sliding windows and weekly-rhythm metrics are job-maintained (they
|
||||
-- shift even on days a user doesn't pay). Powers churn / dip / re-engagement
|
||||
-- dashboards in O(1). Derived from (and fully rebuildable from) user_spend_events
|
||||
-- and user_daily_spends.
|
||||
CREATE TABLE user_spend_summaries (
|
||||
-- Replication only.
|
||||
id BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
|
||||
|
||||
payments_namespace VARCHAR(32) NOT NULL,
|
||||
user_token VARCHAR(32) NOT NULL,
|
||||
|
||||
-- ===== Lifetime (event-maintained) =====
|
||||
-- Gross / category / refund totals are UNSIGNED. Lifetime net is UNSIGNED too:
|
||||
-- over a lifetime you can never refund more than was paid.
|
||||
lifetime_gross_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
lifetime_subscription_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- revenue mix split
|
||||
lifetime_credits_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- revenue mix split
|
||||
lifetime_refund_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
lifetime_net_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- gross - refund
|
||||
-- Event COUNTS (not money): number of payment events vs refund/chargeback
|
||||
-- events over the user's lifetime. Both event-maintained.
|
||||
lifetime_payment_count INT UNSIGNED NOT NULL DEFAULT 0, -- # of positive money events
|
||||
lifetime_refund_count INT UNSIGNED NOT NULL DEFAULT 0, -- # of refund/chargeback events
|
||||
first_payment_at TIMESTAMP NULL DEFAULT NULL,
|
||||
first_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- amount of the first payment
|
||||
last_payment_at TIMESTAMP NULL DEFAULT NULL,
|
||||
last_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- amount of the most recent payment
|
||||
days_since_first_payment INT UNSIGNED DEFAULT NULL, -- job-refreshed
|
||||
days_since_last_payment INT UNSIGNED DEFAULT NULL, -- job-refreshed
|
||||
|
||||
-- ===== Sliding NET spend windows -- SIGNED, a window can net negative
|
||||
-- (job-refreshed from user_daily_spends) =====
|
||||
-- "prev" = the equally-sized window immediately before, for WoW/MoM deltas.
|
||||
net_spend_7d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_prev_7d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_14d_usd_cents BIGINT NOT NULL DEFAULT 0, -- bi-weekly
|
||||
net_spend_prev_14d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_30d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_prev_30d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_60d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_90d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_this_year_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
|
||||
-- ===== Weekly cadence / rhythm (job-refreshed) -- for the sales team =====
|
||||
-- A "week" is Mon-Sun UTC. These let sales spot users breaking their habit.
|
||||
-- Per-user weekly baselines at two horizons: 4w is responsive (recent norm),
|
||||
-- 12w is stable (established norm). 4w well below 12w is itself a cooling-off
|
||||
-- signal. Mean weekly NET spend over the trailing window (calendar weeks,
|
||||
-- zero-spend weeks included). SIGNED, like the window nets.
|
||||
avg_weekly_net_spend_4w_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
avg_weekly_net_spend_12w_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
-- How many of the last N calendar weeks had any net spend -- a regularity
|
||||
-- measure at several horizons.
|
||||
active_weeks_in_last_4 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..4
|
||||
active_weeks_in_last_8 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..8
|
||||
active_weeks_in_last_12 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..12
|
||||
active_weeks_in_last_24 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..24
|
||||
active_weeks_in_last_52 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..52
|
||||
consecutive_active_weeks INT UNSIGNED NOT NULL DEFAULT 0, -- current "streak"
|
||||
consecutive_inactive_weeks INT UNSIGNED NOT NULL DEFAULT 0, -- current "dip" length
|
||||
-- Recency in weeks. Subsumes the entire "is_active_spender_last_N_weeks"
|
||||
-- family: "active in the last N weeks" == (weeks_since_last_spend < N). So we
|
||||
-- intentionally store NO is_active_* booleans (redundant + poor index
|
||||
-- selectivity); we index this field instead and range-query any N. NULL = never spent.
|
||||
weeks_since_last_spend INT UNSIGNED DEFAULT NULL,
|
||||
|
||||
-- ===== Flags =====
|
||||
is_active_subscriber BOOLEAN NOT NULL DEFAULT FALSE, -- from user_subscriptions
|
||||
maybe_subscription_interval VARCHAR(16) DEFAULT NULL, -- 'month' / 'year' (annual plans behave differently)
|
||||
-- Optional precomputed re-engagement priority (job-set).
|
||||
maybe_reengagement_score INT UNSIGNED DEFAULT NULL,
|
||||
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
|
||||
-- ========== INDICES ==========
|
||||
PRIMARY KEY (id),
|
||||
UNIQUE KEY uq_user (payments_namespace, user_token),
|
||||
KEY idx_lifetime (payments_namespace, lifetime_net_spend_usd_cents),
|
||||
KEY idx_last_payment_at (payments_namespace, last_payment_at),
|
||||
KEY idx_first_payment_at (payments_namespace, first_payment_at),
|
||||
KEY idx_spend_7d (payments_namespace, net_spend_7d_usd_cents),
|
||||
KEY idx_spend_30d (payments_namespace, net_spend_30d_usd_cents),
|
||||
KEY idx_reengagement (payments_namespace, maybe_reengagement_score),
|
||||
-- Sort / range-filter by recency. idx_weeks_since_last_spend also powers every
|
||||
-- "active in the last N weeks" query (weeks_since_last_spend < N).
|
||||
KEY idx_weeks_since_last_spend (payments_namespace, weeks_since_last_spend),
|
||||
KEY idx_days_since_last_payment (payments_namespace, days_since_last_payment)
|
||||
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
|
||||
@@ -0,0 +1,47 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
-- NB: This is a manually squashed view of all the CREATE and ALTER statements,
|
||||
-- with comments attached to the fields for centralized documentation.
|
||||
|
||||
-- Per-user, per-day spend rollup (L1). Net = gross - refunds. Subscription vs
|
||||
-- credit-pack splits expose the revenue mix per day without touching the event
|
||||
-- table. Sparsely populated -- no row for days with zero spend and zero refunds.
|
||||
-- Derived from (and fully rebuildable from) user_spend_events.
|
||||
CREATE TABLE user_daily_spends (
|
||||
-- Replication only.
|
||||
id BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
|
||||
|
||||
user_token VARCHAR(32) NOT NULL,
|
||||
payments_namespace VARCHAR(32) NOT NULL,
|
||||
|
||||
-- Reporting day. Bins user_spend_events.payment_occurred_at in a FIXED tz
|
||||
-- (UTC). DATE, not TIMESTAMP.
|
||||
spend_date DATE NOT NULL,
|
||||
|
||||
-- Revenue mix (subset of gross). Category totals can never go negative, so
|
||||
-- they're UNSIGNED.
|
||||
subscription_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
credits_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
|
||||
gross_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- sum of positive amounts (UNSIGNED)
|
||||
refund_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- refund/chargeback magnitude (UNSIGNED)
|
||||
-- NET is the ONE field here that must be SIGNED: a single day can net negative
|
||||
-- when refunds (often of an earlier day's purchase) exceed that day's purchases.
|
||||
net_spend_usd_cents BIGINT NOT NULL DEFAULT 0, -- gross - refund (SIGNED)
|
||||
|
||||
payment_count INT UNSIGNED NOT NULL DEFAULT 0, -- # positive money events
|
||||
credits_granted BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
|
||||
-- ========== INDICES ==========
|
||||
PRIMARY KEY (id),
|
||||
UNIQUE KEY uq_user_day (user_token, payments_namespace, spend_date),
|
||||
-- "Top spenders / spend in a date range": filter by date, then group by user.
|
||||
KEY idx_day_user_net (user_token, payments_namespace, spend_date, net_spend_usd_cents),
|
||||
KEY idx_user_day (user_token, spend_date)
|
||||
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
|
||||
@@ -0,0 +1,106 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
-- NB: This is a manually squashed view of all the CREATE and ALTER statements,
|
||||
-- with comments attached to the fields for centralized documentation.
|
||||
|
||||
-- Append-only ledger of real money movements (USD cents), one row per settled
|
||||
-- event. Derived from settled Stripe webhooks (invoice.paid, payment_intent.
|
||||
-- succeeded / checkout.session.completed, charge.refunded, charge.dispute.*),
|
||||
-- NOT from every webhook. Idempotent via uq_source_object so retries never
|
||||
-- double-count. We always work in USD via Stripe, so there are no currency or
|
||||
-- fee columns.
|
||||
CREATE TABLE user_spend_events (
|
||||
-- Replication only.
|
||||
id BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
|
||||
|
||||
-- Effective public primary key. Token prefix "spend_".
|
||||
token VARCHAR(32) NOT NULL,
|
||||
|
||||
-- Multi-site segregation. Reuses the existing PaymentsNamespace enum.
|
||||
payments_namespace VARCHAR(32) NOT NULL,
|
||||
|
||||
-- Who paid. Nullable so we can record money we can't yet attribute and
|
||||
-- backfill -- but expect this set ~always.
|
||||
maybe_user_token VARCHAR(32) DEFAULT NULL,
|
||||
|
||||
-- PaymentEventType enum. Splits subscription vs credit-pack revenue:
|
||||
-- 'subscription_initial' -- first paid invoice of a new subscription
|
||||
-- 'subscription_renewal' -- recurring paid invoice (monthly, or annual once/yr)
|
||||
-- 'subscription_proration_upgrade' -- mid-cycle upgrade proration charge (positive)
|
||||
-- 'subscription_proration_downgrade' -- mid-cycle downgrade proration credit (rare as a settled event)
|
||||
-- 'credit_pack_purchase' -- one-time credit pack (the ~10x-volume case)
|
||||
-- 'refund' -- refund (amount negative)
|
||||
-- 'chargeback' -- dispute/chargeback (amount negative)
|
||||
-- 'manual_adjustment' -- staff/console correction (payment_source = 'manual')
|
||||
-- 'subscription_monthly_refill' -- OPTIONAL non-revenue event for annual-plan
|
||||
-- monthly credit refills (amount 0)
|
||||
event_type VARCHAR(32) NOT NULL,
|
||||
|
||||
-- ========== MONEY ==========
|
||||
|
||||
-- USD cents. SIGNED: refunds/chargebacks negative; refills are 0. BIGINT so
|
||||
-- downstream lifetime sums never overflow.
|
||||
amount_usd_cents BIGINT NOT NULL,
|
||||
|
||||
-- For credit-pack purchases / subscription grants/refills: how many of OUR
|
||||
-- credits this granted. Lets us correlate dollars-in vs credits-granted vs
|
||||
-- credits-used. NULL for refunds/chargebacks.
|
||||
maybe_credits_granted INTEGER UNSIGNED DEFAULT NULL,
|
||||
|
||||
-- ========== INTERNAL LINKAGE ==========
|
||||
|
||||
-- The subscription this payment belongs to, if any (-> user_subscriptions.token).
|
||||
maybe_user_subscription_token VARCHAR(32) DEFAULT NULL,
|
||||
|
||||
-- The credit-grant ledger entry this produced, if any
|
||||
-- (-> wallet_ledger_entries.token). Ties a dollar event to its credit effect.
|
||||
maybe_wallet_ledger_entry_token VARCHAR(32) DEFAULT NULL,
|
||||
|
||||
-- ========== SOURCE / PROVIDER ==========
|
||||
|
||||
-- PaymentSource enum: 'stripe' or 'manual'. (USD/Stripe only for the
|
||||
-- foreseeable future; 'manual' covers staff/console adjustments.)
|
||||
payment_source VARCHAR(16) NOT NULL,
|
||||
|
||||
-- The authoritative object this row represents -- the dedup anchor. NULLABLE.
|
||||
-- For Stripe: invoice id (subscriptions), payment_intent id (one-time), or
|
||||
-- refund/dispute id (refunds). For payment_source='manual' (no provider
|
||||
-- object) leave it NULL. MySQL allows multiple NULLs in a unique index, so
|
||||
-- manual rows freely duplicate, while any present id is still deduped.
|
||||
source_object_id VARCHAR(255) DEFAULT NULL,
|
||||
|
||||
-- Convenience copies for joins/forensics (255 + utf8mb4_bin for Stripe IDs).
|
||||
maybe_stripe_customer_id VARCHAR(255) DEFAULT NULL,
|
||||
maybe_stripe_invoice_id VARCHAR(255) DEFAULT NULL,
|
||||
maybe_stripe_payment_intent_id VARCHAR(255) DEFAULT NULL,
|
||||
maybe_stripe_charge_id VARCHAR(255) DEFAULT NULL,
|
||||
-- The webhook event that caused this row (provenance only).
|
||||
maybe_stripe_event_id VARCHAR(255) DEFAULT NULL,
|
||||
|
||||
-- Stripe livemode. Keep test-mode money out of dashboards.
|
||||
is_production BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
|
||||
-- ========== TIMESTAMPS ==========
|
||||
|
||||
-- When the money actually moved (Stripe settlement time). Bin/aggregate on
|
||||
-- THIS, not created_at (which is when we ingested the webhook).
|
||||
payment_occurred_at TIMESTAMP NOT NULL,
|
||||
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
|
||||
-- ========== INDICES ==========
|
||||
PRIMARY KEY (id),
|
||||
UNIQUE KEY (token),
|
||||
-- Idempotency: one row per settled object. NULL source_object_id rows (manual)
|
||||
-- are exempt -- MySQL treats each NULL in a unique index as distinct.
|
||||
UNIQUE KEY uq_source_object (payment_source, source_object_id),
|
||||
KEY idx_user_occurred (maybe_user_token, payment_occurred_at),
|
||||
KEY idx_occurred (payment_occurred_at),
|
||||
KEY idx_namespace_occurred (payments_namespace, payment_occurred_at),
|
||||
KEY idx_event_type (event_type),
|
||||
KEY idx_stripe_customer (maybe_stripe_customer_id),
|
||||
KEY idx_stripe_invoice (maybe_stripe_invoice_id)
|
||||
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
|
||||
@@ -0,0 +1,98 @@
|
||||
-- noinspection SqlDialectInspectionForFile
|
||||
-- noinspection SqlNoDataSourceInspectionForFile
|
||||
-- noinspection SqlResolveForFile
|
||||
|
||||
-- NB: This is a manually squashed view of all the CREATE and ALTER statements,
|
||||
-- with comments attached to the fields for centralized documentation.
|
||||
|
||||
-- Per-user spend summary (L2). Lifetime/recency facts are event-maintained (they
|
||||
-- only grow); sliding windows and weekly-rhythm metrics are job-maintained (they
|
||||
-- shift even on days a user doesn't pay). Powers churn / dip / re-engagement
|
||||
-- dashboards in O(1). Derived from (and fully rebuildable from) user_spend_events
|
||||
-- and user_daily_spends.
|
||||
CREATE TABLE user_spend_summaries (
|
||||
-- Replication only.
|
||||
id BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
|
||||
|
||||
payments_namespace VARCHAR(32) NOT NULL,
|
||||
user_token VARCHAR(32) NOT NULL,
|
||||
|
||||
-- ===== Lifetime (event-maintained) =====
|
||||
-- Gross / category / refund totals are UNSIGNED. Lifetime net is UNSIGNED too:
|
||||
-- over a lifetime you can never refund more than was paid.
|
||||
lifetime_gross_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
lifetime_subscription_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- revenue mix split
|
||||
lifetime_credits_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- revenue mix split
|
||||
lifetime_refund_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||||
lifetime_net_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- gross - refund
|
||||
-- Event COUNTS (not money): number of payment events vs refund/chargeback
|
||||
-- events over the user's lifetime. Both event-maintained.
|
||||
lifetime_payment_count INT UNSIGNED NOT NULL DEFAULT 0, -- # of positive money events
|
||||
lifetime_refund_count INT UNSIGNED NOT NULL DEFAULT 0, -- # of refund/chargeback events
|
||||
first_payment_at TIMESTAMP NULL DEFAULT NULL,
|
||||
first_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- amount of the first payment
|
||||
last_payment_at TIMESTAMP NULL DEFAULT NULL,
|
||||
last_spend_usd_cents BIGINT UNSIGNED NOT NULL DEFAULT 0, -- amount of the most recent payment
|
||||
days_since_first_payment INT UNSIGNED DEFAULT NULL, -- job-refreshed
|
||||
days_since_last_payment INT UNSIGNED DEFAULT NULL, -- job-refreshed
|
||||
|
||||
-- ===== Sliding NET spend windows -- SIGNED, a window can net negative
|
||||
-- (job-refreshed from user_daily_spends) =====
|
||||
-- "prev" = the equally-sized window immediately before, for WoW/MoM deltas.
|
||||
net_spend_7d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_prev_7d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_14d_usd_cents BIGINT NOT NULL DEFAULT 0, -- bi-weekly
|
||||
net_spend_prev_14d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_30d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_prev_30d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_60d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_90d_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
net_spend_this_year_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
|
||||
-- ===== Weekly cadence / rhythm (job-refreshed) -- for the sales team =====
|
||||
-- A "week" is Mon-Sun UTC. These let sales spot users breaking their habit.
|
||||
-- Per-user weekly baselines at two horizons: 4w is responsive (recent norm),
|
||||
-- 12w is stable (established norm). 4w well below 12w is itself a cooling-off
|
||||
-- signal. Mean weekly NET spend over the trailing window (calendar weeks,
|
||||
-- zero-spend weeks included). SIGNED, like the window nets.
|
||||
avg_weekly_net_spend_4w_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
avg_weekly_net_spend_12w_usd_cents BIGINT NOT NULL DEFAULT 0,
|
||||
-- How many of the last N calendar weeks had any net spend -- a regularity
|
||||
-- measure at several horizons.
|
||||
active_weeks_in_last_4 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..4
|
||||
active_weeks_in_last_8 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..8
|
||||
active_weeks_in_last_12 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..12
|
||||
active_weeks_in_last_24 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..24
|
||||
active_weeks_in_last_52 TINYINT UNSIGNED NOT NULL DEFAULT 0, -- 0..52
|
||||
consecutive_active_weeks INT UNSIGNED NOT NULL DEFAULT 0, -- current "streak"
|
||||
consecutive_inactive_weeks INT UNSIGNED NOT NULL DEFAULT 0, -- current "dip" length
|
||||
-- Recency in weeks. Subsumes the entire "is_active_spender_last_N_weeks"
|
||||
-- family: "active in the last N weeks" == (weeks_since_last_spend < N). So we
|
||||
-- intentionally store NO is_active_* booleans (redundant + poor index
|
||||
-- selectivity); we index this field instead and range-query any N. NULL = never spent.
|
||||
weeks_since_last_spend INT UNSIGNED DEFAULT NULL,
|
||||
|
||||
-- ===== Flags =====
|
||||
is_active_subscriber BOOLEAN NOT NULL DEFAULT FALSE, -- from user_subscriptions
|
||||
maybe_subscription_interval VARCHAR(16) DEFAULT NULL, -- 'month' / 'year' (annual plans behave differently)
|
||||
-- Optional precomputed re-engagement priority (job-set).
|
||||
maybe_reengagement_score INT UNSIGNED DEFAULT NULL,
|
||||
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
|
||||
-- ========== INDICES ==========
|
||||
PRIMARY KEY (id),
|
||||
UNIQUE KEY uq_user (payments_namespace, user_token),
|
||||
KEY idx_lifetime (payments_namespace, lifetime_net_spend_usd_cents),
|
||||
KEY idx_last_payment_at (payments_namespace, last_payment_at),
|
||||
KEY idx_first_payment_at (payments_namespace, first_payment_at),
|
||||
KEY idx_spend_7d (payments_namespace, net_spend_7d_usd_cents),
|
||||
KEY idx_spend_30d (payments_namespace, net_spend_30d_usd_cents),
|
||||
KEY idx_reengagement (payments_namespace, maybe_reengagement_score),
|
||||
-- Sort / range-filter by recency. idx_weeks_since_last_spend also powers every
|
||||
-- "active in the last N weeks" query (weeks_since_last_spend < N).
|
||||
KEY idx_weeks_since_last_spend (payments_namespace, weeks_since_last_spend),
|
||||
KEY idx_days_since_last_payment (payments_namespace, days_since_last_payment)
|
||||
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
|
||||
@@ -37,7 +37,13 @@ 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 is queried via plain REST (reqwest) with lenient serde structs — the
|
||||
# typed SDK's miniserde parsing is too strict for production data variety.
|
||||
# rustls 0.23 needs an explicit crypto provider (graph has both ring + aws-lc-rs).
|
||||
rustls = { version = "0.23", features = ["ring"] }
|
||||
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
|
||||
|
||||
# Serialization
|
||||
serde = "1.0.125"
|
||||
|
||||
@@ -20,6 +20,7 @@ pub struct Args {
|
||||
// #[derive(Debug, Deserialize, Copy, Clone)]
|
||||
#[derive(ValueEnum, Clone, Copy, Debug)]
|
||||
pub enum Command {
|
||||
BackfillUserSpendEvents,
|
||||
CalculateLegacyTtsResultsUsages,
|
||||
CalculateModelWeightsUsages,
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use std::io::Write;
|
||||
|
||||
use chrono::Local;
|
||||
use log::{info, LevelFilter};
|
||||
use log::{info, warn, LevelFilter};
|
||||
use sqlx::{MySql, Pool};
|
||||
use sqlx::mysql::MySqlPoolOptions;
|
||||
|
||||
@@ -10,6 +10,9 @@ 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::stripe_api::StripeApi;
|
||||
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;
|
||||
|
||||
@@ -21,6 +24,12 @@ pub mod operations;
|
||||
async fn main() -> AnyhowResult<()> {
|
||||
println!("db-backfill: run backfill or migration operations");
|
||||
|
||||
// rustls 0.23 requires a process-wide crypto provider before the first TLS
|
||||
// handshake (the replica's `ssl-mode=required` connection and Stripe HTTPS).
|
||||
// The dependency graph enables BOTH ring and aws-lc-rs, so rustls refuses to
|
||||
// auto-select and panics — install one explicitly. (`Err` = already installed.)
|
||||
let _ = rustls::crypto::ring::default_provider().install_default();
|
||||
|
||||
// init_all_with_default_logging(None);
|
||||
Builder::new()
|
||||
.format(|buf, record| {
|
||||
@@ -39,27 +48,149 @@ async fn main() -> AnyhowResult<()> {
|
||||
|
||||
// NB: This secrets file differs from the rest because we might want to backfill production from local dev.
|
||||
// (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?;
|
||||
// The path is resolved relative to the current working directory — log the
|
||||
// absolute path so a "not found" is actionable.
|
||||
let secrets_filename = ".env-db-backfill-secrets";
|
||||
let secrets_path = std::env::current_dir()
|
||||
.map(|dir| dir.join(secrets_filename))
|
||||
.unwrap_or_else(|_| std::path::PathBuf::from(secrets_filename));
|
||||
info!("Loading secrets from: {}", secrets_path.display());
|
||||
// NB: We deliberately do NOT use `dotenv` here. Its parser treats `$`, spaces,
|
||||
// and `\` as special INSIDE the value (quoted or not), which mangles DB
|
||||
// connection URLs and generated passwords. This loader takes each value
|
||||
// verbatim, so `MYSQL_READ_URL=mysql://user:p$ss w@host:25060/db?ssl-mode=required`
|
||||
// just works.
|
||||
load_secrets_file(&secrets_path)?;
|
||||
|
||||
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 = StripeApi::new(easyenv::get_env_string_required("STRIPE_ARTCRAFT_SECRET_KEY")?)?;
|
||||
backfill_user_spend_events(&read_pool, &write_pool, &stripe, sub_args).await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Minimal, forgiving `.env`-style loader that sets process env vars from a
|
||||
/// `KEY=VALUE` file. The value is the remainder of the line taken VERBATIM
|
||||
/// (after one optional layer of surrounding quotes is stripped). Blank lines and
|
||||
/// `#` comment lines are ignored. Unlike `dotenv`, it does not interpret `$`,
|
||||
/// spaces, `\`, or `#` inside the value — so DB URLs and generated passwords
|
||||
/// pass through unchanged.
|
||||
fn load_secrets_file(path: &std::path::Path) -> AnyhowResult<()> {
|
||||
let contents = std::fs::read_to_string(path)
|
||||
.map_err(|err| anyhow::anyhow!("Could not read secrets file at {} : {}", path.display(), err))?;
|
||||
|
||||
let mut loaded_keys: Vec<String> = Vec::new();
|
||||
for (line_index, raw_line) in contents.lines().enumerate() {
|
||||
let line = raw_line.trim();
|
||||
if line.is_empty() || line.starts_with('#') {
|
||||
continue;
|
||||
}
|
||||
let (key, value) = match line.split_once('=') {
|
||||
Some((key, value)) => (key.trim(), strip_matching_quotes(value.trim())),
|
||||
None => {
|
||||
warn!("secrets {}:{}: ignoring line without '=': {:?}", path.display(), line_index + 1, line);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
if key.is_empty() {
|
||||
warn!("secrets {}:{}: ignoring empty key", path.display(), line_index + 1);
|
||||
continue;
|
||||
}
|
||||
std::env::set_var(key, value);
|
||||
loaded_keys.push(key.to_string());
|
||||
}
|
||||
|
||||
info!(
|
||||
"Loaded {} key(s) from {} ({} bytes): [{}]",
|
||||
loaded_keys.len(),
|
||||
path.display(),
|
||||
contents.len(),
|
||||
loaded_keys.join(", "),
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod secrets_loader_tests {
|
||||
use super::strip_matching_quotes;
|
||||
|
||||
// The whole point: a DB URL with a password containing `$`, a space, `#`, and a
|
||||
// query string with `=` must survive verbatim (this is exactly what `dotenv` mangled).
|
||||
#[test]
|
||||
fn url_value_is_taken_verbatim() {
|
||||
let line = r#"MYSQL_READ_URL=mysql://user:p$ss w#x@host:25060/db?ssl-mode=required"#;
|
||||
let (key, value) = line.split_once('=').unwrap();
|
||||
assert_eq!(key.trim(), "MYSQL_READ_URL");
|
||||
assert_eq!(
|
||||
strip_matching_quotes(value.trim()),
|
||||
"mysql://user:p$ss w#x@host:25060/db?ssl-mode=required",
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn strips_one_layer_of_matching_quotes_only() {
|
||||
assert_eq!(strip_matching_quotes("\"abc\""), "abc");
|
||||
assert_eq!(strip_matching_quotes("'abc'"), "abc");
|
||||
assert_eq!(strip_matching_quotes("abc"), "abc");
|
||||
assert_eq!(strip_matching_quotes("\"abc'"), "\"abc'"); // mismatched: untouched
|
||||
assert_eq!(strip_matching_quotes("\""), "\""); // single char: untouched
|
||||
}
|
||||
}
|
||||
|
||||
/// Strip a single layer of matching surrounding single or double quotes, if any.
|
||||
fn strip_matching_quotes(value: &str) -> &str {
|
||||
let bytes = value.as_bytes();
|
||||
if bytes.len() >= 2 {
|
||||
let first = bytes[0];
|
||||
let last = bytes[bytes.len() - 1];
|
||||
if (first == b'"' && last == b'"') || (first == b'\'' && last == b'\'') {
|
||||
return &value[1..value.len() - 1];
|
||||
}
|
||||
}
|
||||
value
|
||||
}
|
||||
|
||||
async fn get_mysql(env_var_name: &str) -> AnyhowResult<Pool<MySql>> {
|
||||
info!("Connecting to MySQL {env_var_name}...");
|
||||
let url = easyenv::get_env_string_required(env_var_name)?;
|
||||
// Redacted so you can SEE the host + database + params actually being used
|
||||
// (e.g. whether a database path is present) without leaking the password.
|
||||
info!("Connecting to MySQL {env_var_name}: {}", redact_db_url(&url));
|
||||
|
||||
let pool = MySqlPoolOptions::new()
|
||||
.max_connections(easyenv::get_env_num("MYSQL_MAX_CONNECTIONS", 20)?)
|
||||
.connect(&easyenv::get_env_string_required(env_var_name)?)
|
||||
.connect(&url)
|
||||
.await?;
|
||||
|
||||
Ok(pool)
|
||||
}
|
||||
|
||||
/// Mask the user:password portion of a DB URL for logging, leaving scheme, host,
|
||||
/// port, database path, and query params visible.
|
||||
fn redact_db_url(url: &str) -> String {
|
||||
if let Some(scheme_end) = url.find("://") {
|
||||
let after = &url[scheme_end + 3..];
|
||||
if let Some(at) = after.rfind('@') {
|
||||
return format!("{}://***@{}", &url[..scheme_end], &after[at + 1..]);
|
||||
}
|
||||
}
|
||||
url.to_string()
|
||||
}
|
||||
@@ -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.
|
||||
+515
@@ -0,0 +1,515 @@
|
||||
use std::collections::HashMap;
|
||||
use std::convert::TryFrom;
|
||||
use std::fs::File;
|
||||
use std::io::Write;
|
||||
use std::marker::PhantomData;
|
||||
use std::time::Instant;
|
||||
|
||||
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_get_user_token_by_email::{
|
||||
backfill_get_user_token_by_email, BackfillGetUserTokenByEmailArgs,
|
||||
};
|
||||
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 crate::operations::backfill_user_spend_events::stripe_api::{Invoice, StripeApi};
|
||||
use crate::operations::backfill_user_spend_events::sub_args::BackfillUserSpendEventsArgs;
|
||||
|
||||
/// Backfill `user_spend_events` from historical Stripe payments.
|
||||
///
|
||||
/// Stripe (REST) 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: &StripeApi,
|
||||
args: BackfillUserSpendEventsArgs,
|
||||
) -> AnyhowResult<()> {
|
||||
let started = Instant::now();
|
||||
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();
|
||||
// Stripe customer -> resolved user, cached so deep attribution retrieves each
|
||||
// customer at most once.
|
||||
let mut customer_cache: HashMap<String, Option<UserToken>> = HashMap::new();
|
||||
|
||||
// ---- 2. Credit-pack purchases: succeeded one-off PaymentIntents ----
|
||||
backfill_payment_intents(
|
||||
stripe, read_pool, write_pool, since, &args, &ledger_by_id, &user_by_customer,
|
||||
&mut customer_cache, &mut stats, &mut unattributed,
|
||||
)
|
||||
.await?;
|
||||
|
||||
// ---- 3. Subscription payments: paid invoices (create/cycle) ----
|
||||
backfill_invoices(
|
||||
stripe, read_pool, write_pool, since, &args, &ledger_by_id, &user_by_customer,
|
||||
&mut customer_cache, &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);
|
||||
let elapsed = started.elapsed();
|
||||
info!(" elapsed: {:.1}s ({:.0} payments/sec)",
|
||||
elapsed.as_secs_f64(),
|
||||
(stats.credit_packs + stats.subscriptions) as f64 / elapsed.as_secs_f64().max(0.001),
|
||||
);
|
||||
if args.dry_run {
|
||||
info!(" (dry run — NOTHING was written to the database)");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---- Enumerators ----
|
||||
|
||||
async fn backfill_payment_intents(
|
||||
stripe: &StripeApi,
|
||||
read_pool: &Pool<MySql>,
|
||||
write_pool: &Pool<MySql>,
|
||||
since: DateTime<Utc>,
|
||||
args: &BackfillUserSpendEventsArgs,
|
||||
ledger_by_id: &HashMap<String, LedgerPaymentRef>,
|
||||
user_by_customer: &HashMap<String, UserToken>,
|
||||
customer_cache: &mut HashMap<String, Option<UserToken>>,
|
||||
stats: &mut Stats,
|
||||
unattributed: &mut Vec<UnattributedRow>,
|
||||
) -> AnyhowResult<()> {
|
||||
info!("Enumerating Stripe PaymentIntents (credit packs) since {} …", since.format("%Y-%m-%d"));
|
||||
let mut starting_after: Option<String> = None;
|
||||
let mut processed = 0usize;
|
||||
let mut page_num = 0usize;
|
||||
|
||||
loop {
|
||||
let page = stripe.list_payment_intents(since.timestamp(), starting_after.as_deref()).await?;
|
||||
page_num += 1;
|
||||
if page.data.is_empty() {
|
||||
break;
|
||||
}
|
||||
let last_id = page.data.last().map(|pi| pi.id.clone());
|
||||
|
||||
for pi in &page.data {
|
||||
if let Some(limit) = args.limit {
|
||||
if processed >= limit {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
if pi.status != "succeeded" {
|
||||
continue;
|
||||
}
|
||||
let pi_id = pi.id.clone();
|
||||
// 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.clone();
|
||||
let charge_id = pi.latest_charge.clone();
|
||||
let occurred_at = DateTime::from_timestamp(pi.created, 0).unwrap_or_else(Utc::now);
|
||||
let mut 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,
|
||||
);
|
||||
if maybe_user.is_none() && args.deep_attribution {
|
||||
maybe_user = deep_resolve_user(stripe, read_pool, customer_cache, customer_id.as_deref(), None).await;
|
||||
}
|
||||
|
||||
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,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
info!(
|
||||
" payment_intents page {page_num}: {} fetched | packs={} skipped={} unattributed={}",
|
||||
page.data.len(), stats.credit_packs, stats.skipped, unattributed.len(),
|
||||
);
|
||||
if !page.has_more {
|
||||
break;
|
||||
}
|
||||
starting_after = last_id;
|
||||
}
|
||||
info!("PaymentIntents done: {} credit-pack event(s) across {page_num} page(s).", stats.credit_packs);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn backfill_invoices(
|
||||
stripe: &StripeApi,
|
||||
read_pool: &Pool<MySql>,
|
||||
write_pool: &Pool<MySql>,
|
||||
since: DateTime<Utc>,
|
||||
args: &BackfillUserSpendEventsArgs,
|
||||
ledger_by_id: &HashMap<String, LedgerPaymentRef>,
|
||||
user_by_customer: &HashMap<String, UserToken>,
|
||||
customer_cache: &mut HashMap<String, Option<UserToken>>,
|
||||
stats: &mut Stats,
|
||||
unattributed: &mut Vec<UnattributedRow>,
|
||||
) -> AnyhowResult<()> {
|
||||
info!("Enumerating Stripe paid invoices (subscriptions) since {} …", since.format("%Y-%m-%d"));
|
||||
let mut starting_after: Option<String> = None;
|
||||
let mut processed = 0usize;
|
||||
let mut page_num = 0usize;
|
||||
|
||||
loop {
|
||||
let page = stripe.list_paid_invoices(since.timestamp(), starting_after.as_deref()).await?;
|
||||
page_num += 1;
|
||||
if page.data.is_empty() {
|
||||
break;
|
||||
}
|
||||
let last_id = page.data.last().and_then(|inv| inv.id.clone());
|
||||
|
||||
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.as_deref() {
|
||||
Some("subscription_create") => PaymentEventType::SubscriptionInitial,
|
||||
Some("subscription_cycle") => PaymentEventType::SubscriptionRenewal,
|
||||
_ => {
|
||||
stats.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let inv_id = match inv.id.clone() {
|
||||
Some(id) => id,
|
||||
None => {
|
||||
stats.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
processed += 1;
|
||||
|
||||
let customer_id = inv.customer.clone();
|
||||
let occurred_at = DateTime::from_timestamp(inv.created, 0).unwrap_or_else(Utc::now);
|
||||
let mut maybe_user = resolve_user(
|
||||
invoice_metadata_user_token(inv),
|
||||
&inv_id, customer_id.as_deref(), ledger_by_id, user_by_customer,
|
||||
);
|
||||
if maybe_user.is_none() && args.deep_attribution {
|
||||
maybe_user = deep_resolve_user(
|
||||
stripe, read_pool, customer_cache, customer_id.as_deref(), inv.customer_email.as_deref(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
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,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
info!(
|
||||
" invoices page {page_num}: {} fetched | subscriptions={} skipped={} unattributed={}",
|
||||
page.data.len(), stats.subscriptions, stats.skipped, unattributed.len(),
|
||||
);
|
||||
if !page.has_more {
|
||||
break;
|
||||
}
|
||||
starting_after = last_id;
|
||||
}
|
||||
info!("Invoices done: {} subscription event(s) across {page_num} page(s).", stats.subscriptions);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---- Helpers ----
|
||||
|
||||
/// 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_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(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) {
|
||||
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
|
||||
}
|
||||
|
||||
/// 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(v) = inv.metadata.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(v) = sd.metadata.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
|
||||
}
|
||||
|
||||
/// Deep attribution (opt-in): when the cheap path fails, interrogate Stripe.
|
||||
/// (1) invoice email -> users; (2) retrieve the Stripe Customer and use its
|
||||
/// `user_token` metadata, then its email -> users. Customer retrieval is cached.
|
||||
async fn deep_resolve_user(
|
||||
stripe: &StripeApi,
|
||||
read_pool: &Pool<MySql>,
|
||||
customer_cache: &mut HashMap<String, Option<UserToken>>,
|
||||
customer_id: Option<&str>,
|
||||
invoice_email: Option<&str>,
|
||||
) -> Option<UserToken> {
|
||||
if let Some(email) = invoice_email {
|
||||
if let Some(user) = lookup_user_token_by_email(read_pool, email).await {
|
||||
return Some(user);
|
||||
}
|
||||
}
|
||||
|
||||
let customer_id = customer_id?;
|
||||
if let Some(cached) = customer_cache.get(customer_id) {
|
||||
return cached.clone();
|
||||
}
|
||||
|
||||
let mut resolved: Option<UserToken> = None;
|
||||
match stripe.retrieve_customer(customer_id).await {
|
||||
Ok(customer) if !customer.deleted => {
|
||||
if let Some(user_token) = customer.metadata.get("user_token") {
|
||||
if !user_token.is_empty() {
|
||||
resolved = Some(UserToken(user_token.clone()));
|
||||
}
|
||||
}
|
||||
if resolved.is_none() {
|
||||
if let Some(email) = &customer.email {
|
||||
resolved = lookup_user_token_by_email(read_pool, email).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(_deleted) => {}
|
||||
Err(err) => warn!("deep attribution: retrieve customer {customer_id} failed: {err:?}"),
|
||||
}
|
||||
|
||||
customer_cache.insert(customer_id.to_string(), resolved.clone());
|
||||
resolved
|
||||
}
|
||||
|
||||
async fn lookup_user_token_by_email(read_pool: &Pool<MySql>, email: &str) -> Option<UserToken> {
|
||||
match backfill_get_user_token_by_email(BackfillGetUserTokenByEmailArgs {
|
||||
email_address: email,
|
||||
mysql_executor: read_pool,
|
||||
phantom: PhantomData,
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(user) => user,
|
||||
Err(err) => {
|
||||
warn!("deep attribution: email lookup for {email:?} failed: {err:?}");
|
||||
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>,
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
pub mod backfill_user_spend_events;
|
||||
pub mod stripe_api;
|
||||
pub mod sub_args;
|
||||
+206
@@ -0,0 +1,206 @@
|
||||
use std::collections::HashMap;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::anyhow;
|
||||
use errors::AnyhowResult;
|
||||
use reqwest::Client;
|
||||
use serde::de::DeserializeOwned;
|
||||
use serde_derive::Deserialize;
|
||||
|
||||
const STRIPE_API_BASE: &str = "https://api.stripe.com/v1";
|
||||
const PAGE_LIMIT: &str = "100";
|
||||
const MAX_ERROR_BODY_CHARS: usize = 400;
|
||||
|
||||
/// Minimal Stripe REST client for the backfill.
|
||||
///
|
||||
/// It deliberately does NOT use the typed `async-stripe` SDK: that SDK
|
||||
/// deserializes the entire response with `miniserde` into fully-typed structs,
|
||||
/// so a single object anywhere in a page that has an unexpected shape fails the
|
||||
/// whole request ("error deserializing request data"). A production account has
|
||||
/// far too much variety for that to be reliable. Instead we parse with `serde`
|
||||
/// into small structs holding ONLY the fields we need; unknown fields are
|
||||
/// ignored, so it tolerates whatever else Stripe returns.
|
||||
pub struct StripeApi {
|
||||
http: Client,
|
||||
secret_key: String,
|
||||
}
|
||||
|
||||
/// A Stripe `list` envelope.
|
||||
#[derive(Deserialize)]
|
||||
pub struct StripeList<T> {
|
||||
#[serde(default = "Vec::new")]
|
||||
pub data: Vec<T>,
|
||||
#[serde(default)]
|
||||
pub has_more: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct PaymentIntent {
|
||||
pub id: String,
|
||||
#[serde(default)]
|
||||
pub status: String,
|
||||
#[serde(default)]
|
||||
pub amount_received: i64,
|
||||
#[serde(default)]
|
||||
pub created: i64,
|
||||
#[serde(default)]
|
||||
pub livemode: bool,
|
||||
/// Unexpanded: the customer id string (or null).
|
||||
#[serde(default)]
|
||||
pub customer: Option<String>,
|
||||
/// Unexpanded: the charge id string (or null).
|
||||
#[serde(default)]
|
||||
pub latest_charge: Option<String>,
|
||||
#[serde(default)]
|
||||
pub metadata: HashMap<String, String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct Invoice {
|
||||
#[serde(default)]
|
||||
pub id: Option<String>,
|
||||
#[serde(default)]
|
||||
pub amount_paid: i64,
|
||||
#[serde(default)]
|
||||
pub created: i64,
|
||||
#[serde(default)]
|
||||
pub livemode: bool,
|
||||
/// e.g. "subscription_create", "subscription_cycle".
|
||||
#[serde(default)]
|
||||
pub billing_reason: Option<String>,
|
||||
#[serde(default)]
|
||||
pub customer: Option<String>,
|
||||
#[serde(default)]
|
||||
pub customer_email: Option<String>,
|
||||
#[serde(default)]
|
||||
pub metadata: HashMap<String, String>,
|
||||
#[serde(default)]
|
||||
pub lines: InvoiceLines,
|
||||
#[serde(default)]
|
||||
pub parent: Option<InvoiceParent>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Default)]
|
||||
pub struct InvoiceLines {
|
||||
#[serde(default)]
|
||||
pub data: Vec<InvoiceLine>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct InvoiceLine {
|
||||
#[serde(default)]
|
||||
pub metadata: HashMap<String, String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct InvoiceParent {
|
||||
#[serde(default)]
|
||||
pub subscription_details: Option<SubscriptionDetails>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct SubscriptionDetails {
|
||||
#[serde(default)]
|
||||
pub metadata: HashMap<String, String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct Customer {
|
||||
#[serde(default)]
|
||||
pub email: Option<String>,
|
||||
#[serde(default)]
|
||||
pub metadata: HashMap<String, String>,
|
||||
/// Present and `true` for deleted customers.
|
||||
#[serde(default)]
|
||||
pub deleted: bool,
|
||||
}
|
||||
|
||||
impl StripeApi {
|
||||
pub fn new(secret_key: String) -> AnyhowResult<Self> {
|
||||
let http = Client::builder()
|
||||
.connect_timeout(Duration::from_secs(15))
|
||||
.timeout(Duration::from_secs(30))
|
||||
.build()
|
||||
.map_err(|err| anyhow!("building reqwest client: {err}"))?;
|
||||
Ok(Self { http, secret_key })
|
||||
}
|
||||
|
||||
/// One page of PaymentIntents created at/after `created_gte` (unix seconds).
|
||||
pub async fn list_payment_intents(
|
||||
&self,
|
||||
created_gte: i64,
|
||||
starting_after: Option<&str>,
|
||||
) -> AnyhowResult<StripeList<PaymentIntent>> {
|
||||
let created_gte = created_gte.to_string();
|
||||
let mut query: Vec<(&str, &str)> =
|
||||
vec![("created[gte]", &created_gte), ("limit", PAGE_LIMIT)];
|
||||
if let Some(after) = starting_after {
|
||||
query.push(("starting_after", after));
|
||||
}
|
||||
self.get_list("/payment_intents", &query).await
|
||||
}
|
||||
|
||||
/// One page of PAID invoices created at/after `created_gte` (unix seconds).
|
||||
pub async fn list_paid_invoices(
|
||||
&self,
|
||||
created_gte: i64,
|
||||
starting_after: Option<&str>,
|
||||
) -> AnyhowResult<StripeList<Invoice>> {
|
||||
let created_gte = created_gte.to_string();
|
||||
let mut query: Vec<(&str, &str)> =
|
||||
vec![("created[gte]", &created_gte), ("status", "paid"), ("limit", PAGE_LIMIT)];
|
||||
if let Some(after) = starting_after {
|
||||
query.push(("starting_after", after));
|
||||
}
|
||||
self.get_list("/invoices", &query).await
|
||||
}
|
||||
|
||||
pub async fn retrieve_customer(&self, customer_id: &str) -> AnyhowResult<Customer> {
|
||||
let url = format!("{STRIPE_API_BASE}/customers/{customer_id}");
|
||||
let response = self
|
||||
.http
|
||||
.get(&url)
|
||||
.bearer_auth(&self.secret_key)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| anyhow!("Stripe GET {url}: {err}"))?;
|
||||
let status = response.status();
|
||||
let body = response.text().await.map_err(|err| anyhow!("Stripe read body {url}: {err}"))?;
|
||||
if !status.is_success() {
|
||||
return Err(anyhow!("Stripe {status} retrieving customer {customer_id}: {}", truncate(&body)));
|
||||
}
|
||||
serde_json::from_str::<Customer>(&body).map_err(|err| anyhow!("parsing customer {customer_id}: {err}"))
|
||||
}
|
||||
|
||||
async fn get_list<T: DeserializeOwned>(
|
||||
&self,
|
||||
path: &str,
|
||||
query: &[(&str, &str)],
|
||||
) -> AnyhowResult<StripeList<T>> {
|
||||
let url = format!("{STRIPE_API_BASE}{path}");
|
||||
let response = self
|
||||
.http
|
||||
.get(&url)
|
||||
.query(query)
|
||||
.bearer_auth(&self.secret_key)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| anyhow!("Stripe GET {url}: {err}"))?;
|
||||
let status = response.status();
|
||||
let body = response.text().await.map_err(|err| anyhow!("Stripe read body {url}: {err}"))?;
|
||||
if !status.is_success() {
|
||||
return Err(anyhow!("Stripe {status} on {path}: {}", truncate(&body)));
|
||||
}
|
||||
serde_json::from_str::<StripeList<T>>(&body).map_err(|err| anyhow!("parsing {path} response: {err}"))
|
||||
}
|
||||
}
|
||||
|
||||
/// Truncate (on a char boundary) for safe error logging. Stripe error bodies
|
||||
/// contain messages, not secrets.
|
||||
fn truncate(s: &str) -> String {
|
||||
if s.chars().count() > MAX_ERROR_BODY_CHARS {
|
||||
format!("{}…", s.chars().take(MAX_ERROR_BODY_CHARS).collect::<String>())
|
||||
} else {
|
||||
s.to_string()
|
||||
}
|
||||
}
|
||||
+43
@@ -0,0 +1,43 @@
|
||||
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>,
|
||||
|
||||
/// For payments that don't resolve to a user via metadata / ledger / customer
|
||||
/// links, additionally interrogate Stripe: retrieve the Customer and use its
|
||||
/// `user_token` metadata, then fall back to matching the customer/invoice
|
||||
/// email against our users table. Costs extra Stripe API calls (cached per
|
||||
/// customer), so it's opt-in.
|
||||
#[arg(long, default_value_t = false)]
|
||||
pub deep_attribution: bool,
|
||||
}
|
||||
|
||||
/// 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))
|
||||
}
|
||||
@@ -1,2 +1,3 @@
|
||||
pub mod backfill_user_spend_events;
|
||||
pub mod calculate_legacy_tts_results_usages;
|
||||
pub mod calculate_model_weights_usages;
|
||||
@@ -42,6 +42,7 @@ pub mod uploaded_videos;
|
||||
pub mod user_impersonation_requests;
|
||||
pub mod user_referral_codes;
|
||||
pub mod user_referrals;
|
||||
pub mod user_spend_events;
|
||||
pub mod users;
|
||||
pub mod vocoder;
|
||||
pub mod voice_clone_requests;
|
||||
|
||||
+39
@@ -0,0 +1,39 @@
|
||||
use std::marker::PhantomData;
|
||||
|
||||
use sqlx::{Executor, MySql};
|
||||
|
||||
use tokens::tokens::users::UserToken;
|
||||
|
||||
/// Look up a user token by exact email address — the last-resort deep-attribution
|
||||
/// fallback used by the spend-events backfill when a Stripe payment can't be
|
||||
/// linked any other way (Stripe metadata / wallet ledger / customer links).
|
||||
/// `email_address` is a unique key, so at most one row matches.
|
||||
pub struct BackfillGetUserTokenByEmailArgs<'e, 'c, E>
|
||||
where
|
||||
E: 'e + Executor<'c, Database = MySql>,
|
||||
{
|
||||
pub email_address: &'e str,
|
||||
pub mysql_executor: E,
|
||||
pub phantom: PhantomData<&'c E>,
|
||||
}
|
||||
|
||||
pub async fn backfill_get_user_token_by_email<'e, 'c: 'e, E>(
|
||||
args: BackfillGetUserTokenByEmailArgs<'e, 'c, E>,
|
||||
) -> Result<Option<UserToken>, sqlx::Error>
|
||||
where
|
||||
E: 'e + Executor<'c, Database = MySql>,
|
||||
{
|
||||
let row = sqlx::query!(
|
||||
r#"
|
||||
SELECT token AS `token: UserToken`
|
||||
FROM users
|
||||
WHERE email_address = ?
|
||||
LIMIT 1
|
||||
"#,
|
||||
args.email_address,
|
||||
)
|
||||
.fetch_optional(args.mysql_executor)
|
||||
.await?;
|
||||
|
||||
Ok(row.map(|r| r.token))
|
||||
}
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
use std::marker::PhantomData;
|
||||
|
||||
use sqlx::{Executor, MySql};
|
||||
|
||||
use tokens::tokens::users::UserToken;
|
||||
|
||||
/// 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,
|
||||
}
|
||||
|
||||
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.
|
||||
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_as!(
|
||||
ArtcraftCustomerLink,
|
||||
r#"
|
||||
SELECT
|
||||
stripe_customer_id AS `stripe_customer_id!`,
|
||||
user_token AS `user_token!: UserToken`
|
||||
FROM user_stripe_customer_links
|
||||
WHERE payments_namespace = 'artcraft'
|
||||
"#,
|
||||
)
|
||||
.fetch_all(args.mysql_executor)
|
||||
.await?;
|
||||
|
||||
Ok(rows)
|
||||
}
|
||||
+78
@@ -0,0 +1,78 @@
|
||||
use std::marker::PhantomData;
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
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 (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_`).
|
||||
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: i32,
|
||||
|
||||
/// 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.
|
||||
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_as!(
|
||||
LedgerPaymentRef,
|
||||
r#"
|
||||
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'
|
||||
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\_%')
|
||||
"#,
|
||||
args.since,
|
||||
)
|
||||
.fetch_all(args.mysql_executor)
|
||||
.await?;
|
||||
|
||||
Ok(rows)
|
||||
}
|
||||
+132
@@ -0,0 +1,132 @@
|
||||
use std::marker::PhantomData;
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use sqlx::{Executor, MySql};
|
||||
|
||||
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 tokens::tokens::user_spend_events::UserSpendEventToken;
|
||||
use tokens::tokens::user_subscriptions::UserSubscriptionToken;
|
||||
use tokens::tokens::users::UserToken;
|
||||
use tokens::tokens::wallet_ledger_entries::WalletLedgerEntryToken;
|
||||
|
||||
/// Arguments for [`insert_user_spend_event`].
|
||||
///
|
||||
/// Pass an already-open transaction or connection as `mysql_executor` (e.g.
|
||||
/// `&mut *transaction`) so the spend-event row lands in the SAME transaction as
|
||||
/// the wallet credit it accompanies. The function mints the
|
||||
/// [`UserSpendEventToken`] itself.
|
||||
pub struct InsertUserSpendEventArgs<'e, 'c, E>
|
||||
where
|
||||
E: 'e + Executor<'c, Database = MySql>,
|
||||
{
|
||||
pub payments_namespace: PaymentsNamespace,
|
||||
|
||||
/// Who paid. `None` only when we genuinely can't attribute the money yet.
|
||||
pub maybe_user_token: Option<&'e UserToken>,
|
||||
|
||||
pub event_type: PaymentEventType,
|
||||
|
||||
/// Signed USD cents: positive for purchases, negative for refunds/chargebacks,
|
||||
/// zero for non-revenue events (e.g. annual monthly refills).
|
||||
pub amount_usd_cents: i64,
|
||||
|
||||
/// Credits this payment granted, if any (matches the wallet credit).
|
||||
pub maybe_credits_granted: Option<u32>,
|
||||
|
||||
/// The subscription this payment belongs to, if any.
|
||||
pub maybe_user_subscription_token: Option<&'e UserSubscriptionToken>,
|
||||
|
||||
/// The wallet ledger entry this payment produced, if any. Ties a dollar event
|
||||
/// to its credit effect.
|
||||
pub maybe_wallet_ledger_entry_token: Option<&'e WalletLedgerEntryToken>,
|
||||
|
||||
pub payment_source: PaymentSource,
|
||||
|
||||
/// The dedup anchor: the Stripe object id (invoice / payment_intent / refund).
|
||||
/// For Stripe payments this is ALWAYS set — it's what makes a replayed webhook
|
||||
/// idempotent (see `uq_source_object` and the `ON DUPLICATE KEY UPDATE` below).
|
||||
/// `None` is only for manual/sourceless rows, which intentionally never dedup.
|
||||
pub maybe_source_object_id: Option<&'e str>,
|
||||
|
||||
pub maybe_stripe_customer_id: Option<&'e str>,
|
||||
pub maybe_stripe_invoice_id: Option<&'e str>,
|
||||
pub maybe_stripe_payment_intent_id: Option<&'e str>,
|
||||
pub maybe_stripe_charge_id: Option<&'e str>,
|
||||
pub maybe_stripe_event_id: Option<&'e str>,
|
||||
|
||||
/// Stripe livemode.
|
||||
pub is_production: bool,
|
||||
|
||||
/// When the money actually moved (Stripe settlement time), NOT ingest time.
|
||||
pub payment_occurred_at: DateTime<Utc>,
|
||||
|
||||
pub mysql_executor: E,
|
||||
pub phantom: PhantomData<&'c E>,
|
||||
}
|
||||
|
||||
/// Insert one `user_spend_events` row, idempotently.
|
||||
///
|
||||
/// A replayed Stripe webhook collides on `uq_source_object (payment_source,
|
||||
/// source_object_id)`; the `ON DUPLICATE KEY UPDATE token = token` makes that a
|
||||
/// no-op, so a double delivery neither errors nor double-counts. (Manual rows
|
||||
/// with a `NULL` source_object_id are exempt and always insert, since MySQL
|
||||
/// treats each `NULL` in a unique index as distinct.)
|
||||
///
|
||||
/// 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>
|
||||
where
|
||||
E: 'e + Executor<'c, Database = MySql>,
|
||||
{
|
||||
let token = UserSpendEventToken::generate();
|
||||
|
||||
sqlx::query!(
|
||||
r#"
|
||||
INSERT INTO user_spend_events
|
||||
SET
|
||||
token = ?,
|
||||
payments_namespace = ?,
|
||||
maybe_user_token = ?,
|
||||
event_type = ?,
|
||||
amount_usd_cents = ?,
|
||||
maybe_credits_granted = ?,
|
||||
maybe_user_subscription_token = ?,
|
||||
maybe_wallet_ledger_entry_token = ?,
|
||||
payment_source = ?,
|
||||
source_object_id = ?,
|
||||
maybe_stripe_customer_id = ?,
|
||||
maybe_stripe_invoice_id = ?,
|
||||
maybe_stripe_payment_intent_id = ?,
|
||||
maybe_stripe_charge_id = ?,
|
||||
maybe_stripe_event_id = ?,
|
||||
is_production = ?,
|
||||
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,
|
||||
)
|
||||
.execute(args.mysql_executor)
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
pub mod backfill_get_user_token_by_email;
|
||||
pub mod backfill_list_artcraft_customer_links;
|
||||
pub mod backfill_list_artcraft_ledger_payments;
|
||||
pub mod insert_user_spend_event;
|
||||
@@ -26,6 +26,7 @@ pub mod uploaded_videos;
|
||||
pub mod usages;
|
||||
pub mod user_bookmarks;
|
||||
pub mod user_ratings;
|
||||
pub mod user_spend_events;
|
||||
pub mod users;
|
||||
pub mod voice_conversion_models;
|
||||
pub mod voice_conversion_results;
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
pub mod payment_event_type;
|
||||
pub mod payment_source;
|
||||
@@ -0,0 +1,211 @@
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
#[cfg(test)]
|
||||
use strum::EnumCount;
|
||||
#[cfg(test)]
|
||||
use strum::EnumIter;
|
||||
use utoipa::ToSchema;
|
||||
|
||||
/// Used in the `user_spend_events` table in a `VARCHAR(32)` field.
|
||||
///
|
||||
/// Splits subscription revenue from credit-pack revenue, plus the
|
||||
/// money-out (refund/chargeback) and non-revenue (manual / monthly refill) cases.
|
||||
///
|
||||
/// DO NOT CHANGE VALUES WITHOUT A MIGRATION STRATEGY.
|
||||
#[cfg_attr(test, derive(EnumIter, EnumCount))]
|
||||
#[derive(Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, ToSchema)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum PaymentEventType {
|
||||
/// First paid invoice of a new subscription.
|
||||
#[serde(rename = "subscription_initial")]
|
||||
SubscriptionInitial,
|
||||
|
||||
/// Recurring paid invoice (monthly, or annual once per year).
|
||||
#[serde(rename = "subscription_renewal")]
|
||||
SubscriptionRenewal,
|
||||
|
||||
/// Proration charge from a mid-cycle upgrade (positive amount).
|
||||
#[serde(rename = "subscription_proration_upgrade")]
|
||||
SubscriptionProrationUpgrade,
|
||||
|
||||
/// Proration credit from a mid-cycle downgrade (typically a balance credit;
|
||||
/// rare as a standalone settled event). NB: "subscription_proration_downgrade"
|
||||
/// is 32 chars -- exactly the event_type VARCHAR(32) limit (zero headroom).
|
||||
#[serde(rename = "subscription_proration_downgrade")]
|
||||
SubscriptionProrationDowngrade,
|
||||
|
||||
/// One-time credit pack purchase (the ~10x-volume case).
|
||||
#[serde(rename = "credit_pack_purchase")]
|
||||
CreditPackPurchase,
|
||||
|
||||
/// Refund (amount negative).
|
||||
#[serde(rename = "refund")]
|
||||
Refund,
|
||||
|
||||
/// Dispute / chargeback (amount negative).
|
||||
#[serde(rename = "chargeback")]
|
||||
Chargeback,
|
||||
|
||||
/// Staff / console correction (payment_source = 'manual').
|
||||
#[serde(rename = "manual_adjustment")]
|
||||
ManualAdjustment,
|
||||
|
||||
/// Non-revenue event for annual-plan monthly credit refills (amount 0).
|
||||
#[serde(rename = "subscription_monthly_refill")]
|
||||
SubscriptionMonthlyRefill,
|
||||
}
|
||||
|
||||
impl_enum_display_and_debug_using_to_str!(PaymentEventType);
|
||||
impl_mysql_enum_coders!(PaymentEventType);
|
||||
impl_mysql_from_row!(PaymentEventType);
|
||||
|
||||
/// NB: Legacy API for older code.
|
||||
impl PaymentEventType {
|
||||
pub fn to_str(&self) -> &'static str {
|
||||
match self {
|
||||
Self::SubscriptionInitial => "subscription_initial",
|
||||
Self::SubscriptionRenewal => "subscription_renewal",
|
||||
Self::SubscriptionProrationUpgrade => "subscription_proration_upgrade",
|
||||
Self::SubscriptionProrationDowngrade => "subscription_proration_downgrade",
|
||||
Self::CreditPackPurchase => "credit_pack_purchase",
|
||||
Self::Refund => "refund",
|
||||
Self::Chargeback => "chargeback",
|
||||
Self::ManualAdjustment => "manual_adjustment",
|
||||
Self::SubscriptionMonthlyRefill => "subscription_monthly_refill",
|
||||
}
|
||||
}
|
||||
|
||||
pub fn from_str(value: &str) -> Result<Self, String> {
|
||||
match value {
|
||||
"subscription_initial" => Ok(Self::SubscriptionInitial),
|
||||
"subscription_renewal" => Ok(Self::SubscriptionRenewal),
|
||||
"subscription_proration_upgrade" => Ok(Self::SubscriptionProrationUpgrade),
|
||||
"subscription_proration_downgrade" => Ok(Self::SubscriptionProrationDowngrade),
|
||||
"credit_pack_purchase" => Ok(Self::CreditPackPurchase),
|
||||
"refund" => Ok(Self::Refund),
|
||||
"chargeback" => Ok(Self::Chargeback),
|
||||
"manual_adjustment" => Ok(Self::ManualAdjustment),
|
||||
"subscription_monthly_refill" => Ok(Self::SubscriptionMonthlyRefill),
|
||||
_ => Err(format!("invalid value: {:?}", value)),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn all_variants() -> BTreeSet<Self> {
|
||||
// NB: BTreeSet is sorted
|
||||
// NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc.
|
||||
BTreeSet::from([
|
||||
Self::SubscriptionInitial,
|
||||
Self::SubscriptionRenewal,
|
||||
Self::SubscriptionProrationUpgrade,
|
||||
Self::SubscriptionProrationDowngrade,
|
||||
Self::CreditPackPurchase,
|
||||
Self::Refund,
|
||||
Self::Chargeback,
|
||||
Self::ManualAdjustment,
|
||||
Self::SubscriptionMonthlyRefill,
|
||||
])
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::by_table::user_spend_events::payment_event_type::PaymentEventType;
|
||||
use crate::test_helpers::assert_serialization;
|
||||
|
||||
mod serde {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_serialization() {
|
||||
assert_serialization(PaymentEventType::SubscriptionInitial, "subscription_initial");
|
||||
assert_serialization(PaymentEventType::SubscriptionRenewal, "subscription_renewal");
|
||||
assert_serialization(PaymentEventType::SubscriptionProrationUpgrade, "subscription_proration_upgrade");
|
||||
assert_serialization(PaymentEventType::SubscriptionProrationDowngrade, "subscription_proration_downgrade");
|
||||
assert_serialization(PaymentEventType::CreditPackPurchase, "credit_pack_purchase");
|
||||
assert_serialization(PaymentEventType::Refund, "refund");
|
||||
assert_serialization(PaymentEventType::Chargeback, "chargeback");
|
||||
assert_serialization(PaymentEventType::ManualAdjustment, "manual_adjustment");
|
||||
assert_serialization(PaymentEventType::SubscriptionMonthlyRefill, "subscription_monthly_refill");
|
||||
}
|
||||
}
|
||||
|
||||
mod impl_methods {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn to_str() {
|
||||
assert_eq!(PaymentEventType::SubscriptionInitial.to_str(), "subscription_initial");
|
||||
assert_eq!(PaymentEventType::SubscriptionRenewal.to_str(), "subscription_renewal");
|
||||
assert_eq!(PaymentEventType::SubscriptionProrationUpgrade.to_str(), "subscription_proration_upgrade");
|
||||
assert_eq!(PaymentEventType::SubscriptionProrationDowngrade.to_str(), "subscription_proration_downgrade");
|
||||
assert_eq!(PaymentEventType::CreditPackPurchase.to_str(), "credit_pack_purchase");
|
||||
assert_eq!(PaymentEventType::Refund.to_str(), "refund");
|
||||
assert_eq!(PaymentEventType::Chargeback.to_str(), "chargeback");
|
||||
assert_eq!(PaymentEventType::ManualAdjustment.to_str(), "manual_adjustment");
|
||||
assert_eq!(PaymentEventType::SubscriptionMonthlyRefill.to_str(), "subscription_monthly_refill");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_str() {
|
||||
assert_eq!(PaymentEventType::from_str("subscription_initial").unwrap(), PaymentEventType::SubscriptionInitial);
|
||||
assert_eq!(PaymentEventType::from_str("subscription_renewal").unwrap(), PaymentEventType::SubscriptionRenewal);
|
||||
assert_eq!(PaymentEventType::from_str("subscription_proration_upgrade").unwrap(), PaymentEventType::SubscriptionProrationUpgrade);
|
||||
assert_eq!(PaymentEventType::from_str("subscription_proration_downgrade").unwrap(), PaymentEventType::SubscriptionProrationDowngrade);
|
||||
assert_eq!(PaymentEventType::from_str("credit_pack_purchase").unwrap(), PaymentEventType::CreditPackPurchase);
|
||||
assert_eq!(PaymentEventType::from_str("refund").unwrap(), PaymentEventType::Refund);
|
||||
assert_eq!(PaymentEventType::from_str("chargeback").unwrap(), PaymentEventType::Chargeback);
|
||||
assert_eq!(PaymentEventType::from_str("manual_adjustment").unwrap(), PaymentEventType::ManualAdjustment);
|
||||
assert_eq!(PaymentEventType::from_str("subscription_monthly_refill").unwrap(), PaymentEventType::SubscriptionMonthlyRefill);
|
||||
assert!(PaymentEventType::from_str("foo").is_err());
|
||||
}
|
||||
}
|
||||
|
||||
mod manual_variant_checks {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn all_variants() {
|
||||
let mut variants = PaymentEventType::all_variants();
|
||||
assert_eq!(variants.len(), 9);
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::SubscriptionInitial));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::SubscriptionRenewal));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::SubscriptionProrationUpgrade));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::SubscriptionProrationDowngrade));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::CreditPackPurchase));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::Refund));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::Chargeback));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::ManualAdjustment));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentEventType::SubscriptionMonthlyRefill));
|
||||
assert_eq!(variants.pop_first(), None);
|
||||
}
|
||||
}
|
||||
|
||||
mod mechanical_checks {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn variant_length() {
|
||||
use strum::IntoEnumIterator;
|
||||
assert_eq!(PaymentEventType::all_variants().len(), PaymentEventType::iter().len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn round_trip() {
|
||||
for variant in PaymentEventType::all_variants() {
|
||||
assert_eq!(variant, PaymentEventType::from_str(variant.to_str()).unwrap());
|
||||
assert_eq!(variant, PaymentEventType::from_str(&format!("{}", variant)).unwrap());
|
||||
assert_eq!(variant, PaymentEventType::from_str(&format!("{:?}", variant)).unwrap());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serialized_length_ok_for_database() {
|
||||
const MAX_LENGTH: usize = 32;
|
||||
for variant in PaymentEventType::all_variants() {
|
||||
let serialized = variant.to_str();
|
||||
assert!(serialized.len() > 0, "variant {:?} is too short", variant);
|
||||
assert!(serialized.len() <= MAX_LENGTH, "variant {:?} is too long", variant);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
#[cfg(test)]
|
||||
use strum::EnumCount;
|
||||
#[cfg(test)]
|
||||
use strum::EnumIter;
|
||||
use utoipa::ToSchema;
|
||||
|
||||
/// Used in the `user_spend_events` table in a `VARCHAR(16)` field.
|
||||
///
|
||||
/// Where a spend event came from. USD/Stripe only for the foreseeable future;
|
||||
/// `manual` covers staff/console adjustments (which carry a NULL source_object_id).
|
||||
///
|
||||
/// DO NOT CHANGE VALUES WITHOUT A MIGRATION STRATEGY.
|
||||
#[cfg_attr(test, derive(EnumIter, EnumCount))]
|
||||
#[derive(Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, ToSchema)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum PaymentSource {
|
||||
/// Stripe (the source for essentially all rows).
|
||||
#[serde(rename = "stripe")]
|
||||
Stripe,
|
||||
|
||||
/// Manual staff / console adjustment (no provider object).
|
||||
#[serde(rename = "manual")]
|
||||
Manual,
|
||||
}
|
||||
|
||||
impl_enum_display_and_debug_using_to_str!(PaymentSource);
|
||||
impl_mysql_enum_coders!(PaymentSource);
|
||||
impl_mysql_from_row!(PaymentSource);
|
||||
|
||||
/// NB: Legacy API for older code.
|
||||
impl PaymentSource {
|
||||
pub fn to_str(&self) -> &'static str {
|
||||
match self {
|
||||
Self::Stripe => "stripe",
|
||||
Self::Manual => "manual",
|
||||
}
|
||||
}
|
||||
|
||||
pub fn from_str(value: &str) -> Result<Self, String> {
|
||||
match value {
|
||||
"stripe" => Ok(Self::Stripe),
|
||||
"manual" => Ok(Self::Manual),
|
||||
_ => Err(format!("invalid value: {:?}", value)),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn all_variants() -> BTreeSet<Self> {
|
||||
// NB: BTreeSet is sorted
|
||||
// NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc.
|
||||
BTreeSet::from([
|
||||
Self::Stripe,
|
||||
Self::Manual,
|
||||
])
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::by_table::user_spend_events::payment_source::PaymentSource;
|
||||
use crate::test_helpers::assert_serialization;
|
||||
|
||||
mod serde {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_serialization() {
|
||||
assert_serialization(PaymentSource::Stripe, "stripe");
|
||||
assert_serialization(PaymentSource::Manual, "manual");
|
||||
}
|
||||
}
|
||||
|
||||
mod impl_methods {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn to_str() {
|
||||
assert_eq!(PaymentSource::Stripe.to_str(), "stripe");
|
||||
assert_eq!(PaymentSource::Manual.to_str(), "manual");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_str() {
|
||||
assert_eq!(PaymentSource::from_str("stripe").unwrap(), PaymentSource::Stripe);
|
||||
assert_eq!(PaymentSource::from_str("manual").unwrap(), PaymentSource::Manual);
|
||||
assert!(PaymentSource::from_str("foo").is_err());
|
||||
}
|
||||
}
|
||||
|
||||
mod manual_variant_checks {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn all_variants() {
|
||||
let mut variants = PaymentSource::all_variants();
|
||||
assert_eq!(variants.len(), 2);
|
||||
assert_eq!(variants.pop_first(), Some(PaymentSource::Stripe));
|
||||
assert_eq!(variants.pop_first(), Some(PaymentSource::Manual));
|
||||
assert_eq!(variants.pop_first(), None);
|
||||
}
|
||||
}
|
||||
|
||||
mod mechanical_checks {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn variant_length() {
|
||||
use strum::IntoEnumIterator;
|
||||
assert_eq!(PaymentSource::all_variants().len(), PaymentSource::iter().len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn round_trip() {
|
||||
for variant in PaymentSource::all_variants() {
|
||||
assert_eq!(variant, PaymentSource::from_str(variant.to_str()).unwrap());
|
||||
assert_eq!(variant, PaymentSource::from_str(&format!("{}", variant)).unwrap());
|
||||
assert_eq!(variant, PaymentSource::from_str(&format!("{:?}", variant)).unwrap());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serialized_length_ok_for_database() {
|
||||
const MAX_LENGTH: usize = 16;
|
||||
for variant in PaymentSource::all_variants() {
|
||||
let serialized = variant.to_str();
|
||||
assert!(serialized.len() > 0, "variant {:?} is too short", variant);
|
||||
assert!(serialized.len() <= MAX_LENGTH, "variant {:?} is too long", variant);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -53,6 +53,7 @@ pub(crate) enum TokenPrefix {
|
||||
UserBookmark,
|
||||
UserReferralCode,
|
||||
UserSession,
|
||||
UserSpendEvent,
|
||||
UserSubscription,
|
||||
VoiceConversionModel,
|
||||
VoiceConversionResult,
|
||||
@@ -145,6 +146,7 @@ impl PrefixGenerator for TokenPrefix {
|
||||
Self::UserBookmark => "ub_",
|
||||
Self::UserReferralCode => "urc_",
|
||||
Self::UserSession => "session_",
|
||||
Self::UserSpendEvent => "spend_",
|
||||
Self::UserSubscription => "subscription_", // NB: Previously "SUB:"
|
||||
Self::VoiceConversionModel => "vcm_",
|
||||
Self::VoiceConversionResult => "vcr_",
|
||||
|
||||
@@ -37,6 +37,7 @@ pub mod user_bookmarks;
|
||||
pub mod user_impersonation_requests;
|
||||
pub mod user_referral_codes;
|
||||
pub mod user_sessions;
|
||||
pub mod user_spend_events;
|
||||
pub mod user_subscriptions;
|
||||
pub mod users;
|
||||
pub mod vocoder_models;
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
use std::fmt::Debug;
|
||||
|
||||
use serde::Deserialize;
|
||||
use serde::Serialize;
|
||||
use utoipa::ToSchema;
|
||||
|
||||
use crate::prefixes::TokenPrefix;
|
||||
|
||||
/// The primary key for user_spend_events
|
||||
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Serialize, Deserialize, ToSchema)]
|
||||
#[cfg_attr(feature = "database", derive(sqlx::Type))]
|
||||
#[cfg_attr(feature = "database", sqlx(transparent))]
|
||||
pub struct UserSpendEventToken(pub String);
|
||||
|
||||
impl_string_token!(UserSpendEventToken);
|
||||
impl_mysql_token_from_row!(UserSpendEventToken);
|
||||
impl_crockford_generator!(UserSpendEventToken, 32usize, TokenPrefix::UserSpendEvent, CrockfordLower);
|
||||
+29
-1
@@ -1,6 +1,7 @@
|
||||
use crate::configs::credits_packs::stripe_artcraft_credits_pack_info::StripeArtcraftCreditsPackInfo;
|
||||
use crate::configs::subscriptions::stripe_artcraft_subscription_info::StripeArtcraftSubscriptionInfo;
|
||||
use chrono::NaiveDateTime;
|
||||
use chrono::{DateTime, NaiveDateTime, Utc};
|
||||
use enums::by_table::user_spend_events::payment_event_type::PaymentEventType;
|
||||
use reusable_types::stripe::stripe_recurring_interval::StripeRecurringInterval;
|
||||
use reusable_types::stripe::stripe_subscription_status::StripeSubscriptionStatus;
|
||||
use tokens::tokens::users::UserToken;
|
||||
@@ -50,6 +51,21 @@ pub struct WalletCreditsPurchaseEvent {
|
||||
|
||||
/// Also potentially save to `user_stripe_customer_links` table.
|
||||
pub maybe_stripe_customer_id: Option<String>,
|
||||
|
||||
/// USD cents actually collected (Stripe `payment_intent.amount_received`).
|
||||
pub amount_usd_cents: i64,
|
||||
|
||||
/// Stripe livemode.
|
||||
pub is_production: bool,
|
||||
|
||||
/// When the money moved (Stripe `payment_intent.created`).
|
||||
pub payment_occurred_at: DateTime<Utc>,
|
||||
|
||||
/// The successful charge behind the payment intent, if present.
|
||||
pub maybe_stripe_charge_id: Option<String>,
|
||||
|
||||
/// The Stripe webhook event id (`evt_...`) that produced this row.
|
||||
pub maybe_stripe_event_id: Option<String>,
|
||||
}
|
||||
|
||||
|
||||
@@ -146,4 +162,16 @@ pub struct SubscriptionPaidEvent {
|
||||
/// For eagerly created user records (in the new checkout flow), we need to update the user's email.
|
||||
/// We started creating user accounts without a username, email, and password - so we need to collect it here.
|
||||
pub customer_email: Option<String>,
|
||||
|
||||
/// USD cents actually collected (Stripe `invoice.amount_paid`).
|
||||
pub amount_usd_cents: i64,
|
||||
|
||||
/// Whether this is the first paid invoice (initial) or a renewal.
|
||||
pub payment_event_type: PaymentEventType,
|
||||
|
||||
/// When the money moved (Stripe `invoice.created`).
|
||||
pub payment_occurred_at: DateTime<Utc>,
|
||||
|
||||
/// The Stripe webhook event id (`evt_...`) that produced this row.
|
||||
pub maybe_stripe_event_id: Option<String>,
|
||||
}
|
||||
+36
-1
@@ -8,6 +8,11 @@ use mysql_queries::queries::wallets::add_durable_banked_balance_to_wallet::add_d
|
||||
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_transaction;
|
||||
use tokens::tokens::users::UserToken;
|
||||
use std::marker::PhantomData;
|
||||
use chrono::{DateTime, Utc};
|
||||
use enums::by_table::user_spend_events::payment_event_type::PaymentEventType;
|
||||
use enums::by_table::user_spend_events::payment_source::PaymentSource;
|
||||
use mysql_queries::queries::user_spend_events::insert_user_spend_event::{insert_user_spend_event, InsertUserSpendEventArgs};
|
||||
|
||||
/// Record the credits pack purchase
|
||||
pub async fn complete_credits_pack_purchase(
|
||||
@@ -16,6 +21,11 @@ pub async fn complete_credits_pack_purchase(
|
||||
quantity: u64,
|
||||
maybe_ledger_ref: Option<&str>,
|
||||
maybe_stripe_customer_id: Option<&str>,
|
||||
maybe_stripe_charge_id: Option<&str>,
|
||||
maybe_stripe_event_id: Option<&str>,
|
||||
amount_usd_cents: i64,
|
||||
is_production: bool,
|
||||
payment_occurred_at: DateTime<Utc>,
|
||||
transaction: &mut sqlx::Transaction<'_, sqlx::MySql>,
|
||||
) -> anyhow::Result<()> {
|
||||
|
||||
@@ -37,7 +47,7 @@ pub async fn complete_credits_pack_purchase(
|
||||
|
||||
info!("Adding {} credits to wallet: {}", credits_purchased, wallet_token.as_str());
|
||||
|
||||
let _result = add_durable_banked_balance_to_wallet(
|
||||
let wallet_update = add_durable_banked_balance_to_wallet(
|
||||
&wallet_token,
|
||||
credits_purchased,
|
||||
maybe_ledger_ref,
|
||||
@@ -45,6 +55,31 @@ pub async fn complete_credits_pack_purchase(
|
||||
transaction,
|
||||
).await?;
|
||||
|
||||
// Record the money movement in the spend-events ledger, in this SAME
|
||||
// transaction. Idempotent on (payment_source, source_object_id), so a replayed
|
||||
// `payment_intent.succeeded` webhook is a no-op.
|
||||
insert_user_spend_event(InsertUserSpendEventArgs {
|
||||
payments_namespace: PaymentsNamespace::Artcraft,
|
||||
maybe_user_token: Some(owner_user_token),
|
||||
event_type: PaymentEventType::CreditPackPurchase,
|
||||
amount_usd_cents,
|
||||
// NULL rather than a clamped/garbage value if it somehow doesn't fit u32.
|
||||
maybe_credits_granted: u32::try_from(credits_purchased).ok(),
|
||||
maybe_user_subscription_token: None,
|
||||
maybe_wallet_ledger_entry_token: Some(&wallet_update.wallet_ledger_entry_token),
|
||||
payment_source: PaymentSource::Stripe,
|
||||
maybe_source_object_id: maybe_ledger_ref,
|
||||
maybe_stripe_customer_id,
|
||||
maybe_stripe_invoice_id: None,
|
||||
maybe_stripe_payment_intent_id: maybe_ledger_ref,
|
||||
maybe_stripe_charge_id,
|
||||
maybe_stripe_event_id,
|
||||
is_production,
|
||||
payment_occurred_at,
|
||||
mysql_executor: &mut **transaction,
|
||||
phantom: PhantomData,
|
||||
}).await?;
|
||||
|
||||
if let Some(stripe_customer_id) = maybe_stripe_customer_id {
|
||||
optionally_link_user_to_stripe_customer(
|
||||
owner_user_token,
|
||||
|
||||
+29
-1
@@ -6,6 +6,9 @@ use mysql_queries::queries::users::user_subscriptions::upsert_user_subscription_
|
||||
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_transaction;
|
||||
use mysql_queries::queries::wallets::refill_monthly_credits_balance_on_wallet::refill_monthly_credits_balance_on_wallet;
|
||||
use mysql_queries::queries::user_spend_events::insert_user_spend_event::{insert_user_spend_event, InsertUserSpendEventArgs};
|
||||
use enums::by_table::user_spend_events::payment_source::PaymentSource;
|
||||
use std::marker::PhantomData;
|
||||
use reusable_types::stripe::stripe_subscription_status::StripeSubscriptionStatus;
|
||||
|
||||
pub async fn mark_subscription_as_paid(
|
||||
@@ -87,12 +90,37 @@ pub async fn mark_subscription_as_paid(
|
||||
|
||||
info!("Adding {} monthly credits to wallet: {}", monthly_credits , wallet_token.as_str());
|
||||
|
||||
let _result = refill_monthly_credits_balance_on_wallet(
|
||||
let wallet_update = refill_monthly_credits_balance_on_wallet(
|
||||
&wallet_token,
|
||||
monthly_credits,
|
||||
maybe_ledger_ref,
|
||||
transaction
|
||||
).await?;
|
||||
|
||||
// Record the money movement in the spend-events ledger, in this SAME
|
||||
// transaction. Idempotent on (payment_source, source_object_id), so a replayed
|
||||
// `invoice.paid` webhook is a no-op.
|
||||
insert_user_spend_event(InsertUserSpendEventArgs {
|
||||
payments_namespace: PaymentsNamespace::Artcraft,
|
||||
maybe_user_token: Some(&details.owner_user_token),
|
||||
event_type: details.payment_event_type,
|
||||
amount_usd_cents: details.amount_usd_cents,
|
||||
// NULL rather than a clamped/garbage value if it somehow doesn't fit u32.
|
||||
maybe_credits_granted: u32::try_from(monthly_credits).ok(),
|
||||
maybe_user_subscription_token: None,
|
||||
maybe_wallet_ledger_entry_token: Some(&wallet_update.wallet_ledger_entry_token),
|
||||
payment_source: PaymentSource::Stripe,
|
||||
maybe_source_object_id: maybe_ledger_ref,
|
||||
maybe_stripe_customer_id: Some(&details.stripe_customer_id),
|
||||
maybe_stripe_invoice_id: maybe_ledger_ref,
|
||||
maybe_stripe_payment_intent_id: None,
|
||||
maybe_stripe_charge_id: None,
|
||||
maybe_stripe_event_id: details.maybe_stripe_event_id.as_deref(),
|
||||
is_production: details.stripe_is_production,
|
||||
payment_occurred_at: details.payment_occurred_at,
|
||||
mysql_executor: &mut **transaction,
|
||||
phantom: PhantomData,
|
||||
}).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
+5
@@ -26,6 +26,11 @@ pub async fn transactionally_fulfill_artcraft_billing_action(
|
||||
purchase.quantity,
|
||||
purchase.ledger_event_ref.as_deref(),
|
||||
purchase.maybe_stripe_customer_id.as_deref(),
|
||||
purchase.maybe_stripe_charge_id.as_deref(),
|
||||
purchase.maybe_stripe_event_id.as_deref(),
|
||||
purchase.amount_usd_cents,
|
||||
purchase.is_production,
|
||||
purchase.payment_occurred_at,
|
||||
transaction,
|
||||
).await?;
|
||||
}
|
||||
|
||||
+1
@@ -51,6 +51,7 @@ pub async fn handle_webhook_event_enrichment(
|
||||
info!("Event {}, data: {:?}", stripe_event_descriptor, payment_intent);
|
||||
|
||||
return payment_intent_succeeded_extractor(
|
||||
&stripe_event_descriptor,
|
||||
&payment_intent,
|
||||
server_environment,
|
||||
stripe_client,
|
||||
|
||||
+13
@@ -9,6 +9,8 @@ use crate::utils::expand_ids::expand_customer_id::expand_customer_id;
|
||||
use crate::utils::expand_ids::expand_subscription_id::expand_subscription_id;
|
||||
use crate::utils::metadata::get_metadata_user_token::get_metadata_user_token;
|
||||
use crate::utils::stripe_event_descriptor::StripeEventDescriptor;
|
||||
use chrono::{DateTime, Utc};
|
||||
use enums::by_table::user_spend_events::payment_event_type::PaymentEventType;
|
||||
use log::{error, info, warn};
|
||||
use server_environment::ServerEnvironment;
|
||||
use stripe::Client;
|
||||
@@ -134,6 +136,13 @@ pub async fn invoice_paid_extractor(
|
||||
.map(|id| id.to_string())
|
||||
.unwrap_or_else(|| stripe_event_descriptor.stripe_event_id.clone());
|
||||
|
||||
// The first paid invoice of a new subscription is the "initial"; everything
|
||||
// else that reaches here (subscription_cycle) is a renewal.
|
||||
let payment_event_type = match invoice.billing_reason {
|
||||
Some(InvoiceBillingReason::SubscriptionCreate) => PaymentEventType::SubscriptionInitial,
|
||||
_ => PaymentEventType::SubscriptionRenewal,
|
||||
};
|
||||
|
||||
info!("{} : invoice paid is for subscription.", stripe_event_descriptor);
|
||||
|
||||
|
||||
@@ -157,6 +166,10 @@ pub async fn invoice_paid_extractor(
|
||||
maybe_canceled_at: subscription.maybe_canceled_at,
|
||||
ledger_event_ref: Some(ledger_event_ref),
|
||||
customer_email: invoice.customer_email.clone(),
|
||||
amount_usd_cents: invoice.amount_paid,
|
||||
payment_event_type,
|
||||
payment_occurred_at: DateTime::from_timestamp(invoice.created, 0).unwrap_or_else(Utc::now),
|
||||
maybe_stripe_event_id: Some(stripe_event_descriptor.stripe_event_id.clone()),
|
||||
})),
|
||||
webhook_event_log_summary: event_log_summary,
|
||||
})
|
||||
|
||||
+8
@@ -9,6 +9,8 @@ use crate::stripe_requests::stripe_lookup_purchase_from_payment_intent_success::
|
||||
use crate::utils::expand_ids::expand_customer_id::expand_customer_id;
|
||||
use crate::utils::metadata::get_metadata_user_token::get_metadata_user_token;
|
||||
use crate::utils::metadata::get_metadata_wallet_token::get_metadata_wallet_token;
|
||||
use crate::utils::stripe_event_descriptor::StripeEventDescriptor;
|
||||
use chrono::{DateTime, Utc};
|
||||
use log::{error, info, warn};
|
||||
use server_environment::ServerEnvironment;
|
||||
use sqlx::{MySql, Transaction};
|
||||
@@ -20,6 +22,7 @@ use tokens::tokens::wallets::WalletToken;
|
||||
|
||||
// Handle event type: 'payment_intent.succeeded'
|
||||
pub async fn payment_intent_succeeded_extractor(
|
||||
stripe_event_descriptor: &StripeEventDescriptor,
|
||||
payment_intent: &PaymentIntent,
|
||||
server_environment: ServerEnvironment,
|
||||
stripe_client: &Client,
|
||||
@@ -112,6 +115,11 @@ pub async fn payment_intent_succeeded_extractor(
|
||||
quantity: purchase.quantity,
|
||||
ledger_event_ref: Some(payment_intent_id),
|
||||
maybe_stripe_customer_id,
|
||||
amount_usd_cents: payment_intent.amount_received,
|
||||
is_production: payment_intent.livemode,
|
||||
payment_occurred_at: DateTime::from_timestamp(payment_intent.created, 0).unwrap_or_else(Utc::now),
|
||||
maybe_stripe_charge_id: payment_intent.latest_charge.as_ref().map(|charge| charge.id().to_string()),
|
||||
maybe_stripe_event_id: Some(stripe_event_descriptor.stripe_event_id.clone()),
|
||||
})),
|
||||
webhook_event_log_summary: event_log_summary,
|
||||
})
|
||||
|
||||
Reference in new issue
Block a user