Circuit Breakers.

import time
import threading
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
from collections import deque

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreakerConfig:
    """Configuration for circuit breaker behavior."""
    
    # How many failures before opening
    failure_threshold: int = 5
    
    # Time window for counting failures (seconds)
    failure_window: int = 60
    
    # How long to stay open before testing (seconds)
    recovery_timeout: int = 30
    
    # How many successes needed in half-open to close
    success_threshold: int = 1
    
    # What exceptions count as failures
    failure_exceptions: tuple = (Exception,)

class CircuitBreaker:
    """
    Circuit breaker implementation.
    
    Wraps calls to external services and fails fast when the service
    is detected as unhealthy.
    """
    
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        
        self.state = CircuitState.CLOSED
        self.failures = deque()  # Timestamps of recent failures
        self.last_failure_time: Optional[float] = None
        self.half_open_successes = 0
        
        self.lock = threading.Lock()
    
    def call(self, func: Callable, *args, **kwargs) -> Any:
        """
        Execute function through circuit breaker.
        
        Raises CircuitOpenError if circuit is open.
        """
        with self.lock:
            self._check_state_transition()
            
            if self.state == CircuitState.OPEN:
                raise CircuitOpenError(
                    f"Circuit breaker '{self.name}' is open"
                )
        
        try:
            result = func(*args, **kwargs)
            self._record_success()
            return result
        
        except self.config.failure_exceptions as e:
            self._record_failure()
            raise
    
    def _check_state_transition(self):
        """Check if we should transition states."""
        
        if self.state == CircuitState.OPEN:
            # Check if recovery timeout has passed
            if self.last_failure_time:
                elapsed = time.time() - self.last_failure_time
                if elapsed >= self.config.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_successes = 0
    
    def _record_success(self):
        """Record successful call."""
        with self.lock:
            if self.state == CircuitState.HALF_OPEN:
                self.half_open_successes += 1
                if self.half_open_successes >= self.config.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.failures.clear()
    
    def _record_failure(self):
        """Record failed call."""
        with self.lock:
            now = time.time()
            self.last_failure_time = now
            
            if self.state == CircuitState.HALF_OPEN:
                # Single failure in half-open → back to open
                self.state = CircuitState.OPEN
                return
            
            if self.state == CircuitState.CLOSED:
                # Add failure to window
                self.failures.append(now)
                
                # Remove old failures outside window
                cutoff = now - self.config.failure_window
                while self.failures and self.failures[0] < cutoff:
                    self.failures.popleft()
                
                # Check if threshold exceeded
                if len(self.failures) >= self.config.failure_threshold:
                    self.state = CircuitState.OPEN
    
    def get_state(self) -> CircuitState:
        """Get current circuit state."""
        with self.lock:
            self._check_state_transition()
            return self.state

class CircuitOpenError(Exception):
    """Raised when circuit breaker is open."""
    pass

# Create circuit breaker for bank API
bank_circuit = CircuitBreaker(
    name="bank_api",
    config=CircuitBreakerConfig(
        failure_threshold=5,      # Open after 5 failures
        failure_window=60,        # Within 60 seconds
        recovery_timeout=30,      # Try again after 30 seconds
        success_threshold=2,      # Need 2 successes to close
    )
)

def call_bank_api(user_id: str, amount: float) -> dict:
    """Call bank API through circuit breaker."""
    
    def _make_request():
        response = requests.post(
            BANK_API_URL,
            json={'user_id': user_id, 'amount': amount},
            timeout=3.5
        )
        response.raise_for_status()
        return response.json()
    
    try:
        return bank_circuit.call(_make_request)
    
    except CircuitOpenError:
        # Circuit is open - fail fast with fallback
        raise PaymentServiceUnavailable(
            "Payment processing is temporarily unavailable. "
            "Please try again in a few minutes."
        )
    
    except requests.Timeout:
        # Will be recorded as failure by circuit breaker
        raise PaymentTimeout("Payment processing timed out")
    
    except requests.HTTPError as e:
        # Will be recorded as failure by circuit breaker
        raise PaymentFailed(f"Payment failed: {e}")

from functools import wraps

