Skip to main content

max / makenotwork

7.4 KB · 196 lines History Blame Raw
1 //! Settlement of platform-funded credits (Fan+ renewal credit reimbursement).
2 //!
3 //! A platform-wide credit reduces what a buyer pays on a creator's Direct Charge;
4 //! MNW owes the creator the discounted amount so they still net the full price
5 //! ("0% platform fee, creators keep everything"). The obligation lives on the
6 //! `transactions` row (`platform_credit_cents`); the scheduler sweep claims it,
7 //! creates a platform -> connected transfer with a deterministic idempotency key,
8 //! then stamps `platform_credit_settled_at`.
9 //!
10 //! Lifecycle mirrors [`super::pending_refunds`]: claim (set `claimed_at`) -> settle
11 //! (set `settled_at`). A graceful transfer failure clears `claimed_at` for retry; a
12 //! process death between claim and settle leaves the row claimed-but-unsettled and
13 //! is surfaced by [`get_stale_credits`] for human reconciliation rather than blindly
14 //! retried.
15
16 use sqlx::PgPool;
17
18 use super::validated_types::Cents;
19 use super::{TransactionId, UserId};
20 use crate::error::Result;
21
22 /// A completed transaction owing a platform-funded credit to its seller.
23 #[derive(Debug, sqlx::FromRow)]
24 pub struct PlatformCredit {
25 pub transaction_id: TransactionId,
26 pub seller_id: UserId,
27 pub amount_cents: Cents,
28 /// The sale's currency. The reimbursing transfer is denominated in it, so
29 /// the creator is made whole in the currency the sale was priced in rather
30 /// than handed a number that only matches in USD.
31 pub currency: crate::currency::SettlementCurrency,
32 }
33
34 /// Atomically claim the oldest unsettled platform credit (set `claimed_at`), so a
35 /// single scheduler worker settles it once. Returns `None` when nothing is owed.
36 /// Settlement is recorded separately by [`mark_settled`] only after the transfer
37 /// succeeds, a claim that never settles (process killed mid-transfer) is escalated
38 /// by [`get_stale_credits`], not auto-retried.
39 pub async fn claim_unsettled_credit(pool: &PgPool) -> Result<Option<PlatformCredit>> {
40 let row = sqlx::query_as!(
41 PlatformCredit,
42 r#"
43 UPDATE transactions
44 SET platform_credit_claimed_at = NOW()
45 WHERE id = (
46 SELECT id FROM transactions
47 WHERE status = 'completed'
48 AND platform_credit_cents > 0
49 AND platform_credit_settled_at IS NULL
50 AND platform_credit_claimed_at IS NULL
51 ORDER BY completed_at
52 LIMIT 1
53 FOR UPDATE SKIP LOCKED
54 )
55 RETURNING id AS "transaction_id!: TransactionId",
56 seller_id AS "seller_id!: UserId",
57 platform_credit_cents AS "amount_cents!: Cents",
58 currency AS "currency!: crate::currency::SettlementCurrency"
59 "#,
60 )
61 .fetch_optional(pool)
62 .await?;
63
64 Ok(row)
65 }
66
67 /// Record that a claimed credit's transfer succeeded, storing the Stripe
68 /// transfer id so the credit can be reversed if the sale is later refunded.
69 /// Idempotent. Runtime-checked query (the `platform_credit_transfer_id` column
70 /// is newer than the offline sqlx cache), per [`get_stale_credits`].
71 pub async fn mark_settled(
72 pool: &PgPool,
73 transaction_id: TransactionId,
74 transfer_id: &str,
75 ) -> Result<()> {
76 sqlx::query(
77 "UPDATE transactions SET platform_credit_settled_at = NOW(), platform_credit_transfer_id = $2 WHERE id = $1",
78 )
79 .bind(transaction_id)
80 .bind(transfer_id)
81 .execute(pool)
82 .await?;
83 Ok(())
84 }
85
86 /// A settled platform credit whose sale was refunded and whose MNW -> creator
87 /// transfer must be reversed to claw the reimbursement back.
88 #[derive(Debug, sqlx::FromRow)]
89 pub struct ReversibleCredit {
90 pub transaction_id: TransactionId,
91 pub amount_cents: Cents,
92 pub transfer_id: String,
93 }
94
95 /// Up to `limit` settled platform credits sitting on refunded transactions that
96 /// haven't been reversed yet, oldest first. The reversal uses a deterministic
97 /// idempotency key, so the single-instance scheduler can process these
98 /// sequentially and safely. Runtime-checked query (new columns).
99 pub async fn get_reversible_credits(pool: &PgPool, limit: i64) -> Result<Vec<ReversibleCredit>> {
100 let rows = sqlx::query_as::<_, ReversibleCredit>(
101 r"
102 SELECT id AS transaction_id,
103 platform_credit_cents AS amount_cents,
104 platform_credit_transfer_id AS transfer_id
105 FROM transactions
106 WHERE status = 'refunded'
107 AND platform_credit_cents > 0
108 AND platform_credit_settled_at IS NOT NULL
109 AND platform_credit_transfer_id IS NOT NULL
110 AND platform_credit_reversed_at IS NULL
111 ORDER BY completed_at
112 LIMIT $1
113 ",
114 )
115 .bind(limit)
116 .fetch_all(pool)
117 .await?;
118
119 Ok(rows)
120 }
121
122 /// Mark a platform credit reversed (funds clawed back). Idempotent.
123 /// Runtime-checked query (new column).
124 pub async fn mark_reversed(pool: &PgPool, transaction_id: TransactionId) -> Result<()> {
125 sqlx::query("UPDATE transactions SET platform_credit_reversed_at = NOW() WHERE id = $1")
126 .bind(transaction_id)
127 .execute(pool)
128 .await?;
129 Ok(())
130 }
131
132 /// Release a claimed-but-unsettled credit back to the queue after a *graceful*
133 /// transfer failure (transient error where nothing external happened). A
134 /// non-graceful failure (process killed) cannot reach here; that row stays
135 /// claimed-but-unsettled and is escalated by the stale sweep. Idempotent.
136 pub async fn unclaim_credit(pool: &PgPool, transaction_id: TransactionId) -> Result<()> {
137 sqlx::query!(
138 "UPDATE transactions SET platform_credit_claimed_at = NULL WHERE id = $1",
139 transaction_id as TransactionId,
140 )
141 .execute(pool)
142 .await?;
143 Ok(())
144 }
145
146 /// Per-tick cap on the stale-credit escalation sweep (mirrors `STALE_REFUND_BATCH`).
147 pub const STALE_CREDIT_BATCH: i64 = 100;
148
149 /// A stale platform credit needing human reconciliation.
150 #[derive(Debug, sqlx::FromRow)]
151 pub struct StaleCredit {
152 pub transaction_id: TransactionId,
153 pub seller_id: UserId,
154 pub amount_cents: Cents,
155 pub completed_at: Option<chrono::DateTime<chrono::Utc>>,
156 }
157
158 /// Up to [`STALE_CREDIT_BATCH`] credits claimed more than `age` ago that never
159 /// settled (the crash-window between claim and transfer), oldest first. The transfer
160 /// uses a deterministic idempotency key so a human can safely re-trigger it.
161 pub async fn get_stale_credits(pool: &PgPool, age: chrono::Duration) -> Result<Vec<StaleCredit>> {
162 let cutoff = chrono::Utc::now() - age;
163 // runtime-checked bind of a chrono cutoff, per db::pending_refunds::get_stale_refunds.
164 let rows = sqlx::query_as::<_, StaleCredit>(
165 r"
166 SELECT id AS transaction_id, seller_id, platform_credit_cents AS amount_cents, completed_at
167 FROM transactions
168 WHERE status = 'completed'
169 AND platform_credit_cents > 0
170 AND platform_credit_settled_at IS NULL
171 AND platform_credit_claimed_at IS NOT NULL
172 AND platform_credit_claimed_at < $1
173 AND platform_credit_escalated_at IS NULL
174 ORDER BY completed_at
175 LIMIT $2
176 ",
177 )
178 .bind(cutoff)
179 .bind(STALE_CREDIT_BATCH)
180 .fetch_all(pool)
181 .await?;
182
183 Ok(rows)
184 }
185
186 /// Mark a stale credit escalated (alert sent, won't be re-alerted).
187 pub async fn mark_escalated(pool: &PgPool, transaction_id: TransactionId) -> Result<()> {
188 sqlx::query!(
189 "UPDATE transactions SET platform_credit_escalated_at = NOW() WHERE id = $1",
190 transaction_id as TransactionId,
191 )
192 .execute(pool)
193 .await?;
194 Ok(())
195 }
196