Idempotency in Practice.

import hashlib
import json
import time
import redis
from typing import Optional, Tuple
from dataclasses import dataclass
from datetime import timedelta

@dataclass
class IdempotencyRecord:
    key: str
    request_hash: str  # To detect conflicting requests
    status: str        # 'processing', 'completed', 'failed'
    response: Optional[dict]
    created_at: float

class IdempotencyStore:
    """
    Stores idempotency keys and their results.
    Uses Redis for fast lookups and automatic expiration.
    """
    
    def __init__(self, redis_client: redis.Redis, ttl_hours: int = 24):
        self.redis = redis_client
        self.ttl = timedelta(hours=ttl_hours)
    
    def _make_redis_key(self, idempotency_key: str) -> str:
        return f"idempotency:{idempotency_key}"
    
    def _hash_request(self, request_body: dict) -> str:
        """Create hash of request to detect conflicting retries."""
        serialized = json.dumps(request_body, sort_keys=True)
        return hashlib.sha256(serialized.encode()).hexdigest()[:16]
    
    def check_and_set(
        self,
        idempotency_key: str,
        request_body: dict
    ) -> Tuple[bool, Optional[dict]]:
        """
        Check if key exists. If not, claim it.
        
        Returns:
            (is_new, existing_response)
            - (True, None) if this is a new key
            - (False, response) if key exists with completed response
            - Raises ConflictError if key exists with different request
        """
        redis_key = self._make_redis_key(idempotency_key)
        request_hash = self._hash_request(request_body)
        
        # Try to get existing record
        existing = self.redis.get(redis_key)
        
        if existing:
            record = json.loads(existing)
            
            # Check if request matches
            if record['request_hash'] != request_hash:
                raise IdempotencyConflictError(
                    f"Key {idempotency_key} used with different request"
                )
            
            # If completed, return cached response
            if record['status'] == 'completed':
                return (False, record['response'])
            
            # If still processing, client should wait
            if record['status'] == 'processing':
                raise IdempotencyInProgressError(
                    f"Request {idempotency_key} is still being processed"
                )
            
            # If failed, allow retry
            if record['status'] == 'failed':
                # Update to processing and allow retry
                record['status'] = 'processing'
                self.redis.setex(
                    redis_key,
                    self.ttl,
                    json.dumps(record)
                )
                return (True, None)
        
        # New key - claim it
        record = {
            'key': idempotency_key,
            'request_hash': request_hash,
            'status': 'processing',
            'response': None,
            'created_at': time.time()
        }
        
        # Atomic claim with TTL in one command (SET NX EX). A separate
        # SETNX + EXPIRE could crash in between and leave a key stuck in
        # 'processing' forever.
        claimed = self.redis.set(redis_key, json.dumps(record), nx=True, ex=self.ttl)
        
        if not claimed:
            # Race condition - another request claimed it first
            # Recurse to handle the existing record
            return self.check_and_set(idempotency_key, request_body)
        
        return (True, None)
    
    def complete(self, idempotency_key: str, response: dict):
        """Mark request as completed with response."""
        redis_key = self._make_redis_key(idempotency_key)
        
        existing = self.redis.get(redis_key)
        if not existing:
            raise ValueError(f"No record for key {idempotency_key}")
        
        record = json.loads(existing)
        record['status'] = 'completed'
        record['response'] = response
        
        self.redis.setex(redis_key, self.ttl, json.dumps(record))
    
    def fail(self, idempotency_key: str, error: str):
        """Mark request as failed."""
        redis_key = self._make_redis_key(idempotency_key)
        
        existing = self.redis.get(redis_key)
        if not existing:
            return
        
        record = json.loads(existing)
        record['status'] = 'failed'
        record['response'] = {'error': error}
        
        self.redis.setex(redis_key, self.ttl, json.dumps(record))


class IdempotencyConflictError(Exception):
    """Raised when idempotency key is reused with different request."""
    pass

class IdempotencyInProgressError(Exception):
    """Raised when request is still being processed."""
    pass

from typing import Callable

def with_idempotency(
    idempotency_store: IdempotencyStore,
    idempotency_key: str,
    request_body: dict,
    process_func: Callable[[], dict]
) -> dict:
    """
    Execute function with idempotency protection.
    
    If key was seen before with same request, return cached response.
    If key is new, execute function and cache result.
    """
    
    # Check if we've seen this request
    is_new, cached_response = idempotency_store.check_and_set(
        idempotency_key,
        request_body
    )
    
    if not is_new:
        # Return cached response (this is a retry)
        return cached_response
    
    # Process the request
    try:
        response = process_func()
        idempotency_store.complete(idempotency_key, response)
        return response
    
    except Exception as e:
        idempotency_store.fail(idempotency_key, str(e))
        raise