def circuit_breaker(
    name: str,
    failure_threshold: int = 5,
    failure_window: int = 60,
    recovery_timeout: int = 30
):
    """Decorator to wrap function with circuit breaker."""
    
    config = CircuitBreakerConfig(
        failure_threshold=failure_threshold,
        failure_window=failure_window,
        recovery_timeout=recovery_timeout,
    )
    breaker = CircuitBreaker(name, config)
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return breaker.call(func, *args, **kwargs)
        
        wrapper.circuit_breaker = breaker  # Expose for monitoring
        return wrapper
    
    return decorator

# Usage
@circuit_breaker("bank_api", failure_threshold=5, recovery_timeout=30)
def charge_bank(user_id: str, amount: float) -> dict:
    response = requests.post(BANK_API_URL, ...)
    return response.json()

# Calling the function - circuit breaker handles failures
try:
    result = charge_bank("user_123", 99.00)
except CircuitOpenError:
    # Handle gracefully
    return "Payment service temporarily unavailable"

class RateBasedCircuitBreaker:
    """Circuit breaker that uses failure rate, not count."""
    
    def __init__(
        self,
        name: str,
        failure_rate_threshold: float = 0.5,  # 50% failure rate
        minimum_calls: int = 10,               # Need at least 10 calls
        window_size: int = 60,                 # 60 second window
        recovery_timeout: int = 30
    ):
        self.name = name
        self.failure_rate_threshold = failure_rate_threshold
        self.minimum_calls = minimum_calls
        self.window_size = window_size
        self.recovery_timeout = recovery_timeout
        
        self.calls = deque()  # (timestamp, success: bool)
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
        self.lock = threading.Lock()
    
    def _get_stats(self) -> tuple[int, int]:
        """Get success and failure counts in current window."""
        now = time.time()
        cutoff = now - self.window_size
        
        # Remove old entries
        while self.calls and self.calls[0][0] < cutoff:
            self.calls.popleft()
        
        successes = sum(1 for _, success in self.calls if success)
        failures = sum(1 for _, success in self.calls if not success)
        
        return successes, failures
    
    def _record_call(self, success: bool):
        """Record a call result."""
        with self.lock:
            now = time.time()
            self.calls.append((now, success))
            
            if not success:
                self.last_failure_time = now
            
            if self.state == CircuitState.HALF_OPEN:
                if success:
                    self.state = CircuitState.CLOSED
                    self.calls.clear()
                else:
                    self.state = CircuitState.OPEN
                return
            
            if self.state == CircuitState.CLOSED:
                successes, failures = self._get_stats()
                total = successes + failures
                
                if total >= self.minimum_calls:
                    failure_rate = failures / total
                    if failure_rate >= self.failure_rate_threshold:
                        self.state = CircuitState.OPEN

class MultiCircuitBreaker:
    """Manage multiple circuit breakers for different operations."""
    
    def __init__(self):
        self.breakers: dict[str, CircuitBreaker] = {}
        self.lock = threading.Lock()
    
    def get_breaker(self, operation: str, config: CircuitBreakerConfig = None) -> CircuitBreaker:
        """Get or create circuit breaker for operation."""
        with self.lock:
            if operation not in self.breakers:
                self.breakers[operation] = CircuitBreaker(
                    name=operation,
                    config=config or CircuitBreakerConfig()
                )
            return self.breakers[operation]

# Usage
breakers = MultiCircuitBreaker()

# Different thresholds for different operations
charge_breaker = breakers.get_breaker("bank_charge", CircuitBreakerConfig(
    failure_threshold=3,  # Payments are critical - open fast
    recovery_timeout=60   # But wait longer to recover
))

refund_breaker = breakers.get_breaker("bank_refund", CircuitBreakerConfig(
    failure_threshold=10,  # Refunds less critical
    recovery_timeout=30
))

from typing import Callable, TypeVar, Optional

T = TypeVar('T')

class CircuitBreakerWithFallback:
    """Circuit breaker that can return fallback value when open."""
    
    def __init__(
        self,
        name: str,
        config: CircuitBreakerConfig = None,
        fallback: Optional[Callable[[], T]] = None
    ):
        self.breaker = CircuitBreaker(name, config)
        self.fallback = fallback
    
    def call(
        self,
        func: Callable[[], T],
        fallback: Optional[Callable[[], T]] = None
    ) -> T:
        """
        Execute function with circuit breaker and optional fallback.
        
        If circuit is open and fallback is provided, return fallback value
        instead of raising exception.
        """
        try:
            return self.breaker.call(func)
        
        except CircuitOpenError:
            effective_fallback = fallback or self.fallback
            if effective_fallback:
                return effective_fallback()
            raise

