Skip to main content

max / makenotwork

7.8 KB · 193 lines History Blame Raw
1 //! Stripe v2 thin event webhook handler.
2 //!
3 //! Stripe's v2 event system sends "thin" events that contain only a reference
4 //! to the affected object, not the full snapshot. The handler verifies the
5 //! signature, parses the event type, fetches the full object via the API, and
6 //! delegates to the same business logic used by the v1 handler.
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, db,
18 error::{AppError, Result},
19 payments::{self, ThinEvent},
20 wam_client::WamClient,
21 };
22
23 /// POST /stripe/webhook/v2: Handle Stripe v2 thin events
24 #[tracing::instrument(skip_all, name = "stripe::webhook_v2")]
25 pub(super) async fn webhook_v2(
26 State(db): State<PgPool>,
27 State(integrations): State<Integrations>,
28 State(payments): State<Billing>,
29 State(config): State<crate::config::Config>,
30 headers: HeaderMap,
31 body: Bytes,
32 ) -> Result<impl IntoResponse> {
33 let stripe = payments
34 .stripe
35 .as_ref()
36 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
37
38 let signature = headers
39 .get("stripe-signature")
40 .and_then(|v| v.to_str().ok())
41 .ok_or_else(|| AppError::BadRequest("Missing Stripe signature".to_string()))?;
42
43 let payload = std::str::from_utf8(&body)
44 .map_err(|_| AppError::BadRequest("Invalid payload encoding".to_string()))?;
45
46 // Verify signature and parse JSON. A failure here has no benign cause:
47 // Stripe signs correctly, so it means a wrong signing secret (real events
48 // being dropped) or forged events aimed at the billing path.
49 let body_json = stripe
50 .verify_webhook_v2(payload, signature)
51 .inspect_err(|_| {
52 crate::security_signals::note_webhook_signature_failure("stripe");
53 })?;
54
55 // Parse the thin event
56 let thin: ThinEvent = serde_json::from_value(body_json).map_err(|e| {
57 tracing::warn!(error = ?e, "failed to parse v2 thin event");
58 AppError::BadRequest("Invalid v2 event format".to_string())
59 })?;
60
61 tracing::info!(event_type = %thin.event_type, event_id = %thin.id, "received v2 thin event");
62
63 // Serialize concurrent redeliveries of this event id (see
64 // `db::webhook_events::try_lock_event`). Held across dedup-read -> process ->
65 // mark, it makes the check-then-act read below race-free. We *try* the lock
66 // rather than block: a same-event delivery arriving mid-flight gets `None` and
67 // returns 503 immediately instead of parking a pooled connection (Run 23
68 // Conc/Perf). Dropping `_event_lock` on any return releases the lock.
69 let _event_lock = match db::webhook_events::try_lock_event(&db, &thin.id).await {
70 Ok(Some(tx)) => tx,
71 Ok(None) => {
72 tracing::info!(event_id = %thin.id, "concurrent delivery of this v2 event is in flight; returning 503 for redelivery");
73 return Ok(StatusCode::SERVICE_UNAVAILABLE);
74 }
75 Err(e) => {
76 tracing::error!(event_id = %thin.id, error = ?e, "failed to acquire v2 webhook event lock, returning 503 for retry");
77 return Ok(StatusCode::SERVICE_UNAVAILABLE);
78 }
79 };
80
81 // Deduplicate: skip if this event was already processed. Read-only; the
82 // "processed" row is written only after the handler succeeds (below), so a
83 // crash mid-processing leaves no marker and Stripe redelivers (the handler
84 // is idempotent, it re-fetches and re-applies account state).
85 match db::webhook_events::is_event_processed(&db, &thin.id).await {
86 Ok(true) => {
87 tracing::debug!(event_id = %thin.id, "v2 event already processed, skipping");
88 return Ok(StatusCode::OK);
89 }
90 Err(e) => {
91 // Return 503 so Stripe retries later (matching v1 webhook behavior)
92 tracing::error!(event_id = %thin.id, error = ?e, "v2 dedup check failed, returning 503 for retry");
93 return Ok(StatusCode::SERVICE_UNAVAILABLE);
94 }
95 Ok(false) => {} // first time, proceed
96 }
97
98 if let Err(e) = process_v2_thin_event(
99 &db,
100 integrations.wam.as_ref(),
101 stripe.as_ref(),
102 &config.signing_secret,
103 &thin,
104 )
105 .await
106 {
107 // Persist to the local retry queue (backoff + dead-letter WAM via the
108 // scheduler), matching v1's failure path rather than relying solely on
109 // Stripe's redelivery window. Not marked processed, so a Stripe
110 // redelivery also still re-runs the idempotent handler.
111 //
112 // Deliberate trade (fuzz 2026-07-06, Payments): returning 200 below ACKs
113 // the event to Stripe, so Stripe will not redeliver it on its own 3-day
114 // schedule, the in-house queue + scheduler owns retry from here (richer:
115 // local backoff + dead-lettering). The operational consequence is that
116 // for a persistently-failing money event, **scheduler liveness is the
117 // only retry backstop**; if the scheduler is down, a failed webhook has no
118 // external redelivery. The `insert_failed_event` guard below returns 503
119 // (inviting Stripe redelivery) if even the queue insert fails, so the net
120 // is never "silently dropped".
121 tracing::warn!(event_id = %thin.id, error = ?e, "v2 event processing failed; queueing for retry");
122 if let Err(queue_err) = db::webhook_events::insert_failed_event(
123 &db,
124 "stripe_v2",
125 &thin.event_type,
126 payload,
127 Some(signature),
128 &format!("{e:?}"),
129 )
130 .await
131 {
132 tracing::error!(event_id = %thin.id, error = ?queue_err, "failed to queue v2 event for retry; returning 503 for Stripe redelivery");
133 return Ok(StatusCode::SERVICE_UNAVAILABLE);
134 }
135 return Ok(StatusCode::OK);
136 }
137
138 // Succeeded, record it so a redelivery short-circuits.
139 if let Err(e) = db::webhook_events::mark_event_processed(&db, &thin.id).await {
140 tracing::error!(event_id = %thin.id, error = ?e, "failed to record processed v2 event; returning 503 for redelivery");
141 return Ok(StatusCode::SERVICE_UNAVAILABLE);
142 }
143
144 Ok(StatusCode::OK)
145 }
146
147 /// Route a verified v2 thin event to its handler. Shared by the live webhook and
148 /// the scheduler retry worker (which re-parses the stored payload, the
149 /// signature was already verified when the event was first received).
150 pub(crate) async fn process_v2_thin_event(
151 db: &PgPool,
152 wam: Option<&WamClient>,
153 stripe: &dyn payments::PaymentProvider,
154 signing_secret: &str,
155 thin: &ThinEvent,
156 ) -> Result<()> {
157 if thin.event_type.starts_with("v2.core.account") {
158 handle_account_thin_event(db, wam, stripe, signing_secret, thin).await
159 } else {
160 tracing::debug!(event_type = %thin.event_type, "unhandled v2 event type");
161 Ok(())
162 }
163 }
164
165 /// Fetch the full account object and delegate to the shared account-updated handler.
166 async fn handle_account_thin_event(
167 db: &PgPool,
168 wam: Option<&WamClient>,
169 stripe: &dyn payments::PaymentProvider,
170 signing_secret: &str,
171 thin: &ThinEvent,
172 ) -> Result<()> {
173 let account_id = match &thin.related_object {
174 Some(obj) => &obj.id,
175 None => {
176 tracing::warn!(event_id = %thin.id, "v2 account event missing related_object");
177 return Ok(()); // nothing to fetch, acknowledge
178 }
179 };
180
181 let update = stripe.fetch_account(account_id).await.map_err(|e| {
182 tracing::warn!(account_id = %account_id, error = ?e, "failed to fetch account for v2 event");
183 e
184 })?;
185
186 super::webhook::handle_account_updated_from_v2(db, wam, signing_secret, &update).await.map_err(|e| {
187 tracing::warn!(account_id = %account_id, error = ?e, "failed to process account update from v2 event");
188 e
189 })?;
190
191 Ok(())
192 }
193