max / makenotwork

Split purchases.rs by transaction lifecycle 1408 lines and 39 functions across four lifecycles that share a table and little else: the pending-to-completed checkout path, free claims, guest purchases, and refunds, with the read side under all of them. The module doc stays at the top of the tree: it documents the partial unique indexes every sibling's `ON CONFLICT` clause names, and changing one of those means revisiting every claim and checkout path here. No projection macro. The shared column list appears 12 times, but sqlx parses the SQL argument of `query_as!` as a literal string with no macro expansion, so deduplicating it means giving up compile-time checking on money queries. A 13th copy lives in revenue_stats.rs, outside any home a macro here could have. The two `pub(crate)` refund functions keep that visibility. The reason is in their doc: the payment-intent-wide UPDATE would refund a whole cart from one line's event, so only in-crate webhook handlers may call it. Also drops a doc block above `get_seller_transactions_for_export_page` that was superseded when a second one was inserted below it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EEmeiSJnmyL98QzA5Dwsvz

Author: Max Johnson <me@maxj.phd> · 2026-09-05 02:44 UTC

Signed with PGP, not checked

Commit: d6a9bb5deb4bfdc5cd8340494a8399a538caee05

Parent: 2a58d2b

7 files changed, +1450 insertions, -500 deletions

D server/src/db/transactions/purchases.rs -500

		@@ -1,1408 +1,0 @@