# Usage
fraud_circuit = CircuitBreakerWithFallback(
    name="fraud_service",
    config=CircuitBreakerConfig(failure_threshold=5),
    fallback=lambda: {"risk_score": 0.5, "fallback": True}  # Default medium risk
)

def check_fraud(user_id: str, amount: float) -> dict:
    def _call():
        return fraud_service.check(user_id, amount)
    
    # If fraud service is down, return default risk score
    # This allows payments to continue with extra verification
    return fraud_circuit.call(_call)

# Bad: Opens on 5 failures (broken at high traffic)
CircuitBreakerConfig(failure_threshold=5)

# Good: Opens on 50% failure rate with minimum sample
RateBasedCircuitBreaker(
    failure_rate_threshold=0.5,  # 50% failure rate
    minimum_calls=100,            # Need 100 calls before deciding
)

class GradualRecoveryCircuitBreaker:
    """Circuit breaker with gradual traffic ramp-up on recovery."""
    
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        self.breaker = CircuitBreaker(name, config)
        self.recovery_percentage = 100  # % of traffic to allow
        self.recovery_start_time = None
        self.recovery_duration = 60  # seconds to reach 100%
    
    def call(self, func: Callable) -> Any:
        state = self.breaker.get_state()
        
        if state == CircuitState.CLOSED and self.recovery_percentage < 100:
            # We're in recovery ramp-up
            if self._should_allow_request():
                return self.breaker.call(func)
            else:
                raise CircuitOpenError("Request shed during recovery")
        
        return self.breaker.call(func)
    
    def _should_allow_request(self) -> bool:
        """Probabilistically allow request based on recovery percentage."""
        self._update_recovery_percentage()
        return random.random() * 100 < self.recovery_percentage
    
    def _update_recovery_percentage(self):
        """Update recovery percentage based on time since recovery started."""
        if self.recovery_start_time is None:
            self.recovery_start_time = time.time()
        
        elapsed = time.time() - self.recovery_start_time
        self.recovery_percentage = min(100, (elapsed / self.recovery_duration) * 100)
        
        if self.recovery_percentage >= 100:
            self.recovery_start_time = None  # Reset for next recovery

class DistributedCircuitBreaker:
    """Circuit breaker with shared state in Redis."""
    
    def __init__(self, name: str, redis_client, config: CircuitBreakerConfig = None):
        self.name = name
        self.redis = redis_client
        self.config = config or CircuitBreakerConfig()
    
    def _state_key(self) -> str:
        return f"circuit_breaker:{self.name}:state"
    
    def _tripped_key(self) -> str:
        return f"circuit_breaker:{self.name}:tripped"
    
    def _failures_key(self) -> str:
        return f"circuit_breaker:{self.name}:failures"
    
    def get_state(self) -> CircuitState:
        """Get circuit state from Redis."""
        state = self.redis.get(self._state_key())
        if state:
            return CircuitState(state.decode())   # OPEN until its TTL expires
        if self.redis.exists(self._tripped_key()):
            return CircuitState.HALF_OPEN          # OPEN expired, not yet recovered
        return CircuitState.CLOSED
    
    def _open(self):
        """OPEN for recovery_timeout; the 'tripped' marker outlives it."""
        self.redis.setex(self._state_key(), self.config.recovery_timeout,
                         CircuitState.OPEN.value)
        self.redis.set(self._tripped_key(), 1)
    
    def record_failure(self):
        """Record failure in Redis."""
        if self.get_state() == CircuitState.HALF_OPEN:
            self._open()  # Probe failed - back to OPEN
            return
        
        pipe = self.redis.pipeline()
        pipe.lpush(self._failures_key(), time.time())
        pipe.ltrim(self._failures_key(), 0, 99)  # Keep last 100
        pipe.expire(self._failures_key(), self.config.failure_window)
        pipe.execute()
        
        # Check if should open
        failures = self.redis.lrange(self._failures_key(), 0, -1)
        recent = [f for f in failures 
                  if float(f) > time.time() - self.config.failure_window]
        
        if len(recent) >= self.config.failure_threshold:
            self._open()
    
    def record_success(self):
        """Record success, potentially closing circuit."""
        if self.get_state() == CircuitState.HALF_OPEN:
            # Probe succeeded - CLOSED
            self.redis.delete(self._tripped_key(), self._failures_key())

