Webhook Delivery System.

def send_webhook_at_most_once(url, payload):
    try:
        response = requests.post(url, json=payload, timeout=5)
        return 200 <= response.status_code < 300
    except:
        return False  # Don't retry, accept the loss

def send_webhook_at_least_once(url, payload, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = requests.post(url, json=payload, timeout=5)
            if 200 <= response.status_code < 300:
                return True  # Success (any 2xx)
        except:
            pass
        
        # Exponential backoff before retry
        time.sleep(2 ** attempt)
    
    return False  # Move to dead letter queue

# Webhook Receiver Requirements

1. **Idempotency**: Your endpoint must handle duplicate deliveries.
   Use the `event_id` field to deduplicate.

2. **Response**: Return a 2xx status code to acknowledge receipt.
   Any other status or timeout will trigger retry.

3. **Speed**: Respond within 30 seconds.
   Do heavy processing asynchronously.

4. **Verification**: Verify webhook signatures.
   Check the X-Webhook-Signature header.

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    # Verify signature first
    if not verify_signature(request):
        abort(401)
    
    event = request.json
    event_id = event['id']
    
    # Idempotency check
    if db.exists(f"processed:{event_id}"):
        return jsonify({"status": "already_processed"}), 200
    
    # Mark as processing (prevent concurrent duplicates)
    if not db.set(f"processing:{event_id}", "1", nx=True, ex=300):
        # Non-2xx so the sender retries later: if the in-flight attempt
        # fails, a 200 here would have acknowledged an event nobody processed
        return jsonify({"status": "processing"}), 409
    
    try:
        # Process the event
        process_event(event)
        
        # Mark as complete
        db.set(f"processed:{event_id}", "1", ex=86400*7)
        db.delete(f"processing:{event_id}")
        
        return jsonify({"status": "processed"}), 200
    
    except Exception as e:
        db.delete(f"processing:{event_id}")
        raise

class OrderService:
    def create_order(self, order_data):
        order = self.db.create(order_data)
        
        # Emit event for webhook system
        self.event_bus.publish(
            event_type="order.created",
            data=order.to_dict()
        )
        
        return order

class WebhookDispatcher:
    async def handle_event(self, event):
        # Find all subscriptions for this event type
        subscriptions = await self.db.find_subscriptions(
            event_type=event['type']
        )
        
        # Create delivery task for each
        for sub in subscriptions:
            if sub.active:
                await self.queue.enqueue({
                    'subscription_id': sub.id,
                    'event': event
                })

def get_partition(subscription_id: str) -> int:
    """Stable hash partitioning (md5 is the same in every process; Python's built-in hash() of a str is randomised per process)."""
    return int(hashlib.md5(subscription_id.encode()).hexdigest(), 16) % NUM_PARTITIONS

class DeliveryWorker:
    async def process_task(self, task):
        subscription = await self.db.get_subscription(task['subscription_id'])
        
        try:
            response = await self.http.post(
                subscription.url,
                json=self.build_payload(task['event']),
                headers=self.sign_headers(subscription, task['event']),
                timeout=30
            )
            
            if 200 <= response.status < 300:
                await self.log_success(task)
            else:
                await self.schedule_retry(task)
        
        except Exception as e:
            await self.schedule_retry(task)

from datetime import datetime, timezone

async def move_to_dlq(self, task, error):
    await self.dlq.push({
        'task': task,
        'error': error,
        'failed_at': datetime.now(timezone.utc),
        'attempts': task['attempt_count']
    })
    
    # Alert operations
    await self.alert_service.notify(
        f"Webhook failed after {task['attempt_count']} attempts",
        subscription_id=task['subscription_id']
    )

{
    "id": "evt_1234567890abcdef",
    "type": "order.created",
    "api_version": "2024-01-15",
    "created_at": "2024-01-15T10:30:00.123Z",
    "data": {
        "object": "order",
        "id": "ord_abc123",
        "amount": 9900,
        "currency": "usd",
        "status": "pending",
        "customer": {
            "id": "cus_xyz",
            "email": "customer@example.com"
        },
        "items": [...]
    },
    "request_id": "req_original_request_id"
}

def should_retry(response_or_error) -> bool:
    if isinstance(response_or_error, Exception):
        # Network errors - always retry
        if isinstance(response_or_error, (Timeout, ConnectionError)):
            return True
        # Unknown errors - retry cautiously
        return True
    
    status = response_or_error.status_code
    
    # Success (any 2xx) - don't retry
    if 200 <= status < 300:
        return False
    
    # Client errors - don't retry (except rate limit)
    if 400 <= status < 500:
        return status == 429  # Retry rate limits
    
    # Server errors - retry
    if status >= 500:
        return True
    
    # Redirects - don't retry
    if 300 <= status < 400:
        return False
    
    return False

def handle_rate_limit(response, task):
    retry_after = response.headers.get('Retry-After')
    
    if retry_after:
        try:
            delay = int(retry_after)  # delay-seconds form
        except ValueError:
            delay = 60  # HTTP-date form (e.g. "Wed, 21 Oct 2015 07:28:00 GMT"): parse it in production
    else:
        delay = 60  # Default backoff
    
    schedule_retry(task, delay=delay)

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Optional
import random

@dataclass
class RetryConfig:
    max_retries: int = 8  # 9 deliveries in all: the first try plus 8 retries
    # Explicit schedule in seconds: 1m, 5m, 15m, 1h, 2h, 4h, 8h, 24h (~39 h in total)
    delays: tuple = (60, 300, 900, 3600, 7200, 14400, 28800, 86400)
    jitter: float = 0.1  # ±10%

class RetryScheduler:
    def __init__(self, config: RetryConfig, queue):
        self.config = config
        self.queue = queue
    
    def calculate_delay(self, attempt: int) -> int:
        """Look up the delay for this retry in the schedule and add jitter."""
        
        # Delays grow roughly 4-5x at first, then double, and end at 24 hours
        delay = self.config.delays[min(attempt, len(self.config.delays) - 1)]
        
        # Add jitter to prevent thundering herd
        jitter_range = delay * self.config.jitter
        delay += random.uniform(-jitter_range, jitter_range)
        
        return int(delay)
    
    async def schedule_retry(
        self,
        task: dict,
        error: str,
        force_delay: Optional[int] = None
    ) -> bool:
        """
        Schedule a retry for failed delivery.
        Returns True if retry scheduled, False if moved to DLQ.
        """
        
        attempt = task.get('attempt', 0)
        
        if attempt >= self.config.max_retries:
            await self.move_to_dlq(task, error)
            return False
        
        delay = force_delay or self.calculate_delay(attempt)
        
        task['attempt'] = attempt + 1
        task['last_error'] = error
        task['retry_at'] = (datetime.now(timezone.utc) + timedelta(seconds=delay)).isoformat()
        
        await self.queue.enqueue_delayed(task, delay)
        
        return True
    
    async def move_to_dlq(self, task: dict, error: str):
        """Move to dead letter queue after exhausting retries."""
        
        dlq_entry = {
            'task': task,
            'final_error': error,
            'attempts': task.get('attempt', 0),
            'first_attempt': task.get('first_attempt'),
            'last_attempt': datetime.now(timezone.utc).isoformat(),
            'status': 'pending_review'
        }
        
        await self.dlq.push(dlq_entry)
        await self.metrics.increment('webhooks_dlq_total')
        await self.alert_ops(task, error)

class AdaptiveRetryScheduler:
    def __init__(self):
        self.endpoint_health = {}  # endpoint -> health status
    
    async def check_endpoint_health(self, endpoint: str) -> bool:
        """Periodic health check for failing endpoints."""
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.head(endpoint, timeout=5, allow_redirects=False) as resp:
                    return resp.status < 500
        except:
            return False
    
    async def on_endpoint_recovery(self, endpoint: str):
        """When endpoint recovers, accelerate pending retries."""
        
        # Find all pending retries for this endpoint
        pending = await self.queue.get_pending_for_endpoint(endpoint)
        
        # Move them to immediate queue
        for task in pending:
            await self.queue.move_to_immediate(task)
        
        logger.info(f"Accelerated {len(pending)} retries for {endpoint}")
    
    async def run_health_checker(self):
        """Background task to check unhealthy endpoints."""
        
        while True:
            for endpoint, status in self.endpoint_health.items():
                if not status.healthy:
                    is_healthy = await self.check_endpoint_health(endpoint)
                    
                    if is_healthy:
                        logger.info(f"Endpoint recovered: {endpoint}")
                        await self.on_endpoint_recovery(endpoint)
                        status.healthy = True
            
            await asyncio.sleep(30)  # Check every 30 seconds

import hmac
import hashlib
import time
import json

class WebhookSigner:
    def sign_payload(self, payload: dict, secret: str) -> dict:
        """Generate signature headers for webhook."""
        
        timestamp = int(time.time())
        
        # Create signed content: timestamp.payload
        payload_str = json.dumps(payload, separators=(',', ':'), sort_keys=True)
        signed_content = f"{timestamp}.{payload_str}"
        
        # Generate HMAC-SHA256 signature
        signature = hmac.new(
            secret.encode('utf-8'),
            signed_content.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        
        return {
            'X-Webhook-Timestamp': str(timestamp),
            'X-Webhook-Signature': f"sha256={signature}",
            'Content-Type': 'application/json'
        }

class WebhookVerifier:
    def verify(
        self,
        payload_body: str,
        signature_header: str,
        timestamp_header: str,
        secret: str,
        tolerance: int = 300  # 5 minutes
    ) -> bool:
        """Verify webhook signature."""
        
        # Check timestamp is recent (prevent replay attacks)
        try:
            timestamp = int(timestamp_header)
        except (ValueError, TypeError):
            return False
        
        if abs(time.time() - timestamp) > tolerance:
            return False  # Too old or too far in future
        
        # Verify signature
        if not signature_header.startswith('sha256='):
            return False
        
        received_sig = signature_header[7:]
        
        signed_content = f"{timestamp}.{payload_body}"
        expected_sig = hmac.new(
            secret.encode('utf-8'),
            signed_content.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        
        # Constant-time comparison (prevent timing attacks)
        return hmac.compare_digest(received_sig, expected_sig)

from datetime import datetime, timezone

class SecretManager:
    def rotate_secret(self, subscription_id: str) -> dict:
        """Rotate webhook secret with grace period."""
        
        subscription = self.db.get(subscription_id)
        new_secret = secrets.token_hex(32)
        
        # Keep old secret valid for 24 hours
        self.db.update(subscription_id, {
            'secret': new_secret,
            'previous_secret': subscription.secret,
            'previous_secret_expires': datetime.now(timezone.utc) + timedelta(hours=24)
        })
        
        return {
            'new_secret': new_secret,
            'old_secret_valid_until': (datetime.now(timezone.utc) + timedelta(hours=24)).isoformat()
        }
    
    def verify_with_rotation(
        self,
        subscription_id: str,
        payload: str,
        signature: str,
        timestamp: str
    ) -> bool:
        """Verify allowing both current and previous secrets."""
        
        subscription = self.db.get(subscription_id)
        
        # Try current secret
        if self.verifier.verify(payload, signature, timestamp, subscription.secret):
            return True
        
        # Try previous secret if still valid
        if (subscription.previous_secret and 
            subscription.previous_secret_expires > datetime.now(timezone.utc)):
            if self.verifier.verify(payload, signature, timestamp, subscription.previous_secret):
                return True
        
        return False

from datetime import datetime, timezone

class WebhookSecurityManager:
    def validate_endpoint(self, url: str) -> bool:
        """Validate webhook URL before registration."""
        
        parsed = urllib.parse.urlparse(url)
        
        # Require HTTPS
        if parsed.scheme != 'https':
            raise ValueError("Webhook URL must use HTTPS")
        
        # Block private IPs (SSRF prevention)
        try:
            ip = socket.gethostbyname(parsed.hostname)
            # is_global rejects private, loopback, link-local and other reserved ranges
            if not ipaddress.ip_address(ip).is_global:
                raise ValueError("Webhook URL cannot point to a private or reserved IP")
        except socket.gaierror:
            raise ValueError("Cannot resolve webhook hostname")
        
        # Block localhost
        if parsed.hostname in ['localhost', '127.0.0.1', '0.0.0.0']:
            raise ValueError("Webhook URL cannot be localhost")
        
        return True
    
    def test_endpoint(self, url: str, secret: str) -> bool:
        """Send test webhook to verify endpoint works."""
        
        test_event = {
            'id': f'evt_test_{uuid.uuid4().hex[:8]}',
            'type': 'webhook.test',
            'created_at': datetime.now(timezone.utc).isoformat()
        }
        
        try:
            response = requests.post(
                url,
                # Send the exact bytes that were signed
                data=json.dumps(test_event, separators=(',', ':'), sort_keys=True),
                headers=self.signer.sign_payload(test_event, secret),
                timeout=10,
                allow_redirects=False  # a redirect could point at an internal address
            )
            return 200 <= response.status_code < 300
        except:
            return False

# Assumptions:
# - Average HTTP request takes 500ms
# - We want to handle 2,500 requests/second at peak
# - Each worker can handle 1/0.5 = 2 requests/second

# Required workers = 2,500 / 2 = 1,250 workers

# With async workers (10 concurrent requests each):
# Required workers = 2,500 / 20 = 125 workers

# With safety margin (2x):
# Deploy 250 async workers

class WebhookHTTPClient:
    def __init__(self):
        # Connection pool settings
        self.connector = aiohttp.TCPConnector(
            limit=1000,           # Total connections
            limit_per_host=10,    # Per-endpoint limit (fairness)
            ttl_dns_cache=300,    # DNS cache TTL
            keepalive_timeout=30  # Keep connections alive
        )
        
        self.timeout = aiohttp.ClientTimeout(
            total=30,      # Total request timeout
            connect=5,     # Connection timeout
            sock_read=25   # Read timeout
        )
        
        self.session = aiohttp.ClientSession(
            connector=self.connector,
            timeout=self.timeout
        )

class EndpointRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.default_limit = 800  # requests per minute (~13/second)
    
    async def acquire(self, endpoint: str, limit: int = None) -> bool:
        """Try to acquire a rate limit slot."""
        
        limit = limit or self.default_limit
        key = f"ratelimit:{endpoint}"
        
        current = await self.redis.incr(key)
        
        if current == 1:
            await self.redis.expire(key, 60)
        
        if current > limit:
            return False
        
        return True
    
    async def wait_for_slot(self, endpoint: str, timeout: float = 30) -> bool:
        """Wait until rate limit slot available."""
        
        start = time.time()
        
        while time.time() - start < timeout:
            if await self.acquire(endpoint):
                return True
            await asyncio.sleep(0.1)
        
        return False

class PartitionedQueue:
    """
    Partition by subscription for:
    - Parallel processing of different subscriptions
    - Ordered delivery per subscription
    - Isolation (slow subscriber doesn't block others)
    """
    
    NUM_PARTITIONS = 64
    
    def get_partition(self, subscription_id: str) -> int:
        """Stable hash (modulo) partitioning."""
        hash_val = int(hashlib.md5(subscription_id.encode()).hexdigest(), 16)
        return hash_val % self.NUM_PARTITIONS
    
    async def enqueue(self, task: dict):
        """Add task to appropriate partition."""
        partition = self.get_partition(task['subscription_id'])
        stream = f"webhooks:p{partition}"
        await self.redis.xadd(stream, task)
    
    async def start_workers(self):
        """Start workers for all partitions."""
        tasks = []
        
        for partition in range(self.NUM_PARTITIONS):
            # Multiple workers per partition for throughput
            for worker_num in range(4):
                task = asyncio.create_task(
                    self.partition_worker(partition, worker_num)
                )
                tasks.append(task)
        
        await asyncio.gather(*tasks)

class SmartRetryManager:
    async def run_health_monitor(self):
        """Monitor unhealthy endpoints and trigger recovery."""
        
        while True:
            # Get all endpoints with recent failures
            unhealthy = await self.db.get_unhealthy_endpoints()
            
            for endpoint in unhealthy:
                # Check if recovered
                if await self.check_health(endpoint.url):
                    logger.info(f"Endpoint recovered: {endpoint.url}")
                    
                    # Accelerate all pending retries
                    await self.accelerate_retries(endpoint.subscription_id)
                    
                    # Update status
                    endpoint.healthy = True
                    endpoint.consecutive_failures = 0
                    await self.db.update(endpoint)
            
            await asyncio.sleep(30)  # Check every 30 seconds
    
    async def accelerate_retries(self, subscription_id: str):
        """Move all pending retries to immediate queue."""
        
        # Find all scheduled retries for this subscription
        pending = await self.queue.get_scheduled_for_subscription(subscription_id)
        
        logger.info(f"Accelerating {len(pending)} retries for {subscription_id}")
        
        for task in pending:
            # Move from delayed queue to immediate queue
            await self.queue.cancel_delayed(task['id'])
            await self.queue.enqueue_immediate(task)

class GradualRecoveryManager:
    async def deliver_with_rampup(
        self,
        subscription_id: str,
        pending_tasks: List[dict]
    ):
        """Deliver pending webhooks with gradual ramp-up."""
        
        # Start slow, increase over time
        initial_rate = 1    # per second
        max_rate = 13       # per second: the per-endpoint cap (800/minute)
        ramp_duration = 300  # 5 minutes to reach max
        
        start_time = time.time()
        delivered = 0
        
        for task in pending_tasks:
            # Calculate current rate based on time
            elapsed = time.time() - start_time
            progress = min(1.0, elapsed / ramp_duration)
            current_rate = initial_rate + (max_rate - initial_rate) * progress
            
            # Deliver
            await self.deliver(task)
            delivered += 1
            
            # Rate limit
            await asyncio.sleep(1.0 / current_rate)
            
            # Log progress
            if delivered % 100 == 0:
                logger.info(f"Recovery: {delivered}/{len(pending_tasks)} at {current_rate:.1f}/sec")

class DLQManager:
    async def get_failed_webhooks(
        self,
        subscription_id: str = None,
        limit: int = 100
    ) -> List[dict]:
        """List failed webhooks for review."""
        
        query = """
            SELECT * FROM webhook_dlq 
            WHERE status = 'pending'
        """
        params = []
        
        if subscription_id:
            params.append(subscription_id)
            query += f" AND subscription_id = ${len(params)}"  # parameterized, not interpolated
        
        params.append(limit)
        query += f" ORDER BY created_at DESC LIMIT ${len(params)}"
        
        return await self.db.fetch(query, *params)
    
    async def retry_dlq_entry(self, entry_id: str, user_id: str) -> dict:
        """Manually retry a DLQ entry."""
        
        entry = await self.db.get_dlq_entry(entry_id)
        
        if not entry:
            raise NotFoundError("DLQ entry not found")
        
        # Mark as retrying
        await self.db.update_dlq_status(entry_id, 'retrying')
        
        # Attempt delivery
        result = await self.delivery_service.deliver(
            entry['subscription_id'],
            entry['event']
        )
        
        if result.success:
            await self.db.update_dlq_status(
                entry_id, 
                'resolved',
                resolved_by=user_id
            )
            return {'status': 'delivered'}
        else:
            await self.db.update_dlq_status(entry_id, 'pending')
            return {'status': 'failed', 'error': result.error}
    
    async def retry_all_for_subscription(self, subscription_id: str, user_id: str):
        """Retry all DLQ entries for a subscription."""
        
        entries = await self.get_failed_webhooks(subscription_id, limit=10000)
        
        results = {'success': 0, 'failed': 0}
        
        for entry in entries:
            result = await self.retry_dlq_entry(entry['id'], user_id)
            
            if result['status'] == 'delivered':
                results['success'] += 1
            else:
                results['failed'] += 1
        
        return results

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/admin/webhooks/subscriptions/{subscription_id}/health")
async def get_subscription_health(subscription_id: str):
    """Get health status for a subscription."""
    
    subscription = await db.get_subscription(subscription_id)
    
    return {
        "subscription_id": subscription_id,
        "endpoint": subscription.url,
        "healthy": subscription.healthy,
        "consecutive_failures": subscription.consecutive_failures,
        "last_success": subscription.last_success,
        "last_failure": subscription.last_failure,
        "pending_retries": await queue.count_pending(subscription_id),
        "dlq_count": await dlq.count_for_subscription(subscription_id)
    }

@app.get("/admin/webhooks/dlq")
async def list_dlq(subscription_id: str = None, limit: int = 100):
    """List DLQ entries."""
    
    entries = await dlq_manager.get_failed_webhooks(subscription_id, limit)
    
    return {
        "entries": entries,
        "total": len(entries)
    }

@app.post("/admin/webhooks/dlq/{entry_id}/retry")
async def retry_dlq_entry(entry_id: str, current_user: User):
    """Retry a single DLQ entry."""
    
    result = await dlq_manager.retry_dlq_entry(entry_id, current_user.id)
    return result

@app.post("/admin/webhooks/subscriptions/{subscription_id}/retry-all-dlq")
async def retry_all_dlq(subscription_id: str, current_user: User):
    """Retry all DLQ entries for a subscription."""
    
    result = await dlq_manager.retry_all_for_subscription(
        subscription_id, 
        current_user.id
    )
    return result

@app.get("/admin/webhooks/stats")
async def get_webhook_stats():
    """Get overall webhook system stats."""
    
    return {
        "queue_depth": await queue.total_size(),
        "retry_queue_depth": await retry_queue.size(),
        "dlq_depth": await dlq.size(),
        "deliveries_last_hour": await metrics.get_deliveries_count(hours=1),
        "success_rate_last_hour": await metrics.get_success_rate(hours=1),
        "unhealthy_endpoints": await db.count_unhealthy_endpoints()
    }

from prometheus_client import Counter, Histogram, Gauge

# Delivery metrics
webhook_deliveries_total = Counter(
    'webhook_deliveries_total',
    'Total webhook deliveries',
    ['status', 'event_type']  # success, failed, retrying
)

webhook_delivery_duration = Histogram(
    'webhook_delivery_duration_seconds',
    'Webhook delivery duration',
    buckets=[0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30]
)

# Queue metrics
webhook_queue_depth = Gauge(
    'webhook_queue_depth',
    'Webhooks pending delivery',
    ['partition']
)

webhook_retry_queue_depth = Gauge(
    'webhook_retry_queue_depth',
    'Webhooks pending retry'
)

webhook_dlq_depth = Gauge(
    'webhook_dlq_depth',
    'Webhooks in dead letter queue'
)

# Endpoint health
webhook_endpoint_healthy = Gauge(
    'webhook_endpoint_healthy',
    'Endpoint health status',
    ['subscription_id']
)

webhook_endpoint_consecutive_failures = Gauge(
    'webhook_endpoint_consecutive_failures',
    'Consecutive failures per endpoint',
    ['subscription_id']
)

groups:
  - name: webhook_alerts
    rules:
      # High failure rate
      - alert: WebhookHighFailureRate
        expr: |
          sum(rate(webhook_deliveries_total{status="failed"}[5m])) /
          sum(rate(webhook_deliveries_total[5m])) > 0.1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Webhook failure rate above 10%"
          description: "{{ $value | humanizePercentage }} of webhooks failing"
      
      # Queue backlog
      - alert: WebhookQueueBacklog
        expr: sum(webhook_queue_depth) > 10000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Webhook queue backlog: {{ $value }}"
      
      # DLQ growing
      - alert: WebhookDLQGrowing
        expr: webhook_dlq_depth > 100
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Webhook DLQ has {{ $value }} entries"
      
      # Endpoint down
      - alert: WebhookEndpointDown
        expr: webhook_endpoint_consecutive_failures > 10
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "Webhook endpoint has {{ $value }} consecutive failures"

import structlog

logger = structlog.get_logger()

class WebhookLogger:
    def log_delivery_attempt(
        self,
        subscription_id: str,
        event_id: str,
        event_type: str,
        attempt: int,
        result: DeliveryResult
    ):
        logger.info(
            "webhook_delivery_attempt",
            subscription_id=subscription_id,
            event_id=event_id,
            event_type=event_type,
            attempt=attempt,
            success=result.success,
            status_code=result.status_code,
            duration_ms=result.duration_ms,
            error=result.error
        )
    
    def log_moved_to_dlq(
        self,
        subscription_id: str,
        event_id: str,
        error: str,
        attempts: int
    ):
        logger.error(
            "webhook_moved_to_dlq",
            subscription_id=subscription_id,
            event_id=event_id,
            error=error,
            total_attempts=attempts
        )
    
    def log_endpoint_status(
        self,
        subscription_id: str,
        healthy: bool,
        consecutive_failures: int
    ):
        logger.info(
            "webhook_endpoint_status",
            subscription_id=subscription_id,
            healthy=healthy,
            consecutive_failures=consecutive_failures
        )

{"sequence": 42, "entity_id": "order_123", ...}

"""
Production webhook delivery system.
Building on Days 1-4: Timeouts, Idempotency, Circuit Breakers, Webhooks.
"""

import asyncio
import hashlib
import hmac
import json
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import List, Optional, Dict, Any
from enum import Enum
import aiohttp
import structlog
from prometheus_client import Counter, Histogram, Gauge

logger = structlog.get_logger()

# =============================================================================
# Configuration
# =============================================================================

@dataclass
class WebhookConfig:
    # Delivery
    delivery_timeout: int = 30
    max_retries: int = 8  # 9 deliveries in all: the first try plus 8 retries
    retry_delays: List[int] = field(default_factory=lambda: [
        60, 300, 900, 3600, 7200, 14400, 28800, 86400
    ])
    
    # Rate limiting
    max_concurrent_per_endpoint: int = 10
    max_requests_per_minute: int = 800
    
    # Health monitoring
    health_check_interval: int = 30
    circuit_breaker_threshold: int = 10
    
    # Security
    signature_tolerance: int = 300

# =============================================================================
# Models
# =============================================================================

class DeliveryStatus(Enum):
    PENDING = "pending"
    SUCCESS = "success" 
    FAILED = "failed"
    RETRYING = "retrying"
    DLQ = "dlq"

@dataclass
class WebhookSubscription:
    id: str
    user_id: str
    url: str
    secret: str
    events: List[str]
    active: bool = True
    healthy: bool = True
    consecutive_failures: int = 0
    last_success: Optional[datetime] = None
    last_failure: Optional[datetime] = None
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

@dataclass
class WebhookEvent:
    id: str
    type: str
    data: Dict[str, Any]
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

@dataclass 
class DeliveryResult:
    success: bool
    status_code: Optional[int] = None
    duration_ms: Optional[int] = None
    error: Optional[str] = None
    should_retry: bool = False

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

deliveries_total = Counter(
    'webhook_deliveries_total',
    'Total deliveries',
    ['status', 'event_type']
)

delivery_duration = Histogram(
    'webhook_delivery_duration_seconds', 
    'Delivery duration',
    buckets=[0.1, 0.5, 1, 2, 5, 10, 30]
)

queue_depth = Gauge('webhook_queue_depth', 'Queue depth')
dlq_depth = Gauge('webhook_dlq_depth', 'DLQ depth')

# =============================================================================
# Webhook Signer
# =============================================================================

class WebhookSigner:
    @staticmethod
    def sign(payload: dict, secret: str, timestamp: int = None) -> Dict[str, str]:
        timestamp = timestamp or int(time.time())
        payload_str = json.dumps(payload, separators=(',', ':'), sort_keys=True)
        signed_content = f"{timestamp}.{payload_str}"
        
        signature = hmac.new(
            secret.encode(),
            signed_content.encode(),
            hashlib.sha256
        ).hexdigest()
        
        return {
            'X-Webhook-Timestamp': str(timestamp),
            'X-Webhook-Signature': f"sha256={signature}",
            'Content-Type': 'application/json'
        }
    
    @staticmethod
    def verify(payload: str, signature: str, timestamp: str, 
               secret: str, tolerance: int = 300) -> bool:
        try:
            ts = int(timestamp)
            if abs(time.time() - ts) > tolerance:
                return False
            
            if not signature.startswith('sha256='):
                return False
            
            expected = hmac.new(
                secret.encode(),
                f"{ts}.{payload}".encode(),
                hashlib.sha256
            ).hexdigest()
            
            return hmac.compare_digest(signature[7:], expected)
        except:
            return False

# =============================================================================
# Retry Manager
# =============================================================================

class RetryManager:
    def __init__(self, config: WebhookConfig, queue, dlq):
        self.config = config
        self.queue = queue
        self.dlq = dlq
    
    async def schedule_retry(self, task: dict, error: str) -> bool:
        attempt = task.get('attempt', 0)
        
        if attempt >= self.config.max_retries:
            await self.move_to_dlq(task, error)
            return False
        
        delay = self.config.retry_delays[min(attempt, len(self.config.retry_delays) - 1)]
        delay += int(delay * 0.1 * (hash(task['event']['id']) % 100 - 50) / 50)
        
        task['attempt'] = attempt + 1
        task['last_error'] = error
        task['retry_at'] = (datetime.now(timezone.utc) + timedelta(seconds=delay)).isoformat()
        
        await self.queue.enqueue_delayed(task, delay)
        
        logger.info("retry_scheduled",
            subscription_id=task['subscription_id'],
            event_id=task['event']['id'],
            attempt=task['attempt'],
            delay=delay
        )
        
        return True
    
    async def move_to_dlq(self, task: dict, error: str):
        dlq_entry = {
            'id': str(uuid.uuid4()),
            'task': task,
            'error': error,
            'attempts': task.get('attempt', 0),
            'failed_at': datetime.now(timezone.utc).isoformat(),
            'status': 'pending'
        }
        
        await self.dlq.push(dlq_entry)
        dlq_depth.inc()
        
        logger.error("moved_to_dlq",
            subscription_id=task['subscription_id'],
            event_id=task['event']['id'],
            error=error,
            attempts=task.get('attempt', 0)
        )

# =============================================================================
# Delivery Service
# =============================================================================

class WebhookDeliveryService:
    def __init__(self, config: WebhookConfig, subscription_store, 
                 retry_manager: RetryManager):
        self.config = config
        self.subscriptions = subscription_store
        self.retry_manager = retry_manager
        self.signer = WebhookSigner()
        self.session: Optional[aiohttp.ClientSession] = None
        self.endpoint_semaphores: Dict[str, asyncio.Semaphore] = {}
    
    async def start(self):
        connector = aiohttp.TCPConnector(
            limit=1000,
            limit_per_host=self.config.max_concurrent_per_endpoint,
            ttl_dns_cache=300
        )
        timeout = aiohttp.ClientTimeout(total=self.config.delivery_timeout)
        self.session = aiohttp.ClientSession(connector=connector, timeout=timeout)
    
    async def stop(self):
        if self.session:
            await self.session.close()
    
    async def deliver(self, task: dict) -> DeliveryResult:
        subscription_id = task['subscription_id']
        event = task['event']
        attempt = task.get('attempt', 0)
        
        subscription = await self.subscriptions.get(subscription_id)
        if not subscription or not subscription.active:
            return DeliveryResult(success=False, error="Subscription inactive")
        
        # Build signed payload
        payload = {
            'id': event['id'],
            'type': event['type'],
            'data': event['data'],
            'created_at': event['timestamp']
        }
        headers = self.signer.sign(payload, subscription.secret)
        # Send exactly the bytes that were signed (json= would re-serialize)
        body = json.dumps(payload, separators=(',', ':'), sort_keys=True, default=str)
        
        # Rate limit per endpoint
        semaphore = self.endpoint_semaphores.setdefault(
            subscription.url,
            asyncio.Semaphore(self.config.max_concurrent_per_endpoint)
        )
        
        start = time.time()
        
        async with semaphore:
            try:
                async with self.session.post(
                    subscription.url,
                    data=body,
                    headers=headers,
                    allow_redirects=False  # never follow redirects (SSRF); a 3xx is a failed delivery
                ) as response:
                    duration_ms = int((time.time() - start) * 1000)
                    
                    delivery_duration.observe(duration_ms / 1000)
                    
                    if 200 <= response.status < 300:
                        deliveries_total.labels(status='success', event_type=event['type']).inc()
                        await self._record_success(subscription)
                        return DeliveryResult(
                            success=True,
                            status_code=response.status,
                            duration_ms=duration_ms
                        )
                    
                    elif 400 <= response.status < 500 and response.status != 429:
                        deliveries_total.labels(status='client_error', event_type=event['type']).inc()
                        return DeliveryResult(
                            success=False,
                            status_code=response.status,
                            error=f"Client error: {response.status}",
                            should_retry=False
                        )
                    
                    else:
                        deliveries_total.labels(status='server_error', event_type=event['type']).inc()
                        await self._record_failure(subscription)
                        return DeliveryResult(
                            success=False,
                            status_code=response.status,
                            error=f"Server error: {response.status}",
                            should_retry=True,
                            duration_ms=duration_ms
                        )
            
            except asyncio.TimeoutError:
                deliveries_total.labels(status='timeout', event_type=event['type']).inc()
                await self._record_failure(subscription)
                return DeliveryResult(
                    success=False,
                    error="Timeout",
                    should_retry=True
                )
            
            except Exception as e:
                deliveries_total.labels(status='error', event_type=event['type']).inc()
                await self._record_failure(subscription)
                return DeliveryResult(
                    success=False,
                    error=str(e),
                    should_retry=True
                )
    
    async def _record_success(self, subscription: WebhookSubscription):
        subscription.healthy = True
        subscription.consecutive_failures = 0
        subscription.last_success = datetime.now(timezone.utc)
        await self.subscriptions.update(subscription)
    
    async def _record_failure(self, subscription: WebhookSubscription):
        subscription.consecutive_failures += 1
        subscription.last_failure = datetime.now(timezone.utc)
        
        if subscription.consecutive_failures >= self.config.circuit_breaker_threshold:
            subscription.healthy = False
        
        await self.subscriptions.update(subscription)

# =============================================================================
# Worker
# =============================================================================

class WebhookWorker:
    def __init__(self, worker_id: str, delivery_service: WebhookDeliveryService,
                 retry_manager: RetryManager, queue):
        self.worker_id = worker_id
        self.delivery = delivery_service
        self.retry_manager = retry_manager
        self.queue = queue
        self.running = False
    
    async def run(self):
        self.running = True
        logger.info("worker_started", worker_id=self.worker_id)
        
        while self.running:
            try:
                tasks = await self.queue.dequeue(count=10, timeout=5)
                
                if tasks:
                    await asyncio.gather(*[
                        self.process_task(task) for task in tasks
                    ])
            except Exception as e:
                logger.error("worker_error", error=str(e))
                await asyncio.sleep(1)
    
    async def process_task(self, task: dict):
        result = await self.delivery.deliver(task)
        
        if result.success:
            logger.info("delivery_success",
                subscription_id=task['subscription_id'],
                event_id=task['event']['id'],
                duration_ms=result.duration_ms
            )
        elif result.should_retry:
            await self.retry_manager.schedule_retry(task, result.error)
        else:
            logger.warning("delivery_permanent_failure",
                subscription_id=task['subscription_id'],
                event_id=task['event']['id'],
                error=result.error
            )
    
    async def stop(self):
        self.running = False

# =============================================================================
# Health Checker
# =============================================================================

class HealthChecker:
    def __init__(self, subscription_store, queue, check_interval: int = 30):
        self.subscriptions = subscription_store
        self.queue = queue
        self.check_interval = check_interval
        self.running = False
    
    async def run(self):
        self.running = True
        
        while self.running:
            try:
                unhealthy = await self.subscriptions.find_unhealthy()
                
                for sub in unhealthy:
                    if await self._check_health(sub.url):
                        logger.info("endpoint_recovered", 
                            subscription_id=sub.id, url=sub.url)
                        
                        sub.healthy = True
                        sub.consecutive_failures = 0
                        await self.subscriptions.update(sub)
                        
                        await self._accelerate_retries(sub.id)
                
                await asyncio.sleep(self.check_interval)
            except Exception as e:
                logger.error("health_check_error", error=str(e))
                await asyncio.sleep(self.check_interval)
    
    async def _check_health(self, url: str) -> bool:
        try:
            async with aiohttp.ClientSession() as session:
                async with session.head(url, timeout=aiohttp.ClientTimeout(total=5), allow_redirects=False) as resp:
                    return resp.status < 500
        except:
            return False
    
    async def _accelerate_retries(self, subscription_id: str):
        pending = await self.queue.get_pending_for_subscription(subscription_id)
        
        for task in pending:
            await self.queue.move_to_immediate(task)
        
        if pending:
            logger.info("retries_accelerated",
                subscription_id=subscription_id,
                count=len(pending)
            )
    
    async def stop(self):
        self.running = False

# =============================================================================
# DLQ Manager
# =============================================================================

class DLQManager:
    def __init__(self, dlq, delivery_service: WebhookDeliveryService):
        self.dlq = dlq
        self.delivery = delivery_service
    
    async def list_entries(self, subscription_id: str = None, 
                          status: str = 'pending', limit: int = 100) -> List[dict]:
        return await self.dlq.find(
            subscription_id=subscription_id,
            status=status,
            limit=limit
        )
    
    async def retry_entry(self, entry_id: str, user_id: str) -> dict:
        entry = await self.dlq.get(entry_id)
        if not entry:
            raise ValueError("Entry not found")
        
        await self.dlq.update_status(entry_id, 'retrying')
        
        result = await self.delivery.deliver(entry['task'])
        
        if result.success:
            await self.dlq.update_status(entry_id, 'resolved', resolved_by=user_id)
            dlq_depth.dec()
            return {'status': 'delivered'}
        else:
            await self.dlq.update_status(entry_id, 'pending')
            return {'status': 'failed', 'error': result.error}
    
    async def retry_all(self, subscription_id: str, user_id: str) -> dict:
        entries = await self.list_entries(subscription_id, limit=10000)
        
        results = {'success': 0, 'failed': 0}
        
        for entry in entries:
            result = await self.retry_entry(entry['id'], user_id)
            if result['status'] == 'delivered':
                results['success'] += 1
            else:
                results['failed'] += 1
        
        return results
    
    async def discard_entry(self, entry_id: str, user_id: str, reason: str):
        await self.dlq.update_status(entry_id, 'discarded', 
            resolved_by=user_id, resolution_note=reason)
        dlq_depth.dec()