# Usage in payment handler
@app.post('/payments')
def create_payment(request):
    idempotency_key = request.headers.get('Idempotency-Key')
    
    if not idempotency_key:
        return Response(status=400, body="Idempotency-Key header required")
    
    def process():
        return payment_service.process_payment(
            user_id=request.json['user_id'],
            amount=request.json['amount']
        )
    
    try:
        result = with_idempotency(
            idempotency_store,
            idempotency_key,
            request.json,
            process
        )
        return Response(status=200, body=result)
    
    except IdempotencyConflictError:
        return Response(status=422, body="Idempotency key reused with different request")
    
    except IdempotencyInProgressError:
        return Response(status=409, body="Request still processing, please wait")

# Client code
import uuid

def pay(amount: float):
    idempotency_key = f"pay_{uuid.uuid4()}"
    
    response = requests.post(
        '/payments',
        headers={'Idempotency-Key': idempotency_key},
        json={'amount': amount}
    )
    
    if response.status_code in [500, 502, 503, 504] or response is timeout:
        # Safe to retry with same key
        response = requests.post(
            '/payments',
            headers={'Idempotency-Key': idempotency_key},  # Same key!
            json={'amount': amount}
        )
    
    return response

# Step 1: Client requests a payment intent
response = requests.post('/payment-intents', json={'amount': 99.00})
payment_intent_id = response.json()['id']  # "pi_abc123"

# Step 2: Client confirms payment (can retry safely)
response = requests.post(
    f'/payment-intents/{payment_intent_id}/confirm',
    json={'payment_method': 'card_xyz'}
)

# If step 2 times out, client can retry with same payment_intent_id
# Server knows whether confirmation already happened

import hashlib

def generate_idempotency_key(user_id: str, order_id: str, amount: float) -> str:
    """Generate deterministic key from request parameters."""
    content = f"{user_id}:{order_id}:{amount}"
    return hashlib.sha256(content.encode()).hexdigest()[:32]

# Same order always generates same key
key = generate_idempotency_key("user_123", "order_456", 99.00)
# → "a7b3c9d2e1f0..."

from datetime import datetime

def generate_idempotency_key(user_id: str, order_id: str, amount: float) -> str:
    # Include hour to allow same purchase later
    # (Caveat: a retry that crosses the hour boundary gets a new key)
    hour = datetime.now().strftime("%Y%m%d%H")
    content = f"{user_id}:{order_id}:{amount}:{hour}"
    return hashlib.sha256(content.encode()).hexdigest()[:32]

# Conservative recommendations

IDEMPOTENCY_TTL = {
    # For synchronous APIs (user waiting)
    'payment': timedelta(hours=24),
    
    # For async operations (processed later)
    'batch_job': timedelta(days=7),
    
    # For webhooks (external systems retry for days)
    'webhook': timedelta(days=3),
    
    # For internal services (known retry behavior)
    'internal_rpc': timedelta(hours=1),
}

@dataclass
class IdempotencyRecord:
    # Required fields
    key: str                    # The idempotency key
    status: str                 # processing, completed, failed
    response: Optional[dict]    # Cached response to return
    created_at: datetime        # When first seen
    
    # Recommended fields
    request_hash: str           # To detect conflicting requests
    completed_at: Optional[datetime]  # When completed
    
    # Optional but useful
    user_id: Optional[str]      # Who made the request
    request_path: str           # Which endpoint
    request_body: dict          # Original request (for debugging)
    processing_time_ms: int     # How long it took

async def _charge_bank(self, budget, user_id, amount):
    try:
        response = await client.post(BANK_URL, json={...}, timeout=timeout)
        return PaymentResult(status=SUCCESS, transaction_id=response['id'])
    
    except TimeoutError:
        # ← This is the problem!
        # Did the bank charge the user or not?
        return PaymentResult(
            status=ERROR,
            error_message='Payment processing timed out'
        )