from prometheus_client import Counter, Gauge

circuit_state_gauge = Gauge(
    'circuit_breaker_state',
    'Current circuit breaker state (0=closed, 1=half-open, 2=open)',
    ['name']
)

circuit_state_changes = Counter(
    'circuit_breaker_state_changes_total',
    'Circuit breaker state transitions',
    ['name', 'from_state', 'to_state']
)

class ObservableCircuitBreaker(CircuitBreaker):
    """Circuit breaker with metrics and alerting."""
    
    def _transition_state(self, new_state: CircuitState):
        old_state = self.state
        self.state = new_state
        
        # Update metrics
        state_value = {'closed': 0, 'half_open': 1, 'open': 2}
        circuit_state_gauge.labels(name=self.name).set(state_value[new_state.value])
        
        circuit_state_changes.labels(
            name=self.name,
            from_state=old_state.value,
            to_state=new_state.value
        ).inc()
        
        # Log for alerting
        if new_state == CircuitState.OPEN:
            logger.error(
                f"Circuit breaker '{self.name}' OPENED",
                extra={'circuit': self.name, 'event': 'circuit_open'}
            )
        elif new_state == CircuitState.CLOSED and old_state == CircuitState.HALF_OPEN:
            logger.info(
                f"Circuit breaker '{self.name}' recovered",
                extra={'circuit': self.name, 'event': 'circuit_recovered'}
            )

groups:
  - name: circuit_breakers
    rules:
      # Alert when any circuit is open
      - alert: CircuitBreakerOpen
        expr: circuit_breaker_state == 2
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Circuit breaker {{ $labels.name }} is open"
          description: "The circuit has been open for over 1 minute. Investigate the downstream service."
          
      # Alert when circuit is flapping (sign of borderline failure)
      - alert: CircuitBreakerFlapping
        expr: increase(circuit_breaker_state_changes_total[10m]) > 5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Circuit breaker {{ $labels.name }} is flapping"
          description: "The circuit has changed state {{ $value }} times in 10 minutes. The downstream service may be unstable."

class PaymentService:
    def process_payment(self, user_id, amount, idempotency_key):
        # Day 2: Check idempotency
        is_new, cached = self.idempotency.check_and_set(idempotency_key, {...})
        if not is_new:
            return cached
        
        # Day 1: Timeout budget
        budget = TimeoutBudget(4500)
        
        # Step 1: Fraud check
        fraud_result = self._check_fraud(budget, user_id, amount)
        
        # Step 2: Bank charge
        bank_result = self._charge_bank(budget, user_id, amount, idempotency_key)
        
        # Step 3: Notification
        self._send_notification(user_id, amount, bank_result.transaction_id)
        
        return bank_result

from typing import Optional
import asyncio

