Skip to main content

max / makenotwork

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