class PaymentServiceWithIdempotency:
    """
    Payment service from Day 1, now with idempotency.
    """
    
    def __init__(self, config: PaymentConfig, idempotency_store: IdempotencyStore):
        self.config = config
        self.idempotency = idempotency_store
        self.logger = logging.getLogger('payment_service')
    
    async def process_payment(
        self,
        user_id: str,
        amount: float,
        idempotency_key: str
    ) -> PaymentResult:
        """
        Process payment with idempotency protection.
        Safe to call multiple times with same idempotency_key.
        """
        
        request_body = {'user_id': user_id, 'amount': amount}
        
        # Step 1: Check idempotency
        try:
            is_new, cached = self.idempotency.check_and_set(
                idempotency_key,
                request_body
            )
            
            if not is_new:
                self.logger.info(f"Returning cached result for {idempotency_key}")
                return PaymentResult(**cached)
        
        except IdempotencyInProgressError:
            # Previous request still processing
            # Could be: our timeout fired, but bank is still working
            return await self._handle_in_progress(idempotency_key, request_body)
        
        except IdempotencyConflictError as e:
            return PaymentResult(status=PaymentStatus.ERROR, error_message=str(e))
        
        # Step 2: Process payment (now safe to proceed)
        return await self._run_and_record(user_id, amount, idempotency_key)
    
    async def _run_and_record(
        self,
        user_id: str,
        amount: float,
        idempotency_key: str
    ) -> PaymentResult:
        """Run the payment for a key this call has claimed, then record the outcome."""
        try:
            result = await self._do_payment(user_id, amount, idempotency_key)
            
            # Step 3: Record the outcome. Definite outcomes (success, rejection)
            # are cached; errors are marked failed so a retry can try again
            # (safe, because the bank deduplicates on the same key).
            if result.status == PaymentStatus.ERROR:
                self.idempotency.fail(idempotency_key, result.error_message or 'error')
            else:
                self.idempotency.complete(idempotency_key, result.__dict__)
            return result
        
        except Exception as e:
            # Step 3: Record failure
            self.idempotency.fail(idempotency_key, str(e))
            raise
    
    async def _do_payment(
        self,
        user_id: str,
        amount: float,
        idempotency_key: str
    ) -> PaymentResult:
        """
        Actual payment processing (from Day 1).
        """
        budget = TimeoutBudget(self.config.total_budget_ms)
        
        # Fraud check
        fraud_result = await self._check_fraud(budget, user_id, amount)
        if fraud_result.status != PaymentStatus.SUCCESS:
            return fraud_result
        
        # Bank charge (pass idempotency key to bank!)
        bank_result = await self._charge_bank(
            budget, user_id, amount, idempotency_key
        )
        
        if bank_result.status == PaymentStatus.SUCCESS:
            # Non-critical notification
            asyncio.create_task(
                self._send_notification(user_id, amount, bank_result.transaction_id)
            )
        
        return bank_result
    
    async def _charge_bank(
        self,
        budget: TimeoutBudget,
        user_id: str,
        amount: float,
        idempotency_key: str
    ) -> PaymentResult:
        """
        Charge bank with idempotency.
        Even if we timeout, we can recover.
        """
        config = self.config.bank_api
        
        try:
            timeout = budget.get_timeout(config.read_timeout * 1000)
            
            # Pass our idempotency key to bank
            # Bank will also deduplicate on their end
            response = await self.http_client.post(
                f"{config.url}/charge",
                headers={'Idempotency-Key': idempotency_key},
                json={'user_id': user_id, 'amount': amount},
                timeout=timeout
            )
            
            return PaymentResult(
                status=PaymentStatus.SUCCESS,
                transaction_id=response.json()['transaction_id']
            )
        
        except TimeoutError:
            # This is where Day 1 left us stuck!
            # Now we can handle it:
            return await self._recover_from_timeout(idempotency_key, user_id, amount)
    
    async def _recover_from_timeout(
        self,
        idempotency_key: str,
        user_id: str,
        amount: float
    ) -> PaymentResult:
        """
        Handle bank timeout by checking transaction status.
        """
        self.logger.warning(f"Bank timeout for {idempotency_key}, checking status")
        
        try:
            # Ask bank: "Did this transaction complete?"
            # (Assumes a hypothetical bank API with lookup by key. Real processors
            # such as Stripe have no such endpoint: you replay the same request
            # with the same Idempotency-Key and get the saved response.)
            status_response = await self.http_client.get(
                f"{self.config.bank_api.url}/transactions",
                params={'idempotency_key': idempotency_key},
                timeout=5.0  # Short timeout for status check
            )
            
            data = status_response.json()
            
            if data.get('found'):
                # Transaction exists at bank
                if data['status'] == 'completed':
                    return PaymentResult(
                        status=PaymentStatus.SUCCESS,
                        transaction_id=data['transaction_id']
                    )
                elif data['status'] == 'failed':
                    return PaymentResult(
                        status=PaymentStatus.ERROR,
                        error_message=data.get('error', 'Bank declined')
                    )
                else:
                    # Still processing
                    return PaymentResult(
                        status=PaymentStatus.ERROR,
                        error_message='Payment is still processing. Please check back shortly.'
                    )
            else:
                # Transaction not found - request never reached the bank (or
                # hasn't arrived yet). Safe to allow a retry only because the
                # retry reuses the same key, which the bank deduplicates.
                return PaymentResult(
                    status=PaymentStatus.ERROR,
                    error_message='Payment could not be processed. Please try again.'
                )
        
        except Exception as e:
            # Can't even check status - be honest with user
            self.logger.error(f"Status check failed: {e}")
            return PaymentResult(
                status=PaymentStatus.ERROR,
                error_message='Unable to confirm payment status. Please check your statement before retrying.'
            )
    
    async def _handle_in_progress(
        self,
        idempotency_key: str,
        request_body: dict
    ) -> PaymentResult:
        """
        Handle case where previous request is still processing.
        This happens when our timeout fired but bank is still working.
        """
        self.logger.info(f"Request {idempotency_key} in progress, waiting")
        
        # Poll for completion
        for _ in range(10):  # Try for 10 seconds
            await asyncio.sleep(1)
            
            try:
                is_new, cached = self.idempotency.check_and_set(
                    idempotency_key,
                    request_body
                )
            except IdempotencyInProgressError:
                continue  # Still processing - keep waiting
            
            if not is_new:
                return PaymentResult(**cached)
            
            # The earlier attempt failed and this call has now claimed the key:
            # run the payment here (the bank deduplicates on the same key)
            return await self._run_and_record(
                request_body['user_id'], request_body['amount'], idempotency_key
            )
        
        # Still processing after 10 seconds
        return PaymentResult(
            status=PaymentStatus.ERROR,
            error_message='Payment is taking longer than expected. Please check back shortly.'
        )