class PaymentServiceWithCircuitBreaker:
    """
    Payment service with all three reliability patterns:
    - Timeouts (Day 1)
    - Idempotency (Day 2)  
    - Circuit Breakers (Day 3)
    """
    
    def __init__(
        self,
        config: PaymentConfig,
        idempotency_store: IdempotencyStore,
        redis_client
    ):
        self.config = config
        self.idempotency = idempotency_store
        
        # Circuit breakers for each downstream service
        self.fraud_circuit = CircuitBreakerWithFallback(
            name="fraud_service",
            config=CircuitBreakerConfig(
                failure_threshold=5,
                failure_window=60,
                recovery_timeout=30,
            ),
            # If fraud service is down, use default medium risk
            fallback=lambda: {"risk_score": 0.5, "source": "fallback"}
        )
        
        self.bank_circuit = CircuitBreaker(
            name="bank_api",
            config=CircuitBreakerConfig(
                failure_threshold=3,     # More sensitive for payments
                failure_window=30,
                recovery_timeout=60,     # Wait longer before retry
            )
        )
        
        self.notification_circuit = CircuitBreakerWithFallback(
            name="notification_service",
            config=CircuitBreakerConfig(
                failure_threshold=10,    # Less sensitive, not critical
                failure_window=60,
                recovery_timeout=15,
            ),
            fallback=lambda: {"status": "queued"}  # Queue for later
        )
    
    async def process_payment(
        self,
        user_id: str,
        amount: float,
        idempotency_key: str
    ) -> PaymentResult:
        """
        Process payment with full reliability stack.
        """
        request_body = {'user_id': user_id, 'amount': amount}
        
        # =====================================================================
        # Layer 1: Idempotency (Day 2)
        # =====================================================================
        try:
            is_new, cached = await self.idempotency.check_and_claim(
                idempotency_key, request_body, user_id
            )
            if not is_new:
                return PaymentResult(**cached)
        except IdempotencyInProgressError:
            return PaymentResult(
                status=PaymentStatus.PENDING,
                error_message="Payment is being processed"
            )
        
        # =====================================================================
        # Layer 2: Timeout Budget (Day 1)
        # =====================================================================
        budget = TimeoutBudget(self.config.total_budget_ms)
        
        try:
            result = await self._process_with_circuit_breakers(
                budget, user_id, amount, idempotency_key
            )
            await self.idempotency.complete(idempotency_key, result.__dict__)
            return result
        
        except Exception as e:
            await self.idempotency.fail(idempotency_key, str(e))
            raise
    
    async def _process_with_circuit_breakers(
        self,
        budget: TimeoutBudget,
        user_id: str,
        amount: float,
        idempotency_key: str
    ) -> PaymentResult:
        """Process payment with circuit breakers on each step."""
        
        # =====================================================================
        # Step 1: Fraud Check (with fallback)
        # =====================================================================
        fraud_result = await self._check_fraud_with_circuit(budget, user_id, amount)
        
        if fraud_result.get('risk_score', 0) > 0.9:
            return PaymentResult(
                status=PaymentStatus.REJECTED,
                error_message="Transaction flagged as high risk"
            )
        
        # If using fallback (fraud service down), require additional verification
        if fraud_result.get('source') == 'fallback':
            if amount > 100:  # High value + no fraud check = reject
                return PaymentResult(
                    status=PaymentStatus.REJECTED,
                    error_message="Unable to verify transaction. Please try a smaller amount."
                )
        
        # =====================================================================
        # Step 2: Bank Charge (no fallback - critical)
        # =====================================================================
        try:
            bank_result = await self._charge_bank_with_circuit(
                budget, user_id, amount, idempotency_key
            )
        except CircuitOpenError:
            # Bank circuit is open - provide clear message
            return PaymentResult(
                status=PaymentStatus.SERVICE_UNAVAILABLE,
                error_message="Payment processing is temporarily unavailable."
            )
        
        if bank_result.status != PaymentStatus.SUCCESS:
            return bank_result
        
        # =====================================================================
        # Step 3: Notification (with fallback)
        # =====================================================================
        await self._notify_with_circuit(user_id, amount, bank_result.transaction_id)
        
        return bank_result

class SmartPaymentCircuitBreaker:
    """
    Circuit breaker designed for high-stakes, high-traffic scenarios.
    """
    
    def __init__(self):
        self.breaker = RateBasedCircuitBreaker(
            name="bank_api",
            failure_rate_threshold=0.5,  # Only open at 50% failure rate
            minimum_calls=100,            # Need 100 calls before deciding
            window_size=30,
            recovery_timeout=30,          # Shorter recovery time
        )
        
        # Track by error type - not all errors should count
        self.retryable_errors = (Timeout, ConnectionError)
        self.non_retryable_errors = (PaymentDeclined, InsufficientFunds)
    
    def call(self, func):
        try:
            result = func()
            self.breaker.record_success()
            return result
        
        except self.retryable_errors as e:
            # These indicate service problems - count them
            self.breaker.record_failure()
            raise
        
        except self.non_retryable_errors as e:
            # These are business errors, not service problems
            # Don't count against circuit breaker
            self.breaker.record_success()  # Service worked, just said "no"
            raise

async def handle_payment_request(request):
    try:
        result = await payment_service.process_payment(
            user_id=request.user_id,
            amount=request.amount,
            idempotency_key=request.idempotency_key
        )
        return PaymentResponse(result)
    
    except CircuitOpenError:
        # Log for monitoring
        logger.warning("Payment circuit open, returning friendly error")
        
        return PaymentResponse(
            status="temporarily_unavailable",
            message="We're experiencing high demand. Your payment could not be processed.",
            retry_after=60,  # Tell client when to retry
            actions=[
                {
                    "type": "retry",
                    "label": "Try Again",
                    "delay_seconds": 60
                },
                {
                    "type": "alternative",
                    "label": "Pay with PayPal",
                    "url": "/checkout/paypal"
                },
                {
                    "type": "save",
                    "label": "Save Cart for Later",
                    "url": "/cart/save"
                }
            ]
        )

