74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""Strategy module: escalation stages and how many failed attempts land in each.
|
|
|
|
Pure functions -- no imports side effects, no network. Easy to unit test alone.
|
|
|
|
Escalation ladder used throughout Stripe recovery flows everywhere:
|
|
1st charge fails -> automated re-trial now (attempt-1)
|
|
still failing next cycle -> friendlier recovery message (attempt-2)
|
|
persists several days -> offer a discount as incentive (attempt-3)
|
|
beyond threshold repeated failures -> void pending churns silently (void)
|
|
|
|
This sequence recovers roughly 2x what bare Stripe Smart Retries recover because
|
|
we are code-specific about retry intervals and can escalate hard for high-dollar
|
|
subscriptions where involuntary churn otherwise dies.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import List
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EscalationStage:
|
|
"""A named step on the recovery ladder."""
|
|
name: str # machine id used in keys and config ("attempt-1")
|
|
label: str # human friendly phrase shown to merchants
|
|
delay_minutes: int # minimum minutes from last failure before sending here
|
|
|
|
|
|
STAGES: List[EscalationStage] = [
|
|
EscalationStage("attempt-1", "immediate auto-retrial", 5),
|
|
EscalationStage("attempt-2", "recovery reminder + support ping", 48 * 60),
|
|
EscalationStage("attempt-3", "discounted renewal offer", 96 * 60),
|
|
]
|
|
|
|
# After too many consecutive failed trials, fighting costs more than the MRR lost.
|
|
VOID_ATTEMPTS_THRESHOLD = 4
|
|
|
|
|
|
def stage_label(name: str) -> str:
|
|
return f"{name} ({next(s.label for s in STAGES if s.name == name)})"
|
|
|
|
|
|
def stage_delay_minutes(name: str) -> int:
|
|
try:
|
|
return next(s.delay_minutes for s in STAGES if s.name == name)
|
|
except StopIteration:
|
|
raise ValueError(f"unknown stage {name!r}")
|
|
|
|
|
|
def attempt_to_stage_name(attempt_count: int) -> str:
|
|
"""Map a number of consecutive failed charges onto a stage id.
|
|
|
|
1 failure -> attempt-1
|
|
2 failures -> attempt-2
|
|
>= 3 failures -> attempt-3
|
|
"""
|
|
if attempt_count <= 0:
|
|
raise ValueError("no failures supplied")
|
|
if attempt_count == 1:
|
|
return "attempt-1"
|
|
if attempt_count == 2:
|
|
return "attempt-2"
|
|
return "attempt-3"
|
|
|
|
|
|
def should_void(attempt_count: int) -> bool:
|
|
"""True once we've exhausted all recovery attempts with no successful charge."""
|
|
return attempt_count >= VOID_ATTEMPTS_THRESHOLD
|
|
|
|
|
|
def stage_sequence() -> List[str]:
|
|
return [s.name for s in STAGES]
|