Webhooks and Event-Driven APIs

# Client polling for new orders (inefficient)
import time
import requests

def poll_for_orders():
    """
    Check for new orders every 10 seconds.

    Problem: 99% of requests return "no new orders"
    Waste: 6 requests/minute * 1000 clients = 6000 req/min
    Cost: Server processing + database queries + bandwidth
    """
    last_order_id = 0

    while True:
        response = requests.get(
            f"https://api.example.com/orders?since={last_order_id}"
        )

        orders = response.json()

        if orders:
            print(f"New orders: {len(orders)}")
            for order in orders:
                process_order(order)
                last_order_id = order["id"]
        else:
            print("No new orders (wasted request)")

        # Wait 10 seconds and poll again
        time.sleep(10)


# Result with 1000 clients:
# - 6000 requests/minute to check for updates
# - 5940 return "no changes" (99% waste)
# - Latency: 0-10 seconds (average 5 seconds)
# - Database load: Constant queries even with no activity

# Webhook flow (efficient)

# Step 1: Client registers webhook URL
POST /api/webhooks/subscribe
{
    "url": "https://client.example.com/webhook",
    "events": ["order.created", "order.shipped"]
}

# Response:
{
    "webhook_id": "wh_abc123",
    "status": "active"
}


# Step 2: Event occurs (order created)
# Server makes HTTP POST to client's webhook URL

POST https://client.example.com/webhook
Headers:
    Content-Type: application/json
    X-Webhook-Signature: abc123...
Body:
{
    "event": "order.created",
    "timestamp": "2024-01-15T10:30:00Z",
    "data": {
        "order_id": 12345,
        "customer": "john@example.com",
        "total": 99.99
    }
}

# Client receives webhook and responds immediately
HTTP/1.1 200 OK


# Result:
# - Zero polling requests (0 req/min instead of 6000!)
# - Instant notification (0 latency instead of 0-10s)
# - No database queries until event occurs
# - Scales effortlessly (1 request per event, not per client)

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, HttpUrl
import requests
from typing import List
import time

app = FastAPI()

# In-memory storage (use database in production)
webhooks = []


class WebhookSubscription(BaseModel):
    url: HttpUrl
    events: List[str]  # e.g., ["order.created", "order.shipped"]


class OrderEvent(BaseModel):
    event: str
    timestamp: str
    data: dict


@app.post("/api/webhooks/subscribe")
async def subscribe_webhook(subscription: WebhookSubscription):
    """
    Subscribe to webhook events.

    Client provides URL where they want to receive event notifications.
    """
    webhook = {
        "id": f"wh_{int(time.time())}",
        "url": str(subscription.url),
        "events": subscription.events,
        "status": "active",
        "created_at": time.time()
    }

    webhooks.append(webhook)

    return {
        "webhook_id": webhook["id"],
        "status": "active",
        "subscribed_events": subscription.events
    }


def send_webhook(webhook_url: str, event_data: dict):
    """
    Send webhook notification to client's URL.

    Makes HTTP POST request with event data.
    """
    try:
        response = requests.post(
            webhook_url,
            json=event_data,
            headers={"Content-Type": "application/json"},
            timeout=5  # 5 second timeout
        )

        if response.status_code == 200:
            print(f"✓ Webhook delivered to {webhook_url}")
            return True
        else:
            print(f"✗ Webhook failed: {response.status_code}")
            return False

    except requests.exceptions.RequestException as e:
        print(f"✗ Webhook error: {e}")
        return False


def notify_webhooks(event_type: str, event_data: dict, background_tasks: BackgroundTasks):
    """
    Notify all subscribed webhooks about an event.

    Filters webhooks by event type and sends notifications in background.
    """
    event_payload = {
        "event": event_type,
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "data": event_data
    }

    # Find webhooks subscribed to this event
    for webhook in webhooks:
        if event_type in webhook["events"] and webhook["status"] == "active":
            # Send webhook in background (non-blocking)
            background_tasks.add_task(
                send_webhook,
                webhook["url"],
                event_payload
            )


