Skip to main content

max / makenotwork

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