1	-	//! Purchase queries: the pending-transaction lifecycle, free claims, refunds
2	-	//! and cart settlement.
3	-	//!
4	-	//! A paid checkout writes a `pending` row before the buyer leaves for Stripe,
5	-	//! and completion flips it to `completed`; nothing else moves a row between
6	-	//! those states. A pending row that is never completed is deleted by the
7	-	//! caller who abandoned it, or by `cleanup_stale_pending` past the age the
8	-	//! scheduler passes it (25h), which returns any `promo_code_id` so the
9	-	//! reservation is released with the row.
10	-	//!
11	-	//! The dedup these queries lean on is a set of partial unique indexes on
12	-	//! `transactions`, one pair per subject:
13	-	//! `(buyer_id, item_id)` and `(buyer_id, project_id)`, each once for
14	-	//! `status = 'pending'` and once for `status = 'completed'`, and
15	-	//! `(guest_email, item_id)` for `status = 'completed'`. All are partial on
16	-	//! the id being NOT NULL, so a project purchase (NULL `item_id`) does not
17	-	//! collide with an item purchase. `ON CONFLICT DO NOTHING` and the 23505
18	-	//! backstops throughout this file name those indexes; changing one means
19	-	//! revisiting every claim and checkout path here.
20	-
21	-	use super::{
22	-	Cents, ClaimToken, DbPurchaseRow, DbTransaction, DbTransactionExportRow, DownloadToken, ItemId,
23	-	KeyCode, PgPool, ProjectId, PromoCodeId, Result, TransactionId, UserId,
24	-	};
25	-
26	-	/// Parameters for creating a pending Stripe checkout transaction.
27	-	pub struct CreateTransactionParams<'a> {
28	-	pub buyer_id: Option<UserId>,
29	-	pub seller_id: UserId,
30	-	/// `None` for project-level purchases (no specific item).
31	-	pub item_id: Option<ItemId>,
32	-	pub amount_cents: Cents,
33	-	pub platform_fee_cents: Cents,
34	-	pub stripe_checkout_session_id: &'a str,
35	-	pub item_title: &'a str,
36	-	pub seller_username: &'a str,
37	-	pub share_contact: bool,
38	-	/// Set for project-level purchases; `None` for item purchases.
39	-	pub project_id: Option<ProjectId>,
40	-	/// Promo code used for this checkout (for releasing reservations on cleanup).
41	-	pub promo_code_id: Option<PromoCodeId>,
42	-	/// Guest buyer's email (set for guest checkouts, None for logged-in).
43	-	pub guest_email: Option<&'a str>,
44	-	/// Cents MNW owes the seller as reimbursement for a platform-funded credit
45	-	/// (the Fan+ renewal credit) applied to this sale; `0` for ordinary sales. The
46	-	/// scheduler settles it via a platform -> connected transfer once the
47	-	/// transaction completes (see `db::platform_credits`).
48	-	pub platform_credit_cents: i64,
49	-	}
50	-
51	-	/// Common parameters for claiming a free item (direct, discount code, or download code).
52	-	pub struct ClaimParams<'a> {
53	-	pub buyer_id: UserId,
54	-	pub item_id: ItemId,
55	-	pub seller_id: UserId,
56	-	pub item_title: &'a str,
57	-	pub seller_username: &'a str,
58	-	pub share_contact: bool,
59	-	/// If this claim was granted via a bundle purchase, the parent transaction ID.
60	-	pub parent_transaction_id: Option<TransactionId>,
61	-	/// Cents MNW owes the seller when a platform-wide (Fan+) credit made this item
62	-	/// free, the seller is reimbursed the item's price via a platform transfer so
63	-	/// they are still paid. `0` for ordinary free claims (genuinely-free items,
64	-	/// seller-issued free-access codes, bundle grants).
65	-	pub platform_credit_cents: i64,
66	-	}
67	-
68	-	/// Record a new pending transaction for a Stripe checkout session.
69	-	#[tracing::instrument(skip_all)]
70	-	pub async fn create_transaction<'e>(
71	-	executor: impl sqlx::PgExecutor<'e>,
72	-	params: &CreateTransactionParams<'_>,
73	-	) -> Result<DbTransaction> {
74	-	let tx = sqlx::query_as!(
75	-	DbTransaction,
76	-	r#"
77	-	INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact, project_id, promo_code_id, guest_email, platform_credit_cents)
78	-	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
79	-	RETURNING
80	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
81	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
82	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
83	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
84	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
85	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
86	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
87	-	download_token AS "download_token: DownloadToken",
88	-	presentment_amount_cents, presentment_currency
89	-	"#,
90	-	params.buyer_id as Option<UserId>,
91	-	params.seller_id as UserId,
92	-	params.item_id as Option<ItemId>,
93	-	params.amount_cents as Cents,
94	-	params.platform_fee_cents as Cents,
95	-	params.stripe_checkout_session_id,
96	-	params.item_title,
97	-	params.seller_username,
98	-	params.share_contact,
99	-	params.project_id as Option<ProjectId>,
100	-	params.promo_code_id as Option<PromoCodeId>,
101	-	params.guest_email,
102	-	params.platform_credit_cents,
103	-	)
104	-	.fetch_one(executor)
105	-	.await?;
106	-
107	-	Ok(tx)
108	-	}
109	-
110	-	/// Complete a guest transaction: mark it completed, record the guest email, and
111	-	/// mint a `claim_token` so the buyer can later attach the purchase to an account.
112	-	///
113	-	/// Guest purchases always land unclaimed (`buyer_id` NULL); attachment to a user
114	-	/// happens out of band via [`attach_guest_purchases_by_email`] at signup/email
115	-	/// verification.
116	-	#[tracing::instrument(skip_all)]
117	-	pub async fn complete_guest_transaction<'e>(
118	-	executor: impl sqlx::PgExecutor<'e>,
119	-	stripe_checkout_session_id: &str,
120	-	stripe_payment_intent_id: Option<&str>,
121	-	guest_email: &str,
122	-	) -> Result<Option<DbTransaction>> {
123	-	let claim_token = ClaimToken::new();
124	-
125	-	let tx = sqlx::query_as!(
126	-	DbTransaction,
127	-	r#"
128	-	UPDATE transactions
129	-	SET status = 'completed',
130	-	stripe_payment_intent_id = $2,
131	-	completed_at = NOW(),
132	-	guest_email = $3,
133	-	claim_token = $4,
134	-	buyer_id = NULL
135	-	WHERE stripe_checkout_session_id = $1
136	-	AND status = 'pending'
137	-	RETURNING
138	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
139	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
140	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
141	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
142	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
143	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
144	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
145	-	download_token AS "download_token: DownloadToken",
146	-	presentment_amount_cents, presentment_currency
147	-	"#,
148	-	stripe_checkout_session_id,
149	-	stripe_payment_intent_id,
150	-	guest_email,
151	-	claim_token as ClaimToken,
152	-	)
153	-	.fetch_optional(executor)
154	-	.await?;
155	-
156	-	Ok(tx)
157	-	}
158	-
159	-	/// Attach all unclaimed guest purchases for an email to a user account.
160	-	/// Called during signup/email verification to auto-claim prior guest purchases.
161	-	#[tracing::instrument(skip_all)]
162	-	pub async fn attach_guest_purchases_by_email(
163	-	pool: &PgPool,
164	-	email: &str,
165	-	user_id: UserId,
166	-	) -> Result<u64> {
167	-	let result = sqlx::query!(
168	-	r#"
169	-	UPDATE transactions
170	-	SET buyer_id = $1, claimed_by = $1, claim_token = NULL
171	-	WHERE LOWER(guest_email) = LOWER($2)
172	-	AND buyer_id IS NULL
173	-	AND status = 'completed'
174	-	"#,
175	-	user_id as UserId,
176	-	email,
177	-	)
178	-	.execute(pool)
179	-	.await?;
180	-
181	-	Ok(result.rows_affected())
182	-	}
183	-
184	-	/// Claim a single guest purchase by claim token.
185	-	#[tracing::instrument(skip_all)]
186	-	pub async fn claim_guest_purchase(
187	-	pool: &PgPool,
188	-	claim_token: ClaimToken,
189	-	user_id: UserId,
190	-	) -> Result<Option<DbTransaction>> {
191	-	let tx = sqlx::query_as!(
192	-	DbTransaction,
193	-	r#"
194	-	UPDATE transactions
195	-	SET buyer_id = $2, claimed_by = $2, claim_token = NULL
196	-	WHERE claim_token = $1
197	-	AND buyer_id IS NULL
198	-	AND status = 'completed'
199	-	RETURNING
200	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
201	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
202	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
203	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
204	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
205	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
206	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
207	-	download_token AS "download_token: DownloadToken",
208	-	presentment_amount_cents, presentment_currency
209	-	"#,
210	-	claim_token as ClaimToken,
211	-	user_id as UserId,
212	-	)
213	-	.fetch_optional(pool)
214	-	.await?;
215	-
216	-	Ok(tx)
217	-	}
218	-
219	-	/// Look up a completed transaction by download token (for guest download links).
220	-	#[tracing::instrument(skip_all)]
221	-	pub async fn get_transaction_by_download_token(
222	-	pool: &PgPool,
223	-	download_token: DownloadToken,
224	-	) -> Result<Option<DbTransaction>> {
225	-	let tx = sqlx::query_as!(
226	-	DbTransaction,
227	-	r#"
228	-	SELECT
229	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
230	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
231	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
232	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
233	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
234	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
235	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
236	-	download_token AS "download_token: DownloadToken",
237	-	presentment_amount_cents, presentment_currency
238	-	FROM transactions WHERE download_token = $1 AND status = 'completed'
239	-	"#,
240	-	download_token as DownloadToken,
241	-	)
242	-	.fetch_optional(pool)
243	-	.await?;
244	-
245	-	Ok(tx)
246	-	}
247	-
248	-	/// Mark a pending transaction as completed (idempotent; returns `None` if already completed).
249	-	///
250	-	/// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
251	-	/// can include this in a larger transaction when needed.
252	-	#[tracing::instrument(skip_all)]
253	-	pub async fn complete_transaction<'e>(
254	-	executor: impl sqlx::PgExecutor<'e>,
255	-	stripe_checkout_session_id: &str,
256	-	stripe_payment_intent_id: Option<&str>,
257	-	presentment: Option<(i64, &str)>,
258	-	) -> Result<Option<DbTransaction>> {
259	-	// Only update if status is 'pending' for idempotency
260	-	// Returns None if transaction was already completed (duplicate webhook)
261	-	let tx = sqlx::query_as!(
262	-	DbTransaction,
263	-	r#"
264	-	UPDATE transactions
265	-	SET status = 'completed',
266	-	stripe_payment_intent_id = $2,
267	-	presentment_amount_cents = $3,
268	-	presentment_currency = $4,
269	-	completed_at = NOW()
270	-	WHERE stripe_checkout_session_id = $1
271	-	AND status = 'pending'
272	-	RETURNING
273	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
274	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
275	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
276	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
277	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
278	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
279	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
280	-	download_token AS "download_token: DownloadToken",
281	-	presentment_amount_cents, presentment_currency
282	-	"#,
283	-	stripe_checkout_session_id,
284	-	stripe_payment_intent_id,
285	-	presentment.map(\|(cents, _)\| cents),
286	-	presentment.map(\|(_, currency)\| currency),
287	-	)
288	-	.fetch_optional(executor)
289	-	.await?;
290	-
291	-	Ok(tx)
292	-	}
293	-
294	-	/// Complete ALL pending transactions for a cart checkout session.
295	-	/// Returns the list of completed transactions (empty if already processed).
296	-	#[tracing::instrument(skip_all)]
297	-	pub async fn complete_cart_transactions<'e>(
298	-	executor: impl sqlx::PgExecutor<'e>,
299	-	stripe_checkout_session_id: &str,
300	-	stripe_payment_intent_id: Option<&str>,
301	-	) -> Result<Vec<DbTransaction>> {
302	-	let txs = sqlx::query_as!(
303	-	DbTransaction,
304	-	r#"
305	-	UPDATE transactions
306	-	SET status = 'completed',
307	-	stripe_payment_intent_id = $2,
308	-	completed_at = NOW()
309	-	WHERE stripe_checkout_session_id = $1
310	-	AND status = 'pending'
311	-	RETURNING
312	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
313	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
314	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
315	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
316	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
317	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
318	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
319	-	download_token AS "download_token: DownloadToken",
320	-	presentment_amount_cents, presentment_currency
321	-	"#,
322	-	stripe_checkout_session_id,
323	-	stripe_payment_intent_id,
324	-	)
325	-	.fetch_all(executor)
326	-	.await?;
327	-
328	-	Ok(txs)
329	-	}
330	-
331	-	/// Fetch all completed transactions for a checkout session.
332	-	///
333	-	/// Used on the crash-recovery branch of the purchase/cart webhook handlers: when
334	-	/// `complete_transaction` / `complete_cart_transactions` return nothing (the
335	-	/// rows were already flipped to completed by a first attempt that crashed before
336	-	/// running finalize), this re-reads those completed rows so finalize can re-run
337	-	/// idempotently. Covers single and cart purchases since both key on the session.
338	-	#[tracing::instrument(skip_all)]
339	-	pub async fn get_completed_transactions_for_session<'e>(
340	-	executor: impl sqlx::PgExecutor<'e>,
341	-	stripe_checkout_session_id: &str,
342	-	) -> Result<Vec<DbTransaction>> {
343	-	let txs = sqlx::query_as!(
344	-	DbTransaction,
345	-	r#"
346	-	SELECT
347	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
348	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
349	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
350	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
351	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
352	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
353	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
354	-	download_token AS "download_token: DownloadToken",
355	-	presentment_amount_cents, presentment_currency
356	-	FROM transactions
357	-	WHERE stripe_checkout_session_id = $1 AND status = 'completed'
358	-	"#,
359	-	stripe_checkout_session_id,
360	-	)
361	-	.fetch_all(executor)
362	-	.await?;
363	-
364	-	Ok(txs)
365	-	}
366	-
367	-	/// List transactions where the user is the buyer, newest first.
368	-	///
369	-	/// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
370	-	#[tracing::instrument(skip_all)]
371	-	pub async fn get_transactions_by_buyer(
372	-	pool: &PgPool,
373	-	buyer_id: UserId,
374	-	limit: Option<i64>,
375	-	) -> Result<Vec<DbTransaction>> {
376	-	let txs = sqlx::query_as!(
377	-	DbTransaction,
378	-	r#"
379	-	SELECT
380	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
381	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
382	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
383	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
384	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
385	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
386	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
387	-	download_token AS "download_token: DownloadToken",
388	-	presentment_amount_cents, presentment_currency
389	-	FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
390	-	"#,
391	-	buyer_id as UserId,
392	-	limit,
393	-	)
394	-	.fetch_all(pool)
395	-	.await?;
396	-
397	-	Ok(txs)
398	-	}
399	-
400	-	/// One page of a buyer's purchases for CSV export, newest first.
401	-	///
402	-	/// Paginated so the purchases export streams in bounded batches rather than
403	-	/// loading the buyer's whole history with `limit: None`.
404	-	/// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
405	-	pub async fn get_buyer_transactions_for_export_page(
406	-	pool: &PgPool,
407	-	buyer_id: UserId,
408	-	limit: i64,
409	-	offset: i64,
410	-	) -> Result<Vec<DbTransaction>> {
411	-	let txs = sqlx::query_as!(
412	-	DbTransaction,
413	-	r#"
414	-	SELECT
415	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
416	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
417	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
418	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
419	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
420	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
421	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
422	-	download_token AS "download_token: DownloadToken",
423	-	presentment_amount_cents, presentment_currency
424	-	FROM transactions WHERE buyer_id = $1
425	-	ORDER BY created_at DESC, id DESC
426	-	LIMIT $2 OFFSET $3
427	-	"#,
428	-	buyer_id as UserId,
429	-	limit,
430	-	offset,
431	-	)
432	-	.fetch_all(pool)
433	-	.await?;
434	-
435	-	Ok(txs)
436	-	}
437	-
438	-	/// List transactions where the user is the seller, newest first.
439	-	///
440	-	/// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
441	-	#[tracing::instrument(skip_all)]
442	-	pub async fn get_transactions_by_seller(
443	-	pool: &PgPool,
444	-	seller_id: UserId,
445	-	limit: Option<i64>,
446	-	) -> Result<Vec<DbTransaction>> {
447	-	let txs = sqlx::query_as!(
448	-	DbTransaction,
449	-	r#"
450	-	SELECT
451	-	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
452	-	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
453	-	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
454	-	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
455	-	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
456	-	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
457	-	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
458	-	download_token AS "download_token: DownloadToken",
459	-	presentment_amount_cents, presentment_currency
460	-	FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
461	-	"#,
462	-	seller_id as UserId,
463	-	limit,
464	-	)
465	-	.fetch_all(pool)
466	-	.await?;
467	-
468	-	Ok(txs)
469	-	}
470	-
471	-	/// Check whether a user has a completed purchase for a given item.
472	-	#[tracing::instrument(skip_all)]
473	-	pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
474	-	let count: i64 = sqlx::query_scalar!(
475	-	r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
476	-	user_id as UserId,
477	-	item_id as ItemId,
478	-	)
479	-	.fetch_one(pool)
480	-	.await?;
481	-
482	-	Ok(count > 0)
483	-	}
484	-
485	-	/// Bulk variant of `has_purchased_item`. Returns the subset of `item_ids` that
486	-	/// the buyer has a completed purchase for. Single DB roundtrip vs. N calls.
487	-	#[tracing::instrument(skip_all)]
488	-	pub async fn purchased_subset(
489	-	pool: &PgPool,
490	-	user_id: UserId,
491	-	item_ids: &[ItemId],
492	-	) -> Result<std::collections::HashSet<ItemId>> {
493	-	if item_ids.is_empty() {
494	-	return Ok(std::collections::HashSet::new());
495	-	}
496	-	let rows = sqlx::query_scalar!(
497	-	r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
498	-	WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
499	-	user_id as UserId,
500	-	item_ids as &[ItemId],

Lines truncated

A server/src/db/transactions/purchases/checkout.rs +393

		@@ -1,0 +1,393 @@
1	+	//! The pending half of the lifecycle: write a row before the buyer leaves for
2	+	//! Stripe, flip it on their return, and delete it if they never come back.
3	+	//!
4	+	//! Every delete path here returns the row's `promo_code_id` so the caller
5	+	//! releases the reservation with the row.
6	+
7	+	use super::super::{
8	+	Cents, ClaimToken, DbTransaction, DownloadToken, ItemId, PgPool, ProjectId, PromoCodeId,
9	+	Result, TransactionId, UserId,
10	+	};
11	+
12	+	/// Parameters for creating a pending Stripe checkout transaction.
13	+	pub struct CreateTransactionParams<'a> {
14	+	pub buyer_id: Option<UserId>,
15	+	pub seller_id: UserId,
16	+	/// `None` for project-level purchases (no specific item).
17	+	pub item_id: Option<ItemId>,
18	+	pub amount_cents: Cents,
19	+	pub platform_fee_cents: Cents,
20	+	pub stripe_checkout_session_id: &'a str,
21	+	pub item_title: &'a str,
22	+	pub seller_username: &'a str,
23	+	pub share_contact: bool,
24	+	/// Set for project-level purchases; `None` for item purchases.
25	+	pub project_id: Option<ProjectId>,
26	+	/// Promo code used for this checkout (for releasing reservations on cleanup).
27	+	pub promo_code_id: Option<PromoCodeId>,
28	+	/// Guest buyer's email (set for guest checkouts, None for logged-in).
29	+	pub guest_email: Option<&'a str>,
30	+	/// Cents MNW owes the seller as reimbursement for a platform-funded credit
31	+	/// (the Fan+ renewal credit) applied to this sale; `0` for ordinary sales. The
32	+	/// scheduler settles it via a platform -> connected transfer once the
33	+	/// transaction completes (see `db::platform_credits`).
34	+	pub platform_credit_cents: i64,
35	+	}
36	+
37	+	/// Record a new pending transaction for a Stripe checkout session.
38	+	#[tracing::instrument(skip_all)]
39	+	pub async fn create_transaction<'e>(
40	+	executor: impl sqlx::PgExecutor<'e>,
41	+	params: &CreateTransactionParams<'_>,
42	+	) -> Result<DbTransaction> {
43	+	let tx = sqlx::query_as!(
44	+	DbTransaction,
45	+	r#"
46	+	INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact, project_id, promo_code_id, guest_email, platform_credit_cents)
47	+	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
48	+	RETURNING
49	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
50	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
51	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
52	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
53	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
54	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
55	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
56	+	download_token AS "download_token: DownloadToken",
57	+	presentment_amount_cents, presentment_currency
58	+	"#,
59	+	params.buyer_id as Option<UserId>,
60	+	params.seller_id as UserId,
61	+	params.item_id as Option<ItemId>,
62	+	params.amount_cents as Cents,
63	+	params.platform_fee_cents as Cents,
64	+	params.stripe_checkout_session_id,
65	+	params.item_title,
66	+	params.seller_username,
67	+	params.share_contact,
68	+	params.project_id as Option<ProjectId>,
69	+	params.promo_code_id as Option<PromoCodeId>,
70	+	params.guest_email,
71	+	params.platform_credit_cents,
72	+	)
73	+	.fetch_one(executor)
74	+	.await?;
75	+
76	+	Ok(tx)
77	+	}
78	+
79	+	/// Mark a pending transaction as completed (idempotent; returns `None` if already completed).
80	+	///
81	+	/// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
82	+	/// can include this in a larger transaction when needed.
83	+	#[tracing::instrument(skip_all)]
84	+	pub async fn complete_transaction<'e>(
85	+	executor: impl sqlx::PgExecutor<'e>,
86	+	stripe_checkout_session_id: &str,
87	+	stripe_payment_intent_id: Option<&str>,
88	+	presentment: Option<(i64, &str)>,
89	+	) -> Result<Option<DbTransaction>> {
90	+	// Only update if status is 'pending' for idempotency
91	+	// Returns None if transaction was already completed (duplicate webhook)
92	+	let tx = sqlx::query_as!(
93	+	DbTransaction,
94	+	r#"
95	+	UPDATE transactions
96	+	SET status = 'completed',
97	+	stripe_payment_intent_id = $2,
98	+	presentment_amount_cents = $3,
99	+	presentment_currency = $4,
100	+	completed_at = NOW()
101	+	WHERE stripe_checkout_session_id = $1
102	+	AND status = 'pending'
103	+	RETURNING
104	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
105	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
106	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
107	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
108	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
109	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
110	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
111	+	download_token AS "download_token: DownloadToken",
112	+	presentment_amount_cents, presentment_currency
113	+	"#,
114	+	stripe_checkout_session_id,
115	+	stripe_payment_intent_id,
116	+	presentment.map(\|(cents, _)\| cents),
117	+	presentment.map(\|(_, currency)\| currency),
118	+	)
119	+	.fetch_optional(executor)
120	+	.await?;
121	+
122	+	Ok(tx)
123	+	}
124	+
125	+	/// Complete ALL pending transactions for a cart checkout session.
126	+	/// Returns the list of completed transactions (empty if already processed).
127	+	#[tracing::instrument(skip_all)]
128	+	pub async fn complete_cart_transactions<'e>(
129	+	executor: impl sqlx::PgExecutor<'e>,
130	+	stripe_checkout_session_id: &str,
131	+	stripe_payment_intent_id: Option<&str>,
132	+	) -> Result<Vec<DbTransaction>> {
133	+	let txs = sqlx::query_as!(
134	+	DbTransaction,
135	+	r#"
136	+	UPDATE transactions
137	+	SET status = 'completed',
138	+	stripe_payment_intent_id = $2,
139	+	completed_at = NOW()
140	+	WHERE stripe_checkout_session_id = $1
141	+	AND status = 'pending'
142	+	RETURNING
143	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
144	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
145	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
146	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
147	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
148	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
149	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
150	+	download_token AS "download_token: DownloadToken",
151	+	presentment_amount_cents, presentment_currency
152	+	"#,
153	+	stripe_checkout_session_id,
154	+	stripe_payment_intent_id,
155	+	)
156	+	.fetch_all(executor)
157	+	.await?;
158	+
159	+	Ok(txs)
160	+	}
161	+
162	+	/// Fetch all completed transactions for a checkout session.
163	+	///
164	+	/// Used on the crash-recovery branch of the purchase/cart webhook handlers: when
165	+	/// `complete_transaction` / `complete_cart_transactions` return nothing (the
166	+	/// rows were already flipped to completed by a first attempt that crashed before
167	+	/// running finalize), this re-reads those completed rows so finalize can re-run
168	+	/// idempotently. Covers single and cart purchases since both key on the session.
169	+	#[tracing::instrument(skip_all)]
170	+	pub async fn get_completed_transactions_for_session<'e>(
171	+	executor: impl sqlx::PgExecutor<'e>,
172	+	stripe_checkout_session_id: &str,
173	+	) -> Result<Vec<DbTransaction>> {
174	+	let txs = sqlx::query_as!(
175	+	DbTransaction,
176	+	r#"
177	+	SELECT
178	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
179	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
180	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
181	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
182	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
183	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
184	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
185	+	download_token AS "download_token: DownloadToken",
186	+	presentment_amount_cents, presentment_currency
187	+	FROM transactions
188	+	WHERE stripe_checkout_session_id = $1 AND status = 'completed'
189	+	"#,
190	+	stripe_checkout_session_id,
191	+	)
192	+	.fetch_all(executor)
193	+	.await?;
194	+
195	+	Ok(txs)
196	+	}
197	+
198	+	/// Parameters for creating a pending project purchase transaction.
199	+	pub struct CreateProjectTransactionParams<'a> {
200	+	pub buyer_id: UserId,
201	+	pub seller_id: UserId,
202	+	pub project_id: ProjectId,
203	+	pub amount_cents: i32,
204	+	pub stripe_checkout_session_id: &'a str,
205	+	pub project_title: &'a str,
206	+	pub seller_username: &'a str,
207	+	pub share_contact: bool,
208	+	}
209	+
210	+	/// Record a new pending transaction for a project purchase.
211	+	#[tracing::instrument(skip_all)]
212	+	pub async fn create_project_transaction(
213	+	pool: &PgPool,
214	+	params: &CreateProjectTransactionParams<'_>,
215	+	) -> Result<DbTransaction> {
216	+	let tx = sqlx::query_as!(
217	+	DbTransaction,
218	+	r#"
219	+	INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact)
220	+	VALUES ($1, $2, $3, $4, 0, $5, $6, $7, $8)
221	+	RETURNING
222	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
223	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
224	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
225	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
226	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
227	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
228	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
229	+	download_token AS "download_token: DownloadToken",
230	+	presentment_amount_cents, presentment_currency
231	+	"#,
232	+	params.buyer_id as UserId,
233	+	params.seller_id as UserId,
234	+	params.project_id as ProjectId,
235	+	params.amount_cents,
236	+	params.stripe_checkout_session_id,
237	+	params.project_title,
238	+	params.seller_username,
239	+	params.share_contact,
240	+	)
241	+	.fetch_one(pool)
242	+	.await?;
243	+
244	+	Ok(tx)
245	+	}
246	+
247	+	/// Create a pending "placeholder" transaction for a subscription checkout that
248	+	/// used a promo code. This row exists solely so `cleanup_stale_pending` can
249	+	/// release the promo code reservation if the buyer abandons the Stripe session.
250	+	/// It is deleted (not completed) when the subscription webhook fires.
251	+	#[tracing::instrument(skip_all)]
252	+	pub async fn create_subscription_pending_transaction(
253	+	pool: &PgPool,
254	+	buyer_id: UserId,
255	+	seller_id: UserId,
256	+	project_id: ProjectId,
257	+	stripe_checkout_session_id: &str,
258	+	promo_code_id: PromoCodeId,
259	+	) -> Result<()> {
260	+	sqlx::query!(
261	+	r#"
262	+	INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents,
263	+	stripe_checkout_session_id, item_title, seller_username, share_contact, promo_code_id)
264	+	VALUES ($1, $2, $3, 0, 0, $4, 'subscription-promo-hold', '', false, $5)
265	+	"#,
266	+	buyer_id as UserId,
267	+	seller_id as UserId,
268	+	project_id as ProjectId,
269	+	stripe_checkout_session_id,
270	+	promo_code_id as PromoCodeId,
271	+	)
272	+	.execute(pool)
273	+	.await?;
274	+
275	+	Ok(())
276	+	}
277	+
278	+	/// Delete a pending subscription promo-hold transaction by checkout session ID.
279	+	/// Called from the subscription webhook after the subscription is created.
280	+	#[tracing::instrument(skip_all)]
281	+	pub async fn delete_subscription_pending_transaction<'e>(
282	+	executor: impl sqlx::PgExecutor<'e>,
283	+	stripe_checkout_session_id: &str,
284	+	) -> Result<()> {
285	+	sqlx::query!(
286	+	"DELETE FROM transactions WHERE stripe_checkout_session_id = $1 AND status = 'pending'",
287	+	stripe_checkout_session_id,
288	+	)
289	+	.execute(executor)
290	+	.await?;
291	+
292	+	Ok(())
293	+	}
294	+
295	+	/// Delete stale pending transactions (older than the given threshold) and return
296	+	/// the promo_code_ids that need their use_count decremented.
297	+	///
298	+	/// Stripe checkout sessions expire after 24 hours, so pending transactions older
299	+	/// than that will never complete. This releases the pending purchase uniqueness
300	+	/// slot and any reserved promo code use_count.
301	+	#[tracing::instrument(skip_all)]
302	+	pub async fn cleanup_stale_pending(
303	+	pool: &PgPool,
304	+	older_than: chrono::Duration,
305	+	) -> Result<Vec<Option<crate::db::PromoCodeId>>> {
306	+	let cutoff = chrono::Utc::now() - older_than;
307	+	// runtime-checked: binds a chrono DateTime<Utc> param; a bind param's type can't be overridden in the macro when sqlx time+chrono features are unified.
308	+	let rows: Vec<(Option<crate::db::PromoCodeId>,)> = sqlx::query_as(
309	+	r"
310	+	DELETE FROM transactions
311	+	WHERE status = 'pending'
312	+	AND created_at < $1
313	+	RETURNING promo_code_id
314	+	",
315	+	)
316	+	.bind(cutoff)
317	+	.fetch_all(pool)
318	+	.await?;
319	+
320	+	Ok(rows.into_iter().map(\|(id,)\| id).collect())
321	+	}
322	+
323	+	/// Bulk variant of `get_pending_item_purchase`. Returns the subset of `item_ids`
324	+	/// for which the buyer already has a `pending` transaction. Used by cart
325	+	/// checkout to abort early when any line item would collide with the partial
326	+	/// unique index on `(buyer_id, item_id) WHERE status = 'pending'`.
327	+	#[tracing::instrument(skip_all)]
328	+	pub async fn pending_subset(
329	+	pool: &PgPool,
330	+	buyer_id: UserId,
331	+	item_ids: &[ItemId],
332	+	) -> Result<std::collections::HashSet<ItemId>> {
333	+	if item_ids.is_empty() {
334	+	return Ok(std::collections::HashSet::new());
335	+	}
336	+	let rows = sqlx::query_scalar!(
337	+	r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
338	+	WHERE buyer_id = $1 AND status = 'pending' AND item_id = ANY($2)"#,
339	+	buyer_id as UserId,
340	+	item_ids as &[ItemId],
341	+	)
342	+	.fetch_all(pool)
343	+	.await?;
344	+	Ok(rows.into_iter().collect())
345	+	}
346	+
347	+	/// Returns the buyer's pending transaction for a specific item, if any.
348	+	/// Used to surface in-progress checkouts on the purchase page.
349	+	#[tracing::instrument(skip_all)]
350	+	pub async fn get_pending_item_purchase(
351	+	pool: &PgPool,
352	+	buyer_id: UserId,
353	+	item_id: ItemId,
354	+	) -> Result<Option<(TransactionId, chrono::DateTime<chrono::Utc>)>> {
355	+	let row = sqlx::query!(
356	+	r#"
357	+	SELECT id AS "id: TransactionId", created_at AS "created_at: chrono::DateTime<chrono::Utc>"
358	+	FROM transactions
359	+	WHERE buyer_id = $1 AND item_id = $2 AND status = 'pending'
360	+	LIMIT 1
361	+	"#,
362	+	buyer_id as UserId,
363	+	item_id as ItemId,
364	+	)
365	+	.fetch_optional(pool)
366	+	.await?;
367	+
368	+	Ok(row.map(\|r\| (r.id, r.created_at)))
369	+	}
370	+
371	+	/// Delete the buyer's pending transaction for a specific item.
372	+	/// Returns any released `promo_code_id` so the caller can release its
373	+	/// reservation.
374	+	#[tracing::instrument(skip_all)]
375	+	pub async fn delete_pending_item_purchase(
376	+	pool: &PgPool,
377	+	buyer_id: UserId,
378	+	item_id: ItemId,
379	+	) -> Result<Option<crate::db::PromoCodeId>> {
380	+	let row: Option<Option<crate::db::PromoCodeId>> = sqlx::query_scalar!(
381	+	r#"
382	+	DELETE FROM transactions
383	+	WHERE buyer_id = $1 AND item_id = $2 AND status = 'pending'
384	+	RETURNING promo_code_id AS "promo_code_id: crate::db::PromoCodeId"
385	+	"#,
386	+	buyer_id as UserId,
387	+	item_id as ItemId,
388	+	)
389	+	.fetch_optional(pool)
390	+	.await?;
391	+
392	+	Ok(row.flatten())
393	+	}

