How to Detect and Recover Failed Stripe Payments Before You Lose the Customer

# failed_payment_monitor.py
import os
import stripe
import requests
from flask import Flask, request, jsonify
from datetime import datetime, timedelta
from dotenv import load_dotenv
 
load_dotenv()
app = Flask(__name__)
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]
SLACK_WEBHOOK = os.environ["SLACK_REVOPS_WEBHOOK"]
 
 
def send_slack_alert(invoice, customer, attempt_count):
    """Send Slack alert for failed payment."""
    urgency = "🟡" if attempt_count <= 2 else "🔴"
    
    requests.post(SLACK_WEBHOOK, json={
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"{urgency} Payment Failed — Attempt #{attempt_count}",
                },
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Customer:*\n{customer.email}"},
                    {"type": "mrkdwn", "text": f"*Amount:*\n${invoice.amount_due / 100:.2f}"},
                    {"type": "mrkdwn", "text": f"*Plan:*\n{get_plan_name(invoice)}"},
                    {"type": "mrkdwn", "text": f"*Failure reason:*\n{get_failure_reason(invoice)}"},
                ],
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "View in Stripe"},
                        "url": f"https://dashboard.stripe.com/invoices/{invoice.id}",
                    },
                ],
            },
        ],
    })
 
 
def get_plan_name(invoice):
    """Extract plan name from invoice lines."""
    for line in invoice.lines.data:
        if line.price and line.price.product:
            product = stripe.Product.retrieve(line.price.product)
            return product.name
    return "Unknown"
 
 
def get_failure_reason(invoice):
    """Get human-readable failure reason."""
    if not invoice.charge:
        return "No charge attempted"
    
    charge = stripe.Charge.retrieve(invoice.charge)
    outcome = charge.outcome or {}
    
    reasons = {
        "card_declined": "Card declined",
        "insufficient_funds": "Insufficient funds",
        "expired_card": "Card expired",
        "processing_error": "Processing error",
        "incorrect_cvc": "Incorrect CVC",
    }
    
    decline_code = charge.failure_code or outcome.get("reason", "unknown")
    return reasons.get(decline_code, decline_code)
 
 
@app.route("/webhooks/stripe", methods=["POST"])
def handle_webhook():
    payload = request.get_data()
    sig = request.headers.get("Stripe-Signature")
    
    try:
        event = stripe.Webhook.construct_event(payload, sig, WEBHOOK_SECRET)
    except (ValueError, stripe.error.SignatureVerificationError):
        return jsonify({"error": "Invalid signature"}), 400
    
    if event.type == "invoice.payment_failed":
        invoice = event.data.object
        customer = stripe.Customer.retrieve(invoice.customer)
        
        send_slack_alert(invoice, customer, invoice.attempt_count)
        
        # On 3rd+ attempt, also email the CSM directly
        if invoice.attempt_count >= 3:
            escalate_to_csm(customer, invoice)
    
    elif event.type == "customer.subscription.updated":
        sub = event.data.object
        prev = event.data.previous_attributes
        
        # Detect transition to past_due
        if prev.get("status") == "active" and sub.status == "past_due":
            customer = stripe.Customer.retrieve(sub.customer)
            send_slack_alert_status_change(customer, "active", "past_due")
    
    return jsonify({"received": True}), 200

# escalation_rules.py
 
ESCALATION_RULES = [
    {
        "condition": lambda inv, cust: inv.attempt_count == 1,
        "action": "log_only",
        "description": "First failure — Stripe will retry. Log and monitor.",
    },
    {
        "condition": lambda inv, cust: (
            inv.attempt_count == 2
            and inv.amount_due >= 10000  # $100+
        ),
        "action": "slack_alert",
        "description": "Second failure on high-value. Alert RevOps.",
    },
    {
        "condition": lambda inv, cust: inv.attempt_count >= 3,
        "action": "human_outreach",
        "description": "Third+ failure. CS reaches out directly.",
    },
    {
        "condition": lambda inv, cust: (
            inv.attempt_count >= 3
            and inv.amount_due >= 50000  # $500+
        ),
        "action": "executive_escalation",
        "description": "Third+ failure on enterprise. VP of CS notified.",
    },
]

# detect_expiring_cards.py
import stripe
from datetime import datetime
 
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
 
current_month = datetime.now().month
current_year = datetime.now().year
 
# Find cards expiring this month or next month
expiring_soon = []
 
for pm in stripe.PaymentMethod.list(type="card", limit=100).auto_paging_iter():
    exp_month = pm.card.exp_month
    exp_year = pm.card.exp_year
    
    # Check if expiring within 60 days
    card_expiry = datetime(exp_year, exp_month, 1)
    days_until_expiry = (card_expiry - datetime.now()).days
    
    if 0 < days_until_expiry <= 60:
        # Find attached customer
        if pm.customer:
            customer = stripe.Customer.retrieve(pm.customer)
            # Check if this is their default payment method
            if customer.invoice_settings.default_payment_method == pm.id:
                expiring_soon.append({
                    "email": customer.email,
                    "card_last4": pm.card.last4,
                    "exp_month": exp_month,
                    "exp_year": exp_year,
                    "days_until_expiry": days_until_expiry,
                })
 
# Send proactive "please update your card" emails
for card in expiring_soon:
    send_card_update_reminder(card)
    
print(f"Found {len(expiring_soon)} cards expiring within 60 days")

-- Monthly dunning recovery rate
WITH failed_invoices AS (
  SELECT
    DATE_TRUNC('month', created) AS month,
    id,
    amount_due,
    status,
    attempt_count
  FROM stripe_invoices
  WHERE attempted = true
    AND paid = false
    AND created >= CURRENT_DATE - INTERVAL '6 months'
),
eventual_outcomes AS (
  SELECT
    fi.month,
    fi.id,
    fi.amount_due,
    CASE 
      WHEN fi.status = 'paid' THEN 'recovered'
      WHEN fi.status = 'void' THEN 'voided'
      ELSE 'lost'
    END AS outcome
  FROM failed_invoices fi
)
SELECT
  month,
  COUNT(*) AS total_failures,
  SUM(CASE WHEN outcome = 'recovered' THEN 1 ELSE 0 END) AS recovered,
  SUM(CASE WHEN outcome = 'lost' THEN 1 ELSE 0 END) AS lost,
  ROUND(
    SUM(CASE WHEN outcome = 'recovered' THEN 1 ELSE 0 END)::numeric 
    / COUNT(*) * 100, 1
  ) AS recovery_rate_pct,
  SUM(CASE WHEN outcome = 'recovered' THEN amount_due ELSE 0 END) / 100 AS recovered_revenue,
  SUM(CASE WHEN outcome = 'lost' THEN amount_due ELSE 0 END) / 100 AS lost_revenue
FROM eventual_outcomes
GROUP BY month
ORDER BY month DESC;