Skip to main content

max / makenotwork

19.1 KB · 469 lines History Blame Raw
1 //! Stripe webhook event processing.
2
3 mod billing;
4 mod checkout;
5 pub(crate) mod checkout_helpers;
6 mod subscriptions;
7
8 use axum::{
9 body::Bytes,
10 extract::State,
11 http::{StatusCode, header::HeaderMap},
12 response::IntoResponse,
13 };
14 use sqlx::PgPool;
15
16 use crate::{
17 Billing, Integrations,
18 config::Config,
19 db,
20 email::EmailClient,
21 error::{AppError, Result, ResultExt},
22 payments::{AccountUpdate, CheckoutCompletion, CheckoutKind, MnwEvent, UntypedEvent},
23 wam_client::WamClient,
24 };
25
26 /// POST /stripe/webhook - Handle Stripe webhook events
27 #[tracing::instrument(skip_all, name = "stripe::webhook")]
28 #[allow(clippy::too_many_arguments)]
29 pub(in crate::routes::stripe) async fn webhook(
30 State(db): State<PgPool>,
31 State(bg): State<crate::background::BackgroundTx>,
32 State(email): State<EmailClient>,
33 State(integrations): State<Integrations>,
34 State(payments): State<Billing>,
35 State(config): State<Config>,
36 headers: HeaderMap,
37 body: Bytes,
38 ) -> Result<impl IntoResponse> {
39 let stripe = payments
40 .stripe
41 .as_ref()
42 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
43
44 // Get the signature header
45 let signature = headers
46 .get("stripe-signature")
47 .and_then(|v| v.to_str().ok())
48 .ok_or_else(|| AppError::BadRequest("Missing Stripe signature".to_string()))?;
49
50 // Parse and verify the webhook
51 let payload = std::str::from_utf8(&body)
52 .map_err(|_| AppError::BadRequest("Invalid payload encoding".to_string()))?;
53
54 // A failure here has no benign cause: Stripe signs correctly, so it means
55 // a wrong signing secret (real events being dropped) or forged events.
56 let event = stripe.verify_webhook(payload, signature).inspect_err(|_| {
57 crate::security_signals::note_webhook_signature_failure("stripe");
58 })?;
59 tracing::info!(event_type = %event.type_, event_id = %event.id, "received webhook event");
60
61 // Serialize concurrent redeliveries of this event id. Held across the whole
62 // dedup-read -> process -> mark sequence below, the lock makes the dedup read
63 // race-free. We *try* the lock rather than block on it: if a second delivery
64 // of the same event arrives while this one is mid-flight, it gets `None` and
65 // returns 503 immediately instead of parking a pooled connection for the
66 // duration of the in-flight handler (Run 23 Conc/Perf). Stripe redelivers
67 // after this one commits its processed-event mark, and the redelivery's dedup
68 // read then short-circuits. Dropping `_event_lock` on any return path rolls
69 // the (write-free) lock transaction back and releases it, it cannot leak.
70 // See `db::webhook_events::try_lock_event`.
71 let _event_lock = match db::webhook_events::try_lock_event(&db, &event.id).await {
72 Ok(Some(tx)) => tx,
73 Ok(None) => {
74 tracing::info!(event_id = %event.id, "concurrent delivery of this webhook event is in flight; returning 503 for redelivery");
75 return Ok(StatusCode::SERVICE_UNAVAILABLE);
76 }
77 Err(e) => {
78 tracing::error!(event_id = %event.id, error = ?e, "failed to acquire webhook event lock, returning 503 for retry");
79 return Ok(StatusCode::SERVICE_UNAVAILABLE);
80 }
81 };
82
83 // Deduplicate: skip if we already processed this event ID. This is a READ;
84 // the "processed" row is written only after the handler succeeds (below), so
85 // a crash mid-processing leaves no marker and Stripe's redelivery reprocesses.
86 // The `_event_lock` above closes the check-then-act race, so this read now
87 // guarantees exactly-once dispatch regardless of handler idempotency; per-
88 // handler atomic guards (status-guarded UPDATEs, ON CONFLICT writes, the
89 // `FanPlusCreditClaim` witness) remain as defense in depth.
90 match db::webhook_events::is_event_processed(&db, &event.id).await {
91 Ok(true) => {
92 tracing::info!(event_id = %event.id, "duplicate webhook event, skipping");
93 return Ok(StatusCode::OK);
94 }
95 Err(e) => {
96 tracing::error!(event_id = %event.id, error = ?e, "webhook dedup check failed, returning 503 for retry");
97 return Ok(StatusCode::SERVICE_UNAVAILABLE);
98 }
99 Ok(false) => {} // First time seeing this event
100 }
101
102 // For retry-queue persistence we need id+type after `event` is consumed.
103 // Move both out without cloning the underlying allocations.
104 let UntypedEvent {
105 id: event_id,
106 type_: event_type_str,
107 data_object,
108 } = event;
109 // Normalize before dispatch, so what follows reasons about an MNW event
110 // rather than a Stripe event-name string. A payload that will not parse
111 // fails here, with the same wording and the same retry-queue treatment it
112 // had when each match arm parsed for itself.
113 let result = match MnwEvent::normalize(&event_type_str, data_object) {
114 Ok(mnw_event) => {
115 process_webhook_event(
116 &db,
117 &bg,
118 &email,
119 integrations.wam.as_ref(),
120 &payments,
121 &config,
122 mnw_event,
123 &event_id,
124 )
125 .await
126 }
127 Err(e) => Err(e),
128 };
129
130 match result {
131 Ok(()) => {
132 // Work is durably committed, now record the event as processed so a
133 // redelivery short-circuits. If this write fails, return 503: Stripe
134 // redelivers, the idempotent handler re-runs, and the mark is retried.
135 // No event is lost; at worst it is processed twice (safe).
136 if let Err(e) = db::webhook_events::mark_event_processed(&db, &event_id).await {
137 tracing::error!(event_id = %event_id, error = ?e, "failed to record processed webhook event; returning 503 for redelivery");
138 return Ok(StatusCode::SERVICE_UNAVAILABLE);
139 }
140 }
141 Err(ref e) => {
142 tracing::error!(
143 event_id = %event_id, event_type = %event_type_str,
144 error = ?e, "webhook handler failed, queueing for retry"
145 );
146 // Not marked processed, so redelivery (or the retry worker) reruns it.
147 if let Err(queue_err) = db::webhook_events::insert_failed_event(
148 &db,
149 "stripe",
150 &event_type_str,
151 payload,
152 Some(signature),
153 &format!("{e:?}"),
154 )
155 .await
156 {
157 tracing::error!(error = ?queue_err, "failed to queue webhook event for retry; returning 503 to trigger Stripe redelivery");
158 return Ok(StatusCode::SERVICE_UNAVAILABLE);
159 }
160 }
161 }
162
163 Ok(StatusCode::OK)
164 }
165
166 /// Dispatch a normalized webhook event.
167 ///
168 /// Extracted so the caller can catch errors and persist to the retry queue;
169 /// shared by the live webhook handler and the scheduler's retry worker, which
170 /// is the point — both normalize through [`MnwEvent`] first, so neither can
171 /// grow its own idea of what an event type means.
172 ///
173 /// The match is on an enum, so an unhandled Stripe type is
174 /// [`MnwEvent::Unhandled`] by construction. A misspelt string can no longer
175 /// become a silently ignored event.
176 #[allow(clippy::too_many_arguments)]
177 pub(crate) async fn process_webhook_event(
178 db: &PgPool,
179 bg: &crate::background::BackgroundTx,
180 email: &EmailClient,
181 wam: Option<&WamClient>,
182 payments: &Billing,
183 config: &Config,
184 event: MnwEvent,
185 event_id: &str,
186 ) -> Result<()> {
187 match event {
188 MnwEvent::Checkout { kind, session } => {
189 dispatch_checkout_session(
190 db, bg, email, wam, payments, config, kind, &session, event_id,
191 )
192 .await?;
193 }
194 // No funds were captured, so there is nothing to deliver; the pending
195 // transaction (and any reserved promo hold) is released by the
196 // stale-pending cleanup sweeper. Logged rather than silently dropped.
197 MnwEvent::CheckoutAsyncPaymentFailed { session_id } => {
198 tracing::warn!(
199 %session_id,
200 "checkout async payment failed; no funds captured, pending rows will be released by cleanup"
201 );
202 }
203 MnwEvent::AccountUpdated(update) => {
204 handle_account_updated(db, wam, &config.signing_secret, &update).await?;
205 }
206 // Direct webhook: queue as pending if the matching payment hasn't landed
207 // yet. Out-of-band (dashboard) FULL refunds are handled here; per-line
208 // refunds land via RefundSettled below. A charge with no payment intent
209 // is out of scope and normalizes to `None`.
210 MnwEvent::ChargeRefunded(Some(refund_data)) => {
211 billing::handle_charge_refunded(db, &refund_data, true).await?;
212 }
213 MnwEvent::ChargeRefunded(None) => {}
214 // Line-scoped self-service refunds tag the Stripe refund with
215 // mnw_transaction_id; this marks/revokes exactly that transaction so a
216 // cart line refund leaves the order's other lines untouched (Run #2
217 // Payments SERIOUS). Untagged refunds are no-ops here.
218 MnwEvent::RefundSettled(refund) => {
219 billing::handle_refund_created(db, &refund).await?;
220 }
221 MnwEvent::SubscriptionUpdated(sub) => {
222 subscriptions::handle_subscription_updated(db, &sub, event_id).await?;
223 }
224 MnwEvent::SubscriptionDeleted(sub) => {
225 subscriptions::handle_subscription_deleted(db, bg, email, &sub, event_id).await?;
226 }
227 MnwEvent::InvoicePaymentSucceeded(invoice) => {
228 billing::handle_invoice_payment_succeeded(db, bg, email, wam, &invoice, event_id)
229 .await?;
230 }
231 MnwEvent::InvoicePaymentFailed(invoice) => {
232 billing::handle_invoice_payment_failed(db, wam, &invoice, event_id).await?;
233 }
234 MnwEvent::Unhandled { stripe_type } => {
235 tracing::debug!(event_type = %stripe_type, "unhandled webhook event type");
236 }
237 }
238
239 Ok(())
240 }
241
242 /// Route a completed checkout to its handler.
243 ///
244 /// The kind was settled during normalization, from the metadata MNW itself
245 /// wrote at checkout creation, so this is a match rather than a ladder of
246 /// predicates.
247 ///
248 /// Subscription-mode sessions (Fan+, creator tier, SyncKit app sub, project
249 /// subscription) capture no funds at checkout — the subscription lifecycle
250 /// bills separately — so they run unconditionally. One-time payment-mode
251 /// sessions (tip, guest, cart, single purchase) capture funds now and are
252 /// therefore gated on `settled`: an async method that reports
253 /// `payment_status = "unpaid"` on `checkout.session.completed` is deferred
254 /// until Stripe re-delivers the settled session via
255 /// `checkout.session.async_payment_succeeded`. Without this gate, enabling any
256 /// async payment method on a connected account would mint license keys and
257 /// grant downloads before funds settle.
258 #[allow(clippy::too_many_arguments)]
259 async fn dispatch_checkout_session(
260 db: &PgPool,
261 bg: &crate::background::BackgroundTx,
262 email: &EmailClient,
263 wam: Option<&WamClient>,
264 payments: &Billing,
265 config: &Config,
266 kind: CheckoutKind,
267 session: &CheckoutCompletion,
268 event_id: &str,
269 ) -> Result<()> {
270 // One-time payment-mode: funds captured now. Deliver only once settled.
271 // Asked before the match so the gate cannot be forgotten on a new
272 // funds-capturing kind: `captures_funds_at_checkout` is exhaustive over
273 // `CheckoutKind`, so adding one is a compile error until it answers.
274 if kind.captures_funds_at_checkout() && !session.settled {
275 tracing::info!(
276 session_id = %session.session_id,
277 ?kind,
278 "one-time checkout not yet settled (async payment); deferring finalize until async_payment_succeeded"
279 );
280 return Ok(());
281 }
282
283 match kind {
284 CheckoutKind::FanPlus => {
285 checkout::handle_fan_plus_checkout_completed(db, bg, email, session, event_id).await
286 }
287 CheckoutKind::CreatorTier => {
288 checkout::handle_creator_tier_checkout_completed(
289 db, bg, wam, payments, session, event_id,
290 )
291 .await
292 }
293 CheckoutKind::SyncKitAppSub => {
294 checkout::handle_synckit_app_sub_checkout_completed(db, session, event_id).await
295 }
296 CheckoutKind::ProjectSubscription => {
297 checkout::handle_subscription_checkout_completed(db, bg, email, session, event_id).await
298 }
299 CheckoutKind::Tip => {
300 checkout::handle_tip_checkout_completed(db, bg, email, wam, config, session, event_id)
301 .await
302 }
303 CheckoutKind::Guest => {
304 checkout::handle_guest_checkout_completed(db, bg, email, wam, config, session, event_id)
305 .await
306 }
307 CheckoutKind::Cart => {
308 checkout::handle_cart_checkout_completed(db, bg, email, wam, config, session, event_id)
309 .await
310 }
311 CheckoutKind::Purchase => {
312 checkout::handle_purchase_checkout_completed(
313 db, bg, email, wam, config, session, event_id,
314 )
315 .await
316 }
317 }
318 }
319
320 /// Handle account.updated from the v2 thin event endpoint.
321 pub(in crate::routes::stripe) async fn handle_account_updated_from_v2(
322 db: &PgPool,
323 wam: Option<&WamClient>,
324 signing_secret: &str,
325 update: &AccountUpdate,
326 ) -> Result<()> {
327 handle_account_updated(db, wam, signing_secret, update).await
328 }
329
330 /// Handle account.updated webhook
331 async fn handle_account_updated(
332 db: &PgPool,
333 wam: Option<&WamClient>,
334 signing_secret: &str,
335 update: &AccountUpdate,
336 ) -> Result<()> {
337 tracing::info!(
338 account_id = %update.account_id, charges_enabled = %update.charges_enabled,
339 payouts_enabled = %update.payouts_enabled, details_submitted = %update.details_submitted,
340 settlement_currency = ?update.settlement_currency,
341 "account updated"
342 );
343
344 // Read the stored currency before overwriting it, so a genuine change can be
345 // told apart from Stripe restating the same value on one of the many
346 // `account.updated` events it sends.
347 let previous_currency =
348 db::users::get_settlement_currency_by_stripe_account(db, &update.account_id)
349 .await
350 .unwrap_or(None);
351
352 // Update the user's Stripe status
353 db::users::update_user_stripe_status(
354 db,
355 &update.account_id,
356 update.details_submitted,
357 update.payouts_enabled,
358 update.charges_enabled,
359 update.settlement_currency,
360 )
361 .await
362 .with_context(|| format!("update Stripe status for account {}", update.account_id))?;
363
364 // A settlement currency change is not a status change: it silently
365 // redenominates every price the creator has set. Their 1000 was ten pounds
366 // and is now ten euros, and only they can decide what the number should be.
367 // Nothing here rewrites their prices, because guessing at a rate is exactly
368 // what this design refuses to do; it raises the alarm so a person acts.
369 //
370 // Two alarms, for two audiences. The pending acknowledgement tells the
371 // creator and keeps telling them until they confirm they read it, because
372 // they are the only one who can fix the prices and a single email that
373 // landed in spam is indistinguishable from one that was ignored. The wam
374 // ticket tells us, immediately, because somebody should know that a
375 // creator's catalogue is mispriced right now rather than in a month.
376 if let (Some(new_currency), Some(old_currency)) =
377 (update.settlement_currency, previous_currency)
378 && new_currency != old_currency
379 {
380 tracing::warn!(
381 account_id = %update.account_id,
382 %old_currency, %new_currency,
383 "settlement currency changed; the creator's existing prices now mean different money"
384 );
385
386 // Keyed on the pair, so a currency that flaps between the same two
387 // values keeps one open alert while a genuinely new change opens its
388 // own. A failure here must not fail the webhook: Stripe would retry the
389 // whole event, and the status update above has already been applied.
390 match db::users::get_user_id_by_stripe_account(db, &update.account_id).await {
391 Ok(Some(user_id)) => {
392 let details = serde_json::json!({
393 "from": old_currency.to_string(),
394 "to": new_currency.to_string(),
395 });
396 let opened = db::acknowledgements::open(
397 db,
398 user_id,
399 db::AckKind::SettlementCurrencyChanged,
400 &format!("{old_currency}->{new_currency}"),
401 details,
402 signing_secret,
403 )
404 .await;
405 match opened {
406 Ok(true) => tracing::info!(
407 %user_id,
408 "opened a settlement-currency acknowledgement for the creator"
409 ),
410 Ok(false) => {}
411 Err(e) => tracing::error!(
412 error = ?e, %user_id,
413 "could not open the settlement-currency acknowledgement; the wam \
414 ticket below is the only alarm left"
415 ),
416 }
417 }
418 Ok(None) => tracing::warn!(
419 account_id = %update.account_id,
420 "settlement currency changed on an account with no user; cannot notify anyone"
421 ),
422 Err(e) => tracing::error!(
423 error = ?e,
424 "could not resolve the user behind the settlement-currency change"
425 ),
426 }
427
428 if let Some(wam) = wam {
429 let title = format!("Settlement currency changed: {}", update.account_id);
430 let body = format!(
431 "This creator's Stripe account moved from {old_currency} to {new_currency}.\n\n\
432 Every price they have already set is stored as a bare number, so those \
433 numbers now mean {new_currency} instead of {old_currency}. Nothing has been \
434 converted and nothing has been rewritten.\n\n\
435 They need to re-check their prices. Contact them."
436 );
437 wam.create_ticket(
438 &title,
439 Some(&body),
440 "high",
441 "settlement-currency-changed",
442 Some(&update.account_id),
443 )
444 .await;
445 }
446 }
447
448 // Alert if charges or payouts became disabled (creator can't receive payments)
449 if (!update.charges_enabled || !update.payouts_enabled)
450 && let Some(wam) = wam
451 {
452 let title = format!("Stripe Connect degraded: {}", update.account_id);
453 let body = format!(
454 "charges_enabled: {}\npayouts_enabled: {}\ndetails_submitted: {}",
455 update.charges_enabled, update.payouts_enabled, update.details_submitted,
456 );
457 wam.create_ticket(
458 &title,
459 Some(&body),
460 "high",
461 "stripe-connect-degraded",
462 Some(&update.account_id),
463 )
464 .await;
465 }
466
467 Ok(())
468 }
469