@app.post("/api/orders")
async def create_order(
    customer_email: str,
    total: float,
    background_tasks: BackgroundTasks
):
    """
    Create order and trigger webhook notifications.

    When order is created, notify all webhooks subscribed to 'order.created' event.
    """
    # Create order in database
    order = {
        "id": 12345,
        "customer": customer_email,
        "total": total,
        "status": "pending"
    }

    # Trigger webhook notifications
    notify_webhooks(
        event_type="order.created",
        event_data=order,
        background_tasks=background_tasks
    )

    return {
        "message": "Order created",
        "order": order,
        "webhooks_notified": True
    }


# Example flow:
# 1. Client subscribes: POST /api/webhooks/subscribe
#    Body: {"url": "https://client.com/webhook", "events": ["order.created"]}
#    Response: {"webhook_id": "wh_1234", "status": "active"}
#
# 2. Order created: POST /api/orders
#    Body: {"customer_email": "john@example.com", "total": 99.99}
#    Server sends webhook to https://client.com/webhook
#
# 3. Client receives:
#    POST https://client.com/webhook
#    Body: {"event": "order.created", "data": {...}}
#    Client responds: 200 OK

from fastapi import FastAPI, Request
from pydantic import BaseModel

app = FastAPI()


class WebhookPayload(BaseModel):
    event: str
    timestamp: str
    data: dict


@app.post("/webhook")
async def receive_webhook(payload: WebhookPayload):
    """
    Receive webhook from external API.

    Important:
    - Respond quickly (< 5 seconds) with 200 OK
    - Process event asynchronously if needed
    - Don't make slow API calls here
    """
    print(f"Received webhook: {payload.event}")

    # Route to appropriate handler based on event type
    if payload.event == "order.created":
        handle_order_created(payload.data)
    elif payload.event == "order.shipped":
        handle_order_shipped(payload.data)
    elif payload.event == "order.cancelled":
        handle_order_cancelled(payload.data)
    else:
        print(f"Unknown event: {payload.event}")

    # Always return 200 OK quickly
    return {"status": "received"}


def handle_order_created(order_data: dict):
    """Process new order event"""
    print(f"New order: {order_data['id']}")
    # Send email notification
    # Update inventory
    # Log to analytics


def handle_order_shipped(order_data: dict):
    """Process order shipped event"""
    print(f"Order shipped: {order_data['id']}")
    # Send tracking email
    # Update order status


def handle_order_cancelled(order_data: dict):
    """Process order cancelled event"""
    print(f"Order cancelled: {order_data['id']}")
    # Refund payment
    # Restore inventory


# Webhook receiver logs:
# INFO: Received webhook: order.created
# INFO: New order: 12345
# INFO: Sending confirmation email to john@example.com
# INFO: Response sent: 200 OK (35ms)

import hmac
import hashlib
import secrets
import json
import requests
from typing import Optional


def generate_webhook_secret() -> str:
    """
    Generate secure webhook secret key.

    Store this securely and share with client.
    """
    return secrets.token_urlsafe(32)


def sign_webhook_payload(payload: dict, secret: str) -> str:
    """
    Generate HMAC signature for webhook payload.

    Args:
        payload: Webhook event data
        secret: Shared secret key

    Returns:
        HMAC signature (hex string)
    """
    # Convert payload to JSON bytes
    payload_bytes = json.dumps(payload, sort_keys=True).encode('utf-8')

    # Compute HMAC-SHA256
    signature = hmac.new(
        key=secret.encode('utf-8'),
        msg=payload_bytes,
        digestmod=hashlib.sha256
    ).hexdigest()

    return signature


def send_signed_webhook(webhook_url: str, event_data: dict, secret: str) -> bool:
    """
    Send webhook with HMAC signature.

    Includes signature in X-Webhook-Signature header.
    """
    # Create payload
    payload = {
        "event": event_data["event"],
        "timestamp": event_data["timestamp"],
        "data": event_data["data"]
    }

    # Generate signature
    signature = sign_webhook_payload(payload, secret)

    try:
        response = requests.post(
            webhook_url,
            json=payload,
            headers={
                "Content-Type": "application/json",
                "X-Webhook-Signature": signature  # Include signature
            },
            timeout=5
        )

        return response.status_code == 200

    except requests.exceptions.RequestException as e:
        print(f"Webhook delivery failed: {e}")
        return False