A server/src/db/transactions/purchases/claims.rs +307

		@@ -1,0 +1,307 @@
1	+	//! Taking something that costs nothing.
2	+	//!
3	+	//! Every function here writes `amount_cents = 0, status = 'completed'` through
4	+	//! the same `ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND
5	+	//! item_id IS NOT NULL DO NOTHING` idempotency clause, and three carry
6	+	//! `platform_credit_cents` for the Fan+ reimbursement.
7	+
8	+	use super::super::{
9	+	ItemId, KeyCode, PgPool, ProjectId, PromoCodeId, Result, TransactionId, UserId,
10	+	};
11	+
12	+	/// Common parameters for claiming a free item (direct, discount code, or download code).
13	+	pub struct ClaimParams<'a> {
14	+	pub buyer_id: UserId,
15	+	pub item_id: ItemId,
16	+	pub seller_id: UserId,
17	+	pub item_title: &'a str,
18	+	pub seller_username: &'a str,
19	+	pub share_contact: bool,
20	+	/// If this claim was granted via a bundle purchase, the parent transaction ID.
21	+	pub parent_transaction_id: Option<TransactionId>,
22	+	/// Cents MNW owes the seller when a platform-wide (Fan+) credit made this item
23	+	/// free, the seller is reimbursed the item's price via a platform transfer so
24	+	/// they are still paid. `0` for ordinary free claims (genuinely-free items,
25	+	/// seller-issued free-access codes, bundle grants).
26	+	pub platform_credit_cents: i64,
27	+	}
28	+
29	+	/// Claims a free item by creating a zero-cost completed transaction.
30	+	/// Returns true if claimed successfully, false if already in library.
31	+	///
32	+	/// Uses `ON CONFLICT DO NOTHING` against the partial unique index on
33	+	/// `(buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL` to prevent duplicate
34	+	/// claims under concurrent requests.
35	+	#[tracing::instrument(skip_all)]
36	+	pub async fn claim_free_item<'e>(
37	+	executor: impl sqlx::PgExecutor<'e>,
38	+	params: &ClaimParams<'_>,
39	+	) -> Result<bool> {
40	+	let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
41	+	let result = sqlx::query!(
42	+	r#"
43	+	INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, status, completed_at, item_title, seller_username, share_contact, parent_transaction_id, platform_credit_cents)
44	+	VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8, $9)
45	+	ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
46	+	"#,
47	+	params.buyer_id as UserId,
48	+	params.seller_id as UserId,
49	+	params.item_id as ItemId,
50	+	claim_id,
51	+	params.item_title,
52	+	params.seller_username,
53	+	params.share_contact,
54	+	params.parent_transaction_id as Option<TransactionId>,
55	+	params.platform_credit_cents,
56	+	)
57	+	.execute(executor)
58	+	.await?;
59	+
60	+	Ok(result.rows_affected() > 0)
61	+	}
62	+
63	+	/// Batch variant of [`claim_free_item`] for bundle grants: claims every child
64	+	/// item for one buyer in a single INSERT instead of N round-trips (each with its
65	+	/// own pool acquire) on the Stripe webhook / checkout hot path.
66	+	/// Each row is idempotent via the same partial-unique ON CONFLICT as the single
67	+	/// claim, and child items deliberately do not increment `sales_count`. Returns the
68	+	/// number of rows actually inserted. No-op on an empty slice.
69	+	#[tracing::instrument(skip_all)]
70	+	pub async fn claim_free_items_batch<'e>(
71	+	executor: impl sqlx::PgExecutor<'e>,
72	+	buyer_id: UserId,
73	+	seller_id: UserId,
74	+	seller_username: &str,
75	+	parent_transaction_id: Option<TransactionId>,
76	+	items: &[(ItemId, &str)],
77	+	) -> Result<u64> {
78	+	if items.is_empty() {
79	+	return Ok(0);
80	+	}
81	+	let item_ids: Vec<ItemId> = items.iter().map(\|(id, _)\| *id).collect();
82	+	let item_titles: Vec<&str> = items.iter().map(\|(_, title)\| *title).collect();
83	+
84	+	// The per-row claim id mirrors the single claim's `free-claim-{buyer}-{item}`
85	+	// so a later single claim of the same item still collides idempotently.
86	+	let result = sqlx::query(
87	+	r"
88	+	INSERT INTO transactions
89	+	(buyer_id, seller_id, item_id, amount_cents, platform_fee_cents,
90	+	stripe_checkout_session_id, status, completed_at, item_title,
91	+	seller_username, share_contact, parent_transaction_id)
92	+	SELECT
93	+	$1, $2, t.item_id, 0, 0,
94	+	'free-claim-' \|\| $1::text \|\| '-' \|\| t.item_id::text,
95	+	'completed', NOW(), t.item_title, $3, false, $4
96	+	FROM UNNEST($5::uuid[], $6::text[]) AS t(item_id, item_title)
97	+	ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
98	+	",
99	+	)
100	+	.bind(buyer_id)
101	+	.bind(seller_id)
102	+	.bind(seller_username)
103	+	.bind(parent_transaction_id)
104	+	.bind(&item_ids)
105	+	.bind(&item_titles)
106	+	.execute(executor)
107	+	.await?;
108	+
109	+	Ok(result.rows_affected())
110	+	}
111	+
112	+	/// Optional parameters for generating a license key inside a claim transaction.
113	+	pub struct LicenseKeyParams<'a> {
114	+	pub key_code: &'a KeyCode,
115	+	pub max_activations: Option<i32>,
116	+	}
117	+
118	+	/// Atomically claim a free item and increment the promo code's use count.
119	+	///
120	+	/// Claims the item FIRST (INSERT transaction), then increments use_count.
121	+	/// If the user already owns the item (rows_affected == 0), rolls back without
122	+	/// consuming the code. If the code limit is reached, rolls back the claim too.
123	+	///
124	+	/// When `license_key_params` is `Some`, a license key is created inside the
125	+	/// same transaction so that the claim and key are always consistent.
126	+	///
127	+	/// Returns `(code_accepted, item_claimed)`:
128	+	/// - `code_accepted = false` → promo code hit its usage limit (nothing changed)
129	+	/// - `item_claimed = false` → user already owns the item (code was NOT consumed)
130	+	#[tracing::instrument(skip_all)]
131	+	pub async fn claim_free_with_promo_code(
132	+	pool: &PgPool,
133	+	promo_code_id: PromoCodeId,
134	+	params: &ClaimParams<'_>,
135	+	license_key_params: Option<&LicenseKeyParams<'_>>,
136	+	) -> Result<(bool, bool)> {
137	+	let mut tx = pool.begin().await?;
138	+
139	+	// Claim the item first. When a platform-wide credit (Fan+)
140	+	// made the item free, `platform_credit_cents` carries the item's full price so
141	+	// the scheduler reimburses the creator via transfer, the fan pays nothing but
142	+	// the creator is still paid (MNW funds it).
143	+	let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
144	+	let result = sqlx::query!(
145	+	r#"
146	+	INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, status, completed_at, item_title, seller_username, share_contact, promo_code_id, platform_credit_cents)
147	+	VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8, $9)
148	+	ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
149	+	"#,
150	+	params.buyer_id as UserId,
151	+	params.seller_id as UserId,
152	+	params.item_id as ItemId,
153	+	claim_id,
154	+	params.item_title,
155	+	params.seller_username,
156	+	params.share_contact,
157	+	promo_code_id as PromoCodeId,
158	+	params.platform_credit_cents,
159	+	)
160	+	.execute(&mut *tx)
161	+	.await?;
162	+
163	+	let claimed = result.rows_affected() > 0;
164	+
165	+	if !claimed {
166	+	tx.rollback().await?;
167	+	return Ok((true, false));
168	+	}
169	+
170	+	// Increment the promo code use count. Re-check the full validity
171	+	// window (max_uses AND starts_at/expires_at) inside the atomic UPDATE, the
172	+	// pre-flight check ran outside this transaction, so a code expiring in the
173	+	// sub-ms gap must still be rejected here, not just on use-count (Run 11 Pay
174	+	// MINOR / TOCTOU).
175	+	let code_result = sqlx::query!(
176	+	r#"
177	+	UPDATE promo_codes SET use_count = use_count + 1
178	+	WHERE id = $1
179	+	AND (max_uses IS NULL OR use_count < max_uses)
180	+	AND (starts_at IS NULL OR starts_at <= NOW())
181	+	AND (expires_at IS NULL OR expires_at > NOW())
182	+	"#,
183	+	promo_code_id as PromoCodeId,
184	+	)
185	+	.execute(&mut *tx)
186	+	.await?;
187	+
188	+	if code_result.rows_affected() == 0 {
189	+	tx.rollback().await?;
190	+	return Ok((false, false));
191	+	}
192	+
193	+	crate::db::items::increment_sales_count(&mut *tx, params.item_id).await?;
194	+
195	+	// Create the license key inside the same transaction if requested.
196	+	// Retry once on a unique-violation: the wordlist generator has ~6B-coin-
197	+	// flip headroom, so an actual collision is vanishingly rare, but the
198	+	// alternative is surfacing a 500 to a buyer mid-claim, cheap to handle.
199	+	if let Some(lk) = license_key_params {
200	+	let attempt = sqlx::query!(
201	+	r#"
202	+	INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
203	+	VALUES ($1, $2, NULL, $3, $4)
204	+	"#,
205	+	params.item_id as ItemId,
206	+	params.buyer_id as UserId,
207	+	lk.key_code as &KeyCode,
208	+	lk.max_activations,
209	+	)
210	+	.execute(&mut *tx)
211	+	.await;
212	+
213	+	if let Err(sqlx::Error::Database(e)) = &attempt
214	+	&& e.code().as_deref() == Some("23505")
215	+	{
216	+	let retry_code = crate::helpers::generate_key_code();
217	+	tracing::warn!(item_id = %params.item_id, "license key 23505 collision; retrying once");
218	+	sqlx::query!(
219	+	r#"
220	+	INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
221	+	VALUES ($1, $2, NULL, $3, $4)
222	+	"#,
223	+	params.item_id as ItemId,
224	+	params.buyer_id as UserId,
225	+	retry_code as KeyCode,
226	+	lk.max_activations,
227	+	)
228	+	.execute(&mut *tx)
229	+	.await?;
230	+	} else {
231	+	attempt?;
232	+	}
233	+	}
234	+
235	+	tx.commit().await?;
236	+	Ok((true, true))
237	+	}
238	+
239	+	/// Remove a free item from library (deletes the claim transaction).
240	+	/// If the claim was via a promo code, decrements the code's use_count.
241	+	#[tracing::instrument(skip_all)]
242	+	pub async fn remove_free_item_from_library(
243	+	pool: &PgPool,
244	+	user_id: UserId,
245	+	item_id: ItemId,
246	+	) -> Result<bool> {
247	+	// Delete the free claim and return the promo_code_id if one was used
248	+	let row: Option<Option<crate::db::PromoCodeId>> = sqlx::query_scalar!(
249	+	r#"
250	+	DELETE FROM transactions
251	+	WHERE buyer_id = $1 AND item_id = $2 AND amount_cents = 0 AND status = 'completed'
252	+	RETURNING promo_code_id AS "promo_code_id: crate::db::PromoCodeId"
253	+	"#,
254	+	user_id as UserId,
255	+	item_id as ItemId,
256	+	)
257	+	.fetch_optional(pool)
258	+	.await?;
259	+
260	+	let deleted = row.is_some();
261	+
262	+	if let Some(Some(pc_id)) = row {
263	+	crate::db::promo_codes::release_use_count(pool, pc_id)
264	+	.await
265	+	.ok();
266	+	}
267	+
268	+	Ok(deleted)
269	+	}
270	+
271	+	/// Record a free project claim (PWYW with $0 min or free project).
272	+	///
273	+	/// Returns `true` if the claim was actually inserted, `false` if the buyer
274	+	/// already owned the project. Mirrors the `claim_free_item` shape so callers
275	+	/// can gate downstream side-effects (contact-revocation clear, sale-notification
276	+	/// email, etc.) on the winner of a concurrent-claim race, without this signal,
277	+	/// two concurrent `/checkout/project` POSTs both fire those side-effects
278	+	/// regardless of which one's INSERT actually landed.
279	+	#[tracing::instrument(skip_all)]
280	+	pub async fn claim_free_project(
281	+	pool: &PgPool,
282	+	buyer_id: UserId,
283	+	seller_id: UserId,
284	+	project_id: ProjectId,
285	+	item_title: &str,
286	+	seller_username: &str,
287	+	share_contact: bool,
288	+	) -> Result<bool> {
289	+	let result = sqlx::query!(
290	+	r#"
291	+	INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents,
292	+	status, completed_at, item_title, seller_username, share_contact)
293	+	VALUES ($1, $2, $3, 0, 0, 'completed', NOW(), $4, $5, $6)
294	+	ON CONFLICT (buyer_id, project_id) WHERE status = 'completed' AND project_id IS NOT NULL DO NOTHING
295	+	"#,
296	+	buyer_id as UserId,
297	+	seller_id as UserId,
298	+	project_id as ProjectId,
299	+	item_title,
300	+	seller_username,
301	+	share_contact,
302	+	)
303	+	.execute(pool)
304	+	.await?;
305	+
306	+	Ok(result.rows_affected() > 0)
307	+	}

