Skip to main content

max / makenotwork

15.6 KB · 375 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 let event = stripe.verify_webhook(payload, signature)?;
58 tracing::info!(event_type = %event.type_, event_id = %event.id, "received webhook event");
59
60 // Serialize concurrent redeliveries of this event id. Held across the whole
61 // dedup-read -> process -> mark sequence below, the lock makes the dedup read
62 // race-free. We *try* the lock rather than block on it: if a second delivery
63 // of the same event arrives while this one is mid-flight, it gets `None` and
64 // returns 503 immediately instead of parking a pooled connection for the
65 // duration of the in-flight handler (Run 23 Conc/Perf). Stripe redelivers
66 // after this one commits its processed-event mark, and the redelivery's dedup
67 // read then short-circuits. Dropping `_event_lock` on any return path rolls
68 // the (write-free) lock transaction back and releases it, it cannot leak.
69 // See `db::webhook_events::try_lock_event`.
70 let _event_lock = match db::webhook_events::try_lock_event(&db, &event.id).await {
71 Ok(Some(tx)) => tx,
72 Ok(None) => {
73 tracing::info!(event_id = %event.id, "concurrent delivery of this webhook event is in flight; returning 503 for redelivery");
74 return Ok(StatusCode::SERVICE_UNAVAILABLE);
75 }
76 Err(e) => {
77 tracing::error!(event_id = %event.id, error = ?e, "failed to acquire webhook event lock, returning 503 for retry");
78 return Ok(StatusCode::SERVICE_UNAVAILABLE);
79 }
80 };
81
82 // Deduplicate: skip if we already processed this event ID. This is a READ;
83 // the "processed" row is written only after the handler succeeds (below), so
84 // a crash mid-processing leaves no marker and Stripe's redelivery reprocesses.
85 // The `_event_lock` above closes the check-then-act race, so this read now
86 // guarantees exactly-once dispatch regardless of handler idempotency; per-
87 // handler atomic guards (status-guarded UPDATEs, ON CONFLICT writes, the
88 // `FanPlusCreditClaim` witness) remain as defense in depth.
89 match db::webhook_events::is_event_processed(&db, &event.id).await {
90 Ok(true) => {
91 tracing::info!(event_id = %event.id, "duplicate webhook event, skipping");
92 return Ok(StatusCode::OK);
93 }
94 Err(e) => {
95 tracing::error!(event_id = %event.id, error = ?e, "webhook dedup check failed, returning 503 for retry");
96 return Ok(StatusCode::SERVICE_UNAVAILABLE);
97 }
98 Ok(false) => {} // First time seeing this event
99 }
100
101 // For retry-queue persistence we need id+type after `event` is consumed.
102 // Move both out without cloning the underlying allocations.
103 let UntypedEvent {
104 id: event_id,
105 type_: event_type_str,
106 data_object,
107 } = event;
108 let result = process_webhook_event(
109 &db,
110 &bg,
111 &email,
112 integrations.wam.as_ref(),
113 &payments,
114 &config,
115 &event_type_str,
116 &event_id,
117 data_object,
118 )
119 .await;
120
121 match result {
122 Ok(()) => {
123 // Work is durably committed, now record the event as processed so a
124 // redelivery short-circuits. If this write fails, return 503: Stripe
125 // redelivers, the idempotent handler re-runs, and the mark is retried.
126 // No event is lost; at worst it is processed twice (safe).
127 if let Err(e) = db::webhook_events::mark_event_processed(&db, &event_id).await {
128 tracing::error!(event_id = %event_id, error = ?e, "failed to record processed webhook event; returning 503 for redelivery");
129 return Ok(StatusCode::SERVICE_UNAVAILABLE);
130 }
131 }
132 Err(ref e) => {
133 tracing::error!(
134 event_id = %event_id, event_type = %event_type_str,
135 error = ?e, "webhook handler failed, queueing for retry"
136 );
137 // Not marked processed, so redelivery (or the retry worker) reruns it.
138 if let Err(queue_err) = db::webhook_events::insert_failed_event(
139 &db,
140 "stripe",
141 &event_type_str,
142 payload,
143 Some(signature),
144 &format!("{e:?}"),
145 )
146 .await
147 {
148 tracing::error!(error = ?queue_err, "failed to queue webhook event for retry; returning 503 to trigger Stripe redelivery");
149 return Ok(StatusCode::SERVICE_UNAVAILABLE);
150 }
151 }
152 }
153
154 Ok(StatusCode::OK)
155 }
156
157 /// Process a verified Stripe webhook event. Extracted to allow the caller
158 /// to catch errors and persist to the retry queue. Also called by the
159 /// scheduler's webhook retry worker.
160 /// Dispatch a verified Stripe webhook event. Consumes `data_object` exactly
161 /// once into a typed rc.5 struct based on `event_type`. Shared by the live
162 /// webhook handler and the scheduler retry worker.
163 #[allow(clippy::too_many_arguments)]
164 pub(crate) async fn process_webhook_event(
165 db: &PgPool,
166 bg: &crate::background::BackgroundTx,
167 email: &EmailClient,
168 wam: Option<&WamClient>,
169 payments: &Billing,
170 config: &Config,
171 event_type: &str,
172 event_id: &str,
173 data_object: serde_json::Value,
174 ) -> Result<()> {
175 match event_type {
176 // Both events route through the same dispatcher. `completed` fires
177 // immediately; for asynchronous payment methods (ACH/SEPA/Bacs) it
178 // arrives with payment_status="unpaid" and the money-taking handlers
179 // defer until `async_payment_succeeded` re-delivers the settled session.
180 "checkout.session.completed" | "checkout.session.async_payment_succeeded" => {
181 let session: CheckoutSessionView =
182 serde_json::from_value(data_object).map_err(|e| {
183 AppError::BadRequest(format!("Failed to parse CheckoutSession: {e}"))
184 })?;
185 dispatch_checkout_session(db, bg, email, wam, payments, config, &session, event_id)
186 .await?;
187 }
188 // The buyer's async payment (ACH/SEPA/Bacs) never cleared. No funds were
189 // captured, so there is nothing to deliver; the pending transaction (and
190 // any reserved promo hold) is released by the stale-pending cleanup
191 // sweeper. Logged for visibility rather than silently dropped.
192 "checkout.session.async_payment_failed" => {
193 let session: CheckoutSessionView =
194 serde_json::from_value(data_object).map_err(|e| {
195 AppError::BadRequest(format!("Failed to parse CheckoutSession: {e}"))
196 })?;
197 tracing::warn!(
198 session_id = %session.id,
199 "checkout async payment failed; no funds captured, pending rows will be released by cleanup"
200 );
201 }
202 "account.updated" => {
203 let account: AccountView = serde_json::from_value(data_object)
204 .map_err(|e| AppError::BadRequest(format!("Failed to parse Account: {e}")))?;
205 handle_account_updated(db, wam, &AccountUpdate::from(account)).await?;
206 }
207 "charge.refunded" => {
208 let charge: ChargeView = serde_json::from_value(data_object)
209 .map_err(|e| AppError::BadRequest(format!("Failed to parse Charge: {e}")))?;
210 if let Some(refund_data) = ChargeRefundData::from_view(charge) {
211 // Direct webhook: queue as pending if the matching payment hasn't
212 // landed yet. Out-of-band (dashboard) FULL refunds are handled
213 // here; per-line refunds land via refund.created below.
214 billing::handle_charge_refunded(db, &refund_data, true).await?;
215 }
216 }
217 "refund.created" | "refund.updated" => {
218 // Line-scoped self-service refunds tag the Stripe refund with
219 // mnw_transaction_id; this marks/revokes exactly that transaction so
220 // a cart line refund leaves the order's other lines untouched
221 // (Run #2 Payments SERIOUS). Untagged refunds are no-ops here.
222 let refund: RefundView = serde_json::from_value(data_object)
223 .map_err(|e| AppError::BadRequest(format!("Failed to parse Refund: {e}")))?;
224 billing::handle_refund_created(db, &refund).await?;
225 }
226 "customer.subscription.updated" => {
227 let sub: SubscriptionView = serde_json::from_value(data_object)
228 .map_err(|e| AppError::BadRequest(format!("Failed to parse Subscription: {e}")))?;
229 subscriptions::handle_subscription_updated(db, &sub, event_id).await?;
230 }
231 "customer.subscription.deleted" => {
232 let sub: SubscriptionView = serde_json::from_value(data_object)
233 .map_err(|e| AppError::BadRequest(format!("Failed to parse Subscription: {e}")))?;
234 subscriptions::handle_subscription_deleted(db, bg, email, &sub, event_id).await?;
235 }
236 "invoice.payment_succeeded" => {
237 let invoice: InvoiceView = serde_json::from_value(data_object)
238 .map_err(|e| AppError::BadRequest(format!("Failed to parse Invoice: {e}")))?;
239 billing::handle_invoice_payment_succeeded(db, bg, email, wam, &invoice, event_id)
240 .await?;
241 }
242 "invoice.payment_failed" => {
243 let invoice: InvoiceView = serde_json::from_value(data_object)
244 .map_err(|e| AppError::BadRequest(format!("Failed to parse Invoice: {e}")))?;
245 billing::handle_invoice_payment_failed(db, wam, &invoice, event_id).await?;
246 }
247 other => {
248 tracing::debug!(event_type = %other, "unhandled webhook event type");
249 }
250 }
251
252 Ok(())
253 }
254
255 /// Route a checkout session to its handler by metadata shape.
256 ///
257 /// Subscription-mode sessions (Fan+, creator tier, SyncKit app sub, project
258 /// subscription) capture no funds at checkout, the subscription lifecycle bills
259 /// separately, so they run unconditionally. One-time payment-mode sessions
260 /// (tip, guest, cart, single purchase) capture funds now and are therefore
261 /// gated on `payment_settled()`: an async method that reports `payment_status
262 /// = "unpaid"` on `checkout.session.completed` is deferred until Stripe
263 /// re-delivers the settled session via `checkout.session.async_payment_succeeded`.
264 /// Without this gate, enabling any async payment method on a connected account
265 /// would mint license keys and grant downloads before funds settle.
266 #[allow(clippy::too_many_arguments)]
267 async fn dispatch_checkout_session(
268 db: &PgPool,
269 bg: &crate::background::BackgroundTx,
270 email: &EmailClient,
271 wam: Option<&WamClient>,
272 payments: &Billing,
273 config: &Config,
274 session: &CheckoutSessionView,
275 event_id: &str,
276 ) -> Result<()> {
277 let meta = session.metadata.as_ref();
278
279 // Subscription-mode: no funds captured at checkout, no settlement gate.
280 if payments::is_fan_plus_checkout(meta) {
281 return checkout::handle_fan_plus_checkout_completed(db, bg, email, session, event_id)
282 .await;
283 }
284 if payments::is_creator_tier_checkout(meta) {
285 return checkout::handle_creator_tier_checkout_completed(
286 db, bg, wam, payments, session, event_id,
287 )
288 .await;
289 }
290 if payments::is_synckit_app_sub_checkout(meta) {
291 return checkout::handle_synckit_app_sub_checkout_completed(db, session, event_id).await;
292 }
293 if payments::is_subscription_checkout(meta) {
294 return checkout::handle_subscription_checkout_completed(db, bg, email, session, event_id)
295 .await;
296 }
297
298 // One-time payment-mode: funds captured now. Deliver only once settled.
299 if !session.payment_settled() {
300 tracing::info!(
301 session_id = %session.id,
302 payment_status = ?session.payment_status,
303 "one-time checkout not yet settled (async payment); deferring finalize until async_payment_succeeded"
304 );
305 return Ok(());
306 }
307
308 if payments::is_tip_checkout(meta) {
309 checkout::handle_tip_checkout_completed(db, bg, email, wam, config, session, event_id).await
310 } else if payments::is_guest_checkout(meta) {
311 checkout::handle_guest_checkout_completed(db, bg, email, wam, config, session, event_id)
312 .await
313 } else if payments::is_cart_checkout(meta) {
314 checkout::handle_cart_checkout_completed(db, bg, email, wam, config, session, event_id)
315 .await
316 } else {
317 checkout::handle_purchase_checkout_completed(db, bg, email, wam, config, session, event_id)
318 .await
319 }
320 }
321
322 /// Handle account.updated from the v2 thin event endpoint.
323 pub(in crate::routes::stripe) async fn handle_account_updated_from_v2(
324 db: &PgPool,
325 wam: Option<&WamClient>,
326 update: &AccountUpdate,
327 ) -> Result<()> {
328 handle_account_updated(db, wam, update).await
329 }
330
331 /// Handle account.updated webhook
332 async fn handle_account_updated(
333 db: &PgPool,
334 wam: Option<&WamClient>,
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 "account updated"
341 );
342
343 // Update the user's Stripe status
344 db::users::update_user_stripe_status(
345 db,
346 &update.account_id,
347 update.details_submitted,
348 update.payouts_enabled,
349 update.charges_enabled,
350 )
351 .await
352 .with_context(|| format!("update Stripe status for account {}", update.account_id))?;
353
354 // Alert if charges or payouts became disabled (creator can't receive payments)
355 if (!update.charges_enabled || !update.payouts_enabled)
356 && let Some(wam) = wam
357 {
358 let title = format!("Stripe Connect degraded: {}", update.account_id);
359 let body = format!(
360 "charges_enabled: {}\npayouts_enabled: {}\ndetails_submitted: {}",
361 update.charges_enabled, update.payouts_enabled, update.details_submitted,
362 );
363 wam.create_ticket(
364 &title,
365 Some(&body),
366 "high",
367 "stripe-connect-degraded",
368 Some(&update.account_id),
369 )
370 .await;
371 }
372
373 Ok(())
374 }
375