# Dangerous: Same key, different amount

# Request 1
POST /payments
Idempotency-Key: key_abc
{amount: 100}

# Request 2 (bug or malicious)
POST /payments
Idempotency-Key: key_abc
{amount: 1000000}  # Much larger!

# What happens?

def check_and_set(self, key: str, request_body: dict):
    request_hash = hash_request(request_body)
    
    existing = self.get(key)
    if existing:
        if existing.request_hash != request_hash:
            raise IdempotencyConflictError(
                "Key reused with different request parameters"
            )
        # ... rest of logic

def complete(self, key: str, response: dict):
    record = self.get(key)
    record.status = 'completed'
    record.response = response
    
    # Extend TTL to ensure long-lived record for completed transactions
    extended_ttl = max(self.ttl, timedelta(hours=24))
    self.store(key, record, ttl=extended_ttl)

def check_and_set(self, key: str, request_body: dict):
    existing = self.get(key)
    
    if existing:
        if existing.status == 'completed':
            # Definitely don't retry - return cached success
            return (False, existing.response)
        
        elif existing.status == 'processing':
            # Still running - make caller wait
            raise IdempotencyInProgressError()
        
        elif existing.status == 'failed':
            # Previous attempt failed - allow retry
            existing.status = 'processing'
            self.store(key, existing)
            return (True, None)
    
    # New key
    self.create(key, request_body)
    return (True, None)

async def process_payment(self, ...):
    # Begin transaction (only works if both steps are writes in this same
    # database, e.g. an internal ledger; a DB rollback can't undo an
    # external bank charge)
    async with self.db.transaction():
        # Charge customer
        charge_result = await self.charge_customer(...)
        
        # Credit merchant
        credit_result = await self.credit_merchant(...)
        
        # Both succeed or both fail (atomic)

async def process_payment(self, ...):
    charge_result = await self.charge_customer(...)
    
    try:
        credit_result = await self.credit_merchant(...)
    except:
        # Compensating action
        await self.refund_customer(charge_result.id)
        raise

# Good key formats:

# UUID (random, unique)
"550e8400-e29b-41d4-a716-446655440000"

# Prefixed UUID (self-documenting)
"pay_550e8400-e29b-41d4-a716-446655440000"

# Entity-based (deterministic)
"user_123_order_456_v1"

# Timestamp-bucketed (allows intentional retries after window)
"user_123_order_456_2024010112"  # Includes hour

# Bad key formats:

# Sequential (predictable, vulnerable)
"1", "2", "3"

# User-controlled (can be malicious)
"../../../etc/passwd"

# Too short (collision risk)
"abc"

# Timestamp only (same user, same millisecond = collision)
"1704067200000"

import uuid
import hashlib
from datetime import datetime, timezone

