Skip to main content

max / makenotwork

21.5 KB · 518 lines History Blame Raw
1 //! Webhook retry with exponential backoff and stale refund escalation.
2
3 use crate::AppState;
4 use crate::db;
5 use axum::extract::FromRef;
6
7 /// Maximum webhook retry attempts before marking as dead letter.
8 const WEBHOOK_MAX_RETRIES: i32 = 5;
9
10 /// Determine whether a webhook retry attempt should be treated as a dead letter.
11 pub(super) fn is_webhook_dead(attempt: i32) -> bool {
12 attempt >= WEBHOOK_MAX_RETRIES
13 }
14
15 /// Retry failed webhook events with exponential backoff.
16 #[tracing::instrument(skip_all, name = "scheduler::retry_failed_webhooks")]
17 pub(super) async fn retry_failed_webhooks(state: &AppState) {
18 let events = match db::webhook_events::get_retryable_events(&state.db).await {
19 Ok(e) if e.is_empty() => return,
20 Ok(e) => e,
21 Err(e) => {
22 tracing::error!(error = ?e, "failed to fetch retryable webhook events");
23 return;
24 }
25 };
26
27 // The provider is what normalizes a stored payload, so no provider means no
28 // retry to run. Bound here rather than re-checked per event.
29 let Some(provider) = state.payments.as_ref() else {
30 return;
31 };
32
33 for event in events {
34 let attempt = event.attempts + 1;
35 tracing::info!(
36 event_id = %event.id, source = %event.source, event_type = %event.event_type,
37 attempt = attempt, "retrying webhook event"
38 );
39
40 // The Stripe event id (from the stored payload) is the key the live
41 // webhook handler locks and dedups on. Parse it up front so this retry
42 // takes the SAME per-event advisory lock and consults the SAME processed
43 // marker, otherwise a queue retry and a live Stripe redelivery of one
44 // event can run concurrently, degrading exactly-once dispatch to
45 // at-least-once + per-handler idempotency (Run 21 payments).
46 let stripe_event_id: Option<String> = if event.source == "stripe" {
47 crate::payments::UntypedEvent::from_payload(&event.payload)
48 .ok()
49 .map(|p| p.id)
50 } else if event.source == "stripe_v2" {
51 serde_json::from_str::<crate::payments::ThinEvent>(&event.payload)
52 .ok()
53 .map(|t| t.id)
54 } else {
55 None
56 };
57
58 // Hold the per-event lock across dedup-read -> process -> mark, exactly
59 // like the live handler. We *try* the lock: if a live redelivery (or
60 // another tick) already holds it, leave this row queued and move on rather
61 // than blocking the retry loop on a pooled connection. Named binding (not
62 // bare `_`) so the guard lives, and holds the lock, for the whole
63 // iteration.
64 let _event_lock = match &stripe_event_id {
65 Some(eid) => match db::webhook_events::try_lock_event(&state.db, eid).await {
66 Ok(Some(tx)) => Some(tx),
67 Ok(None) => {
68 tracing::info!(event_id = %event.id, "webhook event locked by another worker; leaving queued for next tick");
69 continue;
70 }
71 Err(e) => {
72 tracing::error!(event_id = %event.id, error = ?e, "failed to lock webhook event for retry; will retry next tick");
73 continue;
74 }
75 },
76 None => None,
77 };
78
79 // If a live redelivery already processed this event while it sat in the
80 // queue, don't re-run the handler, just resolve the retry-queue row.
81 if let Some(eid) = &stripe_event_id
82 && matches!(
83 db::webhook_events::is_event_processed(&state.db, eid).await,
84 Ok(true)
85 )
86 {
87 tracing::info!(event_id = %event.id, stripe_event_id = %eid, "webhook already processed (live redelivery won the race); resolving retry row");
88 if let Err(e) = db::webhook_events::mark_processed(&state.db, event.id).await {
89 tracing::error!(error = ?e, "failed to mark webhook event as processed");
90 }
91 continue;
92 }
93
94 // Retry re-runs the full event handler. All handlers must be idempotent
95 // (use ON CONFLICT / WHERE status='pending' guards) since steps completed
96 // before the original failure are not rolled back.
97 let result = if event.source == "stripe" {
98 // Same normalization the live handler runs, through the same trait
99 // member: this path re-parses a stored payload outside
100 // `verify_webhook`, and before the shared `MnwEvent` step it had its
101 // own idea of how a payload became a dispatch.
102 match crate::payments::UntypedEvent::from_payload(&event.payload).and_then(|parsed| {
103 let id = parsed.id.clone();
104 provider.normalize_webhook(parsed).map(|e| (id, e))
105 }) {
106 Ok((id, mnw_event)) => {
107 crate::routes::stripe::process_webhook_event(
108 &state.db,
109 &state.bg,
110 &state.email,
111 state.wam.as_ref(),
112 &crate::Billing::from_ref(state),
113 &state.config,
114 mnw_event,
115 &id,
116 )
117 .await
118 }
119 Err(e) => Err(e),
120 }
121 } else if event.source == "stripe_v2" {
122 // v2 thin events: re-parse the stored payload (signature was verified
123 // at receive time) and re-route. The handler re-fetches the object
124 // from Stripe and re-applies it idempotently.
125 match serde_json::from_str::<crate::payments::ThinEvent>(&event.payload) {
126 Ok(thin) => match state.payment_caps.connect_onboarding.as_ref() {
127 Some(onboarding) => {
128 crate::routes::stripe::process_v2_thin_event(
129 &state.db,
130 state.wam.as_ref(),
131 onboarding.as_ref(),
132 &state.config.signing_secret,
133 &thin,
134 )
135 .await
136 }
137 None => Err(crate::error::AppError::BadRequest(
138 "Stripe not configured".to_string(),
139 )),
140 },
141 Err(e) => Err(crate::error::AppError::BadRequest(format!(
142 "failed to parse stored v2 event: {e}"
143 ))),
144 }
145 } else {
146 Err(crate::error::AppError::BadRequest(format!(
147 "Unknown webhook source: {}",
148 event.source
149 )))
150 };
151
152 match result {
153 Ok(()) => {
154 tracing::info!(event_id = %event.id, "webhook retry succeeded");
155 // Write the shared dedup marker so a later live redelivery of the
156 // same event short-circuits (mirrors the live handler's mark).
157 if let Some(eid) = &stripe_event_id
158 && let Err(e) = db::webhook_events::mark_event_processed(&state.db, eid).await
159 {
160 tracing::error!(event_id = %event.id, error = ?e, "webhook retry succeeded but recording processed-marker failed");
161 }
162 if let Err(e) = db::webhook_events::mark_processed(&state.db, event.id).await {
163 tracing::error!(error = ?e, "failed to mark webhook event as processed");
164 }
165 }
166 Err(e) => {
167 let is_dead = is_webhook_dead(attempt);
168 tracing::warn!(
169 event_id = %event.id, attempt = attempt, error = ?e,
170 dead = is_dead,
171 "webhook retry failed"
172 );
173 if let Err(e) = db::webhook_events::schedule_retry(
174 &state.db,
175 event.id,
176 attempt,
177 &format!("{e:?}"),
178 )
179 .await
180 {
181 tracing::error!(error = ?e, "failed to schedule webhook retry");
182 }
183
184 if is_dead && let Some(ref wam) = state.wam {
185 let title = format!("Dead webhook: {} ({})", event.event_type, event.id);
186 let body = format!(
187 "Webhook event exhausted all {} retry attempts.\n\
188 Source: {}\nType: {}\nLast error: {:?}",
189 attempt, event.source, event.event_type, e,
190 );
191 wam.create_ticket(
192 &title,
193 Some(&body),
194 "high",
195 "webhook-dead-letter",
196 Some(&event.id.to_string()),
197 )
198 .await;
199 }
200 }
201 }
202 }
203 }
204
205 /// Alert the admin about pending refunds that have gone unmatched for >24 hours.
206 #[tracing::instrument(skip_all, name = "scheduler::escalate_stale_refunds")]
207 pub(super) async fn escalate_stale_refunds(state: &AppState) {
208 let stale = match db::pending_refunds::get_stale_refunds(&state.db, chrono::Duration::hours(24))
209 .await
210 {
211 Ok(s) if s.is_empty() => return,
212 Ok(s) => s,
213 Err(e) => {
214 tracing::error!(error = ?e, "failed to query stale pending refunds");
215 return;
216 }
217 };
218
219 let alert_email = std::env::var("ALERT_EMAIL").ok();
220
221 for refund in &stale {
222 // Mark escalated FIRST to prevent duplicate alerts on retry
223 if let Err(e) = db::pending_refunds::mark_escalated(&state.db, refund.id).await {
224 tracing::error!(error = ?e, "failed to mark pending refund as escalated, skipping alerts");
225 continue;
226 }
227
228 tracing::error!(
229 payment_intent_id = %refund.payment_intent_id,
230 amount = refund.amount.as_i64(),
231 amount_refunded = refund.amount_refunded.as_i64(),
232 created_at = %refund.created_at,
233 "STALE PENDING REFUND: not completed within >24h (unmatched, or claimed but \
234 the refund never finished), needs manual investigation"
235 );
236
237 if let Some(ref to) = alert_email {
238 let subject = format!(
239 "Unmatched refund: {} ({}c refunded)",
240 refund.payment_intent_id, refund.amount_refunded
241 );
242 let body = format!(
243 "A charge.refunded webhook for payment intent {} has been pending for >24 hours \
244 without its refund completing.\n\n\
245 Amount: {}c\nAmount refunded: {}c\nReceived: {}\n\n\
246 Either the checkout.session.completed webhook was lost (never matched), or the \
247 refund was claimed but the process died before it finished. \
248 Check the Stripe dashboard for whether the refund was actually issued and \
249 reconcile manually.",
250 refund.payment_intent_id, refund.amount, refund.amount_refunded, refund.created_at,
251 );
252 if let Err(e) = state.email.send_alert(to, &subject, &body).await {
253 tracing::error!(error = ?e, "failed to send stale refund alert email");
254 }
255 }
256
257 if let Some(ref wam) = state.wam {
258 let title = format!(
259 "Unmatched refund: {} ({}c)",
260 refund.payment_intent_id, refund.amount_refunded
261 );
262 let body = format!(
263 "charge.refunded webhook pending >24h with no matching completed transaction.\n\
264 Amount: {}c\nRefunded: {}c\nReceived: {}\n\
265 Check Stripe dashboard and reconcile manually.",
266 refund.amount, refund.amount_refunded, refund.created_at,
267 );
268 wam.create_ticket(
269 &title,
270 Some(&body),
271 "critical",
272 "refund-escalation",
273 Some(&refund.payment_intent_id),
274 )
275 .await;
276 }
277 }
278 }
279
280 /// Per-tick cap on platform-credit settlement, so a backlog drains across ticks
281 /// instead of one unbounded loop stalling the scheduler tick.
282 const SETTLE_CREDITS_PER_TICK: usize = 50;
283
284 /// Settle owed platform-funded credits (Fan+ reimbursements) via platform ->
285 /// connected transfers, making the creator whole for a credit MNW funded.
286 ///
287 /// Each credit is claimed (so one worker settles it), the seller's Stripe account
288 /// resolved, the owed amount transferred with a deterministic idempotency key, then
289 /// marked settled. A seller not yet payable, or a transient transfer failure,
290 /// releases the claim for a later retry; a process death between claim and settle
291 /// leaves the row claimed-but-unsettled and is escalated by
292 /// [`escalate_stale_platform_credits`] rather than blindly retried.
293 #[tracing::instrument(skip_all, name = "scheduler::settle_platform_credits")]
294 pub(super) async fn settle_platform_credits(state: &AppState) {
295 let Some(transfers) = state.payment_caps.platform_transfers.as_ref() else {
296 return;
297 };
298
299 for _ in 0..SETTLE_CREDITS_PER_TICK {
300 let credit = match db::platform_credits::claim_unsettled_credit(&state.db).await {
301 Ok(Some(c)) => c,
302 Ok(None) => break,
303 Err(e) => {
304 tracing::error!(error = ?e, "failed to claim platform credit for settlement");
305 break;
306 }
307 };
308
309 let account_id = match db::users::get_user_by_id(&state.db, credit.seller_id).await {
310 Ok(Some(u)) => u.stripe_account_id,
311 Ok(None) => None,
312 Err(e) => {
313 tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "failed to load seller for platform credit; releasing");
314 db::platform_credits::unclaim_credit(&state.db, credit.transaction_id)
315 .await
316 .ok();
317 continue;
318 }
319 };
320
321 let Some(account) = account_id.as_deref() else {
322 // Seller not (yet) payable, release for a later retry. The stale sweep
323 // escalates any that never become payable.
324 db::platform_credits::unclaim_credit(&state.db, credit.transaction_id)
325 .await
326 .ok();
327 continue;
328 };
329
330 match transfers
331 .create_platform_credit_transfer(
332 account,
333 credit.amount_cents.as_i64(),
334 credit.transaction_id,
335 credit.currency,
336 )
337 .await
338 {
339 Ok(transfer_id) => {
340 if let Err(e) = db::platform_credits::mark_settled(
341 &state.db,
342 credit.transaction_id,
343 &transfer_id,
344 )
345 .await
346 {
347 // The transfer succeeded but the settle write failed. The
348 // deterministic idempotency key makes the next-tick retry safe
349 // (Stripe returns the same transfer, no double-pay).
350 tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "platform credit transferred but marking settled failed");
351 }
352 }
353 Err(e) => {
354 tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "platform credit transfer failed; releasing for retry");
355 db::platform_credits::unclaim_credit(&state.db, credit.transaction_id)
356 .await
357 .ok();
358 }
359 }
360 }
361 }
362
363 /// Reverse the MNW -> creator transfer for settled platform-funded credits whose
364 /// sale was later refunded, clawing the reimbursement back so the platform isn't
365 /// left funding a returned item.
366 ///
367 /// Only *settled* credits need this: an unsettled credit on a refunded
368 /// transaction is never paid out (the settle sweep gates on `status = 'completed'`,
369 /// and refunds flip the row to `refunded`). Each reversal uses a deterministic
370 /// idempotency key, so a redelivery or retry can't claw back twice. Runs after
371 /// settlement in the tick and is single-instance (scheduler advisory lock).
372 #[tracing::instrument(skip_all, name = "scheduler::reverse_refunded_platform_credits")]
373 pub(super) async fn reverse_refunded_platform_credits(state: &AppState) {
374 let Some(transfers) = state.payment_caps.platform_transfers.as_ref() else {
375 return;
376 };
377
378 let reversible = match db::platform_credits::get_reversible_credits(
379 &state.db,
380 SETTLE_CREDITS_PER_TICK as i64,
381 )
382 .await
383 {
384 Ok(r) => r,
385 Err(e) => {
386 tracing::error!(error = ?e, "failed to query reversible platform credits");
387 return;
388 }
389 };
390
391 for credit in reversible {
392 match transfers
393 .create_platform_credit_reversal(
394 &credit.transfer_id,
395 credit.amount_cents.as_i64(),
396 credit.transaction_id,
397 )
398 .await
399 {
400 Ok(()) => {
401 if let Err(e) =
402 db::platform_credits::mark_reversed(&state.db, credit.transaction_id).await
403 {
404 // Reversal succeeded but the mark write failed, the
405 // deterministic idempotency key makes the next-tick retry safe.
406 tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "platform credit reversed but marking reversed failed");
407 } else {
408 tracing::info!(transaction_id = %credit.transaction_id, amount = credit.amount_cents.as_i64(), "reversed platform credit on refund");
409 }
410 }
411 Err(e) => {
412 // Leave it for the next tick; nothing is marked, so it's retried.
413 tracing::error!(error = ?e, transaction_id = %credit.transaction_id, "platform credit reversal failed; will retry next tick");
414 }
415 }
416 }
417 }
418
419 /// Alert the admin about platform-funded credits claimed for settlement but never
420 /// completed (the crash window between claim and transfer). The transfer's
421 /// deterministic idempotency key means a human can safely re-trigger it.
422 #[tracing::instrument(skip_all, name = "scheduler::escalate_stale_platform_credits")]
423 pub(super) async fn escalate_stale_platform_credits(state: &AppState) {
424 let stale =
425 match db::platform_credits::get_stale_credits(&state.db, chrono::Duration::hours(24)).await
426 {
427 Ok(s) if s.is_empty() => return,
428 Ok(s) => s,
429 Err(e) => {
430 tracing::error!(error = ?e, "failed to query stale platform credits");
431 return;
432 }
433 };
434
435 let alert_email = std::env::var("ALERT_EMAIL").ok();
436
437 for credit in &stale {
438 if let Err(e) = db::platform_credits::mark_escalated(&state.db, credit.transaction_id).await
439 {
440 tracing::error!(error = ?e, "failed to mark platform credit escalated, skipping alerts");
441 continue;
442 }
443
444 tracing::error!(
445 transaction_id = %credit.transaction_id,
446 seller_id = %credit.seller_id,
447 amount = credit.amount_cents.as_i64(),
448 "STALE PLATFORM CREDIT: Fan+ reimbursement claimed but not settled within >24h; \
449 verify the transfer in Stripe (idempotency key platform-credit-<transaction_id>) and reconcile"
450 );
451
452 if let Some(ref to) = alert_email {
453 let subject = format!(
454 "Unsettled Fan+ credit: transaction {}",
455 credit.transaction_id
456 );
457 let body = format!(
458 "A platform-funded (Fan+) credit reimbursement was claimed for settlement but the \
459 transfer never completed within >24h.\n\n\
460 Transaction: {}\nSeller: {}\nAmount owed: {}c\n\n\
461 The process likely died between claim and transfer. The transfer uses the \
462 deterministic idempotency key `platform-credit-{}`, so it is safe to re-trigger \
463 or verify in the Stripe dashboard and reconcile manually.",
464 credit.transaction_id,
465 credit.seller_id,
466 credit.amount_cents.as_i64(),
467 credit.transaction_id,
468 );
469 if let Err(e) = state.email.send_alert(to, &subject, &body).await {
470 tracing::error!(error = ?e, "failed to send stale platform credit alert email");
471 }
472 }
473
474 if let Some(ref wam) = state.wam {
475 let title = format!("Unsettled Fan+ credit: {}c", credit.amount_cents.as_i64());
476 let body = format!(
477 "Platform-funded credit claimed >24h ago without settling.\n\
478 Transaction: {}\nSeller: {}\nAmount: {}c\n\
479 Idempotency key: platform-credit-{}. Verify in Stripe and reconcile.",
480 credit.transaction_id,
481 credit.seller_id,
482 credit.amount_cents.as_i64(),
483 credit.transaction_id,
484 );
485 wam.create_ticket(
486 &title,
487 Some(&body),
488 "critical",
489 "platform-credit-escalation",
490 Some(&credit.transaction_id.to_string()),
491 )
492 .await;
493 }
494 }
495 }
496
497 #[cfg(test)]
498 mod tests {
499 use super::*;
500
501 #[test]
502 fn webhook_not_dead_under_threshold() {
503 assert!(!is_webhook_dead(1));
504 assert!(!is_webhook_dead(4));
505 }
506
507 #[test]
508 fn webhook_dead_at_threshold() {
509 assert!(is_webhook_dead(5));
510 }
511
512 #[test]
513 fn webhook_dead_above_threshold() {
514 assert!(is_webhook_dead(6));
515 assert!(is_webhook_dead(100));
516 }
517 }
518