class ResilientClient:
    """
    Client with all three patterns working together.
    """
    
    def __init__(
        self,
        name: str,
        max_concurrent: int = 10,      # Bulkhead
        max_retries: int = 3,          # Retry
        failure_threshold: int = 5,    # Circuit breaker
    ):
        self.name = name
        self.circuit = CircuitBreaker(name, CircuitBreakerConfig(
            failure_threshold=failure_threshold
        ))
        self.semaphore = asyncio.Semaphore(max_concurrent)  # Bulkhead
        self.max_retries = max_retries
    
    async def call(self, func: Callable, timeout: float = 5.0) -> Any:
        """Execute with all resilience patterns."""
        
        # Layer 1: Circuit Breaker (cheapest check first)
        if self.circuit.get_state() == CircuitState.OPEN:
            raise CircuitOpenError(f"{self.name} circuit is open")
        
        # Layer 2: Bulkhead (limit concurrent calls)
        try:
            acquired = await asyncio.wait_for(
                self.semaphore.acquire(),
                timeout=1.0
            )
        except asyncio.TimeoutError:
            raise BulkheadFullError(f"{self.name} bulkhead is full")
        
        try:
            # Layer 3: Retry with backoff
            last_exception = None
            
            for attempt in range(self.max_retries):
                try:
                    result = await asyncio.wait_for(func(), timeout=timeout)
                    self.circuit.record_success()
                    return result
                
                except asyncio.TimeoutError as e:
                    last_exception = e
                    self.circuit.record_failure()
                    
                    if attempt < self.max_retries - 1:
                        delay = (2 ** attempt) + random.uniform(0, 1)
                        await asyncio.sleep(delay)
                
                except Exception as e:
                    last_exception = e
                    self.circuit.record_failure()
                    raise
            
            raise last_exception
        
        finally:
            self.semaphore.release()

"""
Production-ready circuit breaker implementation.
Completes the reliability stack from Days 1-3.
"""

import time
import random
import asyncio
import logging
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Any, Optional, TypeVar, Generic
from collections import deque
import threading
from prometheus_client import Counter, Gauge, Histogram

# =============================================================================
# Metrics
# =============================================================================

circuit_state_gauge = Gauge(
    'circuit_breaker_state',
    'Current circuit state (0=closed, 1=half_open, 2=open)',
    ['name']
)

circuit_calls = Counter(
    'circuit_breaker_calls_total',
    'Circuit breaker call results',
    ['name', 'result']  # success, failure, rejected
)

circuit_state_changes = Counter(
    'circuit_breaker_state_changes_total',
    'Circuit breaker state transitions',
    ['name', 'from_state', 'to_state']
)

# =============================================================================
# Core Types
# =============================================================================

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreakerConfig:
    # Rate-based thresholds
    failure_rate_threshold: float = 0.5
    minimum_calls: int = 10
    
    # Time windows
    sliding_window_size: int = 60  # seconds
    recovery_timeout: int = 30     # seconds
    
    # Recovery behavior
    success_threshold: int = 3     # successes needed to close
    gradual_recovery: bool = True
    recovery_ramp_duration: int = 60  # seconds to reach 100%
    
    # Error classification
    record_exceptions: tuple = (Exception,)
    ignore_exceptions: tuple = ()

T = TypeVar('T')

# =============================================================================
# Production Circuit Breaker
# =============================================================================

