Skip to main content

max / makenotwork

56.8 KB · 1410 lines History Blame Raw
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. (A prior `existing_user_id` auto-attach parameter was always
116 /// passed `None` and has been removed, Run #1 NOTE, dead foot-gun.)
117 #[tracing::instrument(skip_all)]
118 pub async fn complete_guest_transaction<'e>(
119 executor: impl sqlx::PgExecutor<'e>,
120 stripe_checkout_session_id: &str,
121 stripe_payment_intent_id: Option<&str>,
122 guest_email: &str,
123 ) -> Result<Option<DbTransaction>> {
124 let claim_token = ClaimToken::new();
125
126 let tx = sqlx::query_as!(
127 DbTransaction,
128 r#"
129 UPDATE transactions
130 SET status = 'completed',
131 stripe_payment_intent_id = $2,
132 completed_at = NOW(),
133 guest_email = $3,
134 claim_token = $4,
135 buyer_id = NULL
136 WHERE stripe_checkout_session_id = $1
137 AND status = 'pending'
138 RETURNING
139 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
140 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
141 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
142 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
143 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
144 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
145 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
146 download_token AS "download_token: DownloadToken",
147 presentment_amount_cents, presentment_currency
148 "#,
149 stripe_checkout_session_id,
150 stripe_payment_intent_id,
151 guest_email,
152 claim_token as ClaimToken,
153 )
154 .fetch_optional(executor)
155 .await?;
156
157 Ok(tx)
158 }
159
160 /// Attach all unclaimed guest purchases for an email to a user account.
161 /// Called during signup/email verification to auto-claim prior guest purchases.
162 #[tracing::instrument(skip_all)]
163 pub async fn attach_guest_purchases_by_email(
164 pool: &PgPool,
165 email: &str,
166 user_id: UserId,
167 ) -> Result<u64> {
168 let result = sqlx::query!(
169 r#"
170 UPDATE transactions
171 SET buyer_id = $1, claimed_by = $1, claim_token = NULL
172 WHERE LOWER(guest_email) = LOWER($2)
173 AND buyer_id IS NULL
174 AND status = 'completed'
175 "#,
176 user_id as UserId,
177 email,
178 )
179 .execute(pool)
180 .await?;
181
182 Ok(result.rows_affected())
183 }
184
185 /// Claim a single guest purchase by claim token.
186 #[tracing::instrument(skip_all)]
187 pub async fn claim_guest_purchase(
188 pool: &PgPool,
189 claim_token: ClaimToken,
190 user_id: UserId,
191 ) -> Result<Option<DbTransaction>> {
192 let tx = sqlx::query_as!(
193 DbTransaction,
194 r#"
195 UPDATE transactions
196 SET buyer_id = $2, claimed_by = $2, claim_token = NULL
197 WHERE claim_token = $1
198 AND buyer_id IS NULL
199 AND status = 'completed'
200 RETURNING
201 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
202 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
203 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
204 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
205 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
206 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
207 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
208 download_token AS "download_token: DownloadToken",
209 presentment_amount_cents, presentment_currency
210 "#,
211 claim_token as ClaimToken,
212 user_id as UserId,
213 )
214 .fetch_optional(pool)
215 .await?;
216
217 Ok(tx)
218 }
219
220 /// Look up a completed transaction by download token (for guest download links).
221 #[tracing::instrument(skip_all)]
222 pub async fn get_transaction_by_download_token(
223 pool: &PgPool,
224 download_token: DownloadToken,
225 ) -> Result<Option<DbTransaction>> {
226 let tx = sqlx::query_as!(
227 DbTransaction,
228 r#"
229 SELECT
230 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
231 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
232 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
233 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
234 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
235 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
236 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
237 download_token AS "download_token: DownloadToken",
238 presentment_amount_cents, presentment_currency
239 FROM transactions WHERE download_token = $1 AND status = 'completed'
240 "#,
241 download_token as DownloadToken,
242 )
243 .fetch_optional(pool)
244 .await?;
245
246 Ok(tx)
247 }
248
249 /// Mark a pending transaction as completed (idempotent; returns `None` if already completed).
250 ///
251 /// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
252 /// can include this in a larger transaction when needed.
253 #[tracing::instrument(skip_all)]
254 pub async fn complete_transaction<'e>(
255 executor: impl sqlx::PgExecutor<'e>,
256 stripe_checkout_session_id: &str,
257 stripe_payment_intent_id: Option<&str>,
258 presentment: Option<(i64, &str)>,
259 ) -> Result<Option<DbTransaction>> {
260 // Only update if status is 'pending' for idempotency
261 // Returns None if transaction was already completed (duplicate webhook)
262 let tx = sqlx::query_as!(
263 DbTransaction,
264 r#"
265 UPDATE transactions
266 SET status = 'completed',
267 stripe_payment_intent_id = $2,
268 presentment_amount_cents = $3,
269 presentment_currency = $4,
270 completed_at = NOW()
271 WHERE stripe_checkout_session_id = $1
272 AND status = 'pending'
273 RETURNING
274 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
275 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
276 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
277 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
278 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
279 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
280 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
281 download_token AS "download_token: DownloadToken",
282 presentment_amount_cents, presentment_currency
283 "#,
284 stripe_checkout_session_id,
285 stripe_payment_intent_id,
286 presentment.map(|(cents, _)| cents),
287 presentment.map(|(_, currency)| currency),
288 )
289 .fetch_optional(executor)
290 .await?;
291
292 Ok(tx)
293 }
294
295 /// Complete ALL pending transactions for a cart checkout session.
296 /// Returns the list of completed transactions (empty if already processed).
297 #[tracing::instrument(skip_all)]
298 pub async fn complete_cart_transactions<'e>(
299 executor: impl sqlx::PgExecutor<'e>,
300 stripe_checkout_session_id: &str,
301 stripe_payment_intent_id: Option<&str>,
302 ) -> Result<Vec<DbTransaction>> {
303 let txs = sqlx::query_as!(
304 DbTransaction,
305 r#"
306 UPDATE transactions
307 SET status = 'completed',
308 stripe_payment_intent_id = $2,
309 completed_at = NOW()
310 WHERE stripe_checkout_session_id = $1
311 AND status = 'pending'
312 RETURNING
313 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
314 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
315 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
316 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
317 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
318 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
319 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
320 download_token AS "download_token: DownloadToken",
321 presentment_amount_cents, presentment_currency
322 "#,
323 stripe_checkout_session_id,
324 stripe_payment_intent_id,
325 )
326 .fetch_all(executor)
327 .await?;
328
329 Ok(txs)
330 }
331
332 /// Fetch all completed transactions for a checkout session.
333 ///
334 /// Used on the crash-recovery branch of the purchase/cart webhook handlers: when
335 /// `complete_transaction` / `complete_cart_transactions` return nothing (the
336 /// rows were already flipped to completed by a first attempt that crashed before
337 /// running finalize), this re-reads those completed rows so finalize can re-run
338 /// idempotently. Covers single and cart purchases since both key on the session.
339 #[tracing::instrument(skip_all)]
340 pub async fn get_completed_transactions_for_session<'e>(
341 executor: impl sqlx::PgExecutor<'e>,
342 stripe_checkout_session_id: &str,
343 ) -> Result<Vec<DbTransaction>> {
344 let txs = sqlx::query_as!(
345 DbTransaction,
346 r#"
347 SELECT
348 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
349 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
350 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
351 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
352 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
353 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
354 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
355 download_token AS "download_token: DownloadToken",
356 presentment_amount_cents, presentment_currency
357 FROM transactions
358 WHERE stripe_checkout_session_id = $1 AND status = 'completed'
359 "#,
360 stripe_checkout_session_id,
361 )
362 .fetch_all(executor)
363 .await?;
364
365 Ok(txs)
366 }
367
368 /// List transactions where the user is the buyer, newest first.
369 ///
370 /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
371 #[tracing::instrument(skip_all)]
372 pub async fn get_transactions_by_buyer(
373 pool: &PgPool,
374 buyer_id: UserId,
375 limit: Option<i64>,
376 ) -> Result<Vec<DbTransaction>> {
377 let txs = sqlx::query_as!(
378 DbTransaction,
379 r#"
380 SELECT
381 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
382 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
383 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
384 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
385 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
386 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
387 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
388 download_token AS "download_token: DownloadToken",
389 presentment_amount_cents, presentment_currency
390 FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
391 "#,
392 buyer_id as UserId,
393 limit,
394 )
395 .fetch_all(pool)
396 .await?;
397
398 Ok(txs)
399 }
400
401 /// One page of a buyer's purchases for CSV export, newest first.
402 ///
403 /// Paginated so the purchases export streams in bounded batches rather than
404 /// loading the buyer's whole history with `limit: None` (ultra-fuzz Run 4 S1).
405 /// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
406 pub async fn get_buyer_transactions_for_export_page(
407 pool: &PgPool,
408 buyer_id: UserId,
409 limit: i64,
410 offset: i64,
411 ) -> Result<Vec<DbTransaction>> {
412 let txs = sqlx::query_as!(
413 DbTransaction,
414 r#"
415 SELECT
416 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
417 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
418 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
419 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
420 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
421 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
422 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
423 download_token AS "download_token: DownloadToken",
424 presentment_amount_cents, presentment_currency
425 FROM transactions WHERE buyer_id = $1
426 ORDER BY created_at DESC, id DESC
427 LIMIT $2 OFFSET $3
428 "#,
429 buyer_id as UserId,
430 limit,
431 offset,
432 )
433 .fetch_all(pool)
434 .await?;
435
436 Ok(txs)
437 }
438
439 /// List transactions where the user is the seller, newest first.
440 ///
441 /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
442 #[tracing::instrument(skip_all)]
443 pub async fn get_transactions_by_seller(
444 pool: &PgPool,
445 seller_id: UserId,
446 limit: Option<i64>,
447 ) -> Result<Vec<DbTransaction>> {
448 let txs = sqlx::query_as!(
449 DbTransaction,
450 r#"
451 SELECT
452 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
453 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
454 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
455 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
456 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
457 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
458 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
459 download_token AS "download_token: DownloadToken",
460 presentment_amount_cents, presentment_currency
461 FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
462 "#,
463 seller_id as UserId,
464 limit,
465 )
466 .fetch_all(pool)
467 .await?;
468
469 Ok(txs)
470 }
471
472 /// Check whether a user has a completed purchase for a given item.
473 #[tracing::instrument(skip_all)]
474 pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
475 let count: i64 = sqlx::query_scalar!(
476 r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
477 user_id as UserId,
478 item_id as ItemId,
479 )
480 .fetch_one(pool)
481 .await?;
482
483 Ok(count > 0)
484 }
485
486 /// Bulk variant of `has_purchased_item`. Returns the subset of `item_ids` that
487 /// the buyer has a completed purchase for. Single DB roundtrip vs. N calls.
488 #[tracing::instrument(skip_all)]
489 pub async fn purchased_subset(
490 pool: &PgPool,
491 user_id: UserId,
492 item_ids: &[ItemId],
493 ) -> Result<std::collections::HashSet<ItemId>> {
494 if item_ids.is_empty() {
495 return Ok(std::collections::HashSet::new());
496 }
497 let rows = sqlx::query_scalar!(
498 r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
499 WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
500 user_id as UserId,
501 item_ids as &[ItemId],
502 )
503 .fetch_all(pool)
504 .await?;
505 Ok(rows.into_iter().collect())
506 }
507
508 /// Get all item IDs that a user has purchased (for batch access checks)
509 #[tracing::instrument(skip_all)]
510 pub async fn get_user_purchased_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
511 let item_ids: Vec<ItemId> = sqlx::query_scalar!(
512 r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions WHERE buyer_id = $1 AND status = 'completed' AND item_id IS NOT NULL"#,
513 user_id as UserId,
514 )
515 .fetch_all(pool)
516 .await?;
517
518 Ok(item_ids)
519 }
520
521 /// Claims a free item by creating a zero-cost completed transaction.
522 /// Returns true if claimed successfully, false if already in library.
523 ///
524 /// Uses `ON CONFLICT DO NOTHING` against the partial unique index on
525 /// `(buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL` to prevent duplicate
526 /// claims under concurrent requests.
527 #[tracing::instrument(skip_all)]
528 pub async fn claim_free_item<'e>(
529 executor: impl sqlx::PgExecutor<'e>,
530 params: &ClaimParams<'_>,
531 ) -> Result<bool> {
532 let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
533 let result = sqlx::query!(
534 r#"
535 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)
536 VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8, $9)
537 ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
538 "#,
539 params.buyer_id as UserId,
540 params.seller_id as UserId,
541 params.item_id as ItemId,
542 claim_id,
543 params.item_title,
544 params.seller_username,
545 params.share_contact,
546 params.parent_transaction_id as Option<TransactionId>,
547 params.platform_credit_cents,
548 )
549 .execute(executor)
550 .await?;
551
552 Ok(result.rows_affected() > 0)
553 }
554
555 /// Batch variant of [`claim_free_item`] for bundle grants: claims every child
556 /// item for one buyer in a single INSERT instead of N round-trips (each with its
557 /// own pool acquire) on the Stripe webhook / checkout hot path (Perf-S4, Run 9).
558 /// Each row is idempotent via the same partial-unique ON CONFLICT as the single
559 /// claim, and child items deliberately do not increment `sales_count`. Returns the
560 /// number of rows actually inserted. No-op on an empty slice.
561 #[tracing::instrument(skip_all)]
562 pub async fn claim_free_items_batch<'e>(
563 executor: impl sqlx::PgExecutor<'e>,
564 buyer_id: UserId,
565 seller_id: UserId,
566 seller_username: &str,
567 parent_transaction_id: Option<TransactionId>,
568 items: &[(ItemId, &str)],
569 ) -> Result<u64> {
570 if items.is_empty() {
571 return Ok(0);
572 }
573 let item_ids: Vec<ItemId> = items.iter().map(|(id, _)| *id).collect();
574 let item_titles: Vec<&str> = items.iter().map(|(_, title)| *title).collect();
575
576 // The per-row claim id mirrors the single claim's `free-claim-{buyer}-{item}`
577 // so a later single claim of the same item still collides idempotently.
578 let result = sqlx::query(
579 r"
580 INSERT INTO transactions
581 (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents,
582 stripe_checkout_session_id, status, completed_at, item_title,
583 seller_username, share_contact, parent_transaction_id)
584 SELECT
585 $1, $2, t.item_id, 0, 0,
586 'free-claim-' || $1::text || '-' || t.item_id::text,
587 'completed', NOW(), t.item_title, $3, false, $4
588 FROM UNNEST($5::uuid[], $6::text[]) AS t(item_id, item_title)
589 ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
590 ",
591 )
592 .bind(buyer_id)
593 .bind(seller_id)
594 .bind(seller_username)
595 .bind(parent_transaction_id)
596 .bind(&item_ids)
597 .bind(&item_titles)
598 .execute(executor)
599 .await?;
600
601 Ok(result.rows_affected())
602 }
603
604 /// Optional parameters for generating a license key inside a claim transaction.
605 pub struct LicenseKeyParams<'a> {
606 pub key_code: &'a KeyCode,
607 pub max_activations: Option<i32>,
608 }
609
610 /// Atomically claim a free item and increment the promo code's use count.
611 ///
612 /// Claims the item FIRST (INSERT transaction), then increments use_count.
613 /// If the user already owns the item (rows_affected == 0), rolls back without
614 /// consuming the code. If the code limit is reached, rolls back the claim too.
615 ///
616 /// When `license_key_params` is `Some`, a license key is created inside the
617 /// same transaction so that the claim and key are always consistent.
618 ///
619 /// Returns `(code_accepted, item_claimed)`:
620 /// - `code_accepted = false` → promo code hit its usage limit (nothing changed)
621 /// - `item_claimed = false` → user already owns the item (code was NOT consumed)
622 #[tracing::instrument(skip_all)]
623 pub async fn claim_free_with_promo_code(
624 pool: &PgPool,
625 promo_code_id: PromoCodeId,
626 params: &ClaimParams<'_>,
627 license_key_params: Option<&LicenseKeyParams<'_>>,
628 ) -> Result<(bool, bool)> {
629 let mut tx = pool.begin().await?;
630
631 // Claim the item first. When a platform-wide credit (Fan+)
632 // made the item free, `platform_credit_cents` carries the item's full price so
633 // the scheduler reimburses the creator via transfer, the fan pays nothing but
634 // the creator is still paid (MNW funds it).
635 let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
636 let result = sqlx::query!(
637 r#"
638 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)
639 VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8, $9)
640 ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
641 "#,
642 params.buyer_id as UserId,
643 params.seller_id as UserId,
644 params.item_id as ItemId,
645 claim_id,
646 params.item_title,
647 params.seller_username,
648 params.share_contact,
649 promo_code_id as PromoCodeId,
650 params.platform_credit_cents,
651 )
652 .execute(&mut *tx)
653 .await?;
654
655 let claimed = result.rows_affected() > 0;
656
657 if !claimed {
658 tx.rollback().await?;
659 return Ok((true, false));
660 }
661
662 // Increment the promo code use count. Re-check the full validity
663 // window (max_uses AND starts_at/expires_at) inside the atomic UPDATE, the
664 // pre-flight check ran outside this transaction, so a code expiring in the
665 // sub-ms gap must still be rejected here, not just on use-count (Run 11 Pay
666 // MINOR / TOCTOU).
667 let code_result = sqlx::query!(
668 r#"
669 UPDATE promo_codes SET use_count = use_count + 1
670 WHERE id = $1
671 AND (max_uses IS NULL OR use_count < max_uses)
672 AND (starts_at IS NULL OR starts_at <= NOW())
673 AND (expires_at IS NULL OR expires_at > NOW())
674 "#,
675 promo_code_id as PromoCodeId,
676 )
677 .execute(&mut *tx)
678 .await?;
679
680 if code_result.rows_affected() == 0 {
681 tx.rollback().await?;
682 return Ok((false, false));
683 }
684
685 crate::db::items::increment_sales_count(&mut *tx, params.item_id).await?;
686
687 // Create the license key inside the same transaction if requested.
688 // Retry once on a unique-violation: the wordlist generator has ~6B-coin-
689 // flip headroom, so an actual collision is vanishingly rare, but the
690 // alternative is surfacing a 500 to a buyer mid-claim, cheap to handle.
691 if let Some(lk) = license_key_params {
692 let attempt = sqlx::query!(
693 r#"
694 INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
695 VALUES ($1, $2, NULL, $3, $4)
696 "#,
697 params.item_id as ItemId,
698 params.buyer_id as UserId,
699 lk.key_code as &KeyCode,
700 lk.max_activations,
701 )
702 .execute(&mut *tx)
703 .await;
704
705 if let Err(sqlx::Error::Database(e)) = &attempt
706 && e.code().as_deref() == Some("23505")
707 {
708 let retry_code = crate::helpers::generate_key_code();
709 tracing::warn!(item_id = %params.item_id, "license key 23505 collision; retrying once");
710 sqlx::query!(
711 r#"
712 INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
713 VALUES ($1, $2, NULL, $3, $4)
714 "#,
715 params.item_id as ItemId,
716 params.buyer_id as UserId,
717 retry_code as KeyCode,
718 lk.max_activations,
719 )
720 .execute(&mut *tx)
721 .await?;
722 } else {
723 attempt?;
724 }
725 }
726
727 tx.commit().await?;
728 Ok((true, true))
729 }
730
731 // ── Project purchases ──
732
733 /// Check whether a user has a completed purchase for a given project.
734 #[tracing::instrument(skip_all)]
735 pub async fn has_purchased_project(
736 pool: &PgPool,
737 user_id: UserId,
738 project_id: ProjectId,
739 ) -> Result<bool> {
740 let count: i64 = sqlx::query_scalar!(
741 r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND project_id = $2 AND status = 'completed'"#,
742 user_id as UserId,
743 project_id as ProjectId,
744 )
745 .fetch_one(pool)
746 .await?;
747
748 Ok(count > 0)
749 }
750
751 /// Parameters for creating a pending project purchase transaction.
752 pub struct CreateProjectTransactionParams<'a> {
753 pub buyer_id: UserId,
754 pub seller_id: UserId,
755 pub project_id: ProjectId,
756 pub amount_cents: i32,
757 pub stripe_checkout_session_id: &'a str,
758 pub project_title: &'a str,
759 pub seller_username: &'a str,
760 pub share_contact: bool,
761 }
762
763 /// Record a new pending transaction for a project purchase.
764 #[tracing::instrument(skip_all)]
765 pub async fn create_project_transaction(
766 pool: &PgPool,
767 params: &CreateProjectTransactionParams<'_>,
768 ) -> Result<DbTransaction> {
769 let tx = sqlx::query_as!(
770 DbTransaction,
771 r#"
772 INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact)
773 VALUES ($1, $2, $3, $4, 0, $5, $6, $7, $8)
774 RETURNING
775 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
776 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
777 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
778 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
779 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
780 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
781 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
782 download_token AS "download_token: DownloadToken",
783 presentment_amount_cents, presentment_currency
784 "#,
785 params.buyer_id as UserId,
786 params.seller_id as UserId,
787 params.project_id as ProjectId,
788 params.amount_cents,
789 params.stripe_checkout_session_id,
790 params.project_title,
791 params.seller_username,
792 params.share_contact,
793 )
794 .fetch_one(pool)
795 .await?;
796
797 Ok(tx)
798 }
799
800 /// Get items purchased by a user, including any associated license key.
801 ///
802 /// Reads from the `purchases` VIEW (which filters `transactions` to
803 /// `status = 'completed'`), then JOINs through `items → projects → users`
804 /// for display fields. The LEFT JOIN on `license_keys` attaches the most
805 /// recent non-revoked key code so the buyer can see it in their library
806 /// without a separate lookup. Capped at 20 rows for the dashboard summary.
807 #[tracing::instrument(skip_all)]
808 pub async fn get_user_purchases(pool: &PgPool, user_id: UserId) -> Result<Vec<DbPurchaseRow>> {
809 let purchases = sqlx::query_as!(
810 DbPurchaseRow,
811 r#"
812 SELECT
813 transaction_id AS "transaction_id!: TransactionId",
814 item_id AS "item_id!: ItemId",
815 title AS "title!",
816 creator AS "creator!",
817 item_type AS "item_type!: crate::db::ItemType",
818 purchased_at AS "purchased_at!: chrono::DateTime<chrono::Utc>",
819 is_free AS "is_free!",
820 license_key_code AS "license_key_code?: KeyCode",
821 has_new_version AS "has_new_version!"
822 FROM (
823 SELECT DISTINCT ON (p.item_id)
824 p.transaction_id,
825 p.item_id,
826 i.title,
827 u.username as creator,
828 i.item_type,
829 p.purchased_at,
830 -- Badge from what the buyer actually paid, not the item's current
831 -- price: a later re-price to $0 must not retroactively badge a paid
832 -- purchase "Free" (nor vice-versa). The purchases view carries the
833 -- transaction's own amount_cents.
834 (p.amount_cents = 0) as is_free,
835 lk.key_code as license_key_code,
836 (vc.total_versions > 0 AND vc.total_versions > COALESCE(dc.downloaded_count, 0)) as has_new_version
837 FROM purchases p
838 JOIN items i ON p.item_id = i.id
839 JOIN projects proj ON i.project_id = proj.id
840 JOIN users u ON proj.user_id = u.id
841 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
842 LEFT JOIN LATERAL (
843 SELECT COUNT(*) AS total_versions
844 FROM versions v
845 WHERE v.item_id = i.id AND v.s3_key IS NOT NULL
846 ) vc ON true
847 LEFT JOIN LATERAL (
848 SELECT COUNT(*) AS downloaded_count
849 FROM user_downloads ud
850 WHERE ud.user_id = p.buyer_id AND ud.item_id = i.id
851 ) dc ON true
852 WHERE p.buyer_id = $1
853 ORDER BY p.item_id, p.purchased_at DESC
854 ) deduped
855 ORDER BY purchased_at DESC
856 LIMIT 20
857 "#,
858 user_id as UserId,
859 )
860 .fetch_all(pool)
861 .await?;
862
863 Ok(purchases)
864 }
865
866 /// Remove a free item from library (deletes the claim transaction).
867 /// If the claim was via a promo code, decrements the code's use_count.
868 #[tracing::instrument(skip_all)]
869 pub async fn remove_free_item_from_library(
870 pool: &PgPool,
871 user_id: UserId,
872 item_id: ItemId,
873 ) -> Result<bool> {
874 // Delete the free claim and return the promo_code_id if one was used
875 let row: Option<Option<crate::db::PromoCodeId>> = sqlx::query_scalar!(
876 r#"
877 DELETE FROM transactions
878 WHERE buyer_id = $1 AND item_id = $2 AND amount_cents = 0 AND status = 'completed'
879 RETURNING promo_code_id AS "promo_code_id: crate::db::PromoCodeId"
880 "#,
881 user_id as UserId,
882 item_id as ItemId,
883 )
884 .fetch_optional(pool)
885 .await?;
886
887 let deleted = row.is_some();
888
889 if let Some(Some(pc_id)) = row {
890 crate::db::promo_codes::release_use_count(pool, pc_id)
891 .await
892 .ok();
893 }
894
895 Ok(deleted)
896 }
897
898 /// Fetch a single transaction by ID.
899 ///
900 /// # Authorization
901 ///
902 /// This lookup is intentionally **unscoped**, it does not filter by buyer or
903 /// seller, because the two callers need the row to *decide* authorization
904 /// (a receipt page shown to buyer-or-seller; a refund restricted to the
905 /// seller). Every caller MUST therefore check ownership against the returned
906 /// `buyer_id`/`seller_id` before acting on it. A new caller that returns this
907 /// row's contents without such a check would be an IDOR, there is no implicit
908 /// scoping here to lean on.
909 #[tracing::instrument(skip_all)]
910 pub async fn get_transaction_by_id(
911 pool: &PgPool,
912 id: TransactionId,
913 ) -> Result<Option<DbTransaction>> {
914 let tx = sqlx::query_as!(
915 DbTransaction,
916 r#"
917 SELECT
918 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
919 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
920 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
921 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
922 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
923 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
924 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
925 download_token AS "download_token: DownloadToken",
926 presentment_amount_cents, presentment_currency
927 FROM transactions WHERE id = $1
928 "#,
929 id as TransactionId,
930 )
931 .fetch_optional(pool)
932 .await?;
933 Ok(tx)
934 }
935
936 /// Atomically claim a completed transaction for refund (`completed -> refunding`).
937 ///
938 /// Returns `Some(id)` only if THIS call won the transition; returns `None` if the
939 /// row was not `completed` (already refunding, already refunded, or gone). The
940 /// self-service refund handler must call this BEFORE issuing the Stripe refund so
941 /// a rapid double-submit cannot pass the refundability check twice and over-refund
942 /// a shared-cart PaymentIntent (Pay-S1, Run 9). On Stripe error the handler calls
943 /// [`release_refund_claim`] to roll the row back to `completed`; on success the
944 /// `refund.created` webhook finalizes `refunding -> refunded`.
945 #[tracing::instrument(skip_all)]
946 pub async fn claim_transaction_for_refund(
947 pool: &PgPool,
948 id: TransactionId,
949 ) -> Result<Option<TransactionId>> {
950 let row = sqlx::query_scalar!(
951 r#"
952 UPDATE transactions
953 SET status = 'refunding'
954 WHERE id = $1 AND status = 'completed'
955 RETURNING id AS "id: TransactionId"
956 "#,
957 id as TransactionId,
958 )
959 .fetch_optional(pool)
960 .await?;
961
962 Ok(row)
963 }
964
965 /// Release a refund claim (`refunding -> completed`) after a Stripe refund call
966 /// failed, so the creator can retry. Idempotent: only a row still in `refunding`
967 /// transitions; a row the webhook already finalized to `refunded` is left alone.
968 #[tracing::instrument(skip_all)]
969 pub async fn release_refund_claim(pool: &PgPool, id: TransactionId) -> Result<()> {
970 sqlx::query!(
971 r#"
972 UPDATE transactions
973 SET status = 'completed'
974 WHERE id = $1 AND status = 'refunding'
975 "#,
976 id as TransactionId,
977 )
978 .execute(pool)
979 .await?;
980
981 Ok(())
982 }
983
984 /// Mark a transaction as refunded, returning its ID and item_id for downstream cleanup.
985 ///
986 /// The WHERE clause requires `status IN ('completed', 'refunding')` so that
987 /// already-refunded or pending transactions are not double-processed, while a row
988 /// the self-service handler has claimed (`refunding`) still finalizes. Returns an
989 /// empty vec if no matching transactions were found (idempotent for webhook retries).
990 ///
991 /// Returns ALL refunded transactions (handles cart checkouts where multiple
992 /// transactions share the same payment_intent_id).
993 ///
994 /// FULL-INTENT scope, and `pub(crate)` so only in-crate webhook handlers can
995 /// mint it: a single cart line must use the line-scoped
996 /// [`refund_transaction_by_id`] instead, never this PI-wide UPDATE (the Run #2
997 /// Payments SERIOUS that refunded a whole cart from one line's event).
998 #[tracing::instrument(skip_all)]
999 pub(crate) async fn refund_transaction_by_payment_intent<'e>(
1000 executor: impl sqlx::PgExecutor<'e>,
1001 payment_intent_id: &str,
1002 ) -> Result<Vec<(crate::db::TransactionId, Option<ItemId>)>> {
1003 // item_id is nullable on project-level transactions (routes/stripe/checkout/project.rs);
1004 // returning non-Optional ItemId would cause sqlx decode failures and infinite Stripe retries.
1005 let rows = sqlx::query!(
1006 r#"
1007 UPDATE transactions
1008 SET status = 'refunded'
1009 WHERE stripe_payment_intent_id = $1 AND status IN ('completed', 'refunding')
1010 RETURNING id AS "id: crate::db::TransactionId", item_id AS "item_id: ItemId"
1011 "#,
1012 payment_intent_id,
1013 )
1014 .fetch_all(executor)
1015 .await?;
1016
1017 Ok(rows.into_iter().map(|r| (r.id, r.item_id)).collect())
1018 }
1019
1020 /// Mark a SINGLE transaction refunded by id, returning `(id, item_id)` if it
1021 /// transitioned from `completed` or `refunding` (the self-service handler claims
1022 /// the row to `refunding` before calling Stripe). Returns `None` if it was already
1023 /// refunded or otherwise not refundable (idempotent for webhook re-delivery).
1024 ///
1025 /// Used by the line-scoped `refund.created` handler: cart lines share a
1026 /// payment_intent, so refunding one line must touch only its own row, never the
1027 /// PI-wide [`refund_transaction_by_payment_intent`] (Run #2 Payments SERIOUS).
1028 #[tracing::instrument(skip_all)]
1029 pub(crate) async fn refund_transaction_by_id<'e>(
1030 executor: impl sqlx::PgExecutor<'e>,
1031 id: TransactionId,
1032 ) -> Result<Option<(crate::db::TransactionId, Option<ItemId>)>> {
1033 let row = sqlx::query!(
1034 r#"
1035 UPDATE transactions
1036 SET status = 'refunded'
1037 WHERE id = $1 AND status IN ('completed', 'refunding')
1038 RETURNING id AS "id: crate::db::TransactionId", item_id AS "item_id: ItemId"
1039 "#,
1040 id as TransactionId,
1041 )
1042 .fetch_optional(executor)
1043 .await?;
1044
1045 Ok(row.map(|r| (r.id, r.item_id)))
1046 }
1047
1048 /// True if any transaction (any status) references this payment_intent. Lets the
1049 /// `charge.refunded` handler tell "already refunded" (line-scoped refunds marked
1050 /// the rows) apart from "genuinely unmatched" before queuing a pending refund.
1051 pub async fn transaction_exists_for_payment_intent<'e>(
1052 executor: impl sqlx::PgExecutor<'e>,
1053 payment_intent_id: &str,
1054 ) -> Result<bool> {
1055 let exists = sqlx::query_scalar!(
1056 r#"SELECT EXISTS(SELECT 1 FROM transactions WHERE stripe_payment_intent_id = $1) AS "exists!""#,
1057 payment_intent_id,
1058 )
1059 .fetch_one(executor)
1060 .await?;
1061
1062 Ok(exists)
1063 }
1064
1065 /// True if any transaction (any status) references this checkout session. Lets
1066 /// the cart-completion webhook tell a benign duplicate delivery (rows already
1067 /// completed) apart from an ORPHANED paid session (rows never created, buyer
1068 /// charged, got nothing) so the latter is escalated (Run #2 Payments SERIOUS).
1069 pub async fn transaction_exists_for_checkout_session<'e>(
1070 executor: impl sqlx::PgExecutor<'e>,
1071 checkout_session_id: &str,
1072 ) -> Result<bool> {
1073 let exists = sqlx::query_scalar!(
1074 r#"SELECT EXISTS(SELECT 1 FROM transactions WHERE stripe_checkout_session_id = $1) AS "exists!""#,
1075 checkout_session_id,
1076 )
1077 .fetch_one(executor)
1078 .await?;
1079
1080 Ok(exists)
1081 }
1082
1083 /// Revoke all child transactions linked to a parent (bundle) transaction.
1084 ///
1085 /// Returns the item IDs of revoked children so callers can decrement sales counts.
1086 #[tracing::instrument(skip_all)]
1087 pub async fn revoke_child_transactions<'e>(
1088 executor: impl sqlx::PgExecutor<'e>,
1089 parent_transaction_id: TransactionId,
1090 ) -> Result<Vec<ItemId>> {
1091 let item_ids = sqlx::query_scalar!(
1092 r#"
1093 UPDATE transactions
1094 SET status = 'refunded'
1095 WHERE parent_transaction_id = $1 AND status = 'completed'
1096 RETURNING item_id AS "item_id: ItemId"
1097 "#,
1098 parent_transaction_id as TransactionId,
1099 )
1100 .fetch_all(executor)
1101 .await?;
1102
1103 Ok(item_ids.into_iter().flatten().collect())
1104 }
1105
1106 /// Get seller transactions for CSV export, with conditional buyer email.
1107 ///
1108 /// Respects contact revocations: if a buyer revoked sharing, their email
1109 /// is hidden even if `share_contact` was true on the transaction.
1110 #[tracing::instrument(skip_all)]
1111 /// One page of a seller's sales for CSV export, newest first.
1112 ///
1113 /// Paginated (`LIMIT`/`OFFSET`) so the export streams in bounded batches instead
1114 /// of loading the seller's entire transaction history into memory in one query
1115 /// (ultra-fuzz Run 4 S1). The `(created_at, id)` ordering is stable so OFFSET
1116 /// batches don't reorder. (Keyset pagination would avoid OFFSET's deep-scan cost
1117 /// and is the future optimization; OFFSET is sufficient at current scale and
1118 /// keeps peak memory + per-query result bounded, which is the DoS fix.)
1119 pub async fn get_seller_transactions_for_export_page(
1120 pool: &PgPool,
1121 seller_id: UserId,
1122 limit: i64,
1123 offset: i64,
1124 ) -> Result<Vec<DbTransactionExportRow>> {
1125 let rows = sqlx::query_as!(
1126 DbTransactionExportRow,
1127 r#"
1128 SELECT
1129 t.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
1130 t.item_id AS "item_id: ItemId",
1131 t.item_title,
1132 t.amount_cents AS "amount_cents: Cents",
1133 t.status AS "status: crate::db::TransactionStatus",
1134 CASE WHEN t.share_contact AND NOT EXISTS (
1135 SELECT 1 FROM contact_revocations cr
1136 WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
1137 ) THEN u.email ELSE NULL END as buyer_email
1138 FROM transactions t
1139 LEFT JOIN users u ON u.id = t.buyer_id
1140 WHERE t.seller_id = $1
1141 ORDER BY t.created_at DESC, t.id DESC
1142 LIMIT $2 OFFSET $3
1143 "#,
1144 seller_id as UserId,
1145 limit,
1146 offset,
1147 )
1148 .fetch_all(pool)
1149 .await?;
1150
1151 Ok(rows)
1152 }
1153
1154 /// All of a seller's export rows accumulated into one Vec, bounded to
1155 /// `EXPORT_ACCUMULATE_CAP` rows. For admin / internal-API callers that need the
1156 /// full set in memory; the public creator-facing export streams page-by-page via
1157 /// [`get_seller_transactions_for_export_page`] instead of materializing here.
1158 pub async fn get_seller_transactions_for_export(
1159 pool: &PgPool,
1160 seller_id: UserId,
1161 ) -> Result<Vec<DbTransactionExportRow>> {
1162 /// Page size for the accumulating fetch.
1163 const PAGE: i64 = 5_000;
1164 /// Cap so even an admin/internal export can't load an unbounded result set.
1165 const EXPORT_ACCUMULATE_CAP: usize = 1_000_000;
1166
1167 let mut all = Vec::new();
1168 let mut offset = 0i64;
1169 loop {
1170 let page = get_seller_transactions_for_export_page(pool, seller_id, PAGE, offset).await?;
1171 let n = page.len();
1172 all.extend(page);
1173 offset += n as i64;
1174 if (n as i64) < PAGE || all.len() >= EXPORT_ACCUMULATE_CAP {
1175 break;
1176 }
1177 }
1178 Ok(all)
1179 }
1180
1181 /// Create a pending "placeholder" transaction for a subscription checkout that
1182 /// used a promo code. This row exists solely so `cleanup_stale_pending` can
1183 /// release the promo code reservation if the buyer abandons the Stripe session.
1184 /// It is deleted (not completed) when the subscription webhook fires.
1185 #[tracing::instrument(skip_all)]
1186 pub async fn create_subscription_pending_transaction(
1187 pool: &PgPool,
1188 buyer_id: UserId,
1189 seller_id: UserId,
1190 project_id: ProjectId,
1191 stripe_checkout_session_id: &str,
1192 promo_code_id: PromoCodeId,
1193 ) -> Result<()> {
1194 sqlx::query!(
1195 r#"
1196 INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents,
1197 stripe_checkout_session_id, item_title, seller_username, share_contact, promo_code_id)
1198 VALUES ($1, $2, $3, 0, 0, $4, 'subscription-promo-hold', '', false, $5)
1199 "#,
1200 buyer_id as UserId,
1201 seller_id as UserId,
1202 project_id as ProjectId,
1203 stripe_checkout_session_id,
1204 promo_code_id as PromoCodeId,
1205 )
1206 .execute(pool)
1207 .await?;
1208
1209 Ok(())
1210 }
1211
1212 /// Delete a pending subscription promo-hold transaction by checkout session ID.
1213 /// Called from the subscription webhook after the subscription is created.
1214 #[tracing::instrument(skip_all)]
1215 pub async fn delete_subscription_pending_transaction<'e>(
1216 executor: impl sqlx::PgExecutor<'e>,
1217 stripe_checkout_session_id: &str,
1218 ) -> Result<()> {
1219 sqlx::query!(
1220 "DELETE FROM transactions WHERE stripe_checkout_session_id = $1 AND status = 'pending'",
1221 stripe_checkout_session_id,
1222 )
1223 .execute(executor)
1224 .await?;
1225
1226 Ok(())
1227 }
1228
1229 /// Delete stale pending transactions (older than the given threshold) and return
1230 /// the promo_code_ids that need their use_count decremented.
1231 ///
1232 /// Stripe checkout sessions expire after 24 hours, so pending transactions older
1233 /// than that will never complete. This releases the pending purchase uniqueness
1234 /// slot and any reserved promo code use_count.
1235 #[tracing::instrument(skip_all)]
1236 pub async fn cleanup_stale_pending(
1237 pool: &PgPool,
1238 older_than: chrono::Duration,
1239 ) -> Result<Vec<Option<crate::db::PromoCodeId>>> {
1240 let cutoff = chrono::Utc::now() - older_than;
1241 // 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.
1242 let rows: Vec<(Option<crate::db::PromoCodeId>,)> = sqlx::query_as(
1243 r"
1244 DELETE FROM transactions
1245 WHERE status = 'pending'
1246 AND created_at < $1
1247 RETURNING promo_code_id
1248 ",
1249 )
1250 .bind(cutoff)
1251 .fetch_all(pool)
1252 .await?;
1253
1254 Ok(rows.into_iter().map(|(id,)| id).collect())
1255 }
1256
1257 /// Bulk variant of `get_pending_item_purchase`. Returns the subset of `item_ids`
1258 /// for which the buyer already has a `pending` transaction. Used by cart
1259 /// checkout to abort early when any line item would collide with the partial
1260 /// unique index on `(buyer_id, item_id) WHERE status = 'pending'`.
1261 #[tracing::instrument(skip_all)]
1262 pub async fn pending_subset(
1263 pool: &PgPool,
1264 buyer_id: UserId,
1265 item_ids: &[ItemId],
1266 ) -> Result<std::collections::HashSet<ItemId>> {
1267 if item_ids.is_empty() {
1268 return Ok(std::collections::HashSet::new());
1269 }
1270 let rows = sqlx::query_scalar!(
1271 r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
1272 WHERE buyer_id = $1 AND status = 'pending' AND item_id = ANY($2)"#,
1273 buyer_id as UserId,
1274 item_ids as &[ItemId],
1275 )
1276 .fetch_all(pool)
1277 .await?;
1278 Ok(rows.into_iter().collect())
1279 }
1280
1281 /// Returns the buyer's pending transaction for a specific item, if any.
1282 /// Used to surface in-progress checkouts on the purchase page.
1283 #[tracing::instrument(skip_all)]
1284 pub async fn get_pending_item_purchase(
1285 pool: &PgPool,
1286 buyer_id: UserId,
1287 item_id: ItemId,
1288 ) -> Result<Option<(TransactionId, chrono::DateTime<chrono::Utc>)>> {
1289 let row = sqlx::query!(
1290 r#"
1291 SELECT id AS "id: TransactionId", created_at AS "created_at: chrono::DateTime<chrono::Utc>"
1292 FROM transactions
1293 WHERE buyer_id = $1 AND item_id = $2 AND status = 'pending'
1294 LIMIT 1
1295 "#,
1296 buyer_id as UserId,
1297 item_id as ItemId,
1298 )
1299 .fetch_optional(pool)
1300 .await?;
1301
1302 Ok(row.map(|r| (r.id, r.created_at)))
1303 }
1304
1305 /// Delete the buyer's pending transaction for a specific item.
1306 /// Returns any released `promo_code_id` so the caller can release its
1307 /// reservation.
1308 #[tracing::instrument(skip_all)]
1309 pub async fn delete_pending_item_purchase(
1310 pool: &PgPool,
1311 buyer_id: UserId,
1312 item_id: ItemId,
1313 ) -> Result<Option<crate::db::PromoCodeId>> {
1314 let row: Option<Option<crate::db::PromoCodeId>> = sqlx::query_scalar!(
1315 r#"
1316 DELETE FROM transactions
1317 WHERE buyer_id = $1 AND item_id = $2 AND status = 'pending'
1318 RETURNING promo_code_id AS "promo_code_id: crate::db::PromoCodeId"
1319 "#,
1320 buyer_id as UserId,
1321 item_id as ItemId,
1322 )
1323 .fetch_optional(pool)
1324 .await?;
1325
1326 Ok(row.flatten())
1327 }
1328
1329 /// Create a completed free guest transaction.
1330 ///
1331 /// Returns the number of rows inserted (0 if already claimed via ON CONFLICT).
1332 #[allow(clippy::too_many_arguments)]
1333 #[tracing::instrument(skip_all)]
1334 pub async fn create_free_guest_transaction(
1335 pool: &PgPool,
1336 buyer_id: Option<UserId>,
1337 seller_id: UserId,
1338 item_id: ItemId,
1339 checkout_session_id: &str,
1340 item_title: &str,
1341 seller_username: &str,
1342 guest_email: &str,
1343 claim_token: Option<ClaimToken>,
1344 download_token: DownloadToken,
1345 ) -> std::result::Result<u64, sqlx::Error> {
1346 let result = sqlx::query!(
1347 r#"
1348 INSERT INTO transactions (
1349 buyer_id, seller_id, item_id, amount_cents, platform_fee_cents,
1350 stripe_checkout_session_id, status, completed_at,
1351 item_title, seller_username, share_contact,
1352 guest_email, claim_token, download_token
1353 )
1354 VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, false, $7, $8, $9)
1355 ON CONFLICT (guest_email, item_id) WHERE status = 'completed' AND guest_email IS NOT NULL DO NOTHING
1356 "#,
1357 buyer_id as Option<UserId>,
1358 seller_id as UserId,
1359 item_id as ItemId,
1360 checkout_session_id,
1361 item_title,
1362 seller_username,
1363 guest_email,
1364 claim_token as Option<ClaimToken>,
1365 download_token as DownloadToken,
1366 )
1367 .execute(pool)
1368 .await?;
1369
1370 Ok(result.rows_affected())
1371 }
1372
1373 /// Record a free project claim (PWYW with $0 min or free project).
1374 ///
1375 /// Returns `true` if the claim was actually inserted, `false` if the buyer
1376 /// already owned the project. Mirrors the `claim_free_item` shape so callers
1377 /// can gate downstream side-effects (contact-revocation clear, sale-notification
1378 /// email, etc.) on the winner of a concurrent-claim race, without this signal,
1379 /// two concurrent `/checkout/project` POSTs both fire those side-effects
1380 /// regardless of which one's INSERT actually landed (Run #7 deferred SERIOUS).
1381 #[tracing::instrument(skip_all)]
1382 pub async fn claim_free_project(
1383 pool: &PgPool,
1384 buyer_id: UserId,
1385 seller_id: UserId,
1386 project_id: ProjectId,
1387 item_title: &str,
1388 seller_username: &str,
1389 share_contact: bool,
1390 ) -> Result<bool> {
1391 let result = sqlx::query!(
1392 r#"
1393 INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents,
1394 status, completed_at, item_title, seller_username, share_contact)
1395 VALUES ($1, $2, $3, 0, 0, 'completed', NOW(), $4, $5, $6)
1396 ON CONFLICT (buyer_id, project_id) WHERE status = 'completed' AND project_id IS NOT NULL DO NOTHING
1397 "#,
1398 buyer_id as UserId,
1399 seller_id as UserId,
1400 project_id as ProjectId,
1401 item_title,
1402 seller_username,
1403 share_contact,
1404 )
1405 .execute(pool)
1406 .await?;
1407
1408 Ok(result.rows_affected() > 0)
1409 }
1410