Skip to main content

max / makenotwork

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