# Example usage
webhook_secret = generate_webhook_secret()
print(f"Webhook secret (share with client): {webhook_secret}")

event_data = {
    "event": "order.created",
    "timestamp": "2024-01-15T10:30:00Z",
    "data": {"order_id": 12345, "total": 99.99}
}

# Send webhook with signature
success = send_signed_webhook(
    webhook_url="https://client.com/webhook",
    event_data=event_data,
    secret=webhook_secret
)

# Webhook includes header:
# X-Webhook-Signature: a7f2c8e1b9d4c5a8e3f9b1c2d3e4f5a6...

from fastapi import FastAPI, Request, HTTPException, status
import hmac
import hashlib
import json

app = FastAPI()

# Store webhook secret securely (from server during subscription)
WEBHOOK_SECRET = "sk_abc123..."  # Get from server when subscribing


def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """
    Verify webhook signature to ensure authenticity.

    Args:
        payload: Raw request body (bytes)
        signature: Signature from X-Webhook-Signature header
        secret: Shared secret key

    Returns:
        True if signature valid, False otherwise
    """
    # Compute expected signature
    expected_signature = hmac.new(
        key=secret.encode('utf-8'),
        msg=payload,
        digestmod=hashlib.sha256
    ).hexdigest()

    # Constant-time comparison (prevents timing attacks)
    return hmac.compare_digest(signature, expected_signature)


@app.post("/webhook")
async def receive_webhook(request: Request):
    """
    Receive and verify webhook.

    Security:
    - Checks X-Webhook-Signature header
    - Recomputes HMAC and compares
    - Rejects mismatched signatures
    """
    # Get signature from header
    signature = request.headers.get("X-Webhook-Signature")

    if not signature:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing X-Webhook-Signature header"
        )

    # Get raw body
    body = await request.body()

    # Verify signature
    if not verify_webhook_signature(body, signature, WEBHOOK_SECRET):
        print("⚠️ Webhook signature verification FAILED")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid webhook signature"
        )

    print("✓ Webhook signature verified")

    # Parse payload
    payload = json.loads(body)

    # Process event
    print(f"Processing verified event: {payload['event']}")

    return {"status": "received"}


# Example logs:
#
# Authentic webhook:
# INFO: ✓ Webhook signature verified
# INFO: Processing verified event: order.created
# INFO: Response: 200 OK
#
# Fake webhook (no signature):
# WARNING: Missing X-Webhook-Signature header
# INFO: Response: 401 Unauthorized
#
# Fake webhook (wrong signature):
# WARNING: ⚠️ Webhook signature verification FAILED
# INFO: Response: 401 Unauthorized

from fastapi import FastAPI, BackgroundTasks
import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import time
import logging

app = FastAPI()
logger = logging.getLogger(__name__)


class WebhookDeliveryError(Exception):
    """Raised when webhook delivery fails"""
    pass