A server/src/db/transactions/purchases/guest.rs +193

		@@ -1,0 +1,193 @@
1	+	//! Purchases made without an account.
2	+	//!
3	+	//! A guest purchase lands with `buyer_id` NULL, identified by its claim and
4	+	//! download tokens, and is attached to an account out of band once the buyer
5	+	//! verifies the email it was bought with. These are the only queries that
6	+	//! touch `claim_token`, `download_token` or `guest_email`.
7	+
8	+	use super::super::{
9	+	Cents, ClaimToken, DbTransaction, DownloadToken, ItemId, PgPool, ProjectId, PromoCodeId,
10	+	Result, TransactionId, UserId,
11	+	};
12	+
13	+	/// Complete a guest transaction: mark it completed, record the guest email, and
14	+	/// mint a `claim_token` so the buyer can later attach the purchase to an account.
15	+	///
16	+	/// Guest purchases always land unclaimed (`buyer_id` NULL); attachment to a user
17	+	/// happens out of band via [`attach_guest_purchases_by_email`] at signup/email
18	+	/// verification.
19	+	#[tracing::instrument(skip_all)]
20	+	pub async fn complete_guest_transaction<'e>(
21	+	executor: impl sqlx::PgExecutor<'e>,
22	+	stripe_checkout_session_id: &str,
23	+	stripe_payment_intent_id: Option<&str>,
24	+	guest_email: &str,
25	+	) -> Result<Option<DbTransaction>> {
26	+	let claim_token = ClaimToken::new();
27	+
28	+	let tx = sqlx::query_as!(
29	+	DbTransaction,
30	+	r#"
31	+	UPDATE transactions
32	+	SET status = 'completed',
33	+	stripe_payment_intent_id = $2,
34	+	completed_at = NOW(),
35	+	guest_email = $3,
36	+	claim_token = $4,
37	+	buyer_id = NULL
38	+	WHERE stripe_checkout_session_id = $1
39	+	AND status = 'pending'
40	+	RETURNING
41	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
42	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
43	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
44	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
45	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
46	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
47	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
48	+	download_token AS "download_token: DownloadToken",
49	+	presentment_amount_cents, presentment_currency
50	+	"#,
51	+	stripe_checkout_session_id,
52	+	stripe_payment_intent_id,
53	+	guest_email,
54	+	claim_token as ClaimToken,
55	+	)
56	+	.fetch_optional(executor)
57	+	.await?;
58	+
59	+	Ok(tx)
60	+	}
61	+
62	+	/// Attach all unclaimed guest purchases for an email to a user account.
63	+	/// Called during signup/email verification to auto-claim prior guest purchases.
64	+	#[tracing::instrument(skip_all)]
65	+	pub async fn attach_guest_purchases_by_email(
66	+	pool: &PgPool,
67	+	email: &str,
68	+	user_id: UserId,
69	+	) -> Result<u64> {
70	+	let result = sqlx::query!(
71	+	r#"
72	+	UPDATE transactions
73	+	SET buyer_id = $1, claimed_by = $1, claim_token = NULL
74	+	WHERE LOWER(guest_email) = LOWER($2)
75	+	AND buyer_id IS NULL
76	+	AND status = 'completed'
77	+	"#,
78	+	user_id as UserId,
79	+	email,
80	+	)
81	+	.execute(pool)
82	+	.await?;
83	+
84	+	Ok(result.rows_affected())
85	+	}
86	+
87	+	/// Claim a single guest purchase by claim token.
88	+	#[tracing::instrument(skip_all)]
89	+	pub async fn claim_guest_purchase(
90	+	pool: &PgPool,
91	+	claim_token: ClaimToken,
92	+	user_id: UserId,
93	+	) -> Result<Option<DbTransaction>> {
94	+	let tx = sqlx::query_as!(
95	+	DbTransaction,
96	+	r#"
97	+	UPDATE transactions
98	+	SET buyer_id = $2, claimed_by = $2, claim_token = NULL
99	+	WHERE claim_token = $1
100	+	AND buyer_id IS NULL
101	+	AND status = 'completed'
102	+	RETURNING
103	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
104	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
105	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
106	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
107	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
108	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
109	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
110	+	download_token AS "download_token: DownloadToken",
111	+	presentment_amount_cents, presentment_currency
112	+	"#,
113	+	claim_token as ClaimToken,
114	+	user_id as UserId,
115	+	)
116	+	.fetch_optional(pool)
117	+	.await?;
118	+
119	+	Ok(tx)
120	+	}
121	+
122	+	/// Look up a completed transaction by download token (for guest download links).
123	+	#[tracing::instrument(skip_all)]
124	+	pub async fn get_transaction_by_download_token(
125	+	pool: &PgPool,
126	+	download_token: DownloadToken,
127	+	) -> Result<Option<DbTransaction>> {
128	+	let tx = sqlx::query_as!(
129	+	DbTransaction,
130	+	r#"
131	+	SELECT
132	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
133	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
134	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
135	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
136	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
137	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
138	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
139	+	download_token AS "download_token: DownloadToken",
140	+	presentment_amount_cents, presentment_currency
141	+	FROM transactions WHERE download_token = $1 AND status = 'completed'
142	+	"#,
143	+	download_token as DownloadToken,
144	+	)
145	+	.fetch_optional(pool)
146	+	.await?;
147	+
148	+	Ok(tx)
149	+	}
150	+
151	+	/// Create a completed free guest transaction.
152	+	///
153	+	/// Returns the number of rows inserted (0 if already claimed via ON CONFLICT).
154	+	#[allow(clippy::too_many_arguments)]
155	+	#[tracing::instrument(skip_all)]
156	+	pub async fn create_free_guest_transaction(
157	+	pool: &PgPool,
158	+	buyer_id: Option<UserId>,
159	+	seller_id: UserId,
160	+	item_id: ItemId,
161	+	checkout_session_id: &str,
162	+	item_title: &str,
163	+	seller_username: &str,
164	+	guest_email: &str,
165	+	claim_token: Option<ClaimToken>,
166	+	download_token: DownloadToken,
167	+	) -> std::result::Result<u64, sqlx::Error> {
168	+	let result = sqlx::query!(
169	+	r#"
170	+	INSERT INTO transactions (
171	+	buyer_id, seller_id, item_id, amount_cents, platform_fee_cents,
172	+	stripe_checkout_session_id, status, completed_at,
173	+	item_title, seller_username, share_contact,
174	+	guest_email, claim_token, download_token
175	+	)
176	+	VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, false, $7, $8, $9)
177	+	ON CONFLICT (guest_email, item_id) WHERE status = 'completed' AND guest_email IS NOT NULL DO NOTHING
178	+	"#,
179	+	buyer_id as Option<UserId>,
180	+	seller_id as UserId,
181	+	item_id as ItemId,
182	+	checkout_session_id,
183	+	item_title,
184	+	seller_username,
185	+	guest_email,
186	+	claim_token as Option<ClaimToken>,
187	+	download_token as DownloadToken,
188	+	)
189	+	.execute(pool)
190	+	.await?;
191	+
192	+	Ok(result.rows_affected())
193	+	}