class ProductionCircuitBreaker:
    """
    Production circuit breaker with:
    - Rate-based failure detection
    - Gradual recovery
    - Prometheus metrics
    - Distributed state option
    """
    
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        self.logger = logging.getLogger(f'circuit_breaker.{name}')
        
        # State
        self.state = CircuitState.CLOSED
        self.calls: deque = deque()  # (timestamp, success: bool)
        self.last_failure_time: Optional[float] = None
        self.half_open_successes = 0
        self.recovery_start_time: Optional[float] = None
        
        self.lock = threading.Lock()
        
        # Initialize metrics
        circuit_state_gauge.labels(name=name).set(0)
    
    def call(self, func: Callable[[], T]) -> T:
        """Execute function through circuit breaker."""
        
        with self.lock:
            self._maybe_transition_state()
            
            if self.state == CircuitState.OPEN:
                circuit_calls.labels(name=self.name, result='rejected').inc()
                retry_after = self._get_retry_after()
                raise CircuitOpenError(self.name, retry_after)
            
            if self.state == CircuitState.CLOSED and self._in_recovery():
                if not self._should_allow_during_recovery():
                    circuit_calls.labels(name=self.name, result='rejected').inc()
                    raise CircuitOpenError(self.name, retry_after=5)
        
        try:
            result = func()
            self._record_success()
            return result
        
        except self.config.ignore_exceptions:
            self._record_success()
            raise
        
        except self.config.record_exceptions:
            self._record_failure()
            raise
    
    def _maybe_transition_state(self):
        """Check if state should transition."""
        if self.state == CircuitState.OPEN:
            if self.last_failure_time:
                elapsed = time.time() - self.last_failure_time
                if elapsed >= self.config.recovery_timeout:
                    self._transition_to(CircuitState.HALF_OPEN)
    
    def _record_success(self):
        """Record successful call."""
        with self.lock:
            now = time.time()
            self.calls.append((now, True))
            self._clean_old_calls()
            
            circuit_calls.labels(name=self.name, result='success').inc()
            
            if self.state == CircuitState.HALF_OPEN:
                self.half_open_successes += 1
                if self.half_open_successes >= self.config.success_threshold:
                    self._transition_to(CircuitState.CLOSED)
                    self._start_recovery()
    
    def _record_failure(self):
        """Record failed call."""
        with self.lock:
            now = time.time()
            self.last_failure_time = now
            self.calls.append((now, False))
            self._clean_old_calls()
            
            circuit_calls.labels(name=self.name, result='failure').inc()
            
            if self.state == CircuitState.HALF_OPEN:
                self._transition_to(CircuitState.OPEN)
                return
            
            if self.state == CircuitState.CLOSED:
                if self._should_open():
                    self._transition_to(CircuitState.OPEN)
    
    def _should_open(self) -> bool:
        """Check if circuit should open based on failure rate."""
        total = len(self.calls)
        
        if total < self.config.minimum_calls:
            return False
        
        failures = sum(1 for _, success in self.calls if not success)
        failure_rate = failures / total
        
        return failure_rate >= self.config.failure_rate_threshold
    
    def _clean_old_calls(self):
        """Remove calls outside sliding window."""
        cutoff = time.time() - self.config.sliding_window_size
        while self.calls and self.calls[0][0] < cutoff:
            self.calls.popleft()
    
    def _transition_to(self, new_state: CircuitState):
        """Transition to new state with logging and metrics."""
        old_state = self.state
        self.state = new_state
        
        state_values = {
            CircuitState.CLOSED: 0,
            CircuitState.HALF_OPEN: 1,
            CircuitState.OPEN: 2
        }
        
        circuit_state_gauge.labels(name=self.name).set(state_values[new_state])
        circuit_state_changes.labels(
            name=self.name,
            from_state=old_state.value,
            to_state=new_state.value
        ).inc()
        
        if new_state == CircuitState.OPEN:
            self.logger.warning(f"Circuit '{self.name}' OPENED")
        elif new_state == CircuitState.CLOSED:
            self.logger.info(f"Circuit '{self.name}' closed")
        elif new_state == CircuitState.HALF_OPEN:
            self.logger.info(f"Circuit '{self.name}' half-open, testing...")
            self.half_open_successes = 0
    
    def _start_recovery(self):
        """Start gradual recovery period."""
        if self.config.gradual_recovery:
            self.recovery_start_time = time.time()
    
    def _in_recovery(self) -> bool:
        """Check if in gradual recovery period."""
        if not self.recovery_start_time:
            return False
        
        elapsed = time.time() - self.recovery_start_time
        if elapsed >= self.config.recovery_ramp_duration:
            self.recovery_start_time = None
            return False
        
        return True
    
    def _should_allow_during_recovery(self) -> bool:
        """Probabilistically allow request during recovery."""
        elapsed = time.time() - self.recovery_start_time
        recovery_percentage = (elapsed / self.config.recovery_ramp_duration) * 100
        return random.random() * 100 < recovery_percentage
    
    def _get_retry_after(self) -> int:
        """Get seconds until retry might succeed."""
        if self.last_failure_time:
            elapsed = time.time() - self.last_failure_time
            remaining = self.config.recovery_timeout - elapsed
            return max(1, int(remaining))
        return self.config.recovery_timeout
    
    def get_state(self) -> CircuitState:
        """Get current circuit state."""
        with self.lock:
            self._maybe_transition_state()
            return self.state
    
    def get_stats(self) -> dict:
        """Get circuit breaker statistics."""
        with self.lock:
            total = len(self.calls)
            failures = sum(1 for _, success in self.calls if not success)
            
            return {
                'name': self.name,
                'state': self.state.value,
                'total_calls': total,
                'failures': failures,
                'failure_rate': failures / total if total > 0 else 0,
                'in_recovery': self._in_recovery(),
            }


