Skip to main content

max / makenotwork

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