@retry(
    stop=stop_after_attempt(5),  # Try 5 times
    wait=wait_exponential(multiplier=60, min=60, max=3600),  # 1min, 2min, 4min, 8min, 16min
    retry=retry_if_exception_type(WebhookDeliveryError),
    before_sleep=lambda retry_state: logger.warning(
        f"Webhook delivery failed, retrying in {retry_state.next_action.sleep}s... "
        f"(attempt {retry_state.attempt_number}/5)"
    )
)
def deliver_webhook_with_retry(
    webhook_url: str,
    payload: dict,
    signature: str
) -> dict:
    """
    Deliver webhook with automatic retries.

    Retry schedule:
    - Attempt 1: Immediate
    - Attempt 2: Wait 1 minute
    - Attempt 3: Wait 2 minutes
    - Attempt 4: Wait 4 minutes
    - Attempt 5: Wait 8 minutes
    - If all fail: Raise exception

    Retries on:
    - Network errors
    - Timeouts
    - 5xx server errors

    Does NOT retry on:
    - 4xx client errors (indicates bad webhook URL or payload)
    """
    try:
        response = requests.post(
            webhook_url,
            json=payload,
            headers={
                "Content-Type": "application/json",
                "X-Webhook-Signature": signature
            },
            timeout=10
        )

        # Check response
        if response.status_code == 200:
            logger.info(f"✓ Webhook delivered successfully to {webhook_url}")
            return {
                "status": "delivered",
                "status_code": 200,
                "delivered_at": time.time()
            }

        elif 500 <= response.status_code < 600:
            # Server error - retry
            logger.warning(f"Server error {response.status_code}, will retry")
            raise WebhookDeliveryError(f"Server error: {response.status_code}")

        else:
            # Client error (4xx) - don't retry
            logger.error(f"Client error {response.status_code}, not retrying")
            return {
                "status": "failed",
                "status_code": response.status_code,
                "error": "Client error (permanent failure)"
            }

    except requests.exceptions.Timeout:
        logger.warning("Webhook timeout, will retry")
        raise WebhookDeliveryError("Timeout")

    except requests.exceptions.ConnectionError:
        logger.warning("Connection error, will retry")
        raise WebhookDeliveryError("Connection error")


async def send_webhook_async(
    webhook_url: str,
    event_data: dict,
    secret: str,
    background_tasks: BackgroundTasks
):
    """
    Send webhook asynchronously with retries.

    Runs in background task so it doesn't block API response.
    """
    payload = {
        "event": event_data["event"],
        "timestamp": event_data["timestamp"],
        "data": event_data["data"]
    }

    signature = sign_webhook_payload(payload, secret)

    # Add to background tasks
    background_tasks.add_task(
        deliver_webhook_with_retry,
        webhook_url,
        payload,
        signature
    )


@app.post("/api/orders")
async def create_order(
    customer_email: str,
    total: float,
    background_tasks: BackgroundTasks
):
    """Create order and send webhook with automatic retries"""

    order = {
        "id": 12345,
        "customer": customer_email,
        "total": total
    }

    event_data = {
        "event": "order.created",
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "data": order
    }

    # Send webhook with retries (non-blocking)
    await send_webhook_async(
        webhook_url="https://client.com/webhook",
        event_data=event_data,
        secret="sk_abc123",
        background_tasks=background_tasks
    )

    return {"message": "Order created", "webhooks_queued": True}


# Example retry timeline:
# 10:00:00 - Attempt 1: Connection refused (immediate)
# 10:01:00 - Attempt 2: Timeout (wait 1 min)
# 10:03:00 - Attempt 3: 503 Server Error (wait 2 min)
# 10:07:00 - Attempt 4: 503 Server Error (wait 4 min)
# 10:15:00 - Attempt 5: 200 OK Success! (wait 8 min)

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)


def add_to_dead_letter_queue(webhook_url: str, payload: dict, error: str):
    """
    Store failed webhook in dead letter queue.

    Allows manual inspection and redelivery.
    """
    failed_webhook = {
        "webhook_url": webhook_url,
        "payload": payload,
        "error": error,
        "failed_at": time.time(),
        "attempts": 5
    }

    # Store in Redis list
    redis_client.lpush("webhook:dlq", json.dumps(failed_webhook))

    logger.error(f"Webhook permanently failed, added to DLQ: {webhook_url}")


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=60, min=60, max=3600)
)
def deliver_webhook_with_dlq(
    webhook_url: str,
    payload: dict,
    signature: str
) -> dict:
    """
    Deliver webhook with retries and dead letter queue.

    If all retries fail, adds to DLQ instead of losing event.
    """
    try:
        # Attempt delivery with retries
        return deliver_webhook_with_retry(webhook_url, payload, signature)

    except Exception as e:
        # All retries exhausted
        logger.error(f"All retries failed: {e}")

        # Add to dead letter queue
        add_to_dead_letter_queue(webhook_url, payload, str(e))

        return {
            "status": "failed",
            "error": str(e),
            "added_to_dlq": True
        }


