Skip to main content

max / makenotwork

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