Skip to main content

max / makenotwork

41.8 KB · 1039 lines History Blame Raw
1 //! Webhook handlers for checkout.session.completed events.
2
3 use crate::{
4 Billing,
5 config::Config,
6 db,
7 email::EmailClient,
8 error::{AppError, Result, ResultExt},
9 helpers,
10 payments::{
11 CheckoutMetadata, CreatorTierCheckoutMetadata, FanPlusCheckoutMetadata, MnwEventName,
12 SubscriptionCheckoutMetadata, SynckitAppSubCheckoutMetadata, TipCheckoutMetadata,
13 },
14 wam_client::WamClient,
15 };
16 use sqlx::PgPool;
17
18 use super::checkout_helpers::{
19 check_pending_refund, finalize_guest_transaction, finalize_purchase_transaction,
20 record_tip_splits, send_tip_email,
21 };
22
23 /// A `checkout.session.completed` produced no transaction to complete. Tell a
24 /// benign duplicate webhook (rows already completed) apart from an ORPHANED paid
25 /// session (rows never created, buyer charged, got nothing) and escalate the
26 /// latter to WAM instead of silently logging "duplicate" (Run #2 Payments
27 /// SERIOUS). Shared by the purchase, cart, and guest completion handlers so the
28 /// three can't drift.
29 async fn escalate_if_orphaned_session(
30 db: &PgPool,
31 bg: &crate::background::BackgroundTx,
32 wam: Option<&WamClient>,
33 session_id: &str,
34 payment_intent_id: &str,
35 label: &str,
36 ) -> Result<()> {
37 let exists = db::transactions::transaction_exists_for_checkout_session(db, session_id)
38 .await
39 .context("check session transaction existence")?;
40 if exists {
41 tracing::info!(session_id = %session_id, "{label} already completed, ignoring duplicate webhook");
42 } else {
43 tracing::error!(
44 session_id = %session_id, payment_intent_id = %payment_intent_id,
45 "orphaned paid session ({label}): payment completed but no transaction exists, manual reconciliation required"
46 );
47 if let Some(wam) = wam.cloned() {
48 let sid = session_id.to_string();
49 let pi = payment_intent_id.to_string();
50 let label = label.to_string();
51 // Route through the shutdown-drained background pool, NOT a raw
52 // tokio::spawn: this ticket is the only thing that pages a human
53 // about a charged-but-undelivered buyer, so it must not be dropped
54 // when the process is restarted mid-flight during a deploy (Run 9
55 // pattern; audit Run 22).
56 bg.spawn("stripe orphaned-session ticket", async move {
57 let body = format!(
58 "Checkout session {sid} (payment_intent {pi}, {label}) completed at Stripe but has \
59 NO transaction, the pending row(s) were never created. The buyer was charged and \
60 received nothing. Reconcile manually: refund the payment or recreate the order."
61 );
62 wam.create_ticket("Orphaned paid session", Some(&body), "high", "stripe-orphaned-session", Some(&sid)).await;
63 });
64 }
65 }
66 Ok(())
67 }
68
69 /// Defense-in-depth reconciliation of a completed checkout against Stripe's
70 /// reported session totals. Our line items are server-built, so the credited
71 /// total should equal Stripe's pre-tax subtotal and the session should be USD.
72 /// A currency mismatch or an amount mismatch is logged loudly and escalated to
73 /// WAM via the shutdown-drained background pool (so it never runs inside an open
74 /// DB transaction, yet survives a mid-deploy restart rather than being dropped
75 /// like a raw tokio::spawn, audit Run 22); the server-recorded amount stays
76 /// authoritative either way. Shared by the purchase, cart, and guest completion
77 /// handlers so the three can't drift.
78 fn reconcile_checkout_amount(
79 bg: &crate::background::BackgroundTx,
80 wam: Option<&WamClient>,
81 session_id: &str,
82 session: &crate::payments::CheckoutCompletion,
83 credited_cents: i64,
84 label: &str,
85 ) {
86 // Currency guard: a non-USD session makes the integer-cents subtotal
87 // comparison meaningless and should never happen (sessions are built USD).
88 if let Some(currency) = session.currency.as_deref()
89 && !currency.eq_ignore_ascii_case("usd")
90 {
91 tracing::error!(
92 session_id = %session_id, currency = %currency,
93 "checkout session currency is not USD ({label}); integer-cents reconciliation skipped"
94 );
95 if let Some(wam) = wam.cloned() {
96 let session_id = session_id.to_string();
97 let currency = currency.to_string();
98 let label = label.to_string();
99 bg.spawn("stripe non-usd-session ticket", async move {
100 let body = format!(
101 "Checkout session {session_id} ({label}) settled in {currency}, not USD. The \
102 server credits its own USD amount, but investigate how a non-USD session was created."
103 );
104 wam.create_ticket("Non-USD checkout session", Some(&body), "high", "stripe-non-usd-session", Some(&session_id)).await;
105 });
106 }
107 return;
108 }
109
110 if let Some(subtotal) = session.amount_subtotal
111 && subtotal != credited_cents
112 {
113 tracing::error!(
114 session_id = %session_id, credited_cents = %credited_cents, stripe_subtotal_cents = %subtotal,
115 "checkout amount mismatch ({label}): credited amount differs from Stripe session subtotal"
116 );
117 if let Some(wam) = wam.cloned() {
118 let session_id = session_id.to_string();
119 let label = label.to_string();
120 bg.spawn("stripe amount-mismatch ticket", async move {
121 let body = format!(
122 "Credited amount {credited_cents} cents != Stripe session subtotal {subtotal} cents \
123 (session {session_id}, {label}). The server amount is authoritative; investigate a \
124 price-edit / Stripe Tax / currency edge."
125 );
126 wam.create_ticket("Checkout amount mismatch", Some(&body), "high", "stripe-amount-mismatch", Some(&session_id)).await;
127 });
128 }
129 }
130 }
131
132 /// Handle checkout.session.completed for one-time purchases
133 #[tracing::instrument(skip_all, name = "stripe::handle_purchase_checkout")]
134 pub(super) async fn handle_purchase_checkout_completed(
135 db: &PgPool,
136 bg: &crate::background::BackgroundTx,
137 email: &EmailClient,
138 wam: Option<&WamClient>,
139 config: &Config,
140 session: &crate::payments::CheckoutCompletion,
141 event_id: &str,
142 ) -> Result<()> {
143 let session_id = session.session_id.clone();
144
145 tracing::info!(session_id = %session_id, "processing completed purchase checkout");
146
147 // Extract metadata (already typed IDs from CheckoutMetadata)
148 let raw_metadata = CheckoutMetadata::from_metadata(session.metadata.as_ref())?;
149 let buyer_id = raw_metadata.buyer_id;
150 let seller_id = raw_metadata.seller_id;
151 let item_id = raw_metadata.item_id;
152
153 let item_id_display = item_id.map_or_else(|| "project".to_string(), |id| id.to_string());
154
155 // Get the payment intent ID
156 // Display/logging copy only; the DB write below passes `session.payment_intent_id`
157 // directly so a PI-less session stores NULL, not a literal "unknown" that would
158 // collide with other PI-less rows in the money-keyed lookup column (Run 9).
159 let payment_intent_id = session.payment_intent_id.clone().unwrap_or_default();
160
161 // Complete the transaction (idempotent - returns None if already completed).
162 // Steps 1-3 (complete_transaction, increment_sales_count, discount code increment)
163 // are wrapped in a single DB transaction to prevent inconsistent state if any step fails.
164 let mut db_tx = db
165 .begin()
166 .await
167 .context("begin purchase webhook transaction")?;
168
169 // Both halves or neither, matching the column pair's CHECK: a currency with
170 // no amount is not something we can put on a receipt.
171 let presentment = session
172 .presentment
173 .as_ref()
174 .and_then(|p| Some((p.amount?, p.currency.as_deref()?)));
175
176 match db::transactions::complete_transaction(
177 &mut *db_tx,
178 &session_id,
179 session.payment_intent_id.as_deref(),
180 presentment,
181 )
182 .await
183 {
184 Ok(Some(tx)) => {
185 tracing::info!(
186 buyer_id = %buyer_id, seller_id = %seller_id, item_id = %item_id_display, amount_cents = %tx.amount_cents,
187 "transaction completed"
188 );
189
190 // Defense-in-depth reconciliation (currency + subtotal) against the
191 // server-authoritative credited amount.
192 reconcile_checkout_amount(
193 bg,
194 wam,
195 &session_id,
196 session,
197 i64::from(tx.amount_cents),
198 "purchase",
199 );
200
201 // Increment denormalized sales_count (inside transaction)
202 if let Some(iid) = item_id {
203 db::items::increment_sales_count(&mut *db_tx, iid)
204 .await
205 .with_context(|| format!("increment sales count for item {iid}"))?;
206 }
207
208 // Promo code use_count is reserved at checkout time (not here) to prevent
209 // concurrent checkouts from exceeding max_uses. No increment needed in webhook.
210
211 // Commit the critical data integrity operations
212 db_tx
213 .commit()
214 .await
215 .context("commit purchase webhook transaction")?;
216
217 // --- Secondary effects below (outside transaction) ---
218 // Consolidated in one re-runnable finalizer so the purchase and cart
219 // paths can't drift and a crash-recovery redelivery re-runs safely.
220 finalize_purchase_transaction(db, bg, email, wam, config, &tx, buyer_id, seller_id)
221 .await;
222
223 if let Err(e) = db::subscriptions::log_subscription_event(
224 db,
225 None,
226 event_id,
227 MnwEventName::CheckoutCompletedPurchase,
228 &serde_json::json!({"session_id": session_id}),
229 )
230 .await
231 {
232 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
233 }
234
235 // Check for a pending refund that arrived before this payment webhook.
236 // If found, process it now that the transaction is completed.
237 check_pending_refund(db, &payment_intent_id).await;
238 }
239 Ok(None) => {
240 // No row flipped to completed. Either this is a benign duplicate of
241 // an already-finalized purchase, OR the first attempt crashed AFTER
242 // committing the completed status but BEFORE running finalize, in
243 // which case the buyer holds a completed purchase with no license
244 // key / no splits. Re-fetch completed rows for the session and
245 // re-run the idempotent finalizer; only escalate if none exist
246 // (genuinely orphaned: payment took, rows never created).
247 let completed =
248 db::transactions::get_completed_transactions_for_session(db, &session_id)
249 .await
250 .context("re-fetch completed transactions for crash recovery")?;
251 if completed.is_empty() {
252 escalate_if_orphaned_session(
253 db,
254 bg,
255 wam,
256 &session_id,
257 &payment_intent_id,
258 "transaction",
259 )
260 .await?;
261 } else {
262 for tx in &completed {
263 tracing::info!(
264 session_id = %session_id, transaction_id = %tx.id,
265 "crash-recovery: re-running finalize for already-completed session"
266 );
267 finalize_purchase_transaction(
268 db, bg, email, wam, config, tx, buyer_id, seller_id,
269 )
270 .await;
271 }
272 }
273 }
274 Err(e) => {
275 tracing::error!(session_id = %session_id, error = ?e, "failed to complete transaction");
276 return Err(e);
277 }
278 }
279
280 Ok(())
281 }
282
283 /// Handle checkout.session.completed for cart (multi-item) purchases
284 #[tracing::instrument(skip_all, name = "stripe::handle_cart_checkout")]
285 pub(super) async fn handle_cart_checkout_completed(
286 db: &PgPool,
287 bg: &crate::background::BackgroundTx,
288 email: &EmailClient,
289 wam: Option<&WamClient>,
290 config: &Config,
291 session: &crate::payments::CheckoutCompletion,
292 event_id: &str,
293 ) -> Result<()> {
294 let session_id = session.session_id.clone();
295 tracing::info!(session_id = %session_id, "processing completed cart checkout");
296
297 let meta = crate::payments::CartCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
298 let buyer_id = meta.buyer_id;
299 let seller_id = meta.seller_id;
300
301 // Display/logging copy only; the DB write below passes `session.payment_intent_id`
302 // directly so a PI-less session stores NULL, not a literal "unknown" that would
303 // collide with other PI-less rows in the money-keyed lookup column (Run 9).
304 let payment_intent_id = session.payment_intent_id.clone().unwrap_or_default();
305
306 // Complete ALL pending transactions for this session in a single DB transaction
307 let mut db_tx = db.begin().await.context("begin cart webhook transaction")?;
308
309 let completed_txs = db::transactions::complete_cart_transactions(
310 &mut *db_tx,
311 &session_id,
312 session.payment_intent_id.as_deref(),
313 )
314 .await
315 .context("complete cart transactions")?;
316
317 if completed_txs.is_empty() {
318 // Same crash-recovery shape as the single-item handler: re-fetch
319 // completed rows for the session and re-run the idempotent finalizer
320 // before falling through to orphan escalation.
321 db_tx.commit().await.ok();
322 let completed = db::transactions::get_completed_transactions_for_session(db, &session_id)
323 .await
324 .context("re-fetch completed cart transactions for crash recovery")?;
325 if completed.is_empty() {
326 escalate_if_orphaned_session(
327 db,
328 bg,
329 wam,
330 &session_id,
331 &payment_intent_id,
332 "cart transactions",
333 )
334 .await?;
335 } else {
336 for tx in &completed {
337 tracing::info!(
338 session_id = %session_id, transaction_id = %tx.id,
339 "crash-recovery: re-running finalize for already-completed cart session"
340 );
341 finalize_purchase_transaction(db, bg, email, wam, config, tx, buyer_id, seller_id)
342 .await;
343 }
344 }
345 return Ok(());
346 }
347
348 tracing::info!(
349 session_id = %session_id, buyer_id = %buyer_id, seller_id = %seller_id,
350 count = completed_txs.len(), "cart transactions completed"
351 );
352
353 // Defense-in-depth reconciliation (currency + subtotal): the sum of the
354 // credited transactions should equal Stripe's pre-tax subtotal.
355 let cart_credited: i64 = completed_txs
356 .iter()
357 .map(|tx| i64::from(tx.amount_cents))
358 .sum();
359 reconcile_checkout_amount(bg, wam, &session_id, session, cart_credited, "cart");
360
361 // Increment sales count for each item
362 for tx in &completed_txs {
363 if let Some(item_id) = tx.item_id {
364 db::items::increment_sales_count(&mut *db_tx, item_id)
365 .await
366 .with_context(|| format!("increment sales count for item {item_id}"))?;
367 }
368 }
369
370 db_tx
371 .commit()
372 .await
373 .context("commit cart webhook transaction")?;
374
375 // Remove purchased items from cart (items stay in cart until payment succeeds,
376 // so cancelled checkouts don't lose cart contents)
377 db::cart::remove_seller_items_from_cart(db, buyer_id, seller_id)
378 .await
379 .context("remove cart items after successful payment")?;
380
381 // --- Secondary effects (outside transaction) ---
382 // One re-runnable finalizer per transaction; shared with the single-item
383 // path so the effect blocks can't drift. Idempotent on a crash-recovery
384 // redelivery. (clear_contact_revocation runs per-tx that opted in, which is
385 // a no-op once already cleared.)
386 for tx in &completed_txs {
387 finalize_purchase_transaction(db, bg, email, wam, config, tx, buyer_id, seller_id).await;
388 }
389
390 if let Err(e) = db::subscriptions::log_subscription_event(
391 db,
392 None,
393 event_id,
394 MnwEventName::CheckoutCompletedCart,
395 &serde_json::json!({"session_id": session_id, "item_count": completed_txs.len()}),
396 )
397 .await
398 {
399 tracing::warn!(event_id = %event_id, error = ?e, "failed to log cart checkout event");
400 }
401
402 // Check for pending refund
403 check_pending_refund(db, &payment_intent_id).await;
404
405 Ok(())
406 }
407
408 /// Handle checkout.session.completed for subscriptions
409 #[tracing::instrument(skip_all, name = "stripe::handle_subscription_checkout")]
410 pub(super) async fn handle_subscription_checkout_completed(
411 db: &PgPool,
412 bg: &crate::background::BackgroundTx,
413 email: &EmailClient,
414 session: &crate::payments::CheckoutCompletion,
415 event_id: &str,
416 ) -> Result<()> {
417 let session_id = session.session_id.clone();
418 tracing::info!(session_id = %session_id, "processing completed subscription checkout");
419
420 // Extract subscription-specific metadata (already typed IDs)
421 let raw_metadata = SubscriptionCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
422 let subscriber_id = raw_metadata.subscriber_id;
423 let project_id = raw_metadata.project_id;
424 let tier_id = raw_metadata.tier_id;
425
426 // Get the Stripe subscription ID from the session
427 let stripe_subscription_id = session.subscription_id.clone().ok_or_else(|| {
428 tracing::error!("Subscription checkout completed but no subscription ID on session");
429 AppError::BadRequest("Missing subscription ID on session".to_string())
430 })?;
431
432 // Get the Stripe customer ID from the session
433 let stripe_customer_id = session.customer_id.clone().ok_or_else(|| {
434 tracing::error!("Subscription checkout completed but no customer ID on session");
435 AppError::BadRequest("Missing customer ID on session".to_string())
436 })?;
437
438 // Create the subscription record + increment promo code in a single transaction.
439 let mut tx = db
440 .begin()
441 .await
442 .context("begin subscription webhook transaction")?;
443
444 let Some(sub) = db::subscriptions::create_subscription(
445 &mut tx,
446 subscriber_id,
447 tier_id,
448 project_id,
449 &stripe_subscription_id,
450 &stripe_customer_id,
451 )
452 .await
453 .context("create subscription record")?
454 else {
455 tracing::info!(
456 subscriber_id = %subscriber_id, project_id = %project_id,
457 "subscription already exists, ignoring duplicate"
458 );
459 return Ok(());
460 };
461
462 // Promo code use_count is reserved at checkout time (not here) to prevent
463 // concurrent checkouts from exceeding max_uses. No increment needed in webhook.
464
465 // Delete the pending promo-hold transaction (created at checkout time so
466 // cleanup_stale_pending_transactions can release the code if abandoned).
467 db::transactions::delete_subscription_pending_transaction(&mut *tx, &session_id)
468 .await
469 .context("delete subscription pending promo-hold transaction")?;
470
471 tx.commit()
472 .await
473 .context("commit subscription webhook transaction")?;
474
475 tracing::info!(
476 subscription_id = %sub.id, subscriber_id = %subscriber_id, project_id = %project_id, tier_id = %tier_id,
477 "subscription created"
478 );
479
480 // Send subscription started email (fire-and-forget)
481 if let (Ok(Some(subscriber)), Ok(Some(tier)), Ok(Some(project))) = (
482 db::users::get_user_by_id(db, subscriber_id).await,
483 db::subscriptions::get_subscription_tier_by_id(db, tier_id).await,
484 db::projects::get_project_by_id(db, project_id).await,
485 ) {
486 // The tier was priced by the project's owner, in the owner's currency.
487 let creator_currency = db::users::get_user_by_id(db, project.user_id)
488 .await
489 .ok()
490 .flatten()
491 .map(|u| u.settlement_currency)
492 .unwrap_or_default();
493 let price = helpers::format_price(tier.price_cents, creator_currency);
494 let sub_email = subscriber.email.clone();
495 let sub_name = subscriber.display_name;
496 let tier_name = tier.name;
497 let project_title = project.title;
498 let email = email.clone();
499 bg.spawn("subscription started", async move {
500 if let Err(e) = email
501 .send_subscription_started(
502 &sub_email,
503 sub_name.as_deref(),
504 &tier_name,
505 &project_title,
506 &price,
507 )
508 .await
509 {
510 tracing::error!(error = ?e, "failed to send subscription started");
511 }
512 });
513 }
514
515 if let Err(e) = db::subscriptions::log_subscription_event(
516 db, Some(sub.id), event_id, MnwEventName::CheckoutCompletedSubscription,
517 &serde_json::json!({"session_id": session_id, "stripe_subscription_id": stripe_subscription_id}),
518 ).await {
519 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
520 }
521
522 Ok(())
523 }
524
525 /// Handle checkout.session.completed for Fan+ subscriptions
526 #[tracing::instrument(skip_all, name = "stripe::handle_fan_plus_checkout")]
527 pub(super) async fn handle_fan_plus_checkout_completed(
528 db: &PgPool,
529 bg: &crate::background::BackgroundTx,
530 email: &EmailClient,
531 session: &crate::payments::CheckoutCompletion,
532 event_id: &str,
533 ) -> Result<()> {
534 let session_id = session.session_id.clone();
535 tracing::info!(session_id = %session_id, "processing completed Fan+ checkout");
536
537 let metadata = FanPlusCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
538 let user_id = metadata.user_id;
539
540 // Get the Stripe subscription ID from the session
541 let stripe_subscription_id = session.subscription_id.clone().ok_or_else(|| {
542 tracing::error!("Fan+ checkout completed but no subscription ID on session");
543 AppError::BadRequest("Missing subscription ID on session".to_string())
544 })?;
545
546 // Get the Stripe customer ID from the session
547 let stripe_customer_id = session.customer_id.clone().ok_or_else(|| {
548 tracing::error!("Fan+ checkout completed but no customer ID on session");
549 AppError::BadRequest("Missing customer ID on session".to_string())
550 })?;
551
552 // Create the subscription record. Idempotent via ON CONFLICT (user_id) DO
553 // UPDATE with a guard WHERE: a duplicate webhook for an unchanged row updates
554 // nothing and RETURNING yields no row (-> None below, "already exists"); a
555 // genuine re-subscribe updates in place.
556 let Some(sub) = db::fan_plus::create_fan_plus_subscription(
557 db,
558 user_id,
559 &stripe_subscription_id,
560 &stripe_customer_id,
561 )
562 .await
563 .with_context(|| format!("create Fan+ subscription for user {user_id}"))?
564 else {
565 tracing::info!(user_id = %user_id, "Fan+ subscription already exists, ignoring duplicate");
566 return Ok(());
567 };
568
569 tracing::info!(
570 subscription_id = %sub.id, user_id = %user_id,
571 "Fan+ subscription created"
572 );
573
574 // Send welcome email (fire-and-forget)
575 if let Ok(Some(user)) = db::users::get_user_by_id(db, user_id).await {
576 let user_email = user.email.clone();
577 let user_name = user.display_name;
578 let email = email.clone();
579 bg.spawn("Fan+ welcome", async move {
580 if let Err(e) = email
581 .send_fan_plus_welcome(&user_email, user_name.as_deref())
582 .await
583 {
584 tracing::error!(error = ?e, "failed to send Fan+ welcome");
585 }
586 });
587 }
588
589 if let Err(e) = db::subscriptions::log_subscription_event(
590 db, None, event_id, MnwEventName::CheckoutCompletedFanPlus,
591 &serde_json::json!({"session_id": session_id, "stripe_subscription_id": stripe_subscription_id}),
592 ).await {
593 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
594 }
595
596 Ok(())
597 }
598
599 /// Handle checkout.session.completed for creator tier subscriptions
600 #[tracing::instrument(skip_all, name = "stripe::handle_creator_tier_checkout")]
601 pub(super) async fn handle_creator_tier_checkout_completed(
602 db: &PgPool,
603 bg: &crate::background::BackgroundTx,
604 wam: Option<&WamClient>,
605 payments: &Billing,
606 session: &crate::payments::CheckoutCompletion,
607 event_id: &str,
608 ) -> Result<()> {
609 let session_id = session.session_id.clone();
610 tracing::info!(session_id = %session_id, "processing completed creator tier checkout");
611
612 let metadata = CreatorTierCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
613 let user_id = metadata.user_id;
614 let tier: db::CreatorTier = metadata
615 .tier
616 .parse()
617 .map_err(|_| AppError::BadRequest(format!("Invalid tier: {}", metadata.tier)))?;
618
619 // Get the Stripe subscription ID from the session
620 let stripe_subscription_id = session.subscription_id.clone().ok_or_else(|| {
621 tracing::error!("Creator tier checkout completed but no subscription ID on session");
622 AppError::BadRequest("Missing subscription ID on session".to_string())
623 })?;
624
625 // Get the Stripe customer ID from the session
626 let stripe_customer_id = session.customer_id.clone().ok_or_else(|| {
627 tracing::error!("Creator tier checkout completed but no customer ID on session");
628 AppError::BadRequest("Missing customer ID on session".to_string())
629 })?;
630
631 // Create the subscription record. Idempotent via ON CONFLICT (user_id) DO
632 // UPDATE with a guard WHERE (`stripe_subscription_id != EXCLUDED OR status
633 // != 'active'`): a duplicate webhook updates nothing and RETURNING yields no
634 // row (-> None below); a genuine tier-switch or re-subscribe overwrites the
635 // row with the new subscription id and flips it active.
636 let Some(sub) = db::creator_tiers::create_creator_subscription(
637 db,
638 user_id,
639 &stripe_subscription_id,
640 &stripe_customer_id,
641 tier,
642 )
643 .await
644 .with_context(|| format!("create creator tier subscription for user {user_id}"))?
645 else {
646 tracing::info!(user_id = %user_id, "Creator tier subscription already exists, ignoring duplicate");
647 return Ok(());
648 };
649
650 // Sync the denormalized creator_tier column on users
651 db::creator_tiers::sync_user_creator_tier(db, user_id)
652 .await
653 .with_context(|| format!("sync creator tier for user {user_id}"))?;
654
655 // Auto-unhide: restore items hidden by post-grace enforcement
656 match db::items::unhide_all_items_for_user(db, user_id).await {
657 Ok(count) if count > 0 => {
658 tracing::info!(user_id = %user_id, items_unhidden = count, "auto-unhidden items after tier re-subscription");
659 }
660 Err(e) => {
661 tracing::warn!(user_id = %user_id, error = ?e, "failed to unhide items after tier re-subscription");
662 }
663 _ => {}
664 }
665
666 // Auto-unpause: if this creator was paused and just re-subscribed, clear the pause
667 // and un-cancel any fan subscriptions that haven't expired yet.
668 if let Ok(Some(db_user)) = db::users::get_user_by_id(db, user_id).await
669 && db_user.is_creator_paused()
670 {
671 db::users::unpause_creator(db, user_id)
672 .await
673 .with_context(|| format!("unpause creator {user_id}"))?;
674
675 // Un-cancel active fan subscriptions (clear cancel_at_period_end). Fanned
676 // out on the background queue: doing it inline here let a creator with
677 // many fans stall the webhook past Stripe's delivery timeout, which
678 // triggers a retry that re-runs the whole loop.
679 if let (Some(stripe), Some(stripe_account_id)) =
680 (&payments.stripe, &db_user.stripe_account_id)
681 {
682 let fan_subs = db::subscriptions::get_active_subscriptions_by_creator(db, user_id)
683 .await
684 .with_context(|| format!("fetch active fan subs for unpause {user_id}"))?;
685 let ids = fan_subs
686 .into_iter()
687 .map(|s| s.stripe_subscription_id)
688 .collect();
689 crate::payments::fan_ops::spawn_fan_sub_fanout(
690 bg,
691 std::sync::Arc::clone(stripe),
692 stripe_account_id.clone(),
693 ids,
694 crate::payments::fan_ops::FanSubOp::CancelAtPeriodEnd(false),
695 wam.cloned(),
696 );
697 }
698
699 tracing::info!(user_id = %user_id, "creator auto-unpaused after re-subscribing to tier");
700 }
701
702 tracing::info!(
703 user_id = %user_id, tier = %tier,
704 "creator tier subscription created"
705 );
706
707 if let Err(e) = db::subscriptions::log_subscription_event(
708 db,
709 None,
710 event_id,
711 MnwEventName::CheckoutCompletedCreatorTier,
712 &serde_json::json!({
713 "session_id": session_id,
714 "stripe_subscription_id": stripe_subscription_id,
715 "tier": sub.tier,
716 }),
717 )
718 .await
719 {
720 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
721 }
722
723 Ok(())
724 }
725
726 /// Handle checkout.session.completed for tips
727 #[tracing::instrument(skip_all, name = "stripe::handle_tip_checkout")]
728 pub(super) async fn handle_tip_checkout_completed(
729 db: &PgPool,
730 bg: &crate::background::BackgroundTx,
731 email: &EmailClient,
732 wam: Option<&WamClient>,
733 config: &Config,
734 session: &crate::payments::CheckoutCompletion,
735 event_id: &str,
736 ) -> Result<()> {
737 let session_id = session.session_id.clone();
738 tracing::info!(session_id = %session_id, "processing completed tip checkout");
739
740 let metadata = TipCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
741 let tipper_id = metadata.tipper_id;
742 let recipient_id = metadata.recipient_id;
743
744 // Complete the tip (idempotent). A PI-less session stores NULL (not a literal
745 // "unknown") in the money-keyed lookup column (Run 9).
746 match db::tips::complete_tip(db, &session_id, session.payment_intent_id.as_deref())
747 .await
748 .context("complete tip")?
749 {
750 Some(tip) => {
751 tracing::info!(
752 tip_id = %tip.id, tipper_id = %tipper_id, recipient_id = %recipient_id,
753 amount_cents = %tip.amount_cents, "tip completed"
754 );
755
756 // Log the event by id for audit parity with the other checkout
757 // handlers (record_tip_splits mints split revenue, so an event-id
758 // ledger entry matters for reconciliation). MINOR, Run #2 Payments.
759 if let Err(e) = db::subscriptions::log_subscription_event(
760 db,
761 None,
762 event_id,
763 MnwEventName::CheckoutCompletedTip,
764 &serde_json::json!({"session_id": session_id, "tip_id": tip.id}),
765 )
766 .await
767 {
768 tracing::warn!(event_id = %event_id, error = ?e, "failed to log tip event");
769 }
770
771 // Record revenue splits if the tip's project has members
772 if let Some(project_id) = tip.project_id {
773 record_tip_splits(db, tip.id, project_id, tip.amount_cents).await;
774 }
775
776 // Send tip notification email (fire-and-forget)
777 send_tip_email(db, bg, email, config, &tip, tipper_id, recipient_id);
778 }
779 None => {
780 // No pending row flipped to completed. Either a benign duplicate of
781 // an already-finalized tip, OR the first delivery crashed AFTER
782 // completing the tip but BEFORE recording splits, in which case
783 // collaborators hold a completed tip with no split rows. Re-fetch the
784 // tip and re-run the idempotent split write (ON CONFLICT DO NOTHING,
785 // migration 163); a genuine duplicate is a no-op. Run 20 Payments.
786 let recovered = db::tips::get_tip_by_session(db, &session_id)
787 .await
788 .context("re-fetch tip for crash recovery")?;
789 match recovered {
790 // A tip row exists (get_tip_by_session is status-agnostic), so
791 // this is either crash-recovery (re-run the idempotent splits) or
792 // a benign duplicate.
793 Some(tip) => {
794 if let Some(project_id) = tip.project_id {
795 tracing::info!(
796 session_id = %session_id, tip_id = %tip.id,
797 "crash-recovery: re-running tip splits for already-completed tip"
798 );
799 record_tip_splits(db, tip.id, project_id, tip.amount_cents).await;
800 } else {
801 tracing::info!(session_id = %session_id, "tip already completed, ignoring duplicate webhook");
802 }
803 }
804 // No tip row AT ALL for a paid session: the checkout session was
805 // created but the pending_tip insert never landed, so the tipper
806 // was charged and the recipient got nothing. Escalate like the
807 // purchase/cart/guest orphan path instead of logging a benign
808 // "duplicate", tips were the one checkout family without orphan
809 // escalation (Run 21 payments).
810 None => {
811 let pi = session.payment_intent_id.as_deref().unwrap_or("");
812 tracing::error!(
813 session_id = %session_id, payment_intent_id = %pi,
814 "orphaned paid session (tip): payment completed but no tip row exists, manual reconciliation required"
815 );
816 if let Some(wam) = wam.cloned() {
817 let sid = session_id.clone();
818 let pi = pi.to_string();
819 // Drained background pool, not a raw tokio::spawn: a
820 // charged-but-undelivered tip ticket must survive a
821 // mid-deploy restart (audit Run 22).
822 bg.spawn("stripe orphaned-tip ticket", async move {
823 let body = format!(
824 "Tip checkout session {sid} (payment_intent {pi}) completed at Stripe but has \
825 NO tip row, the pending tip was never created. The tipper was charged and the \
826 recipient received nothing. Reconcile manually: refund the payment or recreate the tip."
827 );
828 wam.create_ticket("Orphaned paid tip session", Some(&body), "high", "stripe-orphaned-tip", Some(&sid)).await;
829 });
830 }
831 }
832 }
833 }
834 }
835
836 Ok(())
837 }
838
839 /// Handle checkout.session.completed for guest purchases (no MNW account).
840 ///
841 /// Extracts the buyer's email from Stripe, completes the transaction, and
842 /// auto-attaches to an existing account if the email matches.
843 #[tracing::instrument(skip_all, name = "stripe::handle_guest_checkout")]
844 pub(super) async fn handle_guest_checkout_completed(
845 db: &PgPool,
846 bg: &crate::background::BackgroundTx,
847 email: &EmailClient,
848 wam: Option<&WamClient>,
849 config: &Config,
850 session: &crate::payments::CheckoutCompletion,
851 _event_id: &str,
852 ) -> Result<()> {
853 use crate::payments::GuestCheckoutMetadata;
854
855 let session_id = session.session_id.clone();
856 tracing::info!(session_id = %session_id, "processing completed guest checkout");
857
858 let meta = GuestCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
859
860 // Buyer email as Stripe collected it, flattened out of customer_details
861 // during normalization.
862 let guest_email = session
863 .customer_email
864 .as_deref()
865 .unwrap_or("unknown@guest")
866 .to_string();
867
868 // Display/logging copy only; the DB write below passes `session.payment_intent_id`
869 // directly so a PI-less session stores NULL, not a literal "unknown" that would
870 // collide with other PI-less rows in the money-keyed lookup column (Run 9).
871 let payment_intent_id = session.payment_intent_id.clone().unwrap_or_default();
872
873 // Complete the guest transaction and increment sales count in a single DB transaction
874 // (matching the non-guest path pattern to prevent counter drift on partial failure)
875 let mut db_tx = db
876 .begin()
877 .await
878 .context("begin guest checkout webhook transaction")?;
879
880 // Do NOT auto-attach on an email match. Stripe collects the buyer's email
881 // but does not prove the guest controls it, so matching alone would let
882 // someone drop a purchase into a stranger's verified library. The buyer
883 // always claims via the emailed claim link, which authenticates the
884 // recipient; the license key (if any) is minted at claim time
885 // (Run #21 Payments MINOR-1 / Max's call 2026-06-15).
886 match db::transactions::complete_guest_transaction(
887 &mut *db_tx,
888 &session_id,
889 session.payment_intent_id.as_deref(),
890 &guest_email,
891 )
892 .await?
893 {
894 Some(tx) => {
895 tracing::info!(
896 session_id = %session_id,
897 guest_email = %guest_email,
898 item_id = %meta.item_id,
899 "guest transaction completed"
900 );
901
902 // Defense-in-depth reconciliation (currency + subtotal), mirroring
903 // the single-item and cart paths.
904 reconcile_checkout_amount(
905 bg,
906 wam,
907 &session_id,
908 session,
909 i64::from(tx.amount_cents),
910 "guest",
911 );
912
913 // Increment sales count inside transaction
914 db::items::increment_sales_count(&mut *db_tx, meta.item_id)
915 .await
916 .with_context(|| {
917 format!("increment sales count for guest item {}", meta.item_id)
918 })?;
919
920 db_tx
921 .commit()
922 .await
923 .context("commit guest checkout webhook transaction")?;
924
925 // --- Secondary effects below (outside transaction) ---
926 // One re-runnable finalizer (splits + confirmation + sale email).
927 finalize_guest_transaction(
928 db,
929 bg,
930 email,
931 config,
932 &tx,
933 &guest_email,
934 meta.item_id,
935 meta.seller_id,
936 );
937 }
938 None => {
939 db_tx.commit().await.ok();
940 // Crash-recovery: re-fetch completed guest rows for the session and
941 // re-run the idempotent finalizer; only escalate if none exist.
942 let completed =
943 db::transactions::get_completed_transactions_for_session(db, &session_id)
944 .await
945 .context("re-fetch completed guest transactions for crash recovery")?;
946 if completed.is_empty() {
947 escalate_if_orphaned_session(
948 db,
949 bg,
950 wam,
951 &session_id,
952 &payment_intent_id,
953 "guest transaction",
954 )
955 .await?;
956 } else {
957 for tx in &completed {
958 let guest_email = tx
959 .guest_email
960 .clone()
961 .unwrap_or_else(|| guest_email.clone());
962 tracing::info!(
963 session_id = %session_id, transaction_id = %tx.id,
964 "crash-recovery: re-running finalize for already-completed guest session"
965 );
966 if let Some(item_id) = tx.item_id {
967 finalize_guest_transaction(
968 db,
969 bg,
970 email,
971 config,
972 tx,
973 &guest_email,
974 item_id,
975 meta.seller_id,
976 );
977 }
978 }
979 }
980 }
981 }
982
983 Ok(())
984 }
985
986 /// Handle checkout.session.completed for an end-user SyncKit app subscription.
987 /// Inserts the `app_sync_subscriptions` row; subsequent
988 /// `customer.subscription.updated/.deleted` events keep it in sync.
989 #[tracing::instrument(skip_all, name = "stripe::handle_synckit_app_sub_checkout")]
990 pub(super) async fn handle_synckit_app_sub_checkout_completed(
991 db: &PgPool,
992 session: &crate::payments::CheckoutCompletion,
993 event_id: &str,
994 ) -> Result<()> {
995 let session_id = session.session_id.clone();
996 tracing::info!(session_id = %session_id, "processing completed SyncKit app subscription checkout");
997
998 let meta = SynckitAppSubCheckoutMetadata::from_metadata(session.metadata.as_ref())?;
999
1000 let stripe_subscription_id = session
1001 .subscription_id
1002 .clone()
1003 .ok_or_else(|| AppError::BadRequest("Missing subscription ID on session".to_string()))?;
1004 let stripe_customer_id = session
1005 .customer_id
1006 .clone()
1007 .ok_or_else(|| AppError::BadRequest("Missing customer ID on session".to_string()))?;
1008
1009 let inserted = db::synckit::create_app_sync_subscription(
1010 db,
1011 &db::synckit::NewAppSyncSubscription {
1012 user_id: meta.user_id,
1013 app_id: meta.app_id,
1014 stripe_subscription_id: &stripe_subscription_id,
1015 stripe_customer_id: &stripe_customer_id,
1016 interval: &meta.interval,
1017 storage_limit_bytes: meta.storage_limit_bytes.unwrap_or(0),
1018 },
1019 )
1020 .await
1021 .with_context(|| {
1022 format!(
1023 "create app sync subscription user={} app={}",
1024 meta.user_id, meta.app_id
1025 )
1026 })?;
1027
1028 if !inserted {
1029 tracing::info!(
1030 user_id = %meta.user_id,
1031 app_id = %meta.app_id,
1032 "SyncKit app subscription already exists, ignoring duplicate webhook"
1033 );
1034 }
1035
1036 let _ = event_id;
1037 Ok(())
1038 }
1039