A server/src/db/transactions/purchases/mod.rs +31

		@@ -1,0 +1,31 @@
1	+	//! Purchase queries: the pending-transaction lifecycle, free claims, refunds
2	+	//! and cart settlement.
3	+	//!
4	+	//! A paid checkout writes a `pending` row before the buyer leaves for Stripe,
5	+	//! and completion flips it to `completed`; nothing else moves a row between
6	+	//! those states. A pending row that is never completed is deleted by the
7	+	//! caller who abandoned it, or by `cleanup_stale_pending` past the age the
8	+	//! scheduler passes it (25h), which returns any `promo_code_id` so the
9	+	//! reservation is released with the row.
10	+	//!
11	+	//! The dedup these queries lean on is a set of partial unique indexes on
12	+	//! `transactions`, one pair per subject:
13	+	//! `(buyer_id, item_id)` and `(buyer_id, project_id)`, each once for
14	+	//! `status = 'pending'` and once for `status = 'completed'`, and
15	+	//! `(guest_email, item_id)` for `status = 'completed'`. All are partial on
16	+	//! the id being NOT NULL, so a project purchase (NULL `item_id`) does not
17	+	//! collide with an item purchase. `ON CONFLICT DO NOTHING` and the 23505
18	+	//! backstops throughout this file name those indexes; changing one means
19	+	//! revisiting every claim and checkout path here.
20	+
21	+	mod checkout;
22	+	mod claims;
23	+	mod guest;
24	+	mod reads;
25	+	mod refunds;
26	+
27	+	pub use checkout::*;
28	+	pub use claims::*;
29	+	pub use guest::*;
30	+	pub use reads::*;
31	+	pub use refunds::*;

