Skip to main content

max / makenotwork

Decompose AppState Phase 2: thread slices through stripe dispatcher Complete the full webhook threading. The dispatch boundary no longer holds &AppState: - process_webhook_event and dispatch_checkout_session take the union (db, bg, email, wam, payments(Billing), config) - process_v2_thin_event / handle_account_thin_event take (db, wam, stripe) - handle_account_updated(_from_v2) take (db, wam) - both HTTP webhook handlers extract narrow State<T> slices; sig verify via payments.stripe scheduler/webhooks.rs keeps its own &AppState (not yet migrated) and projects slice refs at the two dispatcher call sites via Billing::from_ref. No AppState reference remains in webhook/mod.rs or webhook_v2.rs. No behavior change; webhook 73 + mock_payment 33 + db_webhook_events 12 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 23:15 UTC
Commit: 6297d7a0c83d0bcb1b295c89197d1ab716f871b4
Parent: 862075b
3 files changed, +87 insertions, -46 deletions
M server/src/routes/stripe/webhook/mod.rs +56 -32
@@ -7,29 +7,39 @@ mod subscriptions;
7 7
8 8 use axum::{
9 9 body::Bytes,
10 - extract::{FromRef, State},
10 + extract::State,
11 11 http::{header::HeaderMap, StatusCode},
12 12 response::IntoResponse,
13 13 };
14 + use sqlx::PgPool;
14 15
15 16 use crate::{
17 + config::Config,
16 18 db,
19 + email::EmailClient,
17 20 error::{AppError, Result, ResultExt},
18 21 payments::{
19 22 self, AccountUpdate, AccountView, ChargeRefundData, ChargeView,
20 23 CheckoutSessionView, InvoiceView, RefundView, SubscriptionView, UntypedEvent,
21 24 },
22 - AppState,
25 + wam_client::WamClient,
26 + Billing, Integrations,
23 27 };
24 28
25 29 /// POST /stripe/webhook - Handle Stripe webhook events
26 30 #[tracing::instrument(skip_all, name = "stripe::webhook")]
31 + #[allow(clippy::too_many_arguments)]
27 32 pub(in crate::routes::stripe) async fn webhook(
28 - State(state): State<AppState>,
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>,
29 39 headers: HeaderMap,
30 40 body: Bytes,
31 41 ) -> Result<impl IntoResponse> {
32 - let stripe = state.stripe.as_ref()
42 + let stripe = payments.stripe.as_ref()
33 43 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
34 44
35 45 // Get the signature header
@@ -55,7 +65,7 @@ pub(in crate::routes::stripe) async fn webhook(
55 65 // read then short-circuits. Dropping `_event_lock` on any return path rolls
56 66 // the (write-free) lock transaction back and releases it — it cannot leak.
57 67 // See `db::webhook_events::try_lock_event`.
58 - let _event_lock = match db::webhook_events::try_lock_event(&state.db, &event.id).await {
68 + let _event_lock = match db::webhook_events::try_lock_event(&db, &event.id).await {
59 69 Ok(Some(tx)) => tx,
60 70 Ok(None) => {
61 71 tracing::info!(event_id = %event.id, "concurrent delivery of this webhook event is in flight; returning 503 for redelivery");
@@ -74,7 +84,7 @@ pub(in crate::routes::stripe) async fn webhook(
74 84 // guarantees exactly-once dispatch regardless of handler idempotency; per-
75 85 // handler atomic guards (status-guarded UPDATEs, ON CONFLICT writes, the
76 86 // `FanPlusCreditClaim` witness) remain as defense in depth.
77 - match db::webhook_events::is_event_processed(&state.db, &event.id).await {
87 + match db::webhook_events::is_event_processed(&db, &event.id).await {
78 88 Ok(true) => {
79 89 tracing::info!(event_id = %event.id, "duplicate webhook event, skipping");
80 90 return Ok(StatusCode::OK);
@@ -89,7 +99,7 @@ pub(in crate::routes::stripe) async fn webhook(
89 99 // For retry-queue persistence we need id+type after `event` is consumed.
90 100 // Move both out without cloning the underlying allocations.
91 101 let UntypedEvent { id: event_id, type_: event_type_str, data_object } = event;
92 - let result = process_webhook_event(&state, &event_type_str, &event_id, data_object).await;
102 + let result = process_webhook_event(&db, &bg, &email, integrations.wam.as_ref(), &payments, &config, &event_type_str, &event_id, data_object).await;
93 103
94 104 match result {
95 105 Ok(()) => {
@@ -97,7 +107,7 @@ pub(in crate::routes::stripe) async fn webhook(
97 107 // redelivery short-circuits. If this write fails, return 503: Stripe
98 108 // redelivers, the idempotent handler re-runs, and the mark is retried.
99 109 // No event is lost; at worst it is processed twice (safe).
100 - if let Err(e) = db::webhook_events::mark_event_processed(&state.db, &event_id).await {
110 + if let Err(e) = db::webhook_events::mark_event_processed(&db, &event_id).await {
101 111 tracing::error!(event_id = %event_id, error = ?e, "failed to record processed webhook event; returning 503 for redelivery");
102 112 return Ok(StatusCode::SERVICE_UNAVAILABLE);
103 113 }
@@ -109,7 +119,7 @@ pub(in crate::routes::stripe) async fn webhook(
109 119 );
110 120 // Not marked processed, so redelivery (or the retry worker) reruns it.
111 121 if let Err(queue_err) = db::webhook_events::insert_failed_event(
112 - &state.db,
122 + &db,
113 123 "stripe",
114 124 &event_type_str,
115 125 payload,
@@ -131,8 +141,14 @@ pub(in crate::routes::stripe) async fn webhook(
131 141 /// Dispatch a verified Stripe webhook event. Consumes `data_object` exactly
132 142 /// once into a typed rc.5 struct based on `event_type`. Shared by the live
133 143 /// webhook handler and the scheduler retry worker.
144 + #[allow(clippy::too_many_arguments)]
134 145 pub(crate) async fn process_webhook_event(
135 - state: &AppState,
146 + db: &PgPool,
147 + bg: &crate::background::BackgroundTx,
148 + email: &EmailClient,
149 + wam: Option<&WamClient>,
150 + payments: &Billing,
151 + config: &Config,
136 152 event_type: &str,
137 153 event_id: &str,
138 154 data_object: serde_json::Value,
@@ -145,7 +161,7 @@ pub(crate) async fn process_webhook_event(
145 161 "checkout.session.completed" | "checkout.session.async_payment_succeeded" => {
146 162 let session: CheckoutSessionView = serde_json::from_value(data_object)
147 163 .map_err(|e| AppError::BadRequest(format!("Failed to parse CheckoutSession: {e}")))?;
148 - dispatch_checkout_session(state, &session, event_id).await?;
164 + dispatch_checkout_session(db, bg, email, wam, payments, config, &session, event_id).await?;
149 165 }
150 166 // The buyer's async payment (ACH/SEPA/Bacs) never cleared. No funds were
151 167 // captured, so there is nothing to deliver; the pending transaction (and
@@ -162,7 +178,7 @@ pub(crate) async fn process_webhook_event(
162 178 "account.updated" => {
163 179 let account: AccountView = serde_json::from_value(data_object)
164 180 .map_err(|e| AppError::BadRequest(format!("Failed to parse Account: {e}")))?;
165 - handle_account_updated(state, &AccountUpdate::from(account)).await?;
181 + handle_account_updated(db, wam, &AccountUpdate::from(account)).await?;
166 182 }
167 183 "charge.refunded" => {
168 184 let charge: ChargeView = serde_json::from_value(data_object)
@@ -171,7 +187,7 @@ pub(crate) async fn process_webhook_event(
171 187 // Direct webhook: queue as pending if the matching payment hasn't
172 188 // landed yet. Out-of-band (dashboard) FULL refunds are handled
173 189 // here; per-line refunds land via refund.created below.
174 - billing::handle_charge_refunded(&state.db, &refund_data, true).await?;
190 + billing::handle_charge_refunded(db, &refund_data, true).await?;
175 191 }
176 192 }
177 193 "refund.created" | "refund.updated" => {
@@ -181,27 +197,27 @@ pub(crate) async fn process_webhook_event(
181 197 // (Run #2 Payments SERIOUS). Untagged refunds are no-ops here.
182 198 let refund: RefundView = serde_json::from_value(data_object)
183 199 .map_err(|e| AppError::BadRequest(format!("Failed to parse Refund: {e}")))?;
184 - billing::handle_refund_created(&state.db, &refund).await?;
200 + billing::handle_refund_created(db, &refund).await?;
185 201 }
186 202 "customer.subscription.updated" => {
187 203 let sub: SubscriptionView = serde_json::from_value(data_object)
188 204 .map_err(|e| AppError::BadRequest(format!("Failed to parse Subscription: {e}")))?;
189 - subscriptions::handle_subscription_updated(&state.db, &sub, event_id).await?;
205 + subscriptions::handle_subscription_updated(db, &sub, event_id).await?;
190 206 }
191 207 "customer.subscription.deleted" => {
192 208 let sub: SubscriptionView = serde_json::from_value(data_object)
193 209 .map_err(|e| AppError::BadRequest(format!("Failed to parse Subscription: {e}")))?;
194 - subscriptions::handle_subscription_deleted(&state.db, &state.bg, &state.email, &sub, event_id).await?;
210 + subscriptions::handle_subscription_deleted(db, bg, email, &sub, event_id).await?;
195 211 }
196 212 "invoice.payment_succeeded" => {
197 213 let invoice: InvoiceView = serde_json::from_value(data_object)
198 214 .map_err(|e| AppError::BadRequest(format!("Failed to parse Invoice: {e}")))?;
199 - billing::handle_invoice_payment_succeeded(&state.db, &state.bg, &state.email, state.wam.as_ref(), &invoice, event_id).await?;
215 + billing::handle_invoice_payment_succeeded(db, bg, email, wam, &invoice, event_id).await?;
200 216 }
201 217 "invoice.payment_failed" => {
202 218 let invoice: InvoiceView = serde_json::from_value(data_object)
203 219 .map_err(|e| AppError::BadRequest(format!("Failed to parse Invoice: {e}")))?;
204 - billing::handle_invoice_payment_failed(&state.db, state.wam.as_ref(), &invoice, event_id).await?;
220 + billing::handle_invoice_payment_failed(db, wam, &invoice, event_id).await?;
205 221 }
206 222 other => {
207 223 tracing::debug!(event_type = %other, "unhandled webhook event type");
@@ -222,8 +238,14 @@ pub(crate) async fn process_webhook_event(
222 238 /// re-delivers the settled session via `checkout.session.async_payment_succeeded`.
223 239 /// Without this gate, enabling any async payment method on a connected account
224 240 /// would mint license keys and grant downloads before funds settle.
241 + #[allow(clippy::too_many_arguments)]
225 242 async fn dispatch_checkout_session(
226 - state: &AppState,
243 + db: &PgPool,
244 + bg: &crate::background::BackgroundTx,
245 + email: &EmailClient,
246 + wam: Option<&WamClient>,
247 + payments: &Billing,
248 + config: &Config,
227 249 session: &CheckoutSessionView,
228 250 event_id: &str,
229 251 ) -> Result<()> {
@@ -231,16 +253,16 @@ async fn dispatch_checkout_session(
231 253
232 254 // Subscription-mode: no funds captured at checkout, no settlement gate.
233 255 if payments::is_fan_plus_checkout(meta) {
234 - return checkout::handle_fan_plus_checkout_completed(&state.db, &state.bg, &state.email, session, event_id).await;
256 + return checkout::handle_fan_plus_checkout_completed(db, bg, email, session, event_id).await;
235 257 }
236 258 if payments::is_creator_tier_checkout(meta) {
237 - return checkout::handle_creator_tier_checkout_completed(&state.db, &state.bg, state.wam.as_ref(), &crate::Billing::from_ref(state), session, event_id).await;
259 + return checkout::handle_creator_tier_checkout_completed(db, bg, wam, payments, session, event_id).await;
238 260 }
239 261 if payments::is_synckit_app_sub_checkout(meta) {
240 - return checkout::handle_synckit_app_sub_checkout_completed(&state.db, session, event_id).await;
262 + return checkout::handle_synckit_app_sub_checkout_completed(db, session, event_id).await;
241 263 }
242 264 if payments::is_subscription_checkout(meta) {
243 - return checkout::handle_subscription_checkout_completed(&state.db, &state.bg, &state.email, session, event_id).await;
265 + return checkout::handle_subscription_checkout_completed(db, bg, email, session, event_id).await;
244 266 }
245 267
246 268 // One-time payment-mode: funds captured now. Deliver only once settled.
@@ -254,27 +276,29 @@ async fn dispatch_checkout_session(
254 276 }
255 277
256 278 if payments::is_tip_checkout(meta) {
257 - checkout::handle_tip_checkout_completed(&state.db, &state.bg, &state.email, state.wam.as_ref(), &state.config, session, event_id).await
279 + checkout::handle_tip_checkout_completed(db, bg, email, wam, config, session, event_id).await
258 280 } else if payments::is_guest_checkout(meta) {
259 - checkout::handle_guest_checkout_completed(&state.db, &state.bg, &state.email, state.wam.as_ref(), &state.config, session, event_id).await
281 + checkout::handle_guest_checkout_completed(db, bg, email, wam, config, session, event_id).await
260 282 } else if payments::is_cart_checkout(meta) {
261 - checkout::handle_cart_checkout_completed(&state.db, &state.bg, &state.email, state.wam.as_ref(), &state.config, session, event_id).await
283 + checkout::handle_cart_checkout_completed(db, bg, email, wam, config, session, event_id).await
262 284 } else {
263 - checkout::handle_purchase_checkout_completed(&state.db, &state.bg, &state.email, state.wam.as_ref(), &state.config, session, event_id).await
285 + checkout::handle_purchase_checkout_completed(db, bg, email, wam, config, session, event_id).await
264 286 }
265 287 }
266 288
267 289 /// Handle account.updated from the v2 thin event endpoint.
268 290 pub(in crate::routes::stripe) async fn handle_account_updated_from_v2(
269 - state: &AppState,
291 + db: &PgPool,
292 + wam: Option<&WamClient>,
270 293 update: &AccountUpdate,
271 294 ) -> Result<()> {
272 - handle_account_updated(state, update).await
295 + handle_account_updated(db, wam, update).await
273 296 }
274 297
275 298 /// Handle account.updated webhook
276 299 async fn handle_account_updated(
277 - state: &AppState,
300 + db: &PgPool,
301 + wam: Option<&WamClient>,
278 302 update: &AccountUpdate,
279 303 ) -> Result<()> {
280 304 tracing::info!(
@@ -285,7 +309,7 @@ async fn handle_account_updated(
285 309
286 310 // Update the user's Stripe status
287 311 db::users::update_user_stripe_status(
288 - &state.db,
312 + db,
289 313 &update.account_id,
290 314 update.details_submitted,
291 315 update.payouts_enabled,
@@ -295,7 +319,7 @@ async fn handle_account_updated(
295 319
296 320 // Alert if charges or payouts became disabled (creator can't receive payments)
297 321 if (!update.charges_enabled || !update.payouts_enabled)
298 - && let Some(ref wam) = state.wam
322 + && let Some(wam) = wam
299 323 {
300 324 let title = format!("Stripe Connect degraded: {}", update.account_id);
301 325 let body = format!(
M server/src/routes/stripe/webhook_v2.rs +18 -12
@@ -11,22 +11,26 @@ use axum::{
11 11 http::{header::HeaderMap, StatusCode},
12 12 response::IntoResponse,
13 13 };
14 + use sqlx::PgPool;
14 15
15 16 use crate::{
16 17 db,
17 18 error::{AppError, Result},
18 19 payments::{self, ThinEvent},
19 - AppState,
20 + wam_client::WamClient,
21 + Billing, Integrations,
20 22 };
21 23
22 24 /// POST /stripe/webhook/v2: Handle Stripe v2 thin events
23 25 #[tracing::instrument(skip_all, name = "stripe::webhook_v2")]
24 26 pub(super) async fn webhook_v2(
25 - State(state): State<AppState>,
27 + State(db): State<PgPool>,
28 + State(integrations): State<Integrations>,
29 + State(payments): State<Billing>,
26 30 headers: HeaderMap,
27 31 body: Bytes,
28 32 ) -> Result<impl IntoResponse> {
29 - let stripe = state.stripe.as_ref()
33 + let stripe = payments.stripe.as_ref()
30 34 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
31 35
32 36 let signature = headers
@@ -54,7 +58,7 @@ pub(super) async fn webhook_v2(
54 58 // rather than block: a same-event delivery arriving mid-flight gets `None` and
55 59 // returns 503 immediately instead of parking a pooled connection (Run 23
56 60 // Conc/Perf). Dropping `_event_lock` on any return releases the lock.
57 - let _event_lock = match db::webhook_events::try_lock_event(&state.db, &thin.id).await {
61 + let _event_lock = match db::webhook_events::try_lock_event(&db, &thin.id).await {
58 62 Ok(Some(tx)) => tx,
59 63 Ok(None) => {
60 64 tracing::info!(event_id = %thin.id, "concurrent delivery of this v2 event is in flight; returning 503 for redelivery");
@@ -70,7 +74,7 @@ pub(super) async fn webhook_v2(
70 74 // "processed" row is written only after the handler succeeds (below), so a
71 75 // crash mid-processing leaves no marker and Stripe redelivers (the handler
72 76 // is idempotent — it re-fetches and re-applies account state).
73 - match db::webhook_events::is_event_processed(&state.db, &thin.id).await {
77 + match db::webhook_events::is_event_processed(&db, &thin.id).await {
74 78 Ok(true) => {
75 79 tracing::debug!(event_id = %thin.id, "v2 event already processed, skipping");
76 80 return Ok(StatusCode::OK);
@@ -83,7 +87,7 @@ pub(super) async fn webhook_v2(
83 87 Ok(false) => {} // first time — proceed
84 88 }
85 89
86 - if let Err(e) = process_v2_thin_event(&state, stripe.as_ref(), &thin).await {
90 + if let Err(e) = process_v2_thin_event(&db, integrations.wam.as_ref(), stripe.as_ref(), &thin).await {
87 91 // Persist to the local retry queue (backoff + dead-letter WAM via the
88 92 // scheduler), matching v1's failure path rather than relying solely on
89 93 // Stripe's redelivery window. Not marked processed, so a Stripe
@@ -100,7 +104,7 @@ pub(super) async fn webhook_v2(
100 104 // is never "silently dropped".
101 105 tracing::warn!(event_id = %thin.id, error = ?e, "v2 event processing failed; queueing for retry");
102 106 if let Err(queue_err) = db::webhook_events::insert_failed_event(
103 - &state.db,
107 + &db,
104 108 "stripe_v2",
105 109 &thin.event_type,
106 110 payload,
@@ -116,7 +120,7 @@ pub(super) async fn webhook_v2(
116 120 }
117 121
118 122 // Succeeded — record it so a redelivery short-circuits.
119 - if let Err(e) = db::webhook_events::mark_event_processed(&state.db, &thin.id).await {
123 + if let Err(e) = db::webhook_events::mark_event_processed(&db, &thin.id).await {
120 124 tracing::error!(event_id = %thin.id, error = ?e, "failed to record processed v2 event; returning 503 for redelivery");
121 125 return Ok(StatusCode::SERVICE_UNAVAILABLE);
122 126 }
@@ -128,12 +132,13 @@ pub(super) async fn webhook_v2(
128 132 /// the scheduler retry worker (which re-parses the stored payload — the
129 133 /// signature was already verified when the event was first received).
130 134 pub(crate) async fn process_v2_thin_event(
131 - state: &AppState,
135 + db: &PgPool,
136 + wam: Option<&WamClient>,
132 137 stripe: &dyn payments::PaymentProvider,
133 138 thin: &ThinEvent,
134 139 ) -> Result<()> {
135 140 if thin.event_type.starts_with("v2.core.account") {
136 - handle_account_thin_event(state, stripe, thin).await
141 + handle_account_thin_event(db, wam, stripe, thin).await
137 142 } else {
138 143 tracing::debug!(event_type = %thin.event_type, "unhandled v2 event type");
139 144 Ok(())
@@ -142,7 +147,8 @@ pub(crate) async fn process_v2_thin_event(
142 147
143 148 /// Fetch the full account object and delegate to the shared account-updated handler.
144 149 async fn handle_account_thin_event(
145 - state: &AppState,
150 + db: &PgPool,
151 + wam: Option<&WamClient>,
146 152 stripe: &dyn payments::PaymentProvider,
147 153 thin: &ThinEvent,
148 154 ) -> Result<()> {
@@ -159,7 +165,7 @@ async fn handle_account_thin_event(
159 165 e
160 166 })?;
161 167
162 - super::webhook::handle_account_updated_from_v2(state, &update).await.map_err(|e| {
168 + super::webhook::handle_account_updated_from_v2(db, wam, &update).await.map_err(|e| {
163 169 tracing::warn!(account_id = %account_id, error = ?e, "failed to process account update from v2 event");
164 170 e
165 171 })?;
M server/src/scheduler/webhooks.rs +13 -2
@@ -2,6 +2,7 @@
2 2
3 3 use crate::db;
4 4 use crate::AppState;
5 + use axum::extract::FromRef;
5 6
6 7 /// Maximum webhook retry attempts before marking as dead letter.
7 8 const WEBHOOK_MAX_RETRIES: i32 = 5;
@@ -86,7 +87,17 @@ pub(super) async fn retry_failed_webhooks(state: &AppState) {
86 87 match crate::payments::UntypedEvent::from_payload(&event.payload) {
87 88 Ok(parsed) => {
88 89 let crate::payments::UntypedEvent { id, type_, data_object } = parsed;
89 - crate::routes::stripe::process_webhook_event(state, &type_, &id, data_object).await
90 + crate::routes::stripe::process_webhook_event(
91 + &state.db,
92 + &state.bg,
93 + &state.email,
94 + state.wam.as_ref(),
95 + &crate::Billing::from_ref(state),
96 + &state.config,
97 + &type_,
98 + &id,
99 + data_object,
100 + ).await
90 101 }
91 102 Err(e) => Err(e),
92 103 }
@@ -97,7 +108,7 @@ pub(super) async fn retry_failed_webhooks(state: &AppState) {
97 108 match serde_json::from_str::<crate::payments::ThinEvent>(&event.payload) {
98 109 Ok(thin) => match state.stripe.as_ref() {
99 110 Some(stripe) => {
100 - crate::routes::stripe::process_v2_thin_event(state, stripe.as_ref(), &thin).await
111 + crate::routes::stripe::process_v2_thin_event(&state.db, state.wam.as_ref(), stripe.as_ref(), &thin).await
101 112 }
102 113 None => Err(crate::error::AppError::BadRequest("Stripe not configured".to_string())),
103 114 },