class CircuitOpenError(Exception):
    """Raised when circuit breaker is open."""
    def __init__(self, name: str, retry_after: int = None):
        self.name = name
        self.retry_after = retry_after
        super().__init__(f"Circuit breaker '{name}' is open")


# =============================================================================
# Complete Payment Service with All Patterns
# =============================================================================

class ResilientPaymentService:
    """
    Payment service demonstrating all three reliability patterns:
    - Day 1: Timeouts
    - Day 2: Idempotency  
    - Day 3: Circuit Breakers
    """
    
    def __init__(self, config, idempotency_store):
        self.config = config
        self.idempotency = idempotency_store
        
        # Different circuit breakers for different services
        self.fraud_circuit = ProductionCircuitBreaker(
            name="fraud_service",
            config=CircuitBreakerConfig(
                failure_rate_threshold=0.3,
                minimum_calls=20,
                recovery_timeout=30,
            )
        )
        
        self.bank_circuit = ProductionCircuitBreaker(
            name="bank_api",
            config=CircuitBreakerConfig(
                failure_rate_threshold=0.5,
                minimum_calls=50,
                recovery_timeout=60,
                success_threshold=3,
                gradual_recovery=True,
            )
        )
        
        self.notification_circuit = ProductionCircuitBreaker(
            name="notification_service",
            config=CircuitBreakerConfig(
                failure_rate_threshold=0.5,
                minimum_calls=10,
                recovery_timeout=15,
            )
        )
    
    async def process_payment(
        self,
        user_id: str,
        amount: float,
        idempotency_key: str
    ):
        """
        Process payment with complete reliability stack.
        
        Layer 1: Idempotency (Day 2) - prevent duplicates
        Layer 2: Timeout Budget (Day 1) - don't wait forever
        Layer 3: Circuit Breakers (Day 3) - fail fast on broken services
        """
        
        # Check idempotency first
        is_new, cached = await self.idempotency.check_and_claim(
            idempotency_key,
            {'user_id': user_id, 'amount': amount}
        )
        
        if not is_new:
            return cached
        
        # Create timeout budget
        budget = TimeoutBudget(self.config.total_budget_ms)
        
        try:
            # Process with circuit breakers
            result = await self._process(budget, user_id, amount, idempotency_key)
            await self.idempotency.complete(idempotency_key, result)
            return result
        
        except Exception as e:
            await self.idempotency.fail(idempotency_key, str(e))
            raise
    
    async def _process(self, budget, user_id, amount, idempotency_key):
        """Internal processing with all patterns."""
        
        # Fraud check with fallback
        try:
            fraud_result = self.fraud_circuit.call(
                lambda: self._check_fraud(budget, user_id, amount)
            )
        except CircuitOpenError:
            # Use fallback risk score
            fraud_result = {"risk_score": 0.5, "fallback": True}
        
        if fraud_result.get('risk_score', 0) > 0.9:
            return {'status': 'rejected', 'reason': 'high_risk'}
        
        # Bank charge - no fallback, critical
        try:
            bank_result = self.bank_circuit.call(
                lambda: self._charge_bank(budget, user_id, amount, idempotency_key)
            )
        except CircuitOpenError as e:
            return {
                'status': 'unavailable',
                'reason': 'payment_service_unavailable',
                'retry_after': e.retry_after
            }
        
        # Notification - with fallback
        try:
            self.notification_circuit.call(
                lambda: self._send_notification(user_id, amount, bank_result['transaction_id'])
            )
        except CircuitOpenError:
            # Queue for later - not critical
            pass
        
        return {
            'status': 'success',
            'transaction_id': bank_result['transaction_id']
        }