Skip to main content

max / makenotwork

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