Skip to main content

max / makenotwork

25.3 KB · 659 lines History Blame Raw
1 //! Webhook handlers for billing events (invoice payments, refunds).
2
3 use crate::{
4 db::{self, SubscriptionStatus},
5 email::EmailClient,
6 error::{Result, ResultExt},
7 helpers,
8 payments::{MnwEventName, SubscriptionProduct},
9 wam_client::WamClient,
10 };
11 use sqlx::PgPool;
12
13 /// Handle invoice.payment_succeeded; update period, send renewal email (not first invoice)
14 pub(super) async fn handle_invoice_payment_succeeded(
15 db: &PgPool,
16 bg: &crate::background::BackgroundTx,
17 email: &EmailClient,
18 wam: Option<&WamClient>,
19 invoice: &crate::payments::InvoiceOutcome,
20 event_id: &str,
21 ) -> Result<()> {
22 let stripe_sub_id = match invoice.subscription_id.as_deref() {
23 Some(s) => s.to_string(),
24 None => return Ok(()), // Not a subscription invoice
25 };
26
27 tracing::info!(stripe_sub_id = %stripe_sub_id, "processing invoice payment succeeded");
28
29 let is_renewal = invoice.is_renewal;
30
31 // End-user SyncKit app subscription? Apply any pending storage-cap change
32 // and refresh the period. Only meaningful on renewals; the first invoice's
33 // cap was set at checkout.
34 if db::synckit::get_subscription_by_stripe_id(db, &stripe_sub_id)
35 .await
36 .context("fetch app sync subscription by stripe id")?
37 .is_some()
38 {
39 db::synckit::update_app_sync_subscription_status(
40 db,
41 &stripe_sub_id,
42 "active",
43 Some(invoice.period_end),
44 )
45 .await
46 .context("refresh app sync subscription period")?;
47 if is_renewal {
48 db::synckit::apply_pending_storage_cap(db, &stripe_sub_id)
49 .await
50 .context("apply pending storage cap")?;
51 }
52 if let Err(e) = db::subscriptions::log_subscription_event(
53 db,
54 None,
55 event_id,
56 MnwEventName::InvoicePaymentSucceeded(SubscriptionProduct::SyncKitAppSub),
57 &serde_json::json!({"stripe_sub_id": stripe_sub_id, "is_renewal": is_renewal}),
58 )
59 .await
60 {
61 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
62 }
63 return Ok(());
64 }
65
66 // SyncKit v2 developer subscription? Identified by the local sync_apps row.
67 if let Some(app_id) = db::synckit_billing::get_app_by_stripe_subscription(db, &stripe_sub_id)
68 .await
69 .context("fetch synckit app by stripe sub id")?
70 {
71 let mut tx = db
72 .begin()
73 .await
74 .context("begin synckit invoice.paid transaction")?;
75 // One guarded write for status + period; only reset usage if the app was
76 // live (a canceled app is refused, so a stray invoice.paid can't refresh
77 // period or usage on it). Raw Stripe period to the sealed writer.
78 let applied = db::synckit_billing::apply_billing_update(
79 &mut *tx,
80 app_id,
81 Some("active"),
82 Some((invoice.period_start, invoice.period_end)),
83 )
84 .await
85 .context("synckit apply_billing_update")?;
86 if applied {
87 db::synckit_billing::reset_period_usage(&mut *tx, app_id)
88 .await
89 .context("synckit reset_period_usage")?;
90 }
91 tx.commit().await.context("commit synckit invoice.paid")?;
92 if let Err(e) = db::subscriptions::log_subscription_event(
93 db, None, event_id, MnwEventName::InvoicePaymentSucceeded(SubscriptionProduct::SyncKit),
94 &serde_json::json!({"stripe_sub_id": stripe_sub_id, "synckit_app_id": app_id.to_string()}),
95 ).await {
96 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
97 }
98 return Ok(());
99 }
100
101 // Check if this is a Fan+ subscription
102 if let Some(fan_sub) = db::fan_plus::get_fan_plus_by_stripe_id(db, &stripe_sub_id)
103 .await
104 .context("fetch fan+ by stripe id")?
105 {
106 // Refresh period (guarded: a canceled Fan+ sub is left untouched). Raw
107 // Stripe period to the sealed writer, which drops a non-positive end.
108 db::fan_plus::apply_stripe_update(
109 db,
110 &stripe_sub_id,
111 None,
112 Some((invoice.period_start, invoice.period_end)),
113 )
114 .await
115 .context("refresh fan+ period")?;
116
117 // On renewal, generate a $5 platform-wide promo code and email it.
118 //
119 // Idempotency: webhook dedup (`webhook/mod.rs`) is a check-then-act read
120 // that two concurrent deliveries of the same `invoice.payment_succeeded`
121 // both pass, so the credit, a money-moving side-effect, serializes on
122 // its own atomic write here. `try_claim_fan_plus_credit` inserts a
123 // `(stripe_sub_id, period_end)` row ON CONFLICT DO NOTHING and hands the
124 // winner a `FanPlusCreditClaim` witness; `issue_fan_plus_credit_code`
125 // requires that witness, so the mint+email path is unreachable without
126 // it. A redelivery (or duplicate concurrent delivery) gets `None` and
127 // skips, so a renewal issues at most one $5 credit.
128 if is_renewal
129 && let Some(claim) =
130 db::promo_codes::try_claim_fan_plus_credit(db, &stripe_sub_id, invoice.period_end)
131 .await
132 .context("claim fan+ credit issuance slot")?
133 {
134 let period_end = chrono::DateTime::from_timestamp(invoice.period_end, 0);
135
136 // Uniqueness of the generated code is enforced by the DB-level
137 // `UNIQUE(creator_id, upper(code))` partial index on `promo_codes`
138 // (see migration 019, idx_promo_codes_creator_code). The wordlist
139 // gives ~66 bits of entropy (6 words × log₂2048) so a collision
140 // within a single creator's history is astronomically unlikely;
141 // if one ever lands, the INSERT errors out as DB error 23505 and
142 // surfaces to the operator log, no silent overwrite.
143 let code = helpers::generate_key_code();
144 match db::promo_codes::issue_fan_plus_credit_code(
145 &claim,
146 db,
147 fan_sub.user_id,
148 code.as_str(),
149 period_end,
150 )
151 .await
152 {
153 Ok(pc) => {
154 tracing::info!(
155 promo_code_id = %pc.id, user_id = %fan_sub.user_id,
156 "Fan+ monthly credit promo code generated"
157 );
158
159 // Email the credit code (fire-and-forget)
160 if let Ok(Some(user)) = db::users::get_user_by_id(db, fan_sub.user_id).await {
161 let code_str = code.to_string();
162 let expiry = period_end;
163 let user_email = user.email.clone();
164 let user_name = user.display_name;
165 let email = email.clone();
166 bg.spawn("Fan+ credit", async move {
167 if let Err(e) = email
168 .send_fan_plus_credit(
169 &user_email,
170 user_name.as_deref(),
171 &code_str,
172 expiry.as_ref(),
173 )
174 .await
175 {
176 tracing::error!(error = ?e, "failed to send Fan+ credit");
177 }
178 });
179 }
180 }
181 Err(e) => {
182 tracing::error!(
183 user_id = %fan_sub.user_id, error = ?e,
184 "failed to generate Fan+ monthly credit promo code"
185 );
186 if let Some(wam) = wam {
187 let title = format!("Fan+ credit not issued: user {}", fan_sub.user_id);
188 let body = format!(
189 "Fan+ subscriber {} paid renewal but $5 credit promo code \
190 generation failed: {e}\n\nManually create a promo code.",
191 fan_sub.user_id,
192 );
193 wam.create_ticket(
194 &title,
195 Some(&body),
196 "high",
197 "fan-plus-credit-failed",
198 Some(&fan_sub.user_id.to_string()),
199 )
200 .await;
201 }
202 }
203 }
204 }
205
206 if let Err(e) = db::subscriptions::log_subscription_event(
207 db,
208 None,
209 event_id,
210 MnwEventName::InvoicePaymentSucceeded(SubscriptionProduct::FanPlus),
211 &serde_json::json!({"stripe_sub_id": stripe_sub_id, "is_renewal": is_renewal}),
212 )
213 .await
214 {
215 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
216 }
217 return Ok(());
218 }
219
220 // Check if this is a creator tier subscription
221 if let Some(_ct_sub) = db::creator_tiers::get_creator_sub_by_stripe_id(db, &stripe_sub_id)
222 .await
223 .context("fetch creator sub by stripe id")?
224 {
225 db::creator_tiers::apply_stripe_update(
226 db,
227 &stripe_sub_id,
228 None,
229 Some((invoice.period_start, invoice.period_end)),
230 )
231 .await
232 .context("refresh creator sub period")?;
233
234 if let Err(e) = db::subscriptions::log_subscription_event(
235 db,
236 None,
237 event_id,
238 MnwEventName::InvoicePaymentSucceeded(SubscriptionProduct::CreatorTier),
239 &serde_json::json!({"stripe_sub_id": stripe_sub_id, "is_renewal": is_renewal}),
240 )
241 .await
242 {
243 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
244 }
245 return Ok(());
246 }
247
248 // Refresh period for fan subscriptions (guarded: canceled rows untouched).
249 db::subscriptions::apply_stripe_update(
250 db,
251 &stripe_sub_id,
252 None,
253 Some((invoice.period_start, invoice.period_end)),
254 )
255 .await
256 .context("refresh subscription period")?;
257
258 // Send renewal email only for renewals (not the first invoice)
259 let db_sub = db::subscriptions::get_subscription_by_stripe_id(db, &stripe_sub_id)
260 .await
261 .context("fetch subscription by stripe id")?;
262
263 if is_renewal
264 && let Some(ref db_sub) = db_sub
265 && let (Ok(Some(subscriber)), Ok(Some(tier))) = (
266 db::users::get_user_by_id(db, db_sub.subscriber_id).await,
267 db::subscriptions::get_subscription_tier_by_id(db, db_sub.tier_id).await,
268 )
269 {
270 // The tier's Stripe Price was minted in the creator's currency, and the
271 // creator is the project owner, not the subscriber whose email this is.
272 let creator_currency = match tier.project_id {
273 Some(pid) => db::projects::get_project_by_id(db, pid)
274 .await
275 .ok()
276 .flatten()
277 .map(|p| p.user_id),
278 None => None,
279 };
280 let creator_currency = match creator_currency {
281 Some(uid) => db::users::get_user_by_id(db, uid)
282 .await
283 .ok()
284 .flatten()
285 .map(|u| u.settlement_currency)
286 .unwrap_or_default(),
287 None => crate::currency::SettlementCurrency::default(),
288 };
289 let price = helpers::format_price(tier.price_cents, creator_currency);
290 let sub_email = subscriber.email.clone();
291 let sub_name = subscriber.display_name;
292 let tier_name = tier.name;
293 let email = email.clone();
294 bg.spawn("subscription renewed", async move {
295 if let Err(e) = email
296 .send_subscription_renewed(&sub_email, sub_name.as_deref(), &tier_name, &price)
297 .await
298 {
299 tracing::error!(error = ?e, "failed to send subscription renewed");
300 }
301 });
302 }
303
304 // Log event
305 let sub_id = db_sub.as_ref().map(|s| s.id);
306 if let Err(e) = db::subscriptions::log_subscription_event(
307 db,
308 sub_id,
309 event_id,
310 MnwEventName::InvoicePaymentSucceeded(SubscriptionProduct::Undetermined),
311 &serde_json::json!({"stripe_sub_id": stripe_sub_id, "is_renewal": is_renewal}),
312 )
313 .await
314 {
315 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
316 }
317
318 Ok(())
319 }
320
321 /// Handle invoice.payment_failed; set status to past_due
322 pub(super) async fn handle_invoice_payment_failed(
323 db: &PgPool,
324 wam: Option<&WamClient>,
325 invoice: &crate::payments::InvoiceOutcome,
326 event_id: &str,
327 ) -> Result<()> {
328 let stripe_sub_id = match invoice.subscription_id.as_deref() {
329 Some(s) => s.to_string(),
330 None => return Ok(()), // Not a subscription invoice
331 };
332
333 tracing::info!(stripe_sub_id = %stripe_sub_id, "processing invoice payment failed");
334
335 // SyncKit v2 developer subscription? Mark suspended_unpaid.
336 if let Some(app_id) = db::synckit_billing::get_app_by_stripe_subscription(db, &stripe_sub_id)
337 .await
338 .context("fetch synckit app by stripe sub id")?
339 {
340 db::synckit_billing::apply_billing_update(db, app_id, Some("suspended_unpaid"), None)
341 .await
342 .context("synckit billing -> suspended_unpaid")?;
343 if let Err(e) = db::subscriptions::log_subscription_event(
344 db, None, event_id, MnwEventName::InvoicePaymentFailed(SubscriptionProduct::SyncKit),
345 &serde_json::json!({"stripe_sub_id": stripe_sub_id, "synckit_app_id": app_id.to_string()}),
346 ).await {
347 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
348 }
349 if let Some(wam) = wam {
350 let title = format!("SyncKit app payment failed: {app_id}");
351 wam.create_ticket(
352 &title,
353 None,
354 "medium",
355 "synckit-payment-failed",
356 Some(&app_id.to_string()),
357 )
358 .await;
359 }
360 return Ok(());
361 }
362
363 // Check if this is a Fan+ subscription
364 if let Some(_fan_sub) = db::fan_plus::get_fan_plus_by_stripe_id(db, &stripe_sub_id)
365 .await
366 .context("fetch fan+ by stripe id")?
367 {
368 db::fan_plus::apply_stripe_update(
369 db,
370 &stripe_sub_id,
371 Some(SubscriptionStatus::PastDue),
372 None,
373 )
374 .await
375 .context("fan+ status -> past_due")?;
376
377 if let Err(e) = db::subscriptions::log_subscription_event(
378 db,
379 None,
380 event_id,
381 MnwEventName::InvoicePaymentFailed(SubscriptionProduct::FanPlus),
382 &serde_json::json!({"stripe_sub_id": stripe_sub_id}),
383 )
384 .await
385 {
386 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
387 }
388 return Ok(());
389 }
390
391 // Check if this is a creator tier subscription
392 if let Some(ct_sub) = db::creator_tiers::get_creator_sub_by_stripe_id(db, &stripe_sub_id)
393 .await
394 .context("fetch creator sub by stripe id")?
395 {
396 db::creator_tiers::apply_stripe_update(
397 db,
398 &stripe_sub_id,
399 Some(SubscriptionStatus::PastDue),
400 None,
401 )
402 .await
403 .context("creator sub status -> past_due")?;
404 db::creator_tiers::sync_user_creator_tier(db, ct_sub.user_id)
405 .await
406 .context("sync user creator tier")?;
407
408 if let Err(e) = db::subscriptions::log_subscription_event(
409 db,
410 None,
411 event_id,
412 MnwEventName::InvoicePaymentFailed(SubscriptionProduct::CreatorTier),
413 &serde_json::json!({"stripe_sub_id": stripe_sub_id}),
414 )
415 .await
416 {
417 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
418 }
419 return Ok(());
420 }
421
422 let updated = db::subscriptions::apply_stripe_update(
423 db,
424 &stripe_sub_id,
425 Some(SubscriptionStatus::PastDue),
426 None,
427 )
428 .await
429 .context("subscription status -> past_due")?;
430
431 // Log event
432 let sub_id = updated.as_ref().map(|s| s.id);
433 if let Err(e) = db::subscriptions::log_subscription_event(
434 db,
435 sub_id,
436 event_id,
437 MnwEventName::InvoicePaymentFailed(SubscriptionProduct::Undetermined),
438 &serde_json::json!({"stripe_sub_id": stripe_sub_id}),
439 )
440 .await
441 {
442 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
443 }
444
445 // Create WAM ticket for subscription payment failures
446 if let Some(wam) = wam {
447 let title = format!("Subscription payment failed: {stripe_sub_id}");
448 wam.create_ticket(
449 &title,
450 None,
451 "medium",
452 "subscription-payment-failed",
453 Some(&stripe_sub_id),
454 )
455 .await;
456 }
457
458 Ok(())
459 }
460
461 /// Revoke a single refunded transaction: decrement its item's sales count,
462 /// revoke its license keys, and revoke + decrement any bundle-child transactions.
463 /// Returns `(keys_revoked, children_revoked)` for logging. Caller must have
464 /// already transitioned the row to `refunded` (so this runs exactly once per
465 /// transaction). Shared by the PI-wide `charge.refunded` path and the
466 /// line-scoped `refund.created` path.
467 async fn revoke_refunded_transaction(
468 db_tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
469 tx_id: db::TransactionId,
470 item_id: Option<db::ItemId>,
471 ) -> Result<(u64, usize)> {
472 // Project-level transactions store item_id IS NULL, skip the item-scoped
473 // updates for those; the project-members split rows aren't sales-counted.
474 if let Some(item_id) = item_id {
475 db::items::decrement_sales_count(&mut **db_tx, item_id)
476 .await
477 .context("decrement sales count")?;
478 }
479
480 let keys = db::license_keys::revoke_keys_by_transaction(db_tx, tx_id)
481 .await
482 .context("revoke license keys")?;
483
484 // Revoke child transactions granted via bundle purchase
485 let revoked_children = db::transactions::revoke_child_transactions(&mut **db_tx, tx_id)
486 .await
487 .context("revoke bundle child transactions")?;
488 for child_item_id in &revoked_children {
489 db::items::decrement_sales_count(&mut **db_tx, *child_item_id)
490 .await
491 .context("decrement child item sales count")?;
492 }
493 Ok((keys, revoked_children.len()))
494 }
495
496 /// Handle a `refund.created` / `refund.updated` webhook for a line-scoped refund.
497 ///
498 /// The self-service refund tags the Stripe refund with `mnw_transaction_id`; we
499 /// mark and revoke exactly that transaction. Cart lines share a PaymentIntent, so
500 /// this is what keeps a single-line refund from touching the order's other lines
501 /// (Run #2 Payments SERIOUS). Refunds without our metadata (e.g. issued from the
502 /// Stripe dashboard) are left to the `charge.refunded` path. Idempotent: the
503 /// `status = 'completed'` transition guard means re-delivery is a no-op, and a
504 /// later `charge.refunded` for the same refund finds nothing left to mark.
505 pub(super) async fn handle_refund_created(
506 db: &PgPool,
507 refund: &crate::payments::RefundOutcome,
508 ) -> Result<()> {
509 if !refund.succeeded {
510 return Ok(());
511 }
512 let Some(tx_id_str) = refund.mnw_transaction_id.as_deref() else {
513 return Ok(()); // out-of-band refund; charge.refunded handles full ones
514 };
515 let Ok(tx_id) = tx_id_str.parse::<db::TransactionId>() else {
516 tracing::warn!(mnw_transaction_id = %tx_id_str, "refund metadata transaction id unparseable; ignoring");
517 return Ok(());
518 };
519
520 let mut db_tx = db.begin().await.context("begin line refund")?;
521 let refunded = db::transactions::refund_transaction_by_id(&mut *db_tx, tx_id)
522 .await
523 .context("refund transaction by id")?;
524 match refunded {
525 Some((tx_id, item_id)) => {
526 let (keys, children) = revoke_refunded_transaction(&mut db_tx, tx_id, item_id).await?;
527 db_tx.commit().await.context("commit line refund")?;
528 tracing::info!(
529 transaction_id = %tx_id,
530 keys_revoked = keys,
531 bundle_children_revoked = children,
532 "line refund processed"
533 );
534 }
535 None => {
536 tracing::info!(
537 transaction_id = %tx_id,
538 "line refund: transaction not in a completed state (already refunded); no-op"
539 );
540 }
541 }
542 Ok(())
543 }
544
545 /// Handle charge.refunded webhook; revoke license keys on full refund,
546 /// log partial refunds without revoking access.
547 /// Process a full refund: revoke transactions/keys/access, or (for the direct
548 /// `charge.refunded` webhook) queue it as pending when no matching transaction
549 /// or tip exists yet. `requeue_if_unmatched` is false when called from the
550 /// pending-refund claim path, that row is already claimed, so re-queuing would
551 /// insert a duplicate (the partial-unique index only covers `matched_at IS
552 /// NULL`); instead we signal "still unmatched" so the caller releases the claim
553 /// and the stale sweep escalates it (MINOR, Run #23).
554 pub(super) async fn handle_charge_refunded(
555 db: &PgPool,
556 refund_data: &crate::payments::ChargeRefundData,
557 requeue_if_unmatched: bool,
558 ) -> Result<()> {
559 let payment_intent_id = &refund_data.payment_intent_id;
560 tracing::info!(
561 payment_intent_id = %payment_intent_id,
562 amount = refund_data.amount.as_i64(),
563 amount_refunded = refund_data.amount_refunded.as_i64(),
564 is_full = refund_data.is_full_refund(),
565 "processing charge refund"
566 );
567
568 // Partial refund: log but do not revoke access or keys
569 if !refund_data.is_full_refund() {
570 tracing::info!(
571 payment_intent_id = %payment_intent_id,
572 "partial refund, access and license keys preserved"
573 );
574 return Ok(());
575 }
576
577 let mut db_tx = db.begin().await.context("begin refund transaction")?;
578
579 // Mark transactions as refunded and get their IDs + item_ids
580 // (cart checkouts can have multiple transactions per payment_intent_id)
581 let refunded =
582 db::transactions::refund_transaction_by_payment_intent(&mut *db_tx, payment_intent_id)
583 .await
584 .context("refund transaction")?;
585
586 if refunded.is_empty() {
587 // No transaction found, check if this was a tip refund
588 let tip_refunded = db::tips::refund_tip_by_payment_intent(db, payment_intent_id)
589 .await
590 .inspect_err(|e| {
591 tracing::error!(
592 payment_intent_id = %payment_intent_id,
593 error = ?e,
594 "tip refund lookup failed"
595 );
596 })
597 .context("refund tip")?;
598 if tip_refunded {
599 tracing::info!(payment_intent_id = %payment_intent_id, "tip refund processed");
600 } else if db::transactions::transaction_exists_for_payment_intent(db, payment_intent_id)
601 .await
602 .context("check transaction existence for refund")?
603 {
604 // Transactions exist for this PI but none were 'completed', they were
605 // already refunded (line-scoped refund.created marked them, or a prior
606 // delivery did). Idempotent no-op; do NOT queue a pending refund.
607 tracing::info!(
608 payment_intent_id = %payment_intent_id,
609 "charge.refunded: transactions already refunded; no-op"
610 );
611 } else if requeue_if_unmatched {
612 // No transaction at all (and no tip), the payment webhook likely
613 // hasn't arrived yet. Queue the refund for later matching rather than
614 // silently dropping it.
615 tracing::warn!(
616 payment_intent_id = %payment_intent_id,
617 "no transaction or tip found, queuing as pending refund"
618 );
619 db::pending_refunds::insert_pending_refund(
620 db,
621 payment_intent_id,
622 refund_data.amount.as_i64(),
623 refund_data.amount_refunded.as_i64(),
624 )
625 .await
626 .context("insert pending refund")?;
627 } else {
628 // Claim path: the pending row is already claimed. Don't insert a
629 // duplicate, report it still-unmatched so the caller releases the
630 // claim and the stale-refund sweep escalates it for manual handling.
631 return Err(crate::error::AppError::Internal(anyhow::anyhow!(
632 "pending refund {payment_intent_id} still has no matching transaction or tip"
633 )));
634 }
635 } else {
636 let mut total_keys_revoked = 0u64;
637 let mut total_children_revoked = 0usize;
638
639 for (tx_id, item_id) in &refunded {
640 let (keys, children) =
641 revoke_refunded_transaction(&mut db_tx, *tx_id, *item_id).await?;
642 total_keys_revoked += keys;
643 total_children_revoked += children;
644 }
645
646 // Commit the refund atomically
647 db_tx.commit().await.context("commit refund transaction")?;
648
649 tracing::info!(
650 transactions_refunded = refunded.len(),
651 keys_revoked = total_keys_revoked,
652 bundle_children_revoked = total_children_revoked,
653 "refund processed"
654 );
655 }
656
657 Ok(())
658 }
659