A server/src/db/transactions/purchases/reads.rs +353

		@@ -1,0 +1,353 @@
1	+	//! Reading purchases back: what a buyer owns, what a seller sold, and the
2	+	//! paginated exports of both.
3	+
4	+	use super::super::{
5	+	Cents, ClaimToken, DbPurchaseRow, DbTransaction, DbTransactionExportRow, DownloadToken, ItemId,
6	+	KeyCode, PgPool, ProjectId, PromoCodeId, Result, TransactionId, UserId,
7	+	};
8	+
9	+	/// List transactions where the user is the buyer, newest first.
10	+	///
11	+	/// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
12	+	#[tracing::instrument(skip_all)]
13	+	pub async fn get_transactions_by_buyer(
14	+	pool: &PgPool,
15	+	buyer_id: UserId,
16	+	limit: Option<i64>,
17	+	) -> Result<Vec<DbTransaction>> {
18	+	let txs = sqlx::query_as!(
19	+	DbTransaction,
20	+	r#"
21	+	SELECT
22	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
23	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
24	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
25	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
26	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
27	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
28	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
29	+	download_token AS "download_token: DownloadToken",
30	+	presentment_amount_cents, presentment_currency
31	+	FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
32	+	"#,
33	+	buyer_id as UserId,
34	+	limit,
35	+	)
36	+	.fetch_all(pool)
37	+	.await?;
38	+
39	+	Ok(txs)
40	+	}
41	+
42	+	/// One page of a buyer's purchases for CSV export, newest first.
43	+	///
44	+	/// Paginated so the purchases export streams in bounded batches rather than
45	+	/// loading the buyer's whole history with `limit: None`.
46	+	/// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
47	+	pub async fn get_buyer_transactions_for_export_page(
48	+	pool: &PgPool,
49	+	buyer_id: UserId,
50	+	limit: i64,
51	+	offset: i64,
52	+	) -> Result<Vec<DbTransaction>> {
53	+	let txs = sqlx::query_as!(
54	+	DbTransaction,
55	+	r#"
56	+	SELECT
57	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
58	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
59	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
60	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
61	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
62	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
63	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
64	+	download_token AS "download_token: DownloadToken",
65	+	presentment_amount_cents, presentment_currency
66	+	FROM transactions WHERE buyer_id = $1
67	+	ORDER BY created_at DESC, id DESC
68	+	LIMIT $2 OFFSET $3
69	+	"#,
70	+	buyer_id as UserId,
71	+	limit,
72	+	offset,
73	+	)
74	+	.fetch_all(pool)
75	+	.await?;
76	+
77	+	Ok(txs)
78	+	}
79	+
80	+	/// List transactions where the user is the seller, newest first.
81	+	///
82	+	/// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
83	+	#[tracing::instrument(skip_all)]
84	+	pub async fn get_transactions_by_seller(
85	+	pool: &PgPool,
86	+	seller_id: UserId,
87	+	limit: Option<i64>,
88	+	) -> Result<Vec<DbTransaction>> {
89	+	let txs = sqlx::query_as!(
90	+	DbTransaction,
91	+	r#"
92	+	SELECT
93	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
94	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
95	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
96	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
97	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
98	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
99	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
100	+	download_token AS "download_token: DownloadToken",
101	+	presentment_amount_cents, presentment_currency
102	+	FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
103	+	"#,
104	+	seller_id as UserId,
105	+	limit,
106	+	)
107	+	.fetch_all(pool)
108	+	.await?;
109	+
110	+	Ok(txs)
111	+	}
112	+
113	+	/// Check whether a user has a completed purchase for a given item.
114	+	#[tracing::instrument(skip_all)]
115	+	pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
116	+	let count: i64 = sqlx::query_scalar!(
117	+	r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
118	+	user_id as UserId,
119	+	item_id as ItemId,
120	+	)
121	+	.fetch_one(pool)
122	+	.await?;
123	+
124	+	Ok(count > 0)
125	+	}
126	+
127	+	/// Bulk variant of `has_purchased_item`. Returns the subset of `item_ids` that
128	+	/// the buyer has a completed purchase for. Single DB roundtrip vs. N calls.
129	+	#[tracing::instrument(skip_all)]
130	+	pub async fn purchased_subset(
131	+	pool: &PgPool,
132	+	user_id: UserId,
133	+	item_ids: &[ItemId],
134	+	) -> Result<std::collections::HashSet<ItemId>> {
135	+	if item_ids.is_empty() {
136	+	return Ok(std::collections::HashSet::new());
137	+	}
138	+	let rows = sqlx::query_scalar!(
139	+	r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
140	+	WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
141	+	user_id as UserId,
142	+	item_ids as &[ItemId],
143	+	)
144	+	.fetch_all(pool)
145	+	.await?;
146	+	Ok(rows.into_iter().collect())
147	+	}
148	+
149	+	/// Get all item IDs that a user has purchased (for batch access checks)
150	+	#[tracing::instrument(skip_all)]
151	+	pub async fn get_user_purchased_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
152	+	let item_ids: Vec<ItemId> = sqlx::query_scalar!(
153	+	r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions WHERE buyer_id = $1 AND status = 'completed' AND item_id IS NOT NULL"#,
154	+	user_id as UserId,
155	+	)
156	+	.fetch_all(pool)
157	+	.await?;
158	+
159	+	Ok(item_ids)
160	+	}
161	+
162	+	/// Check whether a user has a completed purchase for a given project.
163	+	#[tracing::instrument(skip_all)]
164	+	pub async fn has_purchased_project(
165	+	pool: &PgPool,
166	+	user_id: UserId,
167	+	project_id: ProjectId,
168	+	) -> Result<bool> {
169	+	let count: i64 = sqlx::query_scalar!(
170	+	r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND project_id = $2 AND status = 'completed'"#,
171	+	user_id as UserId,
172	+	project_id as ProjectId,
173	+	)
174	+	.fetch_one(pool)
175	+	.await?;
176	+
177	+	Ok(count > 0)
178	+	}
179	+
180	+	/// Get items purchased by a user, including any associated license key.
181	+	///
182	+	/// Reads from the `purchases` VIEW (which filters `transactions` to
183	+	/// `status = 'completed'`), then JOINs through `items → projects → users`
184	+	/// for display fields. The LEFT JOIN on `license_keys` attaches the most
185	+	/// recent non-revoked key code so the buyer can see it in their library
186	+	/// without a separate lookup. Capped at 20 rows for the dashboard summary.
187	+	#[tracing::instrument(skip_all)]
188	+	pub async fn get_user_purchases(pool: &PgPool, user_id: UserId) -> Result<Vec<DbPurchaseRow>> {
189	+	let purchases = sqlx::query_as!(
190	+	DbPurchaseRow,
191	+	r#"
192	+	SELECT
193	+	transaction_id AS "transaction_id!: TransactionId",
194	+	item_id AS "item_id!: ItemId",
195	+	title AS "title!",
196	+	creator AS "creator!",
197	+	item_type AS "item_type!: crate::db::ItemType",
198	+	purchased_at AS "purchased_at!: chrono::DateTime<chrono::Utc>",
199	+	is_free AS "is_free!",
200	+	license_key_code AS "license_key_code?: KeyCode",
201	+	has_new_version AS "has_new_version!"
202	+	FROM (
203	+	SELECT DISTINCT ON (p.item_id)
204	+	p.transaction_id,
205	+	p.item_id,
206	+	i.title,
207	+	u.username as creator,
208	+	i.item_type,
209	+	p.purchased_at,
210	+	-- Badge from what the buyer actually paid, not the item's current
211	+	-- price: a later re-price to $0 must not retroactively badge a paid
212	+	-- purchase "Free" (nor vice-versa). The purchases view carries the
213	+	-- transaction's own amount_cents.
214	+	(p.amount_cents = 0) as is_free,
215	+	lk.key_code as license_key_code,
216	+	(vc.total_versions > 0 AND vc.total_versions > COALESCE(dc.downloaded_count, 0)) as has_new_version
217	+	FROM purchases p
218	+	JOIN items i ON p.item_id = i.id
219	+	JOIN projects proj ON i.project_id = proj.id
220	+	JOIN users u ON proj.user_id = u.id
221	+	LEFT JOIN license_keys lk ON lk.item_id = p.item_id AND lk.owner_id = p.buyer_id AND lk.revoked_at IS NULL
222	+	LEFT JOIN LATERAL (
223	+	SELECT COUNT(*) AS total_versions
224	+	FROM versions v
225	+	WHERE v.item_id = i.id AND v.s3_key IS NOT NULL
226	+	) vc ON true
227	+	LEFT JOIN LATERAL (
228	+	SELECT COUNT(*) AS downloaded_count
229	+	FROM user_downloads ud
230	+	WHERE ud.user_id = p.buyer_id AND ud.item_id = i.id
231	+	) dc ON true
232	+	WHERE p.buyer_id = $1
233	+	ORDER BY p.item_id, p.purchased_at DESC
234	+	) deduped
235	+	ORDER BY purchased_at DESC
236	+	LIMIT 20
237	+	"#,
238	+	user_id as UserId,
239	+	)
240	+	.fetch_all(pool)
241	+	.await?;
242	+
243	+	Ok(purchases)
244	+	}
245	+
246	+	/// Fetch a single transaction by ID.
247	+	///
248	+	/// # Authorization
249	+	///
250	+	/// This lookup is intentionally unscoped, it does not filter by buyer or
251	+	/// seller, because the two callers need the row to decide authorization
252	+	/// (a receipt page shown to buyer-or-seller; a refund restricted to the
253	+	/// seller). Every caller MUST therefore check ownership against the returned
254	+	/// `buyer_id`/`seller_id` before acting on it. A new caller that returns this
255	+	/// row's contents without such a check would be an IDOR, there is no implicit
256	+	/// scoping here to lean on.
257	+	#[tracing::instrument(skip_all)]
258	+	pub async fn get_transaction_by_id(
259	+	pool: &PgPool,
260	+	id: TransactionId,
261	+	) -> Result<Option<DbTransaction>> {
262	+	let tx = sqlx::query_as!(
263	+	DbTransaction,
264	+	r#"
265	+	SELECT
266	+	id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
267	+	item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
268	+	currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
269	+	created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
270	+	item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
271	+	parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
272	+	guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
273	+	download_token AS "download_token: DownloadToken",
274	+	presentment_amount_cents, presentment_currency
275	+	FROM transactions WHERE id = $1
276	+	"#,
277	+	id as TransactionId,
278	+	)
279	+	.fetch_optional(pool)
280	+	.await?;
281	+	Ok(tx)
282	+	}
283	+
284	+	#[tracing::instrument(skip_all)]
285	+	/// One page of a seller's sales for CSV export, newest first.
286	+	///
287	+	/// Paginated (`LIMIT`/`OFFSET`) so the export streams in bounded batches instead
288	+	/// of loading the seller's entire transaction history into memory in one query.
289	+	/// The `(created_at, id)` ordering is stable so OFFSET
290	+	/// batches don't reorder. (Keyset pagination would avoid OFFSET's deep-scan cost
291	+	/// and is the future optimization; OFFSET is sufficient at current scale and
292	+	/// keeps peak memory + per-query result bounded, which is the DoS fix.)
293	+	pub async fn get_seller_transactions_for_export_page(
294	+	pool: &PgPool,
295	+	seller_id: UserId,
296	+	limit: i64,
297	+	offset: i64,
298	+	) -> Result<Vec<DbTransactionExportRow>> {
299	+	let rows = sqlx::query_as!(
300	+	DbTransactionExportRow,
301	+	r#"
302	+	SELECT
303	+	t.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
304	+	t.item_id AS "item_id: ItemId",
305	+	t.item_title,
306	+	t.amount_cents AS "amount_cents: Cents",
307	+	t.status AS "status: crate::db::TransactionStatus",
308	+	CASE WHEN t.share_contact AND NOT EXISTS (
309	+	SELECT 1 FROM contact_revocations cr
310	+	WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
311	+	) THEN u.email ELSE NULL END as buyer_email
312	+	FROM transactions t
313	+	LEFT JOIN users u ON u.id = t.buyer_id
314	+	WHERE t.seller_id = $1
315	+	ORDER BY t.created_at DESC, t.id DESC
316	+	LIMIT $2 OFFSET $3
317	+	"#,
318	+	seller_id as UserId,
319	+	limit,
320	+	offset,
321	+	)
322	+	.fetch_all(pool)
323	+	.await?;
324	+
325	+	Ok(rows)
326	+	}
327	+
328	+	/// All of a seller's export rows accumulated into one Vec, bounded to
329	+	/// `EXPORT_ACCUMULATE_CAP` rows. For admin / internal-API callers that need the
330	+	/// full set in memory; the public creator-facing export streams page-by-page via
331	+	/// [`get_seller_transactions_for_export_page`] instead of materializing here.
332	+	pub async fn get_seller_transactions_for_export(
333	+	pool: &PgPool,
334	+	seller_id: UserId,
335	+	) -> Result<Vec<DbTransactionExportRow>> {
336	+	/// Page size for the accumulating fetch.
337	+	const PAGE: i64 = 5_000;
338	+	/// Cap so even an admin/internal export can't load an unbounded result set.
339	+	const EXPORT_ACCUMULATE_CAP: usize = 1_000_000;
340	+
341	+	let mut all = Vec::new();
342	+	let mut offset = 0i64;
343	+	loop {
344	+	let page = get_seller_transactions_for_export_page(pool, seller_id, PAGE, offset).await?;
345	+	let n = page.len();
346	+	all.extend(page);
347	+	offset += n as i64;
348	+	if (n as i64) < PAGE \|\| all.len() >= EXPORT_ACCUMULATE_CAP {
349	+	break;
350	+	}
351	+	}
352	+	Ok(all)
353	+	}