@app.post("/api/webhooks/dlq/retry/{index}")
async def retry_failed_webhook(index: int):
    """
    Manually retry webhook from dead letter queue.

    Admin can review failed webhooks and retry after fixing issues.
    """
    # Get failed webhook from DLQ
    failed_webhook_json = redis_client.lindex("webhook:dlq", index)

    if not failed_webhook_json:
        return {"error": "Webhook not found in DLQ"}

    failed_webhook = json.loads(failed_webhook_json)

    # Retry delivery
    result = deliver_webhook_with_retry(
        failed_webhook["webhook_url"],
        failed_webhook["payload"],
        sign_webhook_payload(failed_webhook["payload"], "sk_abc123")
    )

    if result["status"] == "delivered":
        # Remove from DLQ
        redis_client.lrem("webhook:dlq", 1, failed_webhook_json)
        return {"message": "Webhook redelivered successfully"}
    else:
        return {"error": "Retry failed", "result": result}


@app.get("/api/webhooks/dlq")
async def list_failed_webhooks():
    """List all webhooks in dead letter queue"""

    dlq_length = redis_client.llen("webhook:dlq")

    failed_webhooks = []
    for i in range(dlq_length):
        webhook_json = redis_client.lindex("webhook:dlq", i)
        failed_webhooks.append(json.loads(webhook_json))

    return {
        "total_failed": dlq_length,
        "webhooks": failed_webhooks
    }


# Example DLQ entry:
# {
#     "webhook_url": "https://client.com/webhook",
#     "payload": {"event": "order.created", "data": {...}},
#     "error": "Connection timeout after 5 attempts",
#     "failed_at": 1705320000,
#     "attempts": 5
# }

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, HttpUrl
from typing import List, Optional
import secrets
import time

app = FastAPI()

# Database (use SQLAlchemy/PostgreSQL in production)
webhooks_db = {}


class WebhookCreate(BaseModel):
    url: HttpUrl
    events: List[str]
    description: Optional[str] = None


class WebhookUpdate(BaseModel):
    url: Optional[HttpUrl] = None
    events: Optional[List[str]] = None
    description: Optional[str] = None
    status: Optional[str] = None  # active, paused


@app.post("/api/webhooks", status_code=status.HTTP_201_CREATED)
async def create_webhook_subscription(webhook: WebhookCreate):
    """
    Create new webhook subscription.

    Returns webhook ID and secret for signature verification.
    """
    webhook_id = f"wh_{secrets.token_urlsafe(16)}"
    secret = f"whsec_{secrets.token_urlsafe(32)}"

    webhook_data = {
        "id": webhook_id,
        "url": str(webhook.url),
        "events": webhook.events,
        "description": webhook.description,
        "secret": secret,
        "status": "active",
        "created_at": time.time(),
        "updated_at": time.time(),
        "delivery_stats": {
            "total_delivered": 0,
            "total_failed": 0,
            "last_delivery": None
        }
    }

    webhooks_db[webhook_id] = webhook_data

    return {
        "id": webhook_id,
        "url": webhook_data["url"],
        "events": webhook_data["events"],
        "secret": secret,  # Client stores this for verification
        "status": "active"
    }


@app.get("/api/webhooks")
async def list_webhooks(status: Optional[str] = None):
    """
    List all webhook subscriptions.

    Query params:
    - status: Filter by status (active, paused)
    """
    webhooks = list(webhooks_db.values())

    if status:
        webhooks = [w for w in webhooks if w["status"] == status]

    # Don't return secrets in list
    return {
        "total": len(webhooks),
        "webhooks": [
            {
                "id": w["id"],
                "url": w["url"],
                "events": w["events"],
                "status": w["status"],
                "created_at": w["created_at"],
                "delivery_stats": w["delivery_stats"]
            }
            for w in webhooks
        ]
    }


@app.get("/api/webhooks/{webhook_id}")
async def get_webhook(webhook_id: str):
    """Get webhook subscription details"""

    webhook = webhooks_db.get(webhook_id)

    if not webhook:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Webhook not found"
        )

    return webhook


