"""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]