Skip to main content

max / makenotwork

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