@app.patch("/api/webhooks/{webhook_id}")
async def update_webhook(webhook_id: str, update: WebhookUpdate):
    """
    Update webhook subscription.

    Can update:
    - URL
    - Event filters
    - Description
    - Status (active/paused)
    """
    webhook = webhooks_db.get(webhook_id)

    if not webhook:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Webhook not found"
        )

    # Update fields
    if update.url:
        webhook["url"] = str(update.url)
    if update.events:
        webhook["events"] = update.events
    if update.description:
        webhook["description"] = update.description
    if update.status:
        if update.status not in ["active", "paused"]:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Status must be 'active' or 'paused'"
            )
        webhook["status"] = update.status

    webhook["updated_at"] = time.time()

    return {
        "id": webhook_id,
        "url": webhook["url"],
        "events": webhook["events"],
        "status": webhook["status"],
        "updated_at": webhook["updated_at"]
    }


@app.delete("/api/webhooks/{webhook_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_webhook(webhook_id: str):
    """Delete webhook subscription"""

    if webhook_id not in webhooks_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Webhook not found"
        )

    del webhooks_db[webhook_id]

    return None  # 204 No Content


@app.post("/api/webhooks/{webhook_id}/pause")
async def pause_webhook(webhook_id: str):
    """Pause webhook delivery temporarily"""

    webhook = webhooks_db.get(webhook_id)

    if not webhook:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Webhook not found"
        )

    webhook["status"] = "paused"
    webhook["updated_at"] = time.time()

    return {"message": "Webhook paused", "status": "paused"}


@app.post("/api/webhooks/{webhook_id}/resume")
async def resume_webhook(webhook_id: str):
    """Resume paused webhook delivery"""

    webhook = webhooks_db.get(webhook_id)

    if not webhook:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Webhook not found"
        )

    webhook["status"] = "active"
    webhook["updated_at"] = time.time()

    return {"message": "Webhook resumed", "status": "active"}


@app.get("/api/webhooks/{webhook_id}/deliveries")
async def get_webhook_deliveries(webhook_id: str, limit: int = 100):
    """
    Get webhook delivery history.

    Shows recent deliveries with status, response code, timestamps.
    """
    webhook = webhooks_db.get(webhook_id)

    if not webhook:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Webhook not found"
        )

    # In production, fetch from deliveries table
    deliveries = [
        {
            "id": "del_123",
            "status": "delivered",
            "status_code": 200,
            "delivered_at": time.time() - 3600,
            "duration_ms": 45
        },
        {
            "id": "del_124",
            "status": "failed",
            "status_code": 503,
            "delivered_at": time.time() - 1800,
            "error": "Server error",
            "retries": 3
        }
    ]

    return {
        "webhook_id": webhook_id,
        "total_deliveries": len(deliveries),
        "deliveries": deliveries[:limit]
    }

# Good webhook payload structure (following Stripe pattern)
{
    "id": "evt_1A2B3C4D5E6F",  # Unique event ID
    "object": "event",  # Object type
    "api_version": "2024-01-15",  # API version for compatibility
    "created": 1705320000,  # Unix timestamp
    "type": "order.created",  # Event type (dot-separated)
    "data": {
        "object": {
            "id": "ord_123456",
            "object": "order",
            "customer": {
                "id": "cus_abc123",
                "email": "john@example.com",
                "name": "John Doe"
            },
            "items": [
                {
                    "id": "item_1",
                    "product": "Widget",
                    "quantity": 2,
                    "price": 1999  # Cents (avoid float precision issues)
                }
            ],
            "total": 3998,
            "currency": "usd",
            "status": "pending",
            "created": 1705320000,
            "metadata": {
                "order_source": "web",
                "campaign_id": "summer_sale"
            }
        },
        "previous_attributes": {}  # For update events
    },
    "livemode": true,  # Production vs test mode
    "pending_webhooks": 2,  # How many webhooks are queued
    "request": {
        "id": "req_xyz789",  # Original API request that triggered event
        "idempotency_key": "idp_abc123"
    }
}

from enum import Enum
from typing import Dict, Any
import time


