Skip to main content

max / makenotwork

24.2 KB · 639 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 let price = helpers::format_price(tier.price_cents);
270 let sub_email = subscriber.email.clone();
271 let sub_name = subscriber.display_name;
272 let tier_name = tier.name;
273 let email = email.clone();
274 bg.spawn("subscription renewed", async move {
275 if let Err(e) = email
276 .send_subscription_renewed(&sub_email, sub_name.as_deref(), &tier_name, &price)
277 .await
278 {
279 tracing::error!(error = ?e, "failed to send subscription renewed");
280 }
281 });
282 }
283
284 // Log event
285 let sub_id = db_sub.as_ref().map(|s| s.id);
286 if let Err(e) = db::subscriptions::log_subscription_event(
287 db,
288 sub_id,
289 event_id,
290 "invoice.payment_succeeded",
291 &serde_json::json!({"stripe_sub_id": stripe_sub_id, "is_renewal": is_renewal}),
292 )
293 .await
294 {
295 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
296 }
297
298 Ok(())
299 }
300
301 /// Handle invoice.payment_failed; set status to past_due
302 pub(super) async fn handle_invoice_payment_failed(
303 db: &PgPool,
304 wam: Option<&WamClient>,
305 invoice: &crate::payments::InvoiceView,
306 event_id: &str,
307 ) -> Result<()> {
308 let stripe_sub_id = match invoice.subscription_id() {
309 Some(s) => s.to_string(),
310 None => return Ok(()), // Not a subscription invoice
311 };
312
313 tracing::info!(stripe_sub_id = %stripe_sub_id, "processing invoice payment failed");
314
315 // SyncKit v2 developer subscription? Mark suspended_unpaid.
316 if let Some(app_id) = db::synckit_billing::get_app_by_stripe_subscription(db, &stripe_sub_id)
317 .await
318 .context("fetch synckit app by stripe sub id")?
319 {
320 db::synckit_billing::apply_billing_update(db, app_id, Some("suspended_unpaid"), None)
321 .await
322 .context("synckit billing -> suspended_unpaid")?;
323 if let Err(e) = db::subscriptions::log_subscription_event(
324 db, None, event_id, "invoice.payment_failed.synckit",
325 &serde_json::json!({"stripe_sub_id": stripe_sub_id, "synckit_app_id": app_id.to_string()}),
326 ).await {
327 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
328 }
329 if let Some(wam) = wam {
330 let title = format!("SyncKit app payment failed: {app_id}");
331 wam.create_ticket(
332 &title,
333 None,
334 "medium",
335 "synckit-payment-failed",
336 Some(&app_id.to_string()),
337 )
338 .await;
339 }
340 return Ok(());
341 }
342
343 // Check if this is a Fan+ subscription
344 if let Some(_fan_sub) = db::fan_plus::get_fan_plus_by_stripe_id(db, &stripe_sub_id)
345 .await
346 .context("fetch fan+ by stripe id")?
347 {
348 db::fan_plus::apply_stripe_update(
349 db,
350 &stripe_sub_id,
351 Some(SubscriptionStatus::PastDue),
352 None,
353 )
354 .await
355 .context("fan+ status -> past_due")?;
356
357 if let Err(e) = db::subscriptions::log_subscription_event(
358 db,
359 None,
360 event_id,
361 "invoice.payment_failed.fan_plus",
362 &serde_json::json!({"stripe_sub_id": stripe_sub_id}),
363 )
364 .await
365 {
366 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
367 }
368 return Ok(());
369 }
370
371 // Check if this is a creator tier subscription
372 if let Some(ct_sub) = db::creator_tiers::get_creator_sub_by_stripe_id(db, &stripe_sub_id)
373 .await
374 .context("fetch creator sub by stripe id")?
375 {
376 db::creator_tiers::apply_stripe_update(
377 db,
378 &stripe_sub_id,
379 Some(SubscriptionStatus::PastDue),
380 None,
381 )
382 .await
383 .context("creator sub status -> past_due")?;
384 db::creator_tiers::sync_user_creator_tier(db, ct_sub.user_id)
385 .await
386 .context("sync user creator tier")?;
387
388 if let Err(e) = db::subscriptions::log_subscription_event(
389 db,
390 None,
391 event_id,
392 "invoice.payment_failed.creator_tier",
393 &serde_json::json!({"stripe_sub_id": stripe_sub_id}),
394 )
395 .await
396 {
397 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
398 }
399 return Ok(());
400 }
401
402 let updated = db::subscriptions::apply_stripe_update(
403 db,
404 &stripe_sub_id,
405 Some(SubscriptionStatus::PastDue),
406 None,
407 )
408 .await
409 .context("subscription status -> past_due")?;
410
411 // Log event
412 let sub_id = updated.as_ref().map(|s| s.id);
413 if let Err(e) = db::subscriptions::log_subscription_event(
414 db,
415 sub_id,
416 event_id,
417 "invoice.payment_failed",
418 &serde_json::json!({"stripe_sub_id": stripe_sub_id}),
419 )
420 .await
421 {
422 tracing::error!(event_id = %event_id, error = ?e, "failed to log subscription event");
423 }
424
425 // Create WAM ticket for subscription payment failures
426 if let Some(wam) = wam {
427 let title = format!("Subscription payment failed: {stripe_sub_id}");
428 wam.create_ticket(
429 &title,
430 None,
431 "medium",
432 "subscription-payment-failed",
433 Some(&stripe_sub_id),
434 )
435 .await;
436 }
437
438 Ok(())
439 }
440
441 /// Revoke a single refunded transaction: decrement its item's sales count,
442 /// revoke its license keys, and revoke + decrement any bundle-child transactions.
443 /// Returns `(keys_revoked, children_revoked)` for logging. Caller must have
444 /// already transitioned the row to `refunded` (so this runs exactly once per
445 /// transaction). Shared by the PI-wide `charge.refunded` path and the
446 /// line-scoped `refund.created` path.
447 async fn revoke_refunded_transaction(
448 db_tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
449 tx_id: db::TransactionId,
450 item_id: Option<db::ItemId>,
451 ) -> Result<(u64, usize)> {
452 // Project-level transactions store item_id IS NULL, skip the item-scoped
453 // updates for those; the project-members split rows aren't sales-counted.
454 if let Some(item_id) = item_id {
455 db::items::decrement_sales_count(&mut **db_tx, item_id)
456 .await
457 .context("decrement sales count")?;
458 }
459
460 let keys = db::license_keys::revoke_keys_by_transaction(db_tx, tx_id)
461 .await
462 .context("revoke license keys")?;
463
464 // Revoke child transactions granted via bundle purchase
465 let revoked_children = db::transactions::revoke_child_transactions(&mut **db_tx, tx_id)
466 .await
467 .context("revoke bundle child transactions")?;
468 for child_item_id in &revoked_children {
469 db::items::decrement_sales_count(&mut **db_tx, *child_item_id)
470 .await
471 .context("decrement child item sales count")?;
472 }
473 Ok((keys, revoked_children.len()))
474 }
475
476 /// Handle a `refund.created` / `refund.updated` webhook for a line-scoped refund.
477 ///
478 /// The self-service refund tags the Stripe refund with `mnw_transaction_id`; we
479 /// mark and revoke exactly that transaction. Cart lines share a PaymentIntent, so
480 /// this is what keeps a single-line refund from touching the order's other lines
481 /// (Run #2 Payments SERIOUS). Refunds without our metadata (e.g. issued from the
482 /// Stripe dashboard) are left to the `charge.refunded` path. Idempotent: the
483 /// `status = 'completed'` transition guard means re-delivery is a no-op, and a
484 /// later `charge.refunded` for the same refund finds nothing left to mark.
485 pub(super) async fn handle_refund_created(
486 db: &PgPool,
487 refund: &crate::payments::RefundView,
488 ) -> Result<()> {
489 if !refund.is_succeeded() {
490 return Ok(());
491 }
492 let Some(tx_id_str) = refund.mnw_transaction_id() else {
493 return Ok(()); // out-of-band refund; charge.refunded handles full ones
494 };
495 let Ok(tx_id) = tx_id_str.parse::<db::TransactionId>() else {
496 tracing::warn!(mnw_transaction_id = %tx_id_str, "refund metadata transaction id unparseable; ignoring");
497 return Ok(());
498 };
499
500 let mut db_tx = db.begin().await.context("begin line refund")?;
501 let refunded = db::transactions::refund_transaction_by_id(&mut *db_tx, tx_id)
502 .await
503 .context("refund transaction by id")?;
504 match refunded {
505 Some((tx_id, item_id)) => {
506 let (keys, children) = revoke_refunded_transaction(&mut db_tx, tx_id, item_id).await?;
507 db_tx.commit().await.context("commit line refund")?;
508 tracing::info!(
509 transaction_id = %tx_id,
510 keys_revoked = keys,
511 bundle_children_revoked = children,
512 "line refund processed"
513 );
514 }
515 None => {
516 tracing::info!(
517 transaction_id = %tx_id,
518 "line refund: transaction not in a completed state (already refunded); no-op"
519 );
520 }
521 }
522 Ok(())
523 }
524
525 /// Handle charge.refunded webhook; revoke license keys on full refund,
526 /// log partial refunds without revoking access.
527 /// Process a full refund: revoke transactions/keys/access, or (for the direct
528 /// `charge.refunded` webhook) queue it as pending when no matching transaction
529 /// or tip exists yet. `requeue_if_unmatched` is false when called from the
530 /// pending-refund claim path, that row is already claimed, so re-queuing would
531 /// insert a duplicate (the partial-unique index only covers `matched_at IS
532 /// NULL`); instead we signal "still unmatched" so the caller releases the claim
533 /// and the stale sweep escalates it (MINOR, Run #23).
534 pub(super) async fn handle_charge_refunded(
535 db: &PgPool,
536 refund_data: &crate::payments::ChargeRefundData,
537 requeue_if_unmatched: bool,
538 ) -> Result<()> {
539 let payment_intent_id = &refund_data.payment_intent_id;
540 tracing::info!(
541 payment_intent_id = %payment_intent_id,
542 amount = refund_data.amount.as_i64(),
543 amount_refunded = refund_data.amount_refunded.as_i64(),
544 is_full = refund_data.is_full_refund(),
545 "processing charge refund"
546 );
547
548 // Partial refund: log but do not revoke access or keys
549 if !refund_data.is_full_refund() {
550 tracing::info!(
551 payment_intent_id = %payment_intent_id,
552 "partial refund, access and license keys preserved"
553 );
554 return Ok(());
555 }
556
557 let mut db_tx = db.begin().await.context("begin refund transaction")?;
558
559 // Mark transactions as refunded and get their IDs + item_ids
560 // (cart checkouts can have multiple transactions per payment_intent_id)
561 let refunded =
562 db::transactions::refund_transaction_by_payment_intent(&mut *db_tx, payment_intent_id)
563 .await
564 .context("refund transaction")?;
565
566 if refunded.is_empty() {
567 // No transaction found, check if this was a tip refund
568 let tip_refunded = db::tips::refund_tip_by_payment_intent(db, payment_intent_id)
569 .await
570 .inspect_err(|e| {
571 tracing::error!(
572 payment_intent_id = %payment_intent_id,
573 error = ?e,
574 "tip refund lookup failed"
575 );
576 })
577 .context("refund tip")?;
578 if tip_refunded {
579 tracing::info!(payment_intent_id = %payment_intent_id, "tip refund processed");
580 } else if db::transactions::transaction_exists_for_payment_intent(db, payment_intent_id)
581 .await
582 .context("check transaction existence for refund")?
583 {
584 // Transactions exist for this PI but none were 'completed', they were
585 // already refunded (line-scoped refund.created marked them, or a prior
586 // delivery did). Idempotent no-op; do NOT queue a pending refund.
587 tracing::info!(
588 payment_intent_id = %payment_intent_id,
589 "charge.refunded: transactions already refunded; no-op"
590 );
591 } else if requeue_if_unmatched {
592 // No transaction at all (and no tip), the payment webhook likely
593 // hasn't arrived yet. Queue the refund for later matching rather than
594 // silently dropping it.
595 tracing::warn!(
596 payment_intent_id = %payment_intent_id,
597 "no transaction or tip found, queuing as pending refund"
598 );
599 db::pending_refunds::insert_pending_refund(
600 db,
601 payment_intent_id,
602 refund_data.amount.as_i64(),
603 refund_data.amount_refunded.as_i64(),
604 )
605 .await
606 .context("insert pending refund")?;
607 } else {
608 // Claim path: the pending row is already claimed. Don't insert a
609 // duplicate, report it still-unmatched so the caller releases the
610 // claim and the stale-refund sweep escalates it for manual handling.
611 return Err(crate::error::AppError::Internal(anyhow::anyhow!(
612 "pending refund {payment_intent_id} still has no matching transaction or tip"
613 )));
614 }
615 } else {
616 let mut total_keys_revoked = 0u64;
617 let mut total_children_revoked = 0usize;
618
619 for (tx_id, item_id) in &refunded {
620 let (keys, children) =
621 revoke_refunded_transaction(&mut db_tx, *tx_id, *item_id).await?;
622 total_keys_revoked += keys;
623 total_children_revoked += children;
624 }
625
626 // Commit the refund atomically
627 db_tx.commit().await.context("commit refund transaction")?;
628
629 tracing::info!(
630 transactions_refunded = refunded.len(),
631 keys_revoked = total_keys_revoked,
632 bundle_children_revoked = total_children_revoked,
633 "refund processed"
634 );
635 }
636
637 Ok(())
638 }
639