Skip to main content

max / makenotwork

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