class EventType(str, Enum):
    """Webhook event types"""
    ORDER_CREATED = "order.created"
    ORDER_UPDATED = "order.updated"
    ORDER_CANCELLED = "order.cancelled"
    ORDER_FULFILLED = "order.fulfilled"
    PAYMENT_SUCCEEDED = "payment.succeeded"
    PAYMENT_FAILED = "payment.failed"
    CUSTOMER_CREATED = "customer.created"
    CUSTOMER_UPDATED = "customer.updated"


def create_webhook_event(
    event_type: EventType,
    data: Dict[str, Any],
    previous_attributes: Dict[str, Any] = None
) -> Dict[str, Any]:
    """
    Create standardized webhook event payload.

    Args:
        event_type: Type of event
        data: Event data (the main object)
        previous_attributes: For update events, what changed

    Returns:
        Formatted webhook payload
    """
    event_id = f"evt_{secrets.token_urlsafe(16)}"

    payload = {
        "id": event_id,
        "object": "event",
        "api_version": "2024-01-15",
        "created": int(time.time()),
        "type": event_type.value,
        "data": {
            "object": data,
            "previous_attributes": previous_attributes or {}
        },
        "livemode": True  # Set based on environment
    }

    return payload


# Example: Order created event
order_data = {
    "id": "ord_123456",
    "object": "order",
    "customer": {
        "id": "cus_abc123",
        "email": "john@example.com",
        "name": "John Doe"
    },
    "total": 9999,
    "currency": "usd",
    "status": "pending",
    "created": int(time.time())
}

event = create_webhook_event(EventType.ORDER_CREATED, order_data)


# Example: Order updated event (with previous attributes)
updated_order = {
    "id": "ord_123456",
    "status": "shipped",  # Changed from "pending"
    "tracking_number": "1Z999AA10123456784"  # New field
}

previous = {
    "status": "pending"  # What changed
}

event = create_webhook_event(
    EventType.ORDER_UPDATED,
    updated_order,
    previous
)

# Payload shows:
# - Current state in data.object
# - What changed in data.previous_attributes
# Client can see: "status changed from pending to shipped"

# Step 1: Install ngrok
# Download from https://ngrok.com or:
brew install ngrok  # macOS
# or
snap install ngrok  # Linux

# Step 2: Start your FastAPI server locally
uvicorn main:app --host 0.0.0.0 --port 8000

# Step 3: Create ngrok tunnel
ngrok http 8000

# Output:
# Forwarding: https://abc123.ngrok.io -> http://localhost:8000

# Step 4: Use ngrok URL for webhook subscription
POST https://api.example.com/api/webhooks/subscribe
{
    "url": "https://abc123.ngrok.io/webhook",  # Public ngrok URL
    "events": ["order.created"]
}

# Step 5: Trigger event
POST https://api.example.com/api/orders
{"customer": "john@example.com", "total": 99.99}

# Step 6: Webhook delivered to your local server!
# Your local FastAPI receives:
POST http://localhost:8000/webhook
{"event": "order.created", "data": {...}}

# ngrok dashboard (http://localhost:4040) shows all webhook requests

from fastapi import FastAPI

app = FastAPI()


@app.post("/api/webhooks/test")
async def trigger_test_webhook(
    webhook_id: str,
    event_type: str,
    test_data: dict = None
):
    """
    Manually trigger webhook for testing.

    Useful for:
    - Local development
    - Integration testing
    - Debugging webhook handlers
    """
    webhook = webhooks_db.get(webhook_id)

    if not webhook:
        raise HTTPException(404, "Webhook not found")

    # Create test event
    if not test_data:
        # Use sample data
        test_data = {
            "id": "test_123",
            "sample": True,
            "message": "This is a test webhook"
        }

    event = create_webhook_event(
        event_type=event_type,
        data=test_data
    )

    # Deliver immediately (synchronous for testing)
    signature = sign_webhook_payload(event, webhook["secret"])

    result = deliver_webhook_with_retry(
        webhook["url"],
        event,
        signature
    )

    return {
        "message": "Test webhook sent",
        "event": event,
        "result": result
    }


