webhook-integration-patterns

What this skill does

# Webhook Integration Patterns

This skill provides guidance for designing and implementing robust webhook systems—both as providers (sending webhooks) and consumers (receiving webhooks).

## Core Competencies

- **Delivery Guarantees**: At-least-once, exactly-once semantics
- **Security**: Signature verification, secret rotation
- **Reliability**: Retry strategies, dead letter handling
- **Scalability**: Queue-based processing, rate limiting

## Webhook Fundamentals

### What Webhooks Solve

```
Polling (inefficient):           Webhooks (efficient):
┌────────┐     ┌────────┐       ┌────────┐     ┌────────┐
│ Client │     │ Server │       │ Client │     │ Server │
└───┬────┘     └───┬────┘       └───┬────┘     └───┬────┘
    │ Any news?    │                │              │
    │─────────────▶│                │              │
    │    No        │                │   Event!     │
    │◀─────────────│                │◀─────────────│
    │ Any news?    │                │  POST /hook  │
    │─────────────▶│                │◀─────────────│
    │    No        │                │   200 OK     │
    │◀─────────────│                │─────────────▶│
    │ Any news?    │
    │─────────────▶│       Push-based notification
    │    YES!      │       instead of polling
    │◀─────────────│
```

### Webhook Anatomy

```http
POST /webhooks/payment HTTP/1.1
Host: your-app.com
Content-Type: application/json
X-Webhook-Signature: sha256=abc123...
X-Webhook-ID: evt_12345
X-Webhook-Timestamp: 1706616000

{
  "id": "evt_12345",
  "type": "payment.completed",
  "created_at": "2024-01-30T12:00:00Z",
  "data": {
    "payment_id": "pay_abc",
    "amount": 1000,
    "currency": "USD"
  }
}
```

Key headers:
- **Signature**: Cryptographic proof of authenticity
- **ID**: Unique event identifier for deduplication
- **Timestamp**: When the event occurred

## Webhook Provider Design

### Event Schema Design

```python
class WebhookEvent:
    """Standard webhook event structure"""

    def __init__(self, event_type, data, idempotency_key=None):
        self.id = self._generate_id()
        self.type = event_type
        self.created_at = datetime.utcnow().isoformat()
        self.api_version = "2024-01-30"
        self.data = data
        self.idempotency_key = idempotency_key or self.id

    def to_payload(self):
        return {
            "id": self.id,
            "type": self.type,
            "created_at": self.created_at,
            "api_version": self.api_version,
            "data": self.data
        }

# Event types follow resource.action pattern
EVENT_TYPES = [
    "payment.created",
    "payment.completed",
    "payment.failed",
    "subscription.created",
    "subscription.updated",
    "subscription.cancelled",
    "customer.created",
    "customer.deleted"
]
```

### Signature Generation

```python
import hmac
import hashlib
import time

class WebhookSigner:
    """Sign webhook payloads for verification"""

    def __init__(self, secret):  # allow-secret
        self.secret = secret.encode()  # allow-secret

    def sign(self, payload, timestamp=None):
        """Generate HMAC signature"""
        timestamp = timestamp or int(time.time())
        payload_str = json.dumps(payload, separators=(',', ':'))

        # Include timestamp to prevent replay attacks
        signed_payload = f"{timestamp}.{payload_str}"

        signature = hmac.new(
            self.secret,
            signed_payload.encode(),
            hashlib.sha256
        ).hexdigest()

        return {
            'signature': f"sha256={signature}",
            'timestamp': timestamp
        }

    def create_headers(self, payload):
        """Generate all webhook headers"""
        sign_data = self.sign(payload)
        return {
            'Content-Type': 'application/json',
            'X-Webhook-Signature': sign_data['signature'],
            'X-Webhook-Timestamp': str(sign_data['timestamp']),
            'X-Webhook-ID': payload['id']
        }
```

### Delivery System

```python
import asyncio
from datetime import datetime, timedelta

class WebhookDeliverySystem:
    """Reliable webhook delivery with retries"""

    RETRY_SCHEDULE = [
        timedelta(seconds=10),
        timedelta(minutes=1),
        timedelta(minutes=5),
        timedelta(minutes=30),
        timedelta(hours=1),
        timedelta(hours=6),
        timedelta(hours=24)
    ]

    def __init__(self, signer, http_client):
        self.signer = signer
        self.http = http_client
        self.delivery_log = []

    async def deliver(self, endpoint, event, attempt=0):
        """Attempt webhook delivery"""
        payload = event.to_payload()
        headers = self.signer.create_headers(payload)

        try:
            response = await self.http.post(
                endpoint.url,
                json=payload,
                headers=headers,
                timeout=30
            )

            self._log_attempt(endpoint, event, attempt, response)

            if response.status_code in (200, 201, 202, 204):
                return {'status': 'delivered', 'attempts': attempt + 1}

            # Non-success status - schedule retry
            return await self._schedule_retry(endpoint, event, attempt)

        except Exception as e:
            self._log_attempt(endpoint, event, attempt, error=e)
            return await self._schedule_retry(endpoint, event, attempt)

    async def _schedule_retry(self, endpoint, event, attempt):
        """Schedule next retry or give up"""
        if attempt >= len(self.RETRY_SCHEDULE):
            self._move_to_dead_letter(endpoint, event)
            return {'status': 'failed', 'attempts': attempt + 1}

        delay = self.RETRY_SCHEDULE[attempt]
        # In production: use job queue with delayed execution
        await asyncio.sleep(delay.total_seconds())
        return await self.deliver(endpoint, event, attempt + 1)

    def _move_to_dead_letter(self, endpoint, event):
        """Store failed webhook for manual review"""
        # Store in dead letter queue/table
        pass
```

### Endpoint Management

```python
class WebhookEndpoint:
    """Subscriber endpoint configuration"""

    def __init__(self, url, events, secret=None):  # allow-secret
        self.id = generate_id()
        self.url = url
        self.events = events  # List of subscribed event types
        self.secret = secret or generate_secret()  # allow-secret
        self.status = 'active'
        self.created_at = datetime.utcnow()

        # Health tracking
        self.consecutive_failures = 0
        self.last_success = None
        self.last_failure = None

    def should_receive(self, event_type):
        """Check if endpoint subscribes to this event"""
        if '*' in self.events:
            return True
        return event_type in self.events

    def record_success(self):
        self.consecutive_failures = 0
        self.last_success = datetime.utcnow()
        if self.status == 'disabled':
            self.status = 'active'

    def record_failure(self):
        self.consecutive_failures += 1
        self.last_failure = datetime.utcnow()

        # Auto-disable after too many failures
        if self.consecutive_failures >= 10:
            self.status = 'disabled'
```

## Webhook Consumer Design

### Signature Verification

```python
class WebhookVerifier:
    """Verify incoming webhook signatures"""

    TIMESTAMP_TOLERANCE = 300  # 5 minutes

    def __init__(self, secret):  # allow-secret
        self.secret = secret.encode()  # allow-secret

    def verify(self, payload, signature, timestamp):
        """Verify webhook authenticity"""
        # Check timestamp freshness
        current_time = int(time.time())
        if abs(current_time - int(timestamp)) > self.TIMESTAMP_TOLERANCE:
            raise WebhookVerificationError("Timestamp too old")

        # Compute expected signature
        signed_payload = f"{timestamp}.{payload}"
        expected = hmac.new(
            self.secret,
            signed_p