Skip to main content

max / makenotwork

14.3 KB · 354 lines History Blame Raw
1 //! Reading purchases back: what a buyer owns, what a seller sold, and the
2 //! paginated exports of both.
3
4 use super::super::{
5 Cents, ClaimToken, DbPurchaseRow, DbTransaction, DbTransactionExportRow, DownloadToken, ItemId,
6 KeyCode, PgPool, ProjectId, PromoCodeId, Result, TransactionId, UserId,
7 };
8
9 /// List transactions where the user is the buyer, newest first.
10 ///
11 /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
12 #[tracing::instrument(skip_all)]
13 pub async fn get_transactions_by_buyer(
14 pool: &PgPool,
15 buyer_id: UserId,
16 limit: Option<i64>,
17 ) -> Result<Vec<DbTransaction>> {
18 let txs = sqlx::query_as!(
19 DbTransaction,
20 r#"
21 SELECT
22 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
23 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
24 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
25 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
26 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
27 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
28 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
29 download_token AS "download_token: DownloadToken",
30 presentment_amount_cents, presentment_currency
31 FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
32 "#,
33 buyer_id as UserId,
34 limit,
35 )
36 .fetch_all(pool)
37 .await?;
38
39 Ok(txs)
40 }
41
42 /// One page of a buyer's purchases for CSV export, newest first.
43 ///
44 /// Paginated so the purchases export streams in bounded batches rather than
45 /// loading the buyer's whole history with `limit: None`.
46 /// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
47 pub async fn get_buyer_transactions_for_export_page(
48 pool: &PgPool,
49 buyer_id: UserId,
50 limit: i64,
51 offset: i64,
52 ) -> Result<Vec<DbTransaction>> {
53 let txs = sqlx::query_as!(
54 DbTransaction,
55 r#"
56 SELECT
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 presentment_amount_cents, presentment_currency
66 FROM transactions WHERE buyer_id = $1
67 ORDER BY created_at DESC, id DESC
68 LIMIT $2 OFFSET $3
69 "#,
70 buyer_id as UserId,
71 limit,
72 offset,
73 )
74 .fetch_all(pool)
75 .await?;
76
77 Ok(txs)
78 }
79
80 /// List transactions where the user is the seller, newest first.
81 ///
82 /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
83 #[tracing::instrument(skip_all)]
84 pub async fn get_transactions_by_seller(
85 pool: &PgPool,
86 seller_id: UserId,
87 limit: Option<i64>,
88 ) -> Result<Vec<DbTransaction>> {
89 let txs = sqlx::query_as!(
90 DbTransaction,
91 r#"
92 SELECT
93 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
94 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
95 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
96 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
97 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
98 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
99 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
100 download_token AS "download_token: DownloadToken",
101 presentment_amount_cents, presentment_currency
102 FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
103 "#,
104 seller_id as UserId,
105 limit,
106 )
107 .fetch_all(pool)
108 .await?;
109
110 Ok(txs)
111 }
112
113 /// Check whether a user has a completed purchase for a given item.
114 #[tracing::instrument(skip_all)]
115 pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
116 let count: i64 = sqlx::query_scalar!(
117 r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
118 user_id as UserId,
119 item_id as ItemId,
120 )
121 .fetch_one(pool)
122 .await?;
123
124 Ok(count > 0)
125 }
126
127 /// Bulk variant of `has_purchased_item`. Returns the subset of `item_ids` that
128 /// the buyer has a completed purchase for. Single DB roundtrip vs. N calls.
129 #[tracing::instrument(skip_all)]
130 pub async fn purchased_subset(
131 pool: &PgPool,
132 user_id: UserId,
133 item_ids: &[ItemId],
134 ) -> Result<std::collections::HashSet<ItemId>> {
135 if item_ids.is_empty() {
136 return Ok(std::collections::HashSet::new());
137 }
138 let rows = sqlx::query_scalar!(
139 r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
140 WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
141 user_id as UserId,
142 item_ids as &[ItemId],
143 )
144 .fetch_all(pool)
145 .await?;
146 Ok(rows.into_iter().collect())
147 }
148
149 /// Get all item IDs that a user has purchased (for batch access checks)
150 #[tracing::instrument(skip_all)]
151 pub async fn get_user_purchased_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
152 let item_ids: Vec<ItemId> = sqlx::query_scalar!(
153 r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions WHERE buyer_id = $1 AND status = 'completed' AND item_id IS NOT NULL"#,
154 user_id as UserId,
155 )
156 .fetch_all(pool)
157 .await?;
158
159 Ok(item_ids)
160 }
161
162 /// Check whether a user has a completed purchase for a given project.
163 #[tracing::instrument(skip_all)]
164 pub async fn has_purchased_project(
165 pool: &PgPool,
166 user_id: UserId,
167 project_id: ProjectId,
168 ) -> Result<bool> {
169 let count: i64 = sqlx::query_scalar!(
170 r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND project_id = $2 AND status = 'completed'"#,
171 user_id as UserId,
172 project_id as ProjectId,
173 )
174 .fetch_one(pool)
175 .await?;
176
177 Ok(count > 0)
178 }
179
180 /// Get items purchased by a user, including any associated license key.
181 ///
182 /// Reads from the `purchases` VIEW (which filters `transactions` to
183 /// `status = 'completed'`), then JOINs through `items → projects → users`
184 /// for display fields. The LEFT JOIN on `license_keys` attaches the most
185 /// recent non-revoked key code so the buyer can see it in their library
186 /// without a separate lookup. Capped at 20 rows for the dashboard summary.
187 #[tracing::instrument(skip_all)]
188 pub async fn get_user_purchases(pool: &PgPool, user_id: UserId) -> Result<Vec<DbPurchaseRow>> {
189 let purchases = sqlx::query_as!(
190 DbPurchaseRow,
191 r#"
192 SELECT
193 transaction_id AS "transaction_id!: TransactionId",
194 item_id AS "item_id!: ItemId",
195 title AS "title!",
196 creator AS "creator!",
197 item_type AS "item_type!: crate::db::ItemType",
198 purchased_at AS "purchased_at!: chrono::DateTime<chrono::Utc>",
199 is_free AS "is_free!",
200 license_key_code AS "license_key_code?: KeyCode",
201 has_new_version AS "has_new_version!"
202 FROM (
203 SELECT DISTINCT ON (p.item_id)
204 p.transaction_id,
205 p.item_id,
206 i.title,
207 u.username as creator,
208 i.item_type,
209 p.purchased_at,
210 -- Badge from what the buyer actually paid, not the item's current
211 -- price: a later re-price to $0 must not retroactively badge a paid
212 -- purchase "Free" (nor vice-versa). The purchases view carries the
213 -- transaction's own amount_cents.
214 (p.amount_cents = 0) as is_free,
215 lk.key_code as license_key_code,
216 (vc.total_versions > 0 AND vc.total_versions > COALESCE(dc.downloaded_count, 0)) as has_new_version
217 FROM purchases p
218 JOIN items i ON p.item_id = i.id
219 JOIN projects proj ON i.project_id = proj.id
220 JOIN users u ON proj.user_id = u.id
221 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
222 LEFT JOIN LATERAL (
223 SELECT COUNT(*) AS total_versions
224 FROM versions v
225 WHERE v.item_id = i.id AND v.s3_key IS NOT NULL
226 ) vc ON true
227 LEFT JOIN LATERAL (
228 SELECT COUNT(*) AS downloaded_count
229 FROM user_downloads ud
230 WHERE ud.user_id = p.buyer_id AND ud.item_id = i.id
231 ) dc ON true
232 WHERE p.buyer_id = $1
233 ORDER BY p.item_id, p.purchased_at DESC
234 ) deduped
235 ORDER BY purchased_at DESC
236 LIMIT 20
237 "#,
238 user_id as UserId,
239 )
240 .fetch_all(pool)
241 .await?;
242
243 Ok(purchases)
244 }
245
246 /// Fetch a single transaction by ID.
247 ///
248 /// # Authorization
249 ///
250 /// This lookup is intentionally **unscoped**, it does not filter by buyer or
251 /// seller, because the two callers need the row to *decide* authorization
252 /// (a receipt page shown to buyer-or-seller; a refund restricted to the
253 /// seller). Every caller MUST therefore check ownership against the returned
254 /// `buyer_id`/`seller_id` before acting on it. A new caller that returns this
255 /// row's contents without such a check would be an IDOR, there is no implicit
256 /// scoping here to lean on.
257 #[tracing::instrument(skip_all)]
258 pub async fn get_transaction_by_id(
259 pool: &PgPool,
260 id: TransactionId,
261 ) -> Result<Option<DbTransaction>> {
262 let tx = sqlx::query_as!(
263 DbTransaction,
264 r#"
265 SELECT
266 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
267 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
268 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
269 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
270 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
271 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
272 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
273 download_token AS "download_token: DownloadToken",
274 presentment_amount_cents, presentment_currency
275 FROM transactions WHERE id = $1
276 "#,
277 id as TransactionId,
278 )
279 .fetch_optional(pool)
280 .await?;
281 Ok(tx)
282 }
283
284 #[tracing::instrument(skip_all)]
285 /// One page of a seller's sales for CSV export, newest first.
286 ///
287 /// Paginated (`LIMIT`/`OFFSET`) so the export streams in bounded batches instead
288 /// of loading the seller's entire transaction history into memory in one query.
289 /// The `(created_at, id)` ordering is stable so OFFSET
290 /// batches don't reorder. (Keyset pagination would avoid OFFSET's deep-scan cost
291 /// and is the future optimization; OFFSET is sufficient at current scale and
292 /// keeps peak memory + per-query result bounded, which is the DoS fix.)
293 pub async fn get_seller_transactions_for_export_page(
294 pool: &PgPool,
295 seller_id: UserId,
296 limit: i64,
297 offset: i64,
298 ) -> Result<Vec<DbTransactionExportRow>> {
299 let rows = sqlx::query_as!(
300 DbTransactionExportRow,
301 r#"
302 SELECT
303 t.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
304 t.item_id AS "item_id: ItemId",
305 t.item_title,
306 t.amount_cents AS "amount_cents: Cents",
307 t.status AS "status: crate::db::TransactionStatus",
308 CASE WHEN t.share_contact AND NOT EXISTS (
309 SELECT 1 FROM contact_revocations cr
310 WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
311 ) THEN u.email ELSE NULL END as buyer_email
312 FROM transactions t
313 LEFT JOIN users u ON u.id = t.buyer_id
314 WHERE t.seller_id = $1
315 ORDER BY t.created_at DESC, t.id DESC
316 LIMIT $2 OFFSET $3
317 "#,
318 seller_id as UserId,
319 limit,
320 offset,
321 )
322 .fetch_all(pool)
323 .await?;
324
325 Ok(rows)
326 }
327
328 /// All of a seller's export rows accumulated into one Vec, bounded to
329 /// `EXPORT_ACCUMULATE_CAP` rows. For admin / internal-API callers that need the
330 /// full set in memory; the public creator-facing export streams page-by-page via
331 /// [`get_seller_transactions_for_export_page`] instead of materializing here.
332 pub async fn get_seller_transactions_for_export(
333 pool: &PgPool,
334 seller_id: UserId,
335 ) -> Result<Vec<DbTransactionExportRow>> {
336 /// Page size for the accumulating fetch.
337 const PAGE: i64 = 5_000;
338 /// Cap so even an admin/internal export can't load an unbounded result set.
339 const EXPORT_ACCUMULATE_CAP: usize = 1_000_000;
340
341 let mut all = Vec::new();
342 let mut offset = 0i64;
343 loop {
344 let page = get_seller_transactions_for_export_page(pool, seller_id, PAGE, offset).await?;
345 let n = page.len();
346 all.extend(page);
347 offset += n as i64;
348 if (n as i64) < PAGE || all.len() >= EXPORT_ACCUMULATE_CAP {
349 break;
350 }
351 }
352 Ok(all)
353 }
354