# Usage:
POST /api/webhooks/test
{
    "webhook_id": "wh_abc123",
    "event_type": "order.created",
    "test_data": {
        "id": "test_order_1",
        "total": 9999
    }
}

# Response:
{
    "message": "Test webhook sent",
    "event": {...},
    "result": {"status": "delivered", "status_code": 200}
}

# mock_webhook_server.py
from fastapi import FastAPI, Request
import json
import time

app = FastAPI()

received_webhooks = []


@app.post("/webhook")
async def mock_webhook_receiver(request: Request):
    """
    Mock webhook receiver for testing.

    Logs all received webhooks for inspection.
    """
    # Get headers
    headers = dict(request.headers)

    # Get body
    body = await request.json()

    # Store webhook
    webhook_data = {
        "timestamp": time.time(),
        "headers": headers,
        "body": body
    }

    received_webhooks.append(webhook_data)

    print("=" * 60)
    print("WEBHOOK RECEIVED")
    print("=" * 60)
    print(f"Event: {body.get('event')}")
    print(f"Signature: {headers.get('x-webhook-signature', 'MISSING')}")
    print(f"Body: {json.dumps(body, indent=2)}")
    print("=" * 60)

    return {"status": "received"}


@app.get("/webhooks/received")
async def list_received_webhooks():
    """View all received webhooks"""
    return {
        "total": len(received_webhooks),
        "webhooks": received_webhooks
    }


@app.delete("/webhooks/received")
async def clear_received_webhooks():
    """Clear webhook history"""
    received_webhooks.clear()
    return {"message": "Cleared"}


# Run mock server:
# uvicorn mock_webhook_server:app --port 8001

# Subscribe to mock server:
POST /api/webhooks/subscribe
{"url": "http://localhost:8001/webhook", "events": ["order.created"]}

# View received webhooks:
GET http://localhost:8001/webhooks/received

from prometheus_client import Counter, Histogram, Gauge

# Webhook metrics
WEBHOOK_DELIVERIES = Counter(
    'webhook_deliveries_total',
    'Total webhook deliveries',
    ['event_type', 'status']  # labels: order.created, delivered/failed
)

WEBHOOK_DELIVERY_DURATION = Histogram(
    'webhook_delivery_duration_seconds',
    'Webhook delivery duration',
    ['event_type'],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

WEBHOOK_QUEUE_SIZE = Gauge(
    'webhook_queue_size',
    'Webhooks waiting for delivery'
)

WEBHOOK_FAILURES = Counter(
    'webhook_failures_total',
    'Failed webhook deliveries',
    ['event_type', 'error_type']  # timeout, server_error, client_error
)


def deliver_webhook_with_metrics(url: str, payload: dict, signature: str):
    """Deliver webhook with metric tracking"""

    event_type = payload.get('type', 'unknown')

    with WEBHOOK_DELIVERY_DURATION.labels(event_type=event_type).time():
        try:
            response = requests.post(url, json=payload, headers={...}, timeout=5)

            if response.status_code == 200:
                WEBHOOK_DELIVERIES.labels(
                    event_type=event_type,
                    status='delivered'
                ).inc()
                return True
            else:
                WEBHOOK_DELIVERIES.labels(
                    event_type=event_type,
                    status='failed'
                ).inc()

                if 400 <= response.status_code < 500:
                    error_type = 'client_error'
                else:
                    error_type = 'server_error'

                WEBHOOK_FAILURES.labels(
                    event_type=event_type,
                    error_type=error_type
                ).inc()

                return False

        except requests.exceptions.Timeout:
            WEBHOOK_FAILURES.labels(
                event_type=event_type,
                error_type='timeout'
            ).inc()
            return False


# Grafana dashboard queries:
# - Delivery success rate:
#   sum(rate(webhook_deliveries_total{status="delivered"}[5m])) /
#   sum(rate(webhook_deliveries_total[5m]))
#
# - P95 delivery latency:
#   histogram_quantile(0.95, rate(webhook_delivery_duration_seconds_bucket[5m]))
#
# - Failure rate by error type:
#   sum by (error_type) (rate(webhook_failures_total[5m]))