max / makenotwork

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