class IdempotencyKeyGenerator:
    """Generate idempotency keys with various strategies."""
    
    @staticmethod
    def random() -> str:
        """Random UUID - use when client controls retry."""
        return f"key_{uuid.uuid4()}"
    
    @staticmethod
    def deterministic(user_id: str, action: str, entity_id: str) -> str:
        """Deterministic key - same input = same key."""
        content = f"{user_id}:{action}:{entity_id}"
        return hashlib.sha256(content.encode()).hexdigest()[:32]
    
    @staticmethod
    def time_bucketed(
        user_id: str,
        action: str,
        entity_id: str,
        bucket_minutes: int = 60
    ) -> str:
        """Allow same action after time window."""
        # Bucket number since the epoch, so any bucket size works (60 = hourly, 1440 = daily)
        now = datetime.now(timezone.utc)
        bucket = int(now.timestamp() // (bucket_minutes * 60))
        content = f"{user_id}:{action}:{entity_id}:{bucket}"
        return hashlib.sha256(content.encode()).hexdigest()[:32]
    
    @staticmethod
    def composite(prefix: str, *parts: str) -> str:
        """Readable composite key."""
        safe_parts = [p.replace('_', '-') for p in parts]
        return f"{prefix}_{'_'.join(safe_parts)}"

# Usage examples:

# For user-initiated payment (use random)
key = IdempotencyKeyGenerator.random()
# → "key_550e8400-e29b-41d4-a716-446655440000"

# For order payment (use deterministic)
key = IdempotencyKeyGenerator.deterministic("user_123", "pay", "order_456")
# → "a1b2c3d4e5f6..."

# For daily charge (use time bucketed - daily)
key = IdempotencyKeyGenerator.time_bucketed("user_123", "subscription", "sub_789", 1440)
# → "x1y2z3..." (changes each day)

# For webhook (use composite for readability)
key = IdempotencyKeyGenerator.composite("webhook", "order_created", "order_456")
# → "webhook_order-created_order-456"

import re

class IdempotencyKeyValidator:
    """Validate idempotency keys."""
    
    MIN_LENGTH = 16
    MAX_LENGTH = 256
    PATTERN = re.compile(r'^[a-zA-Z0-9_\-]+$')
    
    @classmethod
    def validate(cls, key: str) -> tuple[bool, str]:
        """
        Validate key format.
        Returns (is_valid, error_message).
        """
        if not key:
            return False, "Idempotency key is required"
        
        if len(key) < cls.MIN_LENGTH:
            return False, f"Key too short (min {cls.MIN_LENGTH} chars)"
        
        if len(key) > cls.MAX_LENGTH:
            return False, f"Key too long (max {cls.MAX_LENGTH} chars)"
        
        if not cls.PATTERN.match(key):
            return False, "Key contains invalid characters"
        
        return True, ""

# Usage in middleware (FastAPI/Starlette: needs the "http" argument, async, and await call_next)
from fastapi import Response

@app.middleware("http")
async def validate_idempotency_key(request, call_next):
    key = request.headers.get('Idempotency-Key')
    
    if request.method == 'POST':  # Only required for non-idempotent methods
        if not key:
            return Response(status_code=400, content="Idempotency-Key header required")
        
        is_valid, error = IdempotencyKeyValidator.validate(key)
        if not is_valid:
            return Response(status_code=400, content=error)
    
    return await call_next(request)

# Payment Idempotency Strategy

## Overview

All payment operations MUST be idempotent. Users can safely retry any
payment operation without risk of duplicate charges.

## Idempotency Key Format

### Client-Generated Keys (API Consumers)
- Format: `pay_<uuid4>`
- Example: `pay_550e8400-e29b-41d4-a716-446655440000`
- Required header: `Idempotency-Key: <key>`
- Validation: 16-256 alphanumeric characters plus `-` and `_`

### Server-Generated Keys (Internal)
- Format: `<action>_<entity>_<hash>`
- Example: `charge_order_abc123_v1`
- Generated from: user_id + order_id + amount

## Storage

- Primary: PostgreSQL `idempotency_keys` table
- Cache: Redis with 1-hour TTL
- Record TTL: 7 days for completed, 24 hours for failed

## Deduplication Window

| Scenario | Window |
|----------|--------|
| User retry (network timeout) | Immediate |
| Accidental double-click | Immediate |
| Page refresh | 1 hour |
| Intentional re-purchase | After 24 hours |

## Edge Cases

### 1. Same Key, Different Request
- Detection: Hash request body, compare on lookup
- Response: HTTP 422 "Idempotency key reused with different parameters"

### 2. Request In Progress
- Detection: Record exists with status='processing'
- Response: HTTP 409 "Request is being processed"
- Client action: Retry with exponential backoff

### 3. Previous Request Failed
- Detection: Record exists with status='failed'
- Response: Allow new attempt (reset status to 'processing')

### 4. Key Expired
- Detection: Record not found (TTL expired)
- Response: Process as new request
- Risk: Potential duplicate if original eventually completes
- Mitigation: Check bank API for existing transaction

## Reconciliation

Daily job to detect inconsistencies:
1. Find payments where our record differs from bank
2. Find orphaned transactions (bank charged, we have no record)
3. Alert on any discrepancies for manual review

## Monitoring

### Alerts
- Duplicate attempt rate > 5% (user experience issue)
- Conflict rate > 0.1% (potential bug or attack)
- Orphaned transaction detected (immediate page)

### Metrics
- `idempotency_cache_hit_rate` - Should be > 95%
- `idempotency_duplicate_prevented_total` - Growing is good
- `idempotency_conflict_total` - Should be near zero

"""
Production-ready idempotency implementation.
Builds on Day 1's payment service, adding retry safety.
"""

import hashlib
import json
import time
import asyncio
import logging
from dataclasses import dataclass, field, asdict
from typing import Optional, Dict, Any, Callable, Awaitable
from datetime import datetime, timedelta, timezone
from enum import Enum
import redis.asyncio as redis
import asyncpg
from prometheus_client import Counter, Histogram

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

idempotency_checks = Counter(
    'idempotency_checks_total',
    'Idempotency check results',
    ['result']  # new, cached, conflict, in_progress
)

idempotency_latency = Histogram(
    'idempotency_check_seconds',
    'Idempotency check latency',
    buckets=[.001, .005, .01, .025, .05, .1, .25, .5]
)

# =============================================================================
# Types
# =============================================================================

class IdempotencyStatus(Enum):
    PROCESSING = 'processing'
    COMPLETED = 'completed'
    FAILED = 'failed'

@dataclass
class IdempotencyRecord:
    key: str
    request_hash: str
    status: IdempotencyStatus
    response: Optional[Dict] = None
    http_status: Optional[int] = None
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    completed_at: Optional[datetime] = None
    user_id: Optional[str] = None
    
    def to_dict(self) -> Dict:
        return {
            'key': self.key,
            'request_hash': self.request_hash,
            'status': self.status.value,
            'response': self.response,
            'http_status': self.http_status,
            'created_at': self.created_at.isoformat(),
            'completed_at': self.completed_at.isoformat() if self.completed_at else None,
            'user_id': self.user_id,
        }
    
    @classmethod
    def from_dict(cls, data: Dict) -> 'IdempotencyRecord':
        return cls(
            key=data['key'],
            request_hash=data['request_hash'],
            status=IdempotencyStatus(data['status']),
            response=data.get('response'),
            http_status=data.get('http_status'),
            created_at=datetime.fromisoformat(data['created_at']),
            completed_at=datetime.fromisoformat(data['completed_at']) if data.get('completed_at') else None,
            user_id=data.get('user_id'),
        )

# =============================================================================
# Exceptions
# =============================================================================

class IdempotencyError(Exception):
    pass

class IdempotencyConflictError(IdempotencyError):
    """Key reused with different request."""
    pass

class IdempotencyInProgressError(IdempotencyError):
    """Previous request still processing."""
    pass

# =============================================================================
# Main Implementation
# =============================================================================

class IdempotencyStore:
    """
    Production idempotency store with Redis cache and PostgreSQL persistence.
    
    Features:
    - Fast path through Redis for cache hits
    - PostgreSQL for durability and recovery
    - Atomic claim using database locks
    - Automatic TTL management
    """
    
    def __init__(
        self,
        redis_client: redis.Redis,
        pg_pool: asyncpg.Pool,
        cache_ttl: timedelta = timedelta(hours=1),
        record_ttl: timedelta = timedelta(days=7),
    ):
        self.redis = redis_client
        self.pg = pg_pool
        self.cache_ttl = cache_ttl
        self.record_ttl = record_ttl
        self.logger = logging.getLogger('idempotency')
    
    def _hash_request(self, request_body: Dict) -> str:
        """Create deterministic hash of request."""
        serialized = json.dumps(request_body, sort_keys=True, default=str)
        return hashlib.sha256(serialized.encode()).hexdigest()
    
    def _redis_key(self, key: str) -> str:
        return f"idempotency:{key}"
    
    async def check_and_claim(
        self,
        key: str,
        request_body: Dict,
        user_id: Optional[str] = None
    ) -> tuple[bool, Optional[IdempotencyRecord]]:
        """
        Check if key exists and claim it if not.
        
        Returns:
            (is_new, record)
            - (True, None): Key is new, claimed for processing
            - (False, record): Key exists, return cached record
            
        Raises:
            IdempotencyConflictError: Key exists with different request
            IdempotencyInProgressError: Previous request still processing
        """
        start = time.time()
        request_hash = self._hash_request(request_body)
        
        try:
            # Fast path: check Redis cache
            cached = await self._check_cache(key, request_hash)
            if cached:
                idempotency_checks.labels(result='cached').inc()
                return (False, cached)
            
            # Slow path: check and claim in database
            return await self._check_and_claim_db(key, request_hash, user_id)
        
        finally:
            idempotency_latency.observe(time.time() - start)
    
    async def _check_cache(self, key: str, request_hash: str) -> Optional[IdempotencyRecord]:
        """Check Redis cache for completed request."""
        redis_key = self._redis_key(key)
        cached = await self.redis.get(redis_key)
        
        if not cached:
            return None
        
        record = IdempotencyRecord.from_dict(json.loads(cached))
        
        # Validate request hash
        if record.request_hash != request_hash:
            raise IdempotencyConflictError(
                f"Key '{key}' was used with a different request"
            )
        
        if record.status == IdempotencyStatus.COMPLETED:
            return record
        
        if record.status == IdempotencyStatus.PROCESSING:
            raise IdempotencyInProgressError(
                f"Request '{key}' is still being processed"
            )
        
        # Failed - allow retry (will go to DB path)
        return None
    
    async def _check_and_claim_db(
        self,
        key: str,
        request_hash: str,
        user_id: Optional[str]
    ) -> tuple[bool, Optional[IdempotencyRecord]]:
        """Check and claim in database with locking."""
        
        async with self.pg.acquire() as conn:
            async with conn.transaction():
                # Try to get existing record with lock
                existing = await conn.fetchrow("""
                    SELECT * FROM idempotency_records
                    WHERE key = $1
                    FOR UPDATE
                """, key)
                
                if existing:
                    record = self._row_to_record(existing)
                    
                    # Validate request hash
                    if record.request_hash != request_hash:
                        idempotency_checks.labels(result='conflict').inc()
                        raise IdempotencyConflictError(
                            f"Key '{key}' was used with a different request"
                        )
                    
                    if record.status == IdempotencyStatus.COMPLETED:
                        # Cache for future fast-path
                        await self._cache_record(record)
                        idempotency_checks.labels(result='cached').inc()
                        return (False, record)
                    
                    if record.status == IdempotencyStatus.PROCESSING:
                        # Check if stuck (processing for too long)
                        if self._is_stuck(record):
                            # Allow retry - reset to processing
                            await self._reset_record(conn, key)
                            idempotency_checks.labels(result='new').inc()
                            return (True, None)
                        
                        idempotency_checks.labels(result='in_progress').inc()
                        raise IdempotencyInProgressError(
                            f"Request '{key}' is still being processed"
                        )
                    
                    # Failed - allow retry
                    await conn.execute("""
                        UPDATE idempotency_records
                        SET status = 'processing', completed_at = NULL
                        WHERE key = $1
                    """, key)
                    idempotency_checks.labels(result='new').inc()
                    return (True, None)
                
                # New key - insert (ON CONFLICT: FOR UPDATE above can't lock a
                # row that doesn't exist yet, so a concurrent request may win)
                result = await conn.execute("""
                    INSERT INTO idempotency_records
                    (key, request_hash, status, created_at, user_id, expires_at)
                    VALUES ($1, $2, 'processing', $3, $4, $5)
                    ON CONFLICT (key) DO NOTHING
                """, key, request_hash, datetime.now(timezone.utc), user_id,
                     datetime.now(timezone.utc) + self.record_ttl)
                
                if result == 'INSERT 0 0':
                    idempotency_checks.labels(result='in_progress').inc()
                    raise IdempotencyInProgressError(
                        f"Request '{key}' is being claimed by a concurrent request"
                    )
                
                idempotency_checks.labels(result='new').inc()
                return (True, None)
    
    def _is_stuck(self, record: IdempotencyRecord) -> bool:
        """Check if a processing record is stuck (> 5 minutes)."""
        stuck_threshold = timedelta(minutes=5)
        return datetime.now(timezone.utc) - record.created_at > stuck_threshold
    
    async def _reset_record(self, conn, key: str):
        """Reset a stuck record for retry."""
        await conn.execute("""
            UPDATE idempotency_records
            SET status = 'processing',
                created_at = $1,
                completed_at = NULL
            WHERE key = $2
        """, datetime.now(timezone.utc), key)
    
    async def complete(
        self,
        key: str,
        response: Dict,
        http_status: int = 200
    ):
        """Mark request as completed with response."""
        now = datetime.now(timezone.utc)
        
        # Update database
        async with self.pg.acquire() as conn:
            await conn.execute("""
                UPDATE idempotency_records
                SET status = 'completed',
                    response = $1,
                    http_status = $2,
                    completed_at = $3
                WHERE key = $4
            """, json.dumps(response), http_status, now, key)
            
            # Get full record for caching
            row = await conn.fetchrow(
                "SELECT * FROM idempotency_records WHERE key = $1", key
            )
        
        # Update cache
        if row:
            record = self._row_to_record(row)
            await self._cache_record(record)
    
    async def fail(self, key: str, error: str):
        """Mark request as failed."""
        async with self.pg.acquire() as conn:
            await conn.execute("""
                UPDATE idempotency_records
                SET status = 'failed',
                    response = $1,
                    completed_at = $2
                WHERE key = $3
            """, json.dumps({'error': error}), datetime.now(timezone.utc), key)
        
        # Remove from cache (allow retry)
        await self.redis.delete(self._redis_key(key))
    
    async def _cache_record(self, record: IdempotencyRecord):
        """Cache completed record in Redis."""
        redis_key = self._redis_key(record.key)
        await self.redis.setex(
            redis_key,
            int(self.cache_ttl.total_seconds()),
            json.dumps(record.to_dict())
        )
    
    def _row_to_record(self, row) -> IdempotencyRecord:
        """Convert database row to record."""
        return IdempotencyRecord(
            key=row['key'],
            request_hash=row['request_hash'],
            status=IdempotencyStatus(row['status']),
            response=json.loads(row['response']) if row['response'] else None,
            http_status=row['http_status'],
            created_at=row['created_at'],
            completed_at=row['completed_at'],
            user_id=row['user_id'],
        )


# =============================================================================
# High-Level Wrapper
# =============================================================================

async def with_idempotency(
    store: IdempotencyStore,
    key: str,
    request_body: Dict,
    processor: Callable[[], Awaitable[Dict]],
    user_id: Optional[str] = None
) -> tuple[Dict, int]:
    """
    Execute processor with idempotency protection.
    
    Returns: (response_body, http_status)
    """
    # Check/claim
    is_new, record = await store.check_and_claim(key, request_body, user_id)
    
    if not is_new:
        # Return cached response
        return (record.response, record.http_status)
    
    # Process
    try:
        response = await processor()
        await store.complete(key, response, http_status=200)
        return (response, 200)
    
    except Exception as e:
        await store.fail(key, str(e))
        raise


# =============================================================================
# Database Schema
# =============================================================================

SCHEMA = """
CREATE TABLE IF NOT EXISTS idempotency_records (
    key VARCHAR(256) PRIMARY KEY,
    request_hash CHAR(64) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'processing',
    response JSONB,
    http_status INTEGER,
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    completed_at TIMESTAMP,
    user_id VARCHAR(256),
    expires_at TIMESTAMP NOT NULL,
    
    CHECK (status IN ('processing', 'completed', 'failed'))
);

CREATE INDEX IF NOT EXISTS idx_idempotency_expires 
ON idempotency_records(expires_at);

CREATE INDEX IF NOT EXISTS idx_idempotency_user 
ON idempotency_records(user_id);

-- Cleanup job (run daily)
-- DELETE FROM idempotency_records WHERE expires_at < NOW();
"""


# =============================================================================
# Example Usage with Payment Service
# =============================================================================

async def example_usage():
    """Example of using idempotency with payment service."""
    
    # Setup
    redis_client = redis.from_url('redis://localhost')
    pg_pool = await asyncpg.create_pool('postgresql://localhost/payments')
    
    store = IdempotencyStore(redis_client, pg_pool)
    payment_service = PaymentService(...)  # From Day 1
    
    # Handle payment request
    async def handle_payment(request):
        key = request.headers.get('Idempotency-Key')
        
        if not key:
            return Response(status=400, body={'error': 'Idempotency-Key required'})
        
        async def process():
            result = await payment_service.process_payment(
                user_id=request.json['user_id'],
                amount=request.json['amount'],
                idempotency_key=key
            )
            return result.to_dict()
        
        try:
            response, status = await with_idempotency(
                store=store,
                key=key,
                request_body=request.json,
                processor=process,
                user_id=request.json.get('user_id')
            )
            return Response(status=status, body=response)
        
        except IdempotencyConflictError:
            return Response(status=422, body={
                'error': 'Idempotency key was used with different request parameters'
            })
        
        except IdempotencyInProgressError:
            return Response(status=409, body={
                'error': 'Request is still being processed. Please retry shortly.'
            })