Skip to main content

max / makenotwork

Split db/transactions into purchases + seller_contacts + revenue_stats The 1730-line transactions.rs carried three concerns on the transaction table: core purchase/refund/claim flow, buyer-contact/seller-relationship queries, and revenue analytics. Split into a transactions/ directory — purchases.rs (core checkout/claim/refund/pending), seller_contacts.rs (buyer lists, seller contacts, contact sharing), and revenue_stats.rs (per-project/per-user/platform revenue rollups) — with mod.rs re-exporting all three. The db::transactions::* path is preserved (call sites unchanged); inline `super::` type references were rewritten to `crate::db::` for the added module depth. No behavior change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 13:02 UTC
Commit: 1ea57412591d2645ecff335f40c7f0866ea8d4c6
Parent: c431087
5 files changed, +876 insertions, -500 deletions
D server/src/db/transactions.rs -500
@@ -1,1730 +0,0 @@
1 - //! Transaction queries: checkout tracking, purchase history, and free claims.
2 -
3 - use chrono::{DateTime, Utc};
4 - use sqlx::PgPool;
5 -
6 - use super::models::*;
7 - use super::validated_types::KeyCode;
8 - use super::{Cents, ClaimToken, DownloadToken, ItemId, ProjectId, PromoCodeId, TransactionId, UserId};
9 - use crate::error::Result;
10 -
11 - /// Parameters for creating a pending Stripe checkout transaction.
12 - pub struct CreateTransactionParams<'a> {
13 - pub buyer_id: Option<UserId>,
14 - pub seller_id: UserId,
15 - /// `None` for project-level purchases (no specific item).
16 - pub item_id: Option<ItemId>,
17 - pub amount_cents: Cents,
18 - pub platform_fee_cents: Cents,
19 - pub stripe_checkout_session_id: &'a str,
20 - pub item_title: &'a str,
21 - pub seller_username: &'a str,
22 - pub share_contact: bool,
23 - /// Set for project-level purchases; `None` for item purchases.
24 - pub project_id: Option<ProjectId>,
25 - /// Promo code used for this checkout (for releasing reservations on cleanup).
26 - pub promo_code_id: Option<PromoCodeId>,
27 - /// Guest buyer's email (set for guest checkouts, None for logged-in).
28 - pub guest_email: Option<&'a str>,
29 - /// Cents MNW owes the seller as reimbursement for a platform-funded credit
30 - /// (the Fan+ renewal credit) applied to this sale; `0` for ordinary sales. The
31 - /// scheduler settles it via a platform -> connected transfer once the
32 - /// transaction completes (see `db::platform_credits`).
33 - pub platform_credit_cents: i64,
34 - }
35 -
36 - /// Common parameters for claiming a free item (direct, discount code, or download code).
37 - pub struct ClaimParams<'a> {
38 - pub buyer_id: UserId,
39 - pub item_id: ItemId,
40 - pub seller_id: UserId,
41 - pub item_title: &'a str,
42 - pub seller_username: &'a str,
43 - pub share_contact: bool,
44 - /// If this claim was granted via a bundle purchase, the parent transaction ID.
45 - pub parent_transaction_id: Option<TransactionId>,
46 - /// Cents MNW owes the seller when a platform-wide (Fan+) credit made this item
47 - /// free — the seller is reimbursed the item's price via a platform transfer so
48 - /// they are still paid. `0` for ordinary free claims (genuinely-free items,
49 - /// seller-issued free-access codes, bundle grants).
50 - pub platform_credit_cents: i64,
51 - }
52 -
53 - /// Record a new pending transaction for a Stripe checkout session.
54 - #[tracing::instrument(skip_all)]
55 - pub async fn create_transaction<'e>(
56 - executor: impl sqlx::PgExecutor<'e>,
57 - params: &CreateTransactionParams<'_>,
58 - ) -> Result<DbTransaction> {
59 - let tx = sqlx::query_as!(
60 - DbTransaction,
61 - r#"
62 - 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)
63 - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
64 - RETURNING
65 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
66 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
67 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
68 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
69 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
70 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
71 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
72 - download_token AS "download_token: DownloadToken"
73 - "#,
74 - params.buyer_id as Option<UserId>,
75 - params.seller_id as UserId,
76 - params.item_id as Option<ItemId>,
77 - params.amount_cents as Cents,
78 - params.platform_fee_cents as Cents,
79 - params.stripe_checkout_session_id,
80 - params.item_title,
81 - params.seller_username,
82 - params.share_contact,
83 - params.project_id as Option<ProjectId>,
84 - params.promo_code_id as Option<PromoCodeId>,
85 - params.guest_email,
86 - params.platform_credit_cents,
87 - )
88 - .fetch_one(executor)
89 - .await?;
90 -
91 - Ok(tx)
92 - }
93 -
94 - /// Complete a guest transaction: mark it completed, record the guest email, and
95 - /// mint a `claim_token` so the buyer can later attach the purchase to an account.
96 - ///
97 - /// Guest purchases always land unclaimed (`buyer_id` NULL); attachment to a user
98 - /// happens out of band via [`attach_guest_purchases_by_email`] at signup/email
99 - /// verification. (A prior `existing_user_id` auto-attach parameter was always
100 - /// passed `None` and has been removed — Run #1 NOTE, dead foot-gun.)
101 - #[tracing::instrument(skip_all)]
102 - pub async fn complete_guest_transaction<'e>(
103 - executor: impl sqlx::PgExecutor<'e>,
104 - stripe_checkout_session_id: &str,
105 - stripe_payment_intent_id: Option<&str>,
106 - guest_email: &str,
107 - ) -> Result<Option<DbTransaction>> {
108 - let claim_token = ClaimToken::new();
109 -
110 - let tx = sqlx::query_as!(
111 - DbTransaction,
112 - r#"
113 - UPDATE transactions
114 - SET status = 'completed',
115 - stripe_payment_intent_id = $2,
116 - completed_at = NOW(),
117 - guest_email = $3,
118 - claim_token = $4,
119 - buyer_id = NULL
120 - WHERE stripe_checkout_session_id = $1
121 - AND status = 'pending'
122 - RETURNING
123 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
124 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
125 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
126 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
127 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
128 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
129 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
130 - download_token AS "download_token: DownloadToken"
131 - "#,
132 - stripe_checkout_session_id,
133 - stripe_payment_intent_id,
134 - guest_email,
135 - claim_token as ClaimToken,
136 - )
137 - .fetch_optional(executor)
138 - .await?;
139 -
140 - Ok(tx)
141 - }
142 -
143 - /// Attach all unclaimed guest purchases for an email to a user account.
144 - /// Called during signup/email verification to auto-claim prior guest purchases.
145 - #[tracing::instrument(skip_all)]
146 - pub async fn attach_guest_purchases_by_email(
147 - pool: &PgPool,
148 - email: &str,
149 - user_id: UserId,
150 - ) -> Result<u64> {
151 - let result = sqlx::query!(
152 - r#"
153 - UPDATE transactions
154 - SET buyer_id = $1, claimed_by = $1, claim_token = NULL
155 - WHERE LOWER(guest_email) = LOWER($2)
156 - AND buyer_id IS NULL
157 - AND status = 'completed'
158 - "#,
159 - user_id as UserId,
160 - email,
161 - )
162 - .execute(pool)
163 - .await?;
164 -
165 - Ok(result.rows_affected())
166 - }
167 -
168 - /// Claim a single guest purchase by claim token.
169 - #[tracing::instrument(skip_all)]
170 - pub async fn claim_guest_purchase(
171 - pool: &PgPool,
172 - claim_token: ClaimToken,
173 - user_id: UserId,
174 - ) -> Result<Option<DbTransaction>> {
175 - let tx = sqlx::query_as!(
176 - DbTransaction,
177 - r#"
178 - UPDATE transactions
179 - SET buyer_id = $2, claimed_by = $2, claim_token = NULL
180 - WHERE claim_token = $1
181 - AND buyer_id IS NULL
182 - AND status = 'completed'
183 - RETURNING
184 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
185 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
186 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
187 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
188 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
189 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
190 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
191 - download_token AS "download_token: DownloadToken"
192 - "#,
193 - claim_token as ClaimToken,
194 - user_id as UserId,
195 - )
196 - .fetch_optional(pool)
197 - .await?;
198 -
199 - Ok(tx)
200 - }
201 -
202 - /// Look up a completed transaction by download token (for guest download links).
203 - #[tracing::instrument(skip_all)]
204 - pub async fn get_transaction_by_download_token(
205 - pool: &PgPool,
206 - download_token: DownloadToken,
207 - ) -> Result<Option<DbTransaction>> {
208 - let tx = sqlx::query_as!(
209 - DbTransaction,
210 - r#"
211 - SELECT
212 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
213 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
214 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
215 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
216 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
217 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
218 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
219 - download_token AS "download_token: DownloadToken"
220 - FROM transactions WHERE download_token = $1 AND status = 'completed'
221 - "#,
222 - download_token as DownloadToken,
223 - )
224 - .fetch_optional(pool)
225 - .await?;
226 -
227 - Ok(tx)
228 - }
229 -
230 - /// Mark a pending transaction as completed (idempotent; returns `None` if already completed).
231 - ///
232 - /// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
233 - /// can include this in a larger transaction when needed.
234 - #[tracing::instrument(skip_all)]
235 - pub async fn complete_transaction<'e>(
236 - executor: impl sqlx::PgExecutor<'e>,
237 - stripe_checkout_session_id: &str,
238 - stripe_payment_intent_id: Option<&str>,
239 - ) -> Result<Option<DbTransaction>> {
240 - // Only update if status is 'pending' for idempotency
241 - // Returns None if transaction was already completed (duplicate webhook)
242 - let tx = sqlx::query_as!(
243 - DbTransaction,
244 - r#"
245 - UPDATE transactions
246 - SET status = 'completed',
247 - stripe_payment_intent_id = $2,
248 - completed_at = NOW()
249 - WHERE stripe_checkout_session_id = $1
250 - AND status = 'pending'
251 - RETURNING
252 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
253 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
254 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
255 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
256 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
257 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
258 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
259 - download_token AS "download_token: DownloadToken"
260 - "#,
261 - stripe_checkout_session_id,
262 - stripe_payment_intent_id,
263 - )
264 - .fetch_optional(executor)
265 - .await?;
266 -
267 - Ok(tx)
268 - }
269 -
270 - /// Complete ALL pending transactions for a cart checkout session.
271 - /// Returns the list of completed transactions (empty if already processed).
272 - #[tracing::instrument(skip_all)]
273 - pub async fn complete_cart_transactions<'e>(
274 - executor: impl sqlx::PgExecutor<'e>,
275 - stripe_checkout_session_id: &str,
276 - stripe_payment_intent_id: Option<&str>,
277 - ) -> Result<Vec<DbTransaction>> {
278 - let txs = sqlx::query_as!(
279 - DbTransaction,
280 - r#"
281 - UPDATE transactions
282 - SET status = 'completed',
283 - stripe_payment_intent_id = $2,
284 - completed_at = NOW()
285 - WHERE stripe_checkout_session_id = $1
286 - AND status = 'pending'
287 - RETURNING
288 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
289 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
290 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
291 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
292 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
293 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
294 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
295 - download_token AS "download_token: DownloadToken"
296 - "#,
297 - stripe_checkout_session_id,
298 - stripe_payment_intent_id,
299 - )
300 - .fetch_all(executor)
301 - .await?;
302 -
303 - Ok(txs)
304 - }
305 -
306 - /// Fetch all completed transactions for a checkout session.
307 - ///
308 - /// Used on the crash-recovery branch of the purchase/cart webhook handlers: when
309 - /// `complete_transaction` / `complete_cart_transactions` return nothing (the
310 - /// rows were already flipped to completed by a first attempt that crashed before
311 - /// running finalize), this re-reads those completed rows so finalize can re-run
312 - /// idempotently. Covers single and cart purchases since both key on the session.
313 - #[tracing::instrument(skip_all)]
314 - pub async fn get_completed_transactions_for_session<'e>(
315 - executor: impl sqlx::PgExecutor<'e>,
316 - stripe_checkout_session_id: &str,
317 - ) -> Result<Vec<DbTransaction>> {
318 - let txs = sqlx::query_as!(
319 - DbTransaction,
320 - r#"
321 - SELECT
322 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
323 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
324 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
325 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
326 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
327 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
328 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
329 - download_token AS "download_token: DownloadToken"
330 - FROM transactions
331 - WHERE stripe_checkout_session_id = $1 AND status = 'completed'
332 - "#,
333 - stripe_checkout_session_id,
334 - )
335 - .fetch_all(executor)
336 - .await?;
337 -
338 - Ok(txs)
339 - }
340 -
341 - /// List transactions where the user is the buyer, newest first.
342 - ///
343 - /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
344 - #[tracing::instrument(skip_all)]
345 - pub async fn get_transactions_by_buyer(
346 - pool: &PgPool,
347 - buyer_id: UserId,
348 - limit: Option<i64>,
349 - ) -> Result<Vec<DbTransaction>> {
350 - let txs = sqlx::query_as!(
351 - DbTransaction,
352 - r#"
353 - SELECT
354 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
355 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
356 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
357 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
358 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
359 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
360 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
361 - download_token AS "download_token: DownloadToken"
362 - FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
363 - "#,
364 - buyer_id as UserId,
365 - limit,
366 - )
367 - .fetch_all(pool)
368 - .await?;
369 -
370 - Ok(txs)
371 - }
372 -
373 - /// One page of a buyer's purchases for CSV export, newest first.
374 - ///
375 - /// Paginated so the purchases export streams in bounded batches rather than
376 - /// loading the buyer's whole history with `limit: None` (ultra-fuzz Run 4 S1).
377 - /// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
378 - pub async fn get_buyer_transactions_for_export_page(
379 - pool: &PgPool,
380 - buyer_id: UserId,
381 - limit: i64,
382 - offset: i64,
383 - ) -> Result<Vec<DbTransaction>> {
384 - let txs = sqlx::query_as!(
385 - DbTransaction,
386 - r#"
387 - SELECT
388 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
389 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
390 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
391 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
392 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
393 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
394 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
395 - download_token AS "download_token: DownloadToken"
396 - FROM transactions WHERE buyer_id = $1
397 - ORDER BY created_at DESC, id DESC
398 - LIMIT $2 OFFSET $3
399 - "#,
400 - buyer_id as UserId,
401 - limit,
402 - offset,
403 - )
404 - .fetch_all(pool)
405 - .await?;
406 -
407 - Ok(txs)
408 - }
409 -
410 - /// List transactions where the user is the seller, newest first.
411 - ///
412 - /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
413 - #[tracing::instrument(skip_all)]
414 - pub async fn get_transactions_by_seller(
415 - pool: &PgPool,
416 - seller_id: UserId,
417 - limit: Option<i64>,
418 - ) -> Result<Vec<DbTransaction>> {
419 - let txs = sqlx::query_as!(
420 - DbTransaction,
421 - r#"
422 - SELECT
423 - id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
424 - item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
425 - currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
426 - created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
427 - item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
428 - parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
429 - guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
430 - download_token AS "download_token: DownloadToken"
431 - FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
432 - "#,
433 - seller_id as UserId,
434 - limit,
435 - )
436 - .fetch_all(pool)
437 - .await?;
438 -
439 - Ok(txs)
440 - }
441 -
442 - /// Check whether a user has a completed purchase for a given item.
443 - #[tracing::instrument(skip_all)]
444 - pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
445 - let count: i64 = sqlx::query_scalar!(
446 - r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
447 - user_id as UserId,
448 - item_id as ItemId,
449 - )
450 - .fetch_one(pool)
451 - .await?;
452 -
453 - Ok(count > 0)
454 - }
455 -
456 - /// Bulk variant of `has_purchased_item`. Returns the subset of `item_ids` that
457 - /// the buyer has a completed purchase for. Single DB roundtrip vs. N calls.
458 - #[tracing::instrument(skip_all)]
459 - pub async fn purchased_subset(
460 - pool: &PgPool,
461 - user_id: UserId,
462 - item_ids: &[ItemId],
463 - ) -> Result<std::collections::HashSet<ItemId>> {
464 - if item_ids.is_empty() {
465 - return Ok(std::collections::HashSet::new());
466 - }
467 - let rows = sqlx::query_scalar!(
468 - r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
469 - WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
470 - user_id as UserId,
471 - item_ids as &[ItemId],
472 - )
473 - .fetch_all(pool)
474 - .await?;
475 - Ok(rows.into_iter().collect())
476 - }
477 -
478 - /// Get all item IDs that a user has purchased (for batch access checks)
479 - #[tracing::instrument(skip_all)]
480 - pub async fn get_user_purchased_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
481 - let item_ids: Vec<ItemId> = sqlx::query_scalar!(
482 - r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions WHERE buyer_id = $1 AND status = 'completed' AND item_id IS NOT NULL"#,
483 - user_id as UserId,
484 - )
485 - .fetch_all(pool)
486 - .await?;
487 -
488 - Ok(item_ids)
489 - }
490 -
491 - /// Claims a free item by creating a zero-cost completed transaction.
492 - /// Returns true if claimed successfully, false if already in library.
493 - ///
494 - /// Uses `ON CONFLICT DO NOTHING` against the partial unique index on
495 - /// `(buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL` to prevent duplicate
496 - /// claims under concurrent requests.
497 - #[tracing::instrument(skip_all)]
498 - pub async fn claim_free_item<'e>(
499 - executor: impl sqlx::PgExecutor<'e>,
500 - params: &ClaimParams<'_>,
Lines truncated
A server/src/db/transactions/mod.rs +17
@@ -0,0 +1,17 @@
1 + //! Transaction queries: checkout tracking, purchase history, and free claims.
2 +
3 + use chrono::{DateTime, Utc};
4 + use sqlx::PgPool;
5 +
6 + use super::models::*;
7 + use super::validated_types::KeyCode;
8 + use super::{Cents, ClaimToken, DownloadToken, ItemId, ProjectId, PromoCodeId, TransactionId, UserId};
9 + use crate::error::Result;
10 +
11 + mod purchases;
12 + mod revenue_stats;
13 + mod seller_contacts;
14 +
15 + pub use purchases::*;
16 + pub use revenue_stats::*;
17 + pub use seller_contacts::*;
A server/src/db/transactions/purchases.rs +500
@@ -0,0 +1,1365 @@
1 + use super::*;
2 +
3 + /// Parameters for creating a pending Stripe checkout transaction.
4 + pub struct CreateTransactionParams<'a> {
5 + pub buyer_id: Option<UserId>,
6 + pub seller_id: UserId,
7 + /// `None` for project-level purchases (no specific item).
8 + pub item_id: Option<ItemId>,
9 + pub amount_cents: Cents,
10 + pub platform_fee_cents: Cents,
11 + pub stripe_checkout_session_id: &'a str,
12 + pub item_title: &'a str,
13 + pub seller_username: &'a str,
14 + pub share_contact: bool,
15 + /// Set for project-level purchases; `None` for item purchases.
16 + pub project_id: Option<ProjectId>,
17 + /// Promo code used for this checkout (for releasing reservations on cleanup).
18 + pub promo_code_id: Option<PromoCodeId>,
19 + /// Guest buyer's email (set for guest checkouts, None for logged-in).
20 + pub guest_email: Option<&'a str>,
21 + /// Cents MNW owes the seller as reimbursement for a platform-funded credit
22 + /// (the Fan+ renewal credit) applied to this sale; `0` for ordinary sales. The
23 + /// scheduler settles it via a platform -> connected transfer once the
24 + /// transaction completes (see `db::platform_credits`).
25 + pub platform_credit_cents: i64,
26 + }
27 +
28 + /// Common parameters for claiming a free item (direct, discount code, or download code).
29 + pub struct ClaimParams<'a> {
30 + pub buyer_id: UserId,
31 + pub item_id: ItemId,
32 + pub seller_id: UserId,
33 + pub item_title: &'a str,
34 + pub seller_username: &'a str,
35 + pub share_contact: bool,
36 + /// If this claim was granted via a bundle purchase, the parent transaction ID.
37 + pub parent_transaction_id: Option<TransactionId>,
38 + /// Cents MNW owes the seller when a platform-wide (Fan+) credit made this item
39 + /// free — the seller is reimbursed the item's price via a platform transfer so
40 + /// they are still paid. `0` for ordinary free claims (genuinely-free items,
41 + /// seller-issued free-access codes, bundle grants).
42 + pub platform_credit_cents: i64,
43 + }
44 +
45 + /// Record a new pending transaction for a Stripe checkout session.
46 + #[tracing::instrument(skip_all)]
47 + pub async fn create_transaction<'e>(
48 + executor: impl sqlx::PgExecutor<'e>,
49 + params: &CreateTransactionParams<'_>,
50 + ) -> Result<DbTransaction> {
51 + let tx = sqlx::query_as!(
52 + DbTransaction,
53 + r#"
54 + 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)
55 + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
56 + RETURNING
57 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
58 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
59 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
60 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
61 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
62 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
63 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
64 + download_token AS "download_token: DownloadToken"
65 + "#,
66 + params.buyer_id as Option<UserId>,
67 + params.seller_id as UserId,
68 + params.item_id as Option<ItemId>,
69 + params.amount_cents as Cents,
70 + params.platform_fee_cents as Cents,
71 + params.stripe_checkout_session_id,
72 + params.item_title,
73 + params.seller_username,
74 + params.share_contact,
75 + params.project_id as Option<ProjectId>,
76 + params.promo_code_id as Option<PromoCodeId>,
77 + params.guest_email,
78 + params.platform_credit_cents,
79 + )
80 + .fetch_one(executor)
81 + .await?;
82 +
83 + Ok(tx)
84 + }
85 +
86 + /// Complete a guest transaction: mark it completed, record the guest email, and
87 + /// mint a `claim_token` so the buyer can later attach the purchase to an account.
88 + ///
89 + /// Guest purchases always land unclaimed (`buyer_id` NULL); attachment to a user
90 + /// happens out of band via [`attach_guest_purchases_by_email`] at signup/email
91 + /// verification. (A prior `existing_user_id` auto-attach parameter was always
92 + /// passed `None` and has been removed — Run #1 NOTE, dead foot-gun.)
93 + #[tracing::instrument(skip_all)]
94 + pub async fn complete_guest_transaction<'e>(
95 + executor: impl sqlx::PgExecutor<'e>,
96 + stripe_checkout_session_id: &str,
97 + stripe_payment_intent_id: Option<&str>,
98 + guest_email: &str,
99 + ) -> Result<Option<DbTransaction>> {
100 + let claim_token = ClaimToken::new();
101 +
102 + let tx = sqlx::query_as!(
103 + DbTransaction,
104 + r#"
105 + UPDATE transactions
106 + SET status = 'completed',
107 + stripe_payment_intent_id = $2,
108 + completed_at = NOW(),
109 + guest_email = $3,
110 + claim_token = $4,
111 + buyer_id = NULL
112 + WHERE stripe_checkout_session_id = $1
113 + AND status = 'pending'
114 + RETURNING
115 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
116 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
117 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
118 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
119 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
120 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
121 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
122 + download_token AS "download_token: DownloadToken"
123 + "#,
124 + stripe_checkout_session_id,
125 + stripe_payment_intent_id,
126 + guest_email,
127 + claim_token as ClaimToken,
128 + )
129 + .fetch_optional(executor)
130 + .await?;
131 +
132 + Ok(tx)
133 + }
134 +
135 + /// Attach all unclaimed guest purchases for an email to a user account.
136 + /// Called during signup/email verification to auto-claim prior guest purchases.
137 + #[tracing::instrument(skip_all)]
138 + pub async fn attach_guest_purchases_by_email(
139 + pool: &PgPool,
140 + email: &str,
141 + user_id: UserId,
142 + ) -> Result<u64> {
143 + let result = sqlx::query!(
144 + r#"
145 + UPDATE transactions
146 + SET buyer_id = $1, claimed_by = $1, claim_token = NULL
147 + WHERE LOWER(guest_email) = LOWER($2)
148 + AND buyer_id IS NULL
149 + AND status = 'completed'
150 + "#,
151 + user_id as UserId,
152 + email,
153 + )
154 + .execute(pool)
155 + .await?;
156 +
157 + Ok(result.rows_affected())
158 + }
159 +
160 + /// Claim a single guest purchase by claim token.
161 + #[tracing::instrument(skip_all)]
162 + pub async fn claim_guest_purchase(
163 + pool: &PgPool,
164 + claim_token: ClaimToken,
165 + user_id: UserId,
166 + ) -> Result<Option<DbTransaction>> {
167 + let tx = sqlx::query_as!(
168 + DbTransaction,
169 + r#"
170 + UPDATE transactions
171 + SET buyer_id = $2, claimed_by = $2, claim_token = NULL
172 + WHERE claim_token = $1
173 + AND buyer_id IS NULL
174 + AND status = 'completed'
175 + RETURNING
176 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
177 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
178 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
179 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
180 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
181 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
182 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
183 + download_token AS "download_token: DownloadToken"
184 + "#,
185 + claim_token as ClaimToken,
186 + user_id as UserId,
187 + )
188 + .fetch_optional(pool)
189 + .await?;
190 +
191 + Ok(tx)
192 + }
193 +
194 + /// Look up a completed transaction by download token (for guest download links).
195 + #[tracing::instrument(skip_all)]
196 + pub async fn get_transaction_by_download_token(
197 + pool: &PgPool,
198 + download_token: DownloadToken,
199 + ) -> Result<Option<DbTransaction>> {
200 + let tx = sqlx::query_as!(
201 + DbTransaction,
202 + r#"
203 + SELECT
204 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
205 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
206 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
207 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
208 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
209 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
210 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
211 + download_token AS "download_token: DownloadToken"
212 + FROM transactions WHERE download_token = $1 AND status = 'completed'
213 + "#,
214 + download_token as DownloadToken,
215 + )
216 + .fetch_optional(pool)
217 + .await?;
218 +
219 + Ok(tx)
220 + }
221 +
222 + /// Mark a pending transaction as completed (idempotent; returns `None` if already completed).
223 + ///
224 + /// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
225 + /// can include this in a larger transaction when needed.
226 + #[tracing::instrument(skip_all)]
227 + pub async fn complete_transaction<'e>(
228 + executor: impl sqlx::PgExecutor<'e>,
229 + stripe_checkout_session_id: &str,
230 + stripe_payment_intent_id: Option<&str>,
231 + ) -> Result<Option<DbTransaction>> {
232 + // Only update if status is 'pending' for idempotency
233 + // Returns None if transaction was already completed (duplicate webhook)
234 + let tx = sqlx::query_as!(
235 + DbTransaction,
236 + r#"
237 + UPDATE transactions
238 + SET status = 'completed',
239 + stripe_payment_intent_id = $2,
240 + completed_at = NOW()
241 + WHERE stripe_checkout_session_id = $1
242 + AND status = 'pending'
243 + RETURNING
244 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
245 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
246 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
247 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
248 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
249 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
250 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
251 + download_token AS "download_token: DownloadToken"
252 + "#,
253 + stripe_checkout_session_id,
254 + stripe_payment_intent_id,
255 + )
256 + .fetch_optional(executor)
257 + .await?;
258 +
259 + Ok(tx)
260 + }
261 +
262 + /// Complete ALL pending transactions for a cart checkout session.
263 + /// Returns the list of completed transactions (empty if already processed).
264 + #[tracing::instrument(skip_all)]
265 + pub async fn complete_cart_transactions<'e>(
266 + executor: impl sqlx::PgExecutor<'e>,
267 + stripe_checkout_session_id: &str,
268 + stripe_payment_intent_id: Option<&str>,
269 + ) -> Result<Vec<DbTransaction>> {
270 + let txs = sqlx::query_as!(
271 + DbTransaction,
272 + r#"
273 + UPDATE transactions
274 + SET status = 'completed',
275 + stripe_payment_intent_id = $2,
276 + completed_at = NOW()
277 + WHERE stripe_checkout_session_id = $1
278 + AND status = 'pending'
279 + RETURNING
280 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
281 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
282 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
283 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
284 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
285 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
286 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
287 + download_token AS "download_token: DownloadToken"
288 + "#,
289 + stripe_checkout_session_id,
290 + stripe_payment_intent_id,
291 + )
292 + .fetch_all(executor)
293 + .await?;
294 +
295 + Ok(txs)
296 + }
297 +
298 + /// Fetch all completed transactions for a checkout session.
299 + ///
300 + /// Used on the crash-recovery branch of the purchase/cart webhook handlers: when
301 + /// `complete_transaction` / `complete_cart_transactions` return nothing (the
302 + /// rows were already flipped to completed by a first attempt that crashed before
303 + /// running finalize), this re-reads those completed rows so finalize can re-run
304 + /// idempotently. Covers single and cart purchases since both key on the session.
305 + #[tracing::instrument(skip_all)]
306 + pub async fn get_completed_transactions_for_session<'e>(
307 + executor: impl sqlx::PgExecutor<'e>,
308 + stripe_checkout_session_id: &str,
309 + ) -> Result<Vec<DbTransaction>> {
310 + let txs = sqlx::query_as!(
311 + DbTransaction,
312 + r#"
313 + SELECT
314 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
315 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
316 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
317 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
318 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
319 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
320 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
321 + download_token AS "download_token: DownloadToken"
322 + FROM transactions
323 + WHERE stripe_checkout_session_id = $1 AND status = 'completed'
324 + "#,
325 + stripe_checkout_session_id,
326 + )
327 + .fetch_all(executor)
328 + .await?;
329 +
330 + Ok(txs)
331 + }
332 +
333 + /// List transactions where the user is the buyer, newest first.
334 + ///
335 + /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
336 + #[tracing::instrument(skip_all)]
337 + pub async fn get_transactions_by_buyer(
338 + pool: &PgPool,
339 + buyer_id: UserId,
340 + limit: Option<i64>,
341 + ) -> Result<Vec<DbTransaction>> {
342 + let txs = sqlx::query_as!(
343 + DbTransaction,
344 + r#"
345 + SELECT
346 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
347 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
348 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
349 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
350 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
351 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
352 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
353 + download_token AS "download_token: DownloadToken"
354 + FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
355 + "#,
356 + buyer_id as UserId,
357 + limit,
358 + )
359 + .fetch_all(pool)
360 + .await?;
361 +
362 + Ok(txs)
363 + }
364 +
365 + /// One page of a buyer's purchases for CSV export, newest first.
366 + ///
367 + /// Paginated so the purchases export streams in bounded batches rather than
368 + /// loading the buyer's whole history with `limit: None` (ultra-fuzz Run 4 S1).
369 + /// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
370 + pub async fn get_buyer_transactions_for_export_page(
371 + pool: &PgPool,
372 + buyer_id: UserId,
373 + limit: i64,
374 + offset: 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 + FROM transactions WHERE buyer_id = $1
389 + ORDER BY created_at DESC, id DESC
390 + LIMIT $2 OFFSET $3
391 + "#,
392 + buyer_id as UserId,
393 + limit,
394 + offset,
395 + )
396 + .fetch_all(pool)
397 + .await?;
398 +
399 + Ok(txs)
400 + }
401 +
402 + /// List transactions where the user is the seller, newest first.
403 + ///
404 + /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
405 + #[tracing::instrument(skip_all)]
406 + pub async fn get_transactions_by_seller(
407 + pool: &PgPool,
408 + seller_id: UserId,
409 + limit: Option<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 + FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
424 + "#,
425 + seller_id as UserId,
426 + limit,
427 + )
428 + .fetch_all(pool)
429 + .await?;
430 +
431 + Ok(txs)
432 + }
433 +
434 + /// Check whether a user has a completed purchase for a given item.
435 + #[tracing::instrument(skip_all)]
436 + pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
437 + let count: i64 = sqlx::query_scalar!(
438 + r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
439 + user_id as UserId,
440 + item_id as ItemId,
441 + )
442 + .fetch_one(pool)
443 + .await?;
444 +
445 + Ok(count > 0)
446 + }
447 +
448 + /// Bulk variant of `has_purchased_item`. Returns the subset of `item_ids` that
449 + /// the buyer has a completed purchase for. Single DB roundtrip vs. N calls.
450 + #[tracing::instrument(skip_all)]
451 + pub async fn purchased_subset(
452 + pool: &PgPool,
453 + user_id: UserId,
454 + item_ids: &[ItemId],
455 + ) -> Result<std::collections::HashSet<ItemId>> {
456 + if item_ids.is_empty() {
457 + return Ok(std::collections::HashSet::new());
458 + }
459 + let rows = sqlx::query_scalar!(
460 + r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
461 + WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
462 + user_id as UserId,
463 + item_ids as &[ItemId],
464 + )
465 + .fetch_all(pool)
466 + .await?;
467 + Ok(rows.into_iter().collect())
468 + }
469 +
470 + /// Get all item IDs that a user has purchased (for batch access checks)
471 + #[tracing::instrument(skip_all)]
472 + pub async fn get_user_purchased_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
473 + let item_ids: Vec<ItemId> = sqlx::query_scalar!(
474 + r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions WHERE buyer_id = $1 AND status = 'completed' AND item_id IS NOT NULL"#,
475 + user_id as UserId,
476 + )
477 + .fetch_all(pool)
478 + .await?;
479 +
480 + Ok(item_ids)
481 + }
482 +
483 + /// Claims a free item by creating a zero-cost completed transaction.
484 + /// Returns true if claimed successfully, false if already in library.
485 + ///
486 + /// Uses `ON CONFLICT DO NOTHING` against the partial unique index on
487 + /// `(buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL` to prevent duplicate
488 + /// claims under concurrent requests.
489 + #[tracing::instrument(skip_all)]
490 + pub async fn claim_free_item<'e>(
491 + executor: impl sqlx::PgExecutor<'e>,
492 + params: &ClaimParams<'_>,
493 + ) -> Result<bool> {
494 + let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
495 + let result = sqlx::query!(
496 + r#"
497 + 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)
498 + VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8, $9)
499 + ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
500 + "#,
Lines truncated
A server/src/db/transactions/revenue_stats.rs +144
@@ -0,0 +1,144 @@
1 + use super::*;
2 +
3 + /// Sum completed revenue and count sales for all items in a project.
4 + ///
5 + /// Returns `(total_revenue_cents, total_sales)`. Only completed transactions
6 + /// are counted; pending, failed, and refunded are excluded.
7 + #[tracing::instrument(skip_all)]
8 + pub async fn get_revenue_by_project(pool: &PgPool, project_id: ProjectId) -> Result<(i64, i64)> {
9 + let row = sqlx::query!(
10 + r#"
11 + SELECT
12 + COALESCE(SUM(t.amount_cents), 0)::BIGINT AS "total!",
13 + COUNT(*) AS "count!"
14 + FROM transactions t
15 + JOIN items i ON t.item_id = i.id
16 + WHERE i.project_id = $1
17 + AND t.status = 'completed'
18 + "#,
19 + project_id as ProjectId,
20 + )
21 + .fetch_one(pool)
22 + .await?;
23 +
24 + Ok((row.total, row.count))
25 + }
26 +
27 + /// Revenue per project for a given seller, returned as (project_id, title, revenue_cents).
28 + /// Single query replaces N+1 loop in dashboard analytics.
29 + #[tracing::instrument(skip_all)]
30 + pub async fn get_revenue_by_user_projects(
31 + pool: &PgPool,
32 + user_id: UserId,
33 + ) -> Result<Vec<(ProjectId, String, i64)>> {
34 + let rows = sqlx::query!(
35 + r#"
36 + SELECT p.id AS "id: ProjectId", p.title, COALESCE(SUM(t.amount_cents), 0)::BIGINT AS "revenue!"
37 + FROM projects p
38 + LEFT JOIN items i ON i.project_id = p.id
39 + LEFT JOIN transactions t ON t.item_id = i.id AND t.status = 'completed'
40 + WHERE p.user_id = $1
41 + GROUP BY p.id, p.title
42 + HAVING COALESCE(SUM(t.amount_cents), 0) > 0
43 + ORDER BY COALESCE(SUM(t.amount_cents), 0) DESC
44 + "#,
45 + user_id as UserId,
46 + )
47 + .fetch_all(pool)
48 + .await?;
49 +
50 + Ok(rows.into_iter().map(|r| (r.id, r.title, r.revenue)).collect())
51 + }
52 +
53 + /// Revenue and sales per project for a seller within a time range.
54 + ///
55 + /// Used for the cross-project comparison table on the user analytics tab.
56 + #[tracing::instrument(skip_all)]
57 + pub async fn get_revenue_by_user_projects_in_range(
58 + pool: &PgPool,
59 + user_id: UserId,
60 + range: &crate::db::analytics::TimeRange,
61 + ) -> Result<Vec<(ProjectId, String, i64, i64)>> {
62 + let time_filter = match range.interval_sql() {
63 + Some(interval) => format!(
64 + " AND t.completed_at >= NOW() - INTERVAL '{interval}'"
65 + ),
66 + None => String::new(),
67 + };
68 +
69 + let sql = format!(
70 + r#"
71 + SELECT p.id, p.title,
72 + COALESCE(SUM(t.amount_cents), 0)::BIGINT,
73 + COUNT(t.id)::BIGINT
74 + FROM projects p
75 + LEFT JOIN items i ON i.project_id = p.id
76 + LEFT JOIN transactions t ON t.item_id = i.id AND t.status = 'completed'{time_filter}
77 + WHERE p.user_id = $1
78 + GROUP BY p.id, p.title
79 + ORDER BY COALESCE(SUM(t.amount_cents), 0) DESC
80 + "#
81 + );
82 +
83 + // runtime-checked: dynamically-built SQL string — the `time_filter` clause is
84 + // conditionally interpolated via `format!`, so the query text isn't a literal
85 + // and can't be compile-checked by the macro.
86 + let rows: Vec<(ProjectId, String, i64, i64)> = sqlx::query_as(&sql)
87 + .bind(user_id)
88 + .fetch_all(pool)
89 + .await?;
90 +
91 + Ok(rows)
92 + }
93 +
94 + /// Platform-wide revenue stats: total completed revenue, completed count, refunded count.
95 + #[tracing::instrument(skip_all)]
96 + pub async fn get_platform_revenue_stats(pool: &PgPool) -> Result<(i64, i64, i64)> {
97 + let row = sqlx::query!(
98 + r#"
99 + SELECT
100 + COALESCE(SUM(CASE WHEN status = 'completed' THEN amount_cents ELSE 0 END), 0)::BIGINT AS "revenue!",
101 + COALESCE(SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END), 0)::BIGINT AS "completed!",
102 + COALESCE(SUM(CASE WHEN status = 'refunded' THEN 1 ELSE 0 END), 0)::BIGINT AS "refunded!"
103 + FROM transactions
104 + "#,
105 + )
106 + .fetch_one(pool)
107 + .await?;
108 +
109 + Ok((row.revenue, row.completed, row.refunded))
110 + }
111 +
112 + /// Completed and refunded sales for a specific item, for the item dashboard Sales tab.
113 + #[tracing::instrument(skip_all)]
114 + pub async fn get_sales_by_item(
115 + pool: &PgPool,
116 + item_id: ItemId,
117 + seller_id: UserId,
118 + ) -> Result<Vec<DbTransaction>> {
119 + let rows = sqlx::query_as!(
120 + DbTransaction,
121 + r#"
122 + SELECT
123 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
124 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
125 + currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
126 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
127 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
128 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
129 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
130 + download_token AS "download_token: DownloadToken"
131 + FROM transactions
132 + WHERE item_id = $1 AND seller_id = $2
133 + AND status IN ('completed', 'refunded')
134 + ORDER BY created_at DESC
135 + LIMIT 200
136 + "#,
137 + item_id as ItemId,
138 + seller_id as UserId,
139 + )
140 + .fetch_all(pool)
141 + .await?;
142 +
143 + Ok(rows)
144 + }
A server/src/db/transactions/seller_contacts.rs +215
@@ -0,0 +1,215 @@
1 + use super::*;
2 +
3 + /// A buyer's email and display name for platform notifications (e.g. creator departure).
4 + ///
5 + /// Unlike [`DbContactRow`], this includes ALL buyers regardless of contact sharing
6 + /// preference, because platform-initiated notifications (content removal warnings)
7 + /// are distinct from creator-initiated contact.
8 + #[derive(sqlx::FromRow)]
9 + pub struct BuyerNotificationRow {
10 + pub email: String,
11 + pub display_name: Option<String>,
12 + }
13 +
14 + /// Get unique buyers who purchased from a seller, for platform notifications.
15 + ///
16 + /// This bypasses `share_contact` and `contact_revocations` because the notification
17 + /// is sent by the platform (not the creator) to warn buyers about content removal.
18 + ///
19 + /// Capped at `limit` rows to bound memory + outbound email volume on creators
20 + /// with very large historical buyer pools. The caller is responsible for
21 + /// noting when the cap is hit (returned slice length == limit).
22 + #[tracing::instrument(skip_all)]
23 + pub async fn get_all_buyers_for_seller(
24 + pool: &PgPool,
25 + seller_id: UserId,
26 + limit: i64,
27 + ) -> Result<Vec<BuyerNotificationRow>> {
28 + let rows = sqlx::query_as!(
29 + BuyerNotificationRow,
30 + r#"
31 + SELECT DISTINCT u.email, u.display_name
32 + FROM transactions t
33 + JOIN users u ON u.id = t.buyer_id
34 + WHERE t.seller_id = $1
35 + AND t.status = 'completed'
36 + AND t.buyer_id IS NOT NULL
37 + LIMIT $2
38 + "#,
39 + seller_id as UserId,
40 + limit,
41 + )
42 + .fetch_all(pool)
43 + .await?;
44 +
45 + Ok(rows)
46 + }
47 +
48 + /// A contact row: a buyer who opted to share their email with the seller.
49 + #[derive(sqlx::FromRow)]
50 + pub struct DbContactRow {
51 + pub username: String,
52 + pub email: String,
53 + pub total_purchases: i64,
54 + pub total_spent_cents: i64,
55 + pub last_purchase_at: DateTime<Utc>,
56 + }
57 +
58 + /// A creator the fan has actively shared contact info with (no revocation).
59 + #[derive(sqlx::FromRow)]
60 + pub struct SharedCreatorRow {
61 + pub seller_id: UserId,
62 + pub username: String,
63 + pub display_name: Option<String>,
64 + }
65 +
66 + /// Get unique contacts for a seller: buyers who opted in to share their email.
67 + ///
68 + /// Aggregates across all completed transactions where `share_contact = true`,
69 + /// returning one row per buyer with purchase stats. Excludes buyers who have
70 + /// revoked contact sharing.
71 + #[tracing::instrument(skip_all)]
72 + pub async fn get_seller_contacts(
73 + pool: &PgPool,
74 + seller_id: UserId,
75 + ) -> Result<Vec<DbContactRow>> {
76 + let rows = sqlx::query_as!(
77 + DbContactRow,
78 + r#"
79 + SELECT
80 + u.username,
81 + u.email,
82 + COUNT(*) AS "total_purchases!",
83 + COALESCE(SUM(t.amount_cents), 0)::BIGINT AS "total_spent_cents!",
84 + MAX(t.completed_at) AS "last_purchase_at!: chrono::DateTime<chrono::Utc>"
85 + FROM transactions t
86 + JOIN users u ON u.id = t.buyer_id
87 + WHERE t.seller_id = $1
88 + AND t.status = 'completed'
89 + AND t.share_contact = true
90 + AND NOT EXISTS (
91 + SELECT 1 FROM contact_revocations cr
92 + WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
93 + )
94 + GROUP BY t.buyer_id, u.username, u.email
95 + ORDER BY MAX(t.completed_at) DESC
96 + LIMIT 500
97 + "#,
98 + seller_id as UserId,
99 + )
100 + .fetch_all(pool)
101 + .await?;
102 +
103 + Ok(rows)
104 + }
105 +
106 + /// One page of a seller's sharing-opted-in contacts for CSV export, newest
107 + /// first. Paginated so the contacts export streams in bounded batches like its
108 + /// sibling exports instead of buffering one capped query into a single `String`
109 + /// (ultra-fuzz Run 6 R6-Perf-M2). `(MAX(completed_at), buyer_id)` ordering is
110 + /// stable so OFFSET batches don't reorder.
111 + #[tracing::instrument(skip_all)]
112 + pub async fn get_seller_contacts_page(
113 + pool: &PgPool,
114 + seller_id: UserId,
115 + limit: i64,
116 + offset: i64,
117 + ) -> Result<Vec<DbContactRow>> {
118 + let rows = sqlx::query_as::<_, DbContactRow>(
119 + r#"
120 + SELECT
121 + u.username,
122 + u.email,
123 + COUNT(*) AS total_purchases,
124 + COALESCE(SUM(t.amount_cents), 0)::BIGINT AS total_spent_cents,
125 + MAX(t.completed_at) AS last_purchase_at
126 + FROM transactions t
127 + JOIN users u ON u.id = t.buyer_id
128 + WHERE t.seller_id = $1
129 + AND t.status = 'completed'
130 + AND t.share_contact = true
131 + AND NOT EXISTS (
132 + SELECT 1 FROM contact_revocations cr
133 + WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
134 + )
135 + GROUP BY t.buyer_id, u.username, u.email
136 + ORDER BY MAX(t.completed_at) DESC, t.buyer_id
137 + LIMIT $2 OFFSET $3
138 + "#,
139 + )
140 + .bind(seller_id)
141 + .bind(limit)
142 + .bind(offset)
143 + .fetch_all(pool)
144 + .await?;
145 +
146 + Ok(rows)
147 + }
148 +
149 + /// Record a contact revocation (fan withdraws email sharing from a creator).
150 + ///
151 + /// Idempotent: does nothing if already revoked.
152 + #[tracing::instrument(skip_all)]
153 + pub async fn revoke_contact_sharing(
154 + pool: &PgPool,
155 + buyer_id: UserId,
156 + seller_id: UserId,
157 + ) -> Result<()> {
158 + sqlx::query!(
159 + "INSERT INTO contact_revocations (buyer_id, seller_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
160 + buyer_id as UserId,
161 + seller_id as UserId,
162 + )
163 + .execute(pool)
164 + .await?;
165 +
166 + Ok(())
167 + }
168 +
169 + /// Clear a contact revocation (fan re-shares on a new purchase).
170 + #[tracing::instrument(skip_all)]
171 + pub async fn clear_contact_revocation(
172 + pool: &PgPool,
173 + buyer_id: UserId,
174 + seller_id: UserId,
175 + ) -> Result<()> {
176 + sqlx::query!(
177 + "DELETE FROM contact_revocations WHERE buyer_id = $1 AND seller_id = $2",
178 + buyer_id as UserId,
179 + seller_id as UserId,
180 + )
181 + .execute(pool)
182 + .await?;
183 +
184 + Ok(())
185 + }
186 +
187 + /// Get creators the fan has actively shared contact info with (excluding revoked).
188 + #[tracing::instrument(skip_all)]
189 + pub async fn get_shared_creators(
190 + pool: &PgPool,
191 + buyer_id: UserId,
192 + ) -> Result<Vec<SharedCreatorRow>> {
193 + let rows = sqlx::query_as!(
194 + SharedCreatorRow,
195 + r#"
196 + SELECT DISTINCT t.seller_id AS "seller_id!: UserId", u.username, u.display_name
197 + FROM transactions t
198 + JOIN users u ON u.id = t.seller_id
199 + WHERE t.buyer_id = $1
200 + AND t.status = 'completed'
201 + AND t.share_contact = true
202 + AND NOT EXISTS (
203 + SELECT 1 FROM contact_revocations cr
204 + WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
205 + )
206 + ORDER BY u.username
207 + LIMIT 500
208 + "#,
209 + buyer_id as UserId,
210 + )
211 + .fetch_all(pool)
212 + .await?;
213 +
214 + Ok(rows)
215 + }