Skip to main content

max / makenotwork

16.5 KB · 394 lines History Blame Raw
1 //! The pending half of the lifecycle: write a row before the buyer leaves for
2 //! Stripe, flip it on their return, and delete it if they never come back.
3 //!
4 //! Every delete path here returns the row's `promo_code_id` so the caller
5 //! releases the reservation with the row.
6
7 use super::super::{
8 Cents, ClaimToken, DbTransaction, DownloadToken, ItemId, PgPool, ProjectId, PromoCodeId,
9 Result, TransactionId, UserId,
10 };
11
12 /// Parameters for creating a pending Stripe checkout transaction.
13 pub struct CreateTransactionParams<'a> {
14 pub buyer_id: Option<UserId>,
15 pub seller_id: UserId,
16 /// `None` for project-level purchases (no specific item).
17 pub item_id: Option<ItemId>,
18 pub amount_cents: Cents,
19 pub platform_fee_cents: Cents,
20 pub stripe_checkout_session_id: &'a str,
21 pub item_title: &'a str,
22 pub seller_username: &'a str,
23 pub share_contact: bool,
24 /// Set for project-level purchases; `None` for item purchases.
25 pub project_id: Option<ProjectId>,
26 /// Promo code used for this checkout (for releasing reservations on cleanup).
27 pub promo_code_id: Option<PromoCodeId>,
28 /// Guest buyer's email (set for guest checkouts, None for logged-in).
29 pub guest_email: Option<&'a str>,
30 /// Cents MNW owes the seller as reimbursement for a platform-funded credit
31 /// (the Fan+ renewal credit) applied to this sale; `0` for ordinary sales. The
32 /// scheduler settles it via a platform -> connected transfer once the
33 /// transaction completes (see `db::platform_credits`).
34 pub platform_credit_cents: i64,
35 }
36
37 /// Record a new pending transaction for a Stripe checkout session.
38 #[tracing::instrument(skip_all)]
39 pub async fn create_transaction<'e>(
40 executor: impl sqlx::PgExecutor<'e>,
41 params: &CreateTransactionParams<'_>,
42 ) -> Result<DbTransaction> {
43 let tx = sqlx::query_as!(
44 DbTransaction,
45 r#"
46 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)
47 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
48 RETURNING
49 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
50 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
51 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
52 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
53 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
54 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
55 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
56 download_token AS "download_token: DownloadToken",
57 presentment_amount_cents, presentment_currency
58 "#,
59 params.buyer_id as Option<UserId>,
60 params.seller_id as UserId,
61 params.item_id as Option<ItemId>,
62 params.amount_cents as Cents,
63 params.platform_fee_cents as Cents,
64 params.stripe_checkout_session_id,
65 params.item_title,
66 params.seller_username,
67 params.share_contact,
68 params.project_id as Option<ProjectId>,
69 params.promo_code_id as Option<PromoCodeId>,
70 params.guest_email,
71 params.platform_credit_cents,
72 )
73 .fetch_one(executor)
74 .await?;
75
76 Ok(tx)
77 }
78
79 /// Mark a pending transaction as completed (idempotent; returns `None` if already completed).
80 ///
81 /// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
82 /// can include this in a larger transaction when needed.
83 #[tracing::instrument(skip_all)]
84 pub async fn complete_transaction<'e>(
85 executor: impl sqlx::PgExecutor<'e>,
86 stripe_checkout_session_id: &str,
87 stripe_payment_intent_id: Option<&str>,
88 presentment: Option<(i64, &str)>,
89 ) -> Result<Option<DbTransaction>> {
90 // Only update if status is 'pending' for idempotency
91 // Returns None if transaction was already completed (duplicate webhook)
92 let tx = sqlx::query_as!(
93 DbTransaction,
94 r#"
95 UPDATE transactions
96 SET status = 'completed',
97 stripe_payment_intent_id = $2,
98 presentment_amount_cents = $3,
99 presentment_currency = $4,
100 completed_at = NOW()
101 WHERE stripe_checkout_session_id = $1
102 AND status = 'pending'
103 RETURNING
104 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
105 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
106 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
107 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
108 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
109 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
110 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
111 download_token AS "download_token: DownloadToken",
112 presentment_amount_cents, presentment_currency
113 "#,
114 stripe_checkout_session_id,
115 stripe_payment_intent_id,
116 presentment.map(|(cents, _)| cents),
117 presentment.map(|(_, currency)| currency),
118 )
119 .fetch_optional(executor)
120 .await?;
121
122 Ok(tx)
123 }
124
125 /// Complete ALL pending transactions for a cart checkout session.
126 /// Returns the list of completed transactions (empty if already processed).
127 #[tracing::instrument(skip_all)]
128 pub async fn complete_cart_transactions<'e>(
129 executor: impl sqlx::PgExecutor<'e>,
130 stripe_checkout_session_id: &str,
131 stripe_payment_intent_id: Option<&str>,
132 ) -> Result<Vec<DbTransaction>> {
133 let txs = sqlx::query_as!(
134 DbTransaction,
135 r#"
136 UPDATE transactions
137 SET status = 'completed',
138 stripe_payment_intent_id = $2,
139 completed_at = NOW()
140 WHERE stripe_checkout_session_id = $1
141 AND status = 'pending'
142 RETURNING
143 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
144 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
145 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
146 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
147 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
148 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
149 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
150 download_token AS "download_token: DownloadToken",
151 presentment_amount_cents, presentment_currency
152 "#,
153 stripe_checkout_session_id,
154 stripe_payment_intent_id,
155 )
156 .fetch_all(executor)
157 .await?;
158
159 Ok(txs)
160 }
161
162 /// Fetch all completed transactions for a checkout session.
163 ///
164 /// Used on the crash-recovery branch of the purchase/cart webhook handlers: when
165 /// `complete_transaction` / `complete_cart_transactions` return nothing (the
166 /// rows were already flipped to completed by a first attempt that crashed before
167 /// running finalize), this re-reads those completed rows so finalize can re-run
168 /// idempotently. Covers single and cart purchases since both key on the session.
169 #[tracing::instrument(skip_all)]
170 pub async fn get_completed_transactions_for_session<'e>(
171 executor: impl sqlx::PgExecutor<'e>,
172 stripe_checkout_session_id: &str,
173 ) -> Result<Vec<DbTransaction>> {
174 let txs = sqlx::query_as!(
175 DbTransaction,
176 r#"
177 SELECT
178 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
179 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
180 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
181 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
182 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
183 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
184 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
185 download_token AS "download_token: DownloadToken",
186 presentment_amount_cents, presentment_currency
187 FROM transactions
188 WHERE stripe_checkout_session_id = $1 AND status = 'completed'
189 "#,
190 stripe_checkout_session_id,
191 )
192 .fetch_all(executor)
193 .await?;
194
195 Ok(txs)
196 }
197
198 /// Parameters for creating a pending project purchase transaction.
199 pub struct CreateProjectTransactionParams<'a> {
200 pub buyer_id: UserId,
201 pub seller_id: UserId,
202 pub project_id: ProjectId,
203 pub amount_cents: i32,
204 pub stripe_checkout_session_id: &'a str,
205 pub project_title: &'a str,
206 pub seller_username: &'a str,
207 pub share_contact: bool,
208 }
209
210 /// Record a new pending transaction for a project purchase.
211 #[tracing::instrument(skip_all)]
212 pub async fn create_project_transaction(
213 pool: &PgPool,
214 params: &CreateProjectTransactionParams<'_>,
215 ) -> Result<DbTransaction> {
216 let tx = sqlx::query_as!(
217 DbTransaction,
218 r#"
219 INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact)
220 VALUES ($1, $2, $3, $4, 0, $5, $6, $7, $8)
221 RETURNING
222 id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
223 item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
224 currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
225 created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
226 item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
227 parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
228 guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
229 download_token AS "download_token: DownloadToken",
230 presentment_amount_cents, presentment_currency
231 "#,
232 params.buyer_id as UserId,
233 params.seller_id as UserId,
234 params.project_id as ProjectId,
235 params.amount_cents,
236 params.stripe_checkout_session_id,
237 params.project_title,
238 params.seller_username,
239 params.share_contact,
240 )
241 .fetch_one(pool)
242 .await?;
243
244 Ok(tx)
245 }
246
247 /// Create a pending "placeholder" transaction for a subscription checkout that
248 /// used a promo code. This row exists solely so `cleanup_stale_pending` can
249 /// release the promo code reservation if the buyer abandons the Stripe session.
250 /// It is deleted (not completed) when the subscription webhook fires.
251 #[tracing::instrument(skip_all)]
252 pub async fn create_subscription_pending_transaction(
253 pool: &PgPool,
254 buyer_id: UserId,
255 seller_id: UserId,
256 project_id: ProjectId,
257 stripe_checkout_session_id: &str,
258 promo_code_id: PromoCodeId,
259 ) -> Result<()> {
260 sqlx::query!(
261 r#"
262 INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents,
263 stripe_checkout_session_id, item_title, seller_username, share_contact, promo_code_id)
264 VALUES ($1, $2, $3, 0, 0, $4, 'subscription-promo-hold', '', false, $5)
265 "#,
266 buyer_id as UserId,
267 seller_id as UserId,
268 project_id as ProjectId,
269 stripe_checkout_session_id,
270 promo_code_id as PromoCodeId,
271 )
272 .execute(pool)
273 .await?;
274
275 Ok(())
276 }
277
278 /// Delete a pending subscription promo-hold transaction by checkout session ID.
279 /// Called from the subscription webhook after the subscription is created.
280 #[tracing::instrument(skip_all)]
281 pub async fn delete_subscription_pending_transaction<'e>(
282 executor: impl sqlx::PgExecutor<'e>,
283 stripe_checkout_session_id: &str,
284 ) -> Result<()> {
285 sqlx::query!(
286 "DELETE FROM transactions WHERE stripe_checkout_session_id = $1 AND status = 'pending'",
287 stripe_checkout_session_id,
288 )
289 .execute(executor)
290 .await?;
291
292 Ok(())
293 }
294
295 /// Delete stale pending transactions (older than the given threshold) and return
296 /// the promo_code_ids that need their use_count decremented.
297 ///
298 /// Stripe checkout sessions expire after 24 hours, so pending transactions older
299 /// than that will never complete. This releases the pending purchase uniqueness
300 /// slot and any reserved promo code use_count.
301 #[tracing::instrument(skip_all)]
302 pub async fn cleanup_stale_pending(
303 pool: &PgPool,
304 older_than: chrono::Duration,
305 ) -> Result<Vec<Option<crate::db::PromoCodeId>>> {
306 let cutoff = chrono::Utc::now() - older_than;
307 // 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.
308 let rows: Vec<(Option<crate::db::PromoCodeId>,)> = sqlx::query_as(
309 r"
310 DELETE FROM transactions
311 WHERE status = 'pending'
312 AND created_at < $1
313 RETURNING promo_code_id
314 ",
315 )
316 .bind(cutoff)
317 .fetch_all(pool)
318 .await?;
319
320 Ok(rows.into_iter().map(|(id,)| id).collect())
321 }
322
323 /// Bulk variant of `get_pending_item_purchase`. Returns the subset of `item_ids`
324 /// for which the buyer already has a `pending` transaction. Used by cart
325 /// checkout to abort early when any line item would collide with the partial
326 /// unique index on `(buyer_id, item_id) WHERE status = 'pending'`.
327 #[tracing::instrument(skip_all)]
328 pub async fn pending_subset(
329 pool: &PgPool,
330 buyer_id: UserId,
331 item_ids: &[ItemId],
332 ) -> Result<std::collections::HashSet<ItemId>> {
333 if item_ids.is_empty() {
334 return Ok(std::collections::HashSet::new());
335 }
336 let rows = sqlx::query_scalar!(
337 r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
338 WHERE buyer_id = $1 AND status = 'pending' AND item_id = ANY($2)"#,
339 buyer_id as UserId,
340 item_ids as &[ItemId],
341 )
342 .fetch_all(pool)
343 .await?;
344 Ok(rows.into_iter().collect())
345 }
346
347 /// Returns the buyer's pending transaction for a specific item, if any.
348 /// Used to surface in-progress checkouts on the purchase page.
349 #[tracing::instrument(skip_all)]
350 pub async fn get_pending_item_purchase(
351 pool: &PgPool,
352 buyer_id: UserId,
353 item_id: ItemId,
354 ) -> Result<Option<(TransactionId, chrono::DateTime<chrono::Utc>)>> {
355 let row = sqlx::query!(
356 r#"
357 SELECT id AS "id: TransactionId", created_at AS "created_at: chrono::DateTime<chrono::Utc>"
358 FROM transactions
359 WHERE buyer_id = $1 AND item_id = $2 AND status = 'pending'
360 LIMIT 1
361 "#,
362 buyer_id as UserId,
363 item_id as ItemId,
364 )
365 .fetch_optional(pool)
366 .await?;
367
368 Ok(row.map(|r| (r.id, r.created_at)))
369 }
370
371 /// Delete the buyer's pending transaction for a specific item.
372 /// Returns any released `promo_code_id` so the caller can release its
373 /// reservation.
374 #[tracing::instrument(skip_all)]
375 pub async fn delete_pending_item_purchase(
376 pool: &PgPool,
377 buyer_id: UserId,
378 item_id: ItemId,
379 ) -> Result<Option<crate::db::PromoCodeId>> {
380 let row: Option<Option<crate::db::PromoCodeId>> = sqlx::query_scalar!(
381 r#"
382 DELETE FROM transactions
383 WHERE buyer_id = $1 AND item_id = $2 AND status = 'pending'
384 RETURNING promo_code_id AS "promo_code_id: crate::db::PromoCodeId"
385 "#,
386 buyer_id as UserId,
387 item_id as ItemId,
388 )
389 .fetch_optional(pool)
390 .await?;
391
392 Ok(row.flatten())
393 }
394