A server/src/db/transactions/purchases/refunds.rs +173

		@@ -1,0 +1,173 @@
1	+	//! Giving the money back, and proving a webhook has not already been handled.
2	+
3	+	use super::super::{ItemId, PgPool, Result, TransactionId};
4	+
5	+	/// Atomically claim a completed transaction for refund (`completed -> refunding`).
6	+	///
7	+	/// Returns `Some(id)` only if THIS call won the transition; returns `None` if the
8	+	/// row was not `completed` (already refunding, already refunded, or gone). The
9	+	/// self-service refund handler must call this BEFORE issuing the Stripe refund so
10	+	/// a rapid double-submit cannot pass the refundability check twice and over-refund
11	+	/// a shared-cart PaymentIntent. On Stripe error the handler calls
12	+	/// [`release_refund_claim`] to roll the row back to `completed`; on success the
13	+	/// `refund.created` webhook finalizes `refunding -> refunded`.
14	+	#[tracing::instrument(skip_all)]
15	+	pub async fn claim_transaction_for_refund(
16	+	pool: &PgPool,
17	+	id: TransactionId,
18	+	) -> Result<Option<TransactionId>> {
19	+	let row = sqlx::query_scalar!(
20	+	r#"
21	+	UPDATE transactions
22	+	SET status = 'refunding'
23	+	WHERE id = $1 AND status = 'completed'
24	+	RETURNING id AS "id: TransactionId"
25	+	"#,
26	+	id as TransactionId,
27	+	)
28	+	.fetch_optional(pool)
29	+	.await?;
30	+
31	+	Ok(row)
32	+	}
33	+
34	+	/// Release a refund claim (`refunding -> completed`) after a Stripe refund call
35	+	/// failed, so the creator can retry. Idempotent: only a row still in `refunding`
36	+	/// transitions; a row the webhook already finalized to `refunded` is left alone.
37	+	#[tracing::instrument(skip_all)]
38	+	pub async fn release_refund_claim(pool: &PgPool, id: TransactionId) -> Result<()> {
39	+	sqlx::query!(
40	+	r#"
41	+	UPDATE transactions
42	+	SET status = 'completed'
43	+	WHERE id = $1 AND status = 'refunding'
44	+	"#,
45	+	id as TransactionId,
46	+	)
47	+	.execute(pool)
48	+	.await?;
49	+
50	+	Ok(())
51	+	}
52	+
53	+	/// Mark a transaction as refunded, returning its ID and item_id for downstream cleanup.
54	+	///
55	+	/// The WHERE clause requires `status IN ('completed', 'refunding')` so that
56	+	/// already-refunded or pending transactions are not double-processed, while a row
57	+	/// the self-service handler has claimed (`refunding`) still finalizes. Returns an
58	+	/// empty vec if no matching transactions were found (idempotent for webhook retries).
59	+	///
60	+	/// Returns ALL refunded transactions (handles cart checkouts where multiple
61	+	/// transactions share the same payment_intent_id).
62	+	///
63	+	/// FULL-INTENT scope, and `pub(crate)` so only in-crate webhook handlers can
64	+	/// mint it: a single cart line must use the line-scoped
65	+	/// [`refund_transaction_by_id`] instead, never this PI-wide UPDATE, which would
66	+	/// refund a whole cart from one line's event.
67	+	#[tracing::instrument(skip_all)]
68	+	pub(crate) async fn refund_transaction_by_payment_intent<'e>(
69	+	executor: impl sqlx::PgExecutor<'e>,
70	+	payment_intent_id: &str,
71	+	) -> Result<Vec<(crate::db::TransactionId, Option<ItemId>)>> {
72	+	// item_id is nullable on project-level transactions (routes/stripe/checkout/project.rs);
73	+	// returning non-Optional ItemId would cause sqlx decode failures and infinite Stripe retries.
74	+	let rows = sqlx::query!(
75	+	r#"
76	+	UPDATE transactions
77	+	SET status = 'refunded'
78	+	WHERE stripe_payment_intent_id = $1 AND status IN ('completed', 'refunding')
79	+	RETURNING id AS "id: crate::db::TransactionId", item_id AS "item_id: ItemId"
80	+	"#,
81	+	payment_intent_id,
82	+	)
83	+	.fetch_all(executor)
84	+	.await?;
85	+
86	+	Ok(rows.into_iter().map(\|r\| (r.id, r.item_id)).collect())
87	+	}
88	+
89	+	/// Mark a SINGLE transaction refunded by id, returning `(id, item_id)` if it
90	+	/// transitioned from `completed` or `refunding` (the self-service handler claims
91	+	/// the row to `refunding` before calling Stripe). Returns `None` if it was already
92	+	/// refunded or otherwise not refundable (idempotent for webhook re-delivery).
93	+	///
94	+	/// Used by the line-scoped `refund.created` handler: cart lines share a
95	+	/// payment_intent, so refunding one line must touch only its own row, never the
96	+	/// PI-wide [`refund_transaction_by_payment_intent`].
97	+	#[tracing::instrument(skip_all)]
98	+	pub(crate) async fn refund_transaction_by_id<'e>(
99	+	executor: impl sqlx::PgExecutor<'e>,
100	+	id: TransactionId,
101	+	) -> Result<Option<(crate::db::TransactionId, Option<ItemId>)>> {
102	+	let row = sqlx::query!(
103	+	r#"
104	+	UPDATE transactions
105	+	SET status = 'refunded'
106	+	WHERE id = $1 AND status IN ('completed', 'refunding')
107	+	RETURNING id AS "id: crate::db::TransactionId", item_id AS "item_id: ItemId"
108	+	"#,
109	+	id as TransactionId,
110	+	)
111	+	.fetch_optional(executor)
112	+	.await?;
113	+
114	+	Ok(row.map(\|r\| (r.id, r.item_id)))
115	+	}
116	+
117	+	/// True if any transaction (any status) references this payment_intent. Lets the
118	+	/// `charge.refunded` handler tell "already refunded" (line-scoped refunds marked
119	+	/// the rows) apart from "genuinely unmatched" before queuing a pending refund.
120	+	pub async fn transaction_exists_for_payment_intent<'e>(
121	+	executor: impl sqlx::PgExecutor<'e>,
122	+	payment_intent_id: &str,
123	+	) -> Result<bool> {
124	+	let exists = sqlx::query_scalar!(
125	+	r#"SELECT EXISTS(SELECT 1 FROM transactions WHERE stripe_payment_intent_id = $1) AS "exists!""#,
126	+	payment_intent_id,
127	+	)
128	+	.fetch_one(executor)
129	+	.await?;
130	+
131	+	Ok(exists)
132	+	}
133	+
134	+	/// True if any transaction (any status) references this checkout session. Lets
135	+	/// the cart-completion webhook tell a benign duplicate delivery (rows already
136	+	/// completed) apart from an ORPHANED paid session (rows never created, buyer
137	+	/// charged, got nothing) so the latter is escalated.
138	+	pub async fn transaction_exists_for_checkout_session<'e>(
139	+	executor: impl sqlx::PgExecutor<'e>,
140	+	checkout_session_id: &str,
141	+	) -> Result<bool> {
142	+	let exists = sqlx::query_scalar!(
143	+	r#"SELECT EXISTS(SELECT 1 FROM transactions WHERE stripe_checkout_session_id = $1) AS "exists!""#,
144	+	checkout_session_id,
145	+	)
146	+	.fetch_one(executor)
147	+	.await?;
148	+
149	+	Ok(exists)
150	+	}
151	+
152	+	/// Revoke all child transactions linked to a parent (bundle) transaction.
153	+	///
154	+	/// Returns the item IDs of revoked children so callers can decrement sales counts.
155	+	#[tracing::instrument(skip_all)]
156	+	pub async fn revoke_child_transactions<'e>(
157	+	executor: impl sqlx::PgExecutor<'e>,
158	+	parent_transaction_id: TransactionId,
159	+	) -> Result<Vec<ItemId>> {
160	+	let item_ids = sqlx::query_scalar!(
161	+	r#"
162	+	UPDATE transactions
163	+	SET status = 'refunded'
164	+	WHERE parent_transaction_id = $1 AND status = 'completed'
165	+	RETURNING item_id AS "item_id: ItemId"
166	+	"#,
167	+	parent_transaction_id as TransactionId,
168	+	)
169	+	.fetch_all(executor)
170	+	.await?;
171	+
172	+	Ok(item_ids.into_iter().flatten().collect())
173	+	}