Snapshot: full project state

stripe-dunning-app/dunningengine/__init__.py

@@ -0,0 +1,73 @@
+"""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]