Skip to main content

max / makenotwork

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