Multi-Tenant Webhook Delivery Platform

Introduction

30 Second Summary

Every time you receive a payment confirmation from Stripe or a deployment notification from GitHub, a webhook delivery system is working behind the scenes. These systems must handle failures gracefully, prevent duplicate deliveries, and isolate tenants from each other's problems.

In this project, you will build a multi-tenant webhook delivery platform with production-grade reliability patterns running locally via Docker Compose. You'll implement circuit breakers, idempotency, rate limiting, and dead-letter queues.

What You'll Build

A webhook delivery platform that accepts events from tenants, signs payloads with HMAC-SHA256, delivers them to registered endpoints with retry logic, and handles destination failures using per-URL circuit breakers and a Redis Lua token bucket rate limiter.

By the end of this project, you'll have:

• A FastAPI REST API for tenant registration, webhook subscription, and event ingestion with Celery async task processing.
• A delivery engine with HMAC-SHA256 payload signing, exponential backoff retries with jitter, and dead-letter routing to MongoDB.
• Custom Redis-backed reliability patterns including per-destination circuit breakers, idempotency via SET NX, and an atomic Lua token bucket rate limiter for per-tenant isolation.
• Secret Mission: A delivery analytics API that tracks per-tenant success rates, failure counts, and p95 latency using Redis sorted sets.

Are there any prerequisites?

This is an advanced project. You should be comfortable with FastAPI, Celery, Redis, RabbitMQ, MongoDB, and Docker Compose before starting.

You'll also need Python 3.11, PyCharm, Docker Desktop, and Git connected to GitHub already installed on your machine.

Set Up the Project and Docker Compose

A webhook delivery platform requires multiple infrastructure services working together. Your API needs a message broker to queue delivery tasks, a cache layer for rate limiting and circuit breaker state, and a database for persistent storage.

In this step, you will create the project structure, install dependencies, and spin up RabbitMQ, Redis, and MongoDB using Docker Compose. By the end, your FastAPI health endpoint will confirm all services are connected.

In this step, get ready to:

• Create the project directory, virtual environment, and install Python dependencies.
• Write a Docker Compose file to orchestrate RabbitMQ, Redis, and MongoDB.
• Build a FastAPI app skeleton with a health endpoint that verifies all service connections.

Create the project directory and install dependencies

Every file in this project lives inside a single folder on your Desktop. You will also set up a Python virtual environment to isolate your project's packages from the rest of your system.

• Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create the project directory structure by running these commands:

mkdir webhook-platform
cd webhook-platform
mkdir -p app/services
touch app/__init__.py app/services/__init__.py

What does this create?

The app/ directory is your Python package containing all application code. The app/services/ subdirectory will hold the circuit breaker and rate limiter modules. The __init__.py files mark these directories as Python packages so imports work correctly.

• Create and activate a Python virtual environment by running:

python3.11 -m venv venv
source venv/bin/activate

What is a virtual environment?

A virtual environment creates an isolated Python installation for your project. Packages you install here won't conflict with other projects on your machine. Your terminal prompt should now show (venv) at the beginning.

• Create the requirements.txt file that lists all project dependencies. Paste this into a new file called requirements.txt in your project root:

fastapi==0.139.0
uvicorn==0.34.3
celery==5.6.3
httpx==0.28.1
pymongo==4.17.0
redis==5.2.1
pydantic-settings==2.9.1

What are these packages?

• fastapi is your REST API framework.
• uvicorn is the ASGI server that runs FastAPI.
• celery handles async task processing for webhook delivery.
• httpx is the HTTP client for delivering webhooks to endpoints.
• pymongo connects to MongoDB for persistent storage.
• redis connects to Redis for caching, rate limiting, and circuit breaker state.
• pydantic-settings manages configuration using environment variables and defaults.

• Install all dependencies by running:

pip install -r requirements.txt

• Confirm the installation succeeded by running:

pip list | grep fastapi

• You should see fastapi with version 0.139.0 in the output.

Installation failed?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If not, run source venv/bin/activate again. Also confirm you are using Python 3.11 by checking python --version inside the venv.

Help me debug my pip install failure

Write the Docker Compose file

Your webhook platform depends on three infrastructure services. Docker Compose lets you define and run all three with a single command.

• Create a file called docker-compose.yml in the root of your webhook-platform folder and paste the following:

services:
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "check_port_connectivity"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

  mongodb:
    image: mongo:7
    ports:
      - "27017:27017"
    healthcheck:
      test: ["CMD", "mongosh", "--eval", "db.adminCommand('ping')"]
      interval: 10s
      timeout: 5s
      retries: 5

What does this configure?

• rabbitmq:3-management includes the management UI on port 15672 and the AMQP protocol on port 5672. Celery uses AMQP to queue tasks.
• redis:7-alpine is a lightweight Redis image. Redis stores circuit breaker state, rate limiter buckets, and idempotency keys.
• mongo:7 runs MongoDB for tenants, subscriptions, events, and dead-letter storage.
• Each service has a healthcheck so Docker can report whether the service is ready to accept connections.

• Save docker-compose.yml.

Docker Desktop not running?

Make sure Docker Desktop is running before proceeding. Look for the Docker whale icon in your system tray (macOS) or taskbar (Windows). If it is not there, open Docker Desktop from your applications.

Help me get Docker Desktop running

Start Docker services and create the FastAPI app skeleton

With the Compose file saved, you can now start all three infrastructure services. Then you will create the FastAPI application that connects to them.

• Start all services in detached mode by running:

docker compose up -d

• Wait about 15-20 seconds for the health checks to pass, then verify all containers are healthy by running:

docker compose ps

• You should see all three services with a STATUS of Up and (healthy) in the output.

Containers not showing healthy?

Health checks take time. Wait 30 seconds and run docker compose ps again. If a container shows (health: starting), it is still initializing. If it shows Exit, check logs with docker compose logs <service-name>. A common cause is a port conflict if another service is already using ports 5672, 6379, or 27017.

Help me troubleshoot my Docker containers

Now you will create the application files that form the FastAPI skeleton. Start with the configuration module.

• Create app/config.py with the following content:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    mongodb_url: str = "mongodb://localhost:27017"
    mongodb_db: str = "webhook_platform"
    redis_url: str = "redis://localhost:6379/0"
    rabbitmq_url: str = "amqp://guest:guest@localhost:5672//"
    rate_limit_tokens: int = 100
    rate_limit_refill_rate: float = 1.67
    circuit_breaker_failure_threshold: int = 5
    circuit_breaker_recovery_timeout: int = 30


settings = Settings()

What does this code do?

The Settings class uses pydantic-settings to define all configuration values with sensible defaults. Each field can be overridden via environment variables. The settings instance at the bottom is a singleton used across the entire application.

The rate_limit_tokens and rate_limit_refill_rate configure the token bucket (100 tokens refilling at ~1.67 per second = ~100 per minute). The circuit breaker opens after 5 consecutive failures and attempts recovery after 30 seconds.

• Create app/database.py with the following content:

import redis
from pymongo import MongoClient
from app.config import settings

mongo_client = MongoClient(settings.mongodb_url)
db = mongo_client[settings.mongodb_db]

redis_client = redis.Redis.from_url(settings.redis_url, decode_responses=True)

What does this code do?

This module creates the database connections that the rest of the app imports. The db object gives you access to MongoDB collections. The redis_client connects to Redis with decode_responses=True so it returns strings instead of raw bytes.

• Create app/models.py with the Pydantic models that define your API's request and response shapes:

from pydantic import BaseModel, Field
from typing import Optional


class TenantCreate(BaseModel):
    name: str


class TenantResponse(BaseModel):
    tenant_id: str
    name: str
    api_key: str
    webhook_secret: str


class SubscriptionCreate(BaseModel):
    url: str
    event_types: list[str]


class SubscriptionResponse(BaseModel):
    subscription_id: str
    tenant_id: str
    url: str
    event_types: list[str]


class EventCreate(BaseModel):
    event_type: str
    payload: dict


class RateLimitStatus(BaseModel):
    tenant_id: str
    tokens_remaining: float
    allowed: bool
    retry_after_ms: Optional[int] = None

What does this code do?

These Pydantic models define the shape of data flowing in and out of your API. TenantCreate and TenantResponse handle tenant registration. SubscriptionCreate and SubscriptionResponse handle webhook subscriptions. EventCreate is the payload tenants send when publishing events.

• Create app/worker.py as a placeholder. The full delivery engine will be built in Step 3, but the main app needs to import from this file:

from celery import Celery
from app.config import settings

# Initialize the Celery app with RabbitMQ as the broker
celery_app = Celery("webhook_worker", broker=settings.rabbitmq_url)
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,
    worker_prefetch_multiplier=1,
)


@celery_app.task(bind=True, max_retries=5, name="deliver_webhook")
def deliver_webhook(self, **kwargs):
    # Placeholder - full delivery logic implemented in Step 3
    pass

Why a placeholder?

The main FastAPI app imports deliver_webhook to enqueue delivery tasks. This stub defines the Celery task so the import succeeds. You will replace it with the full delivery engine (HMAC signing, retries, dead-letter routing) in Step 3.

• Create app/main.py with the FastAPI app and health endpoint:

import secrets
from datetime import datetime, timezone

from bson import ObjectId
from fastapi import FastAPI, Header, HTTPException

from app.config import settings
from app.database import db, redis_client
from app.models import (
    EventCreate,
    RateLimitStatus,
    SubscriptionCreate,
    SubscriptionResponse,
    TenantCreate,
    TenantResponse,
)
from app.worker import deliver_webhook

app = FastAPI(title="Webhook Delivery Platform")


@app.get("/health")
def health_check():
    try:
        redis_client.ping()
        redis_status = "connected"
    except Exception:
        redis_status = "disconnected"

    try:
        db.command("ping")
        mongo_status = "connected"
    except Exception:
        mongo_status = "disconnected"

    return {"status": "ok", "redis": redis_status, "mongodb": mongo_status}

What does this code do?

The /health endpoint pings both Redis and MongoDB to confirm they are reachable. It returns the connection status for each in a JSON response. The imports at the top include models and worker references that will be used by the API endpoints you build in Step 2.

• Save all files.
• Start the FastAPI development server by running:

uvicorn app.main:app --reload --port 8000

• Open your browser and navigate to http://localhost:8000/health.
• You should see: {"status":"ok","redis":"connected","mongodb":"connected"}.

Seeing disconnected for Redis or MongoDB?

Confirm your Docker containers are still running with docker compose ps. If a container restarted or exited, bring it back with docker compose up -d. Also make sure your app/config.py has the correct default URLs matching the ports in docker-compose.yml.

Help me debug my health endpoint

• Open http://localhost:15672 in your browser to confirm the RabbitMQ management UI is accessible.
• Log in with username guest and password guest.

Why guest/guest?

RabbitMQ ships with a default user guest and password guest that can only connect from localhost. Since your Docker container maps to localhost, these credentials work for local development. In production, you would create dedicated users with strong credentials.

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files and that your health endpoint returns both services as connected.

ⓧ I'd like to double check the full code

Here are all the files you should have at this point:

fastapi==0.139.0
uvicorn==0.34.3
celery==5.6.3
httpx==0.28.1
pymongo==4.17.0
redis==5.2.1
pydantic-settings==2.9.1

services:
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "check_port_connectivity"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

  mongodb:
    image: mongo:7
    ports:
      - "27017:27017"
    healthcheck:
      test: ["CMD", "mongosh", "--eval", "db.adminCommand('ping')"]
      interval: 10s
      timeout: 5s
      retries: 5

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    mongodb_url: str = "mongodb://localhost:27017"
    mongodb_db: str = "webhook_platform"
    redis_url: str = "redis://localhost:6379/0"
    rabbitmq_url: str = "amqp://guest:guest@localhost:5672//"
    rate_limit_tokens: int = 100
    rate_limit_refill_rate: float = 1.67
    circuit_breaker_failure_threshold: int = 5
    circuit_breaker_recovery_timeout: int = 30


settings = Settings()

import redis
from pymongo import MongoClient
from app.config import settings

mongo_client = MongoClient(settings.mongodb_url)
db = mongo_client[settings.mongodb_db]

redis_client = redis.Redis.from_url(settings.redis_url, decode_responses=True)

from pydantic import BaseModel, Field
from typing import Optional


class TenantCreate(BaseModel):
    name: str


class TenantResponse(BaseModel):
    tenant_id: str
    name: str
    api_key: str
    webhook_secret: str


class SubscriptionCreate(BaseModel):
    url: str
    event_types: list[str]


class SubscriptionResponse(BaseModel):
    subscription_id: str
    tenant_id: str
    url: str
    event_types: list[str]


class EventCreate(BaseModel):
    event_type: str
    payload: dict


class RateLimitStatus(BaseModel):
    tenant_id: str
    tokens_remaining: float
    allowed: bool
    retry_after_ms: Optional[int] = None

import secrets
from datetime import datetime, timezone

from bson import ObjectId
from fastapi import FastAPI, Header, HTTPException

from app.config import settings
from app.database import db, redis_client
from app.models import (
    EventCreate,
    RateLimitStatus,
    SubscriptionCreate,
    SubscriptionResponse,
    TenantCreate,
    TenantResponse,
)
from app.worker import deliver_webhook

app = FastAPI(title="Webhook Delivery Platform")


@app.get("/health")
def health_check():
    try:
        redis_client.ping()
        redis_status = "connected"
    except Exception:
        redis_status = "disconnected"

    try:
        db.command("ping")
        mongo_status = "connected"
    except Exception:
        mongo_status = "disconnected"

    return {"status": "ok", "redis": redis_status, "mongodb": mongo_status}

Your foundation is solid. All three infrastructure services are healthy, and your FastAPI app can talk to each of them. Next up, you will build the core API endpoints for registering tenants, creating webhook subscriptions, and ingesting events.

Build the Core API — Tenants, Subscriptions, and Events

Your infrastructure is running and the health endpoint proves connectivity. Now you need the API surface that makes this platform useful to tenants.

A webhook delivery platform must accept three things from the outside world: tenant registrations (who is sending events), subscription configurations (where should events go), and the events themselves. In this step, you will build all three endpoints and wire them into the async delivery pipeline.

In this step, get ready to:

• Build tenant registration and subscription endpoints that store data in MongoDB.
• Build an event ingestion endpoint that validates API keys and enqueues Celery tasks.
• Test the full flow end-to-end and confirm tasks land in RabbitMQ.

Build the tenant registration endpoint

Every tenant needs an API key for authentication and a webhook secret for HMAC signing. The registration endpoint generates both and stores them in MongoDB.

• In app/main.py, find the closing return statement of the health_check function.
• Add the following tenant registration endpoint below it:

@app.post("/tenants", response_model=TenantResponse)
def create_tenant(tenant: TenantCreate):
    # Build the tenant document with auto-generated credentials
    tenant_doc = {
        "name": tenant.name,
        "api_key": secrets.token_hex(16),
        "webhook_secret": secrets.token_hex(32),
        "created_at": datetime.now(timezone.utc),
    }
    result = db.tenants.insert_one(tenant_doc)
    return TenantResponse(
        tenant_id=str(result.inserted_id),
        name=tenant_doc["name"],
        api_key=tenant_doc["api_key"],
        webhook_secret=tenant_doc["webhook_secret"],
    )

What does this code do?

This endpoint receives a tenant name and generates two secrets. The api_key (16 bytes hex) is what tenants send with every event to authenticate. The webhook_secret (32 bytes hex) is used later to sign outgoing payloads with HMAC-SHA256.

Both secrets are generated with secrets.token_hex, which is cryptographically secure. MongoDB's insert_one returns the auto-generated _id which becomes the tenant_id.

Next, tenants need a way to register webhook destinations. The subscription endpoint links a URL and event types to a specific tenant.

• Add the subscription endpoint directly below the create_tenant function:

@app.post("/tenants/{tenant_id}/subscriptions", response_model=SubscriptionResponse)
def create_subscription(tenant_id: str, subscription: SubscriptionCreate):
    # Verify the tenant exists before creating a subscription
    tenant = db.tenants.find_one({"_id": ObjectId(tenant_id)})
    if not tenant:
        raise HTTPException(status_code=404, detail="Tenant not found")

    sub_doc = {
        "tenant_id": tenant_id,
        "url": subscription.url,
        "event_types": subscription.event_types,
        "created_at": datetime.now(timezone.utc),
    }
    result = db.subscriptions.insert_one(sub_doc)
    return SubscriptionResponse(
        subscription_id=str(result.inserted_id),
        tenant_id=tenant_id,
        url=sub_doc["url"],
        event_types=sub_doc["event_types"],
    )

What does this code do?

This endpoint takes a tenant_id path parameter and validates it against MongoDB before creating the subscription. Each subscription stores the destination url and the event_types it listens for.

When an event arrives later, the platform will query subscriptions by tenant_id and event_type to determine which URLs should receive it.

• Save app/main.py.
• Your server should auto-reload. You will not see a visible change yet because we have not tested the endpoints.

Seeing an import error on reload?

If uvicorn shows an error about importing deliver_webhook from app.worker, that is expected at this point. The app/main.py file imports from a worker stub that we will update in the next substep.

Help me debug this import error

Build the event ingestion endpoint

The event ingestion endpoint is the core of the platform. It authenticates the tenant, finds matching subscriptions, and enqueues a delivery task for each one. Before we can call deliver_webhook.delay(), we need a proper Celery task definition in app/worker.py.

• Open app/worker.py and replace its entire contents with the following code:

import hashlib
import hmac
import json
import logging
import random
import time
from datetime import datetime, timezone

import httpx
from celery import Celery

from app.config import settings
from app.database import db, redis_client

logger = logging.getLogger(__name__)

# Create the Celery app connected to RabbitMQ
celery_app = Celery("webhook_worker", broker=settings.rabbitmq_url)
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,
    worker_prefetch_multiplier=1,
)


@celery_app.task(bind=True, max_retries=5, name="deliver_webhook")
def deliver_webhook(
    self,
    tenant_id: str,
    delivery_id: str,
    url: str,
    event_type: str,
    payload: dict,
    webhook_secret: str,
):
    pass

What does this code do?

This creates the Celery application and connects it to RabbitMQ using the broker URL from your settings. The configuration sets JSON serialization and enables task_acks_late so tasks are only acknowledged after completion (preventing message loss on crashes).

The deliver_webhook task is a placeholder for now. It accepts all the parameters needed for delivery but does nothing yet. The bind=True parameter gives the task access to self for retry logic we will add in the next step.

• Save app/worker.py.

✔️ Awesome, I've got everything!

Great. Double check that you have saved app/worker.py.

ⓧ I'd like to double check the full code

import hashlib
import hmac
import json
import logging
import random
import time
from datetime import datetime, timezone

import httpx
from celery import Celery

from app.config import settings
from app.database import db, redis_client

logger = logging.getLogger(__name__)

celery_app = Celery("webhook_worker", broker=settings.rabbitmq_url)
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,
    worker_prefetch_multiplier=1,
)


@celery_app.task(bind=True, max_retries=5, name="deliver_webhook")
def deliver_webhook(
    self,
    tenant_id: str,
    delivery_id: str,
    url: str,
    event_type: str,
    payload: dict,
    webhook_secret: str,
):
    pass

Now add the event ingestion endpoint to app/main.py.

• In app/main.py, add the following endpoint below the create_subscription function:

@app.post("/events", status_code=202)
def ingest_event(event: EventCreate, x_api_key: str = Header()):
    # Authenticate the tenant by API key
    tenant = db.tenants.find_one({"api_key": x_api_key})
    if not tenant:
        raise HTTPException(status_code=401, detail="Invalid API key")

    tenant_id = str(tenant["_id"])
    # Find all subscriptions matching this event type
    subscriptions = db.subscriptions.find(
        {"tenant_id": tenant_id, "event_types": event.event_type}
    )

    # Store the event in MongoDB
    event_doc = {
        "tenant_id": tenant_id,
        "event_type": event.event_type,
        "payload": event.payload,
        "created_at": datetime.now(timezone.utc),
    }
    event_result = db.events.insert_one(event_doc)
    event_id = str(event_result.inserted_id)

    # Enqueue a delivery task for each matching subscription
    delivery_count = 0
    for sub in subscriptions:
        delivery_id = f"{event_id}:{sub['_id']}"
        deliver_webhook.delay(
            tenant_id=tenant_id,
            delivery_id=delivery_id,
            url=sub["url"],
            event_type=event.event_type,
            payload=event.payload,
            webhook_secret=tenant["webhook_secret"],
        )
        delivery_count += 1

    return {"event_id": event_id, "deliveries_queued": delivery_count}

What does this code do?

The endpoint reads the X-Api-Key header and looks up the tenant in MongoDB. If the key is invalid, it returns a 401.

It queries the subscriptions collection for entries matching both the tenant_id and the incoming event_type. For each match, it calls deliver_webhook.delay() which publishes a task message to RabbitMQ without blocking the HTTP response.

The response returns HTTP 202 (Accepted) because delivery is asynchronous. The caller knows the event was received, but delivery happens in the background.

• Save app/main.py.
• Uvicorn should auto-reload without errors now that app/worker.py exports the proper Celery task.

Server still crashing on reload?

Make sure you saved both app/worker.py and app/main.py. The import chain requires worker.py to define deliver_webhook before main.py can import it.

Also confirm your Docker containers are still running. The Celery app connects to RabbitMQ on import.

Help me fix my server crash

✔️ Awesome, I've got everything!

Great. Double check that you have saved both files.

ⓧ I'd like to double check the full code

import secrets
from datetime import datetime, timezone

from bson import ObjectId
from fastapi import FastAPI, Header, HTTPException

from app.config import settings
from app.database import db, redis_client
from app.models import (
    EventCreate,
    RateLimitStatus,
    SubscriptionCreate,
    SubscriptionResponse,
    TenantCreate,
    TenantResponse,
)
from app.worker import deliver_webhook

app = FastAPI(title="Webhook Delivery Platform")


@app.get("/health")
def health_check():
    try:
        redis_client.ping()
        redis_status = "connected"
    except Exception:
        redis_status = "disconnected"

    try:
        db.command("ping")
        mongo_status = "connected"
    except Exception:
        mongo_status = "disconnected"

    return {"status": "ok", "redis": redis_status, "mongodb": mongo_status}


@app.post("/tenants", response_model=TenantResponse)
def create_tenant(tenant: TenantCreate):
    tenant_doc = {
        "name": tenant.name,
        "api_key": secrets.token_hex(16),
        "webhook_secret": secrets.token_hex(32),
        "created_at": datetime.now(timezone.utc),
    }
    result = db.tenants.insert_one(tenant_doc)
    return TenantResponse(
        tenant_id=str(result.inserted_id),
        name=tenant_doc["name"],
        api_key=tenant_doc["api_key"],
        webhook_secret=tenant_doc["webhook_secret"],
    )


@app.post("/tenants/{tenant_id}/subscriptions", response_model=SubscriptionResponse)
def create_subscription(tenant_id: str, subscription: SubscriptionCreate):
    tenant = db.tenants.find_one({"_id": ObjectId(tenant_id)})
    if not tenant:
        raise HTTPException(status_code=404, detail="Tenant not found")

    sub_doc = {
        "tenant_id": tenant_id,
        "url": subscription.url,
        "event_types": subscription.event_types,
        "created_at": datetime.now(timezone.utc),
    }
    result = db.subscriptions.insert_one(sub_doc)
    return SubscriptionResponse(
        subscription_id=str(result.inserted_id),
        tenant_id=tenant_id,
        url=sub_doc["url"],
        event_types=sub_doc["event_types"],
    )


@app.post("/events", status_code=202)
def ingest_event(event: EventCreate, x_api_key: str = Header()):
    tenant = db.tenants.find_one({"api_key": x_api_key})
    if not tenant:
        raise HTTPException(status_code=401, detail="Invalid API key")

    tenant_id = str(tenant["_id"])
    subscriptions = db.subscriptions.find(
        {"tenant_id": tenant_id, "event_types": event.event_type}
    )

    event_doc = {
        "tenant_id": tenant_id,
        "event_type": event.event_type,
        "payload": event.payload,
        "created_at": datetime.now(timezone.utc),
    }
    event_result = db.events.insert_one(event_doc)
    event_id = str(event_result.inserted_id)

    delivery_count = 0
    for sub in subscriptions:
        delivery_id = f"{event_id}:{sub['_id']}"
        deliver_webhook.delay(
            tenant_id=tenant_id,
            delivery_id=delivery_id,
            url=sub["url"],
            event_type=event.event_type,
            payload=event.payload,
            webhook_secret=tenant["webhook_secret"],
        )
        delivery_count += 1

    return {"event_id": event_id, "deliveries_queued": delivery_count}

Test the flow end-to-end

Time to prove the whole chain works. You will register a tenant, add a subscription, publish an event, and confirm the task lands in RabbitMQ.

• Open a new terminal in your project directory (keep uvicorn running in the first one).
• Register a test tenant by running:

curl -X POST http://localhost:8000/tenants -H "Content-Type: application/json" -d '{"name": "test-company"}'

You should see a JSON response containing tenant_id, api_key, and webhook_secret. Copy the tenant_id and api_key values for the next commands.

• Create a subscription for that tenant by running:

curl -X POST http://localhost:8000/tenants/[[TENANT_ID="your tenant_id from above"]]/subscriptions -H "Content-Type: application/json" -d '{"url": "https://httpbin.org/post", "event_types": ["order.created"]}'

You should see a response with subscription_id, url, and event_types.

• Publish an event to trigger the delivery pipeline by running:

curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -H "X-Api-Key: [[API_KEY="your api_key from the tenant response"]]" -d '{"event_type": "order.created", "payload": {"order_id": "123", "amount": 99.99}}'

You should see a response like {"event_id": "...", "deliveries_queued": 1}. This confirms one task was enqueued.

• Open the RabbitMQ management UI in your browser at http://localhost:15672.
• Log in with username guest and password guest.
• Click the Queues tab. You should see a queue with at least one message ready (the task you just published).

Why is the task sitting in the queue?

You have not started a Celery worker yet. When deliver_webhook.delay() is called, it publishes a message to RabbitMQ. The message stays in the queue until a worker consumes it. You will start the worker in the next step when you build the delivery engine.

Not seeing any messages in the queue?

Make sure the api_key in your curl command matches the one returned by POST /tenants. A mismatched key returns 401 and no task is enqueued.

Also verify your event_type matches what you registered in the subscription (order.created). If the types do not match, zero subscriptions are found and zero tasks are enqueued.

Help me debug my missing queue messages

Your API surface is complete. Tenants can register, subscribe to event types, and publish events that get queued for delivery. Next up, you will build the delivery engine that actually sends those webhooks with HMAC signatures and retry logic.

Build the Delivery Engine — HMAC Signing, Retry, and Backoff

Your API can accept events and queue them for delivery, but the delivery task itself is still a stub.

In this step, you will build the actual delivery engine. This means signing each webhook payload with HMAC-SHA256 so recipients can verify authenticity, delivering it over HTTP with proper headers, and handling every failure scenario gracefully with retries, backoff, and a dead-letter queue.

In this step, get ready to:

• Implement HMAC-SHA256 payload signing and HTTP delivery with custom headers.
• Add retry logic with exponential backoff, jitter, and status-code-aware handling.
• Route permanently failed deliveries to a dead-letter queue in MongoDB.

Implement HMAC-SHA256 signing and HTTP delivery

Webhook recipients need a way to verify that an incoming payload actually came from your platform and was not tampered with in transit. The standard approach is to sign the serialized payload body with a shared secret using HMAC-SHA256, then include the signature in a request header.

• In app/worker.py, find the line that says pass inside the deliver_webhook function.
• Delete pass entirely. You will replace it with real delivery logic across this step.
• First, add the sign_payload helper function. Place it between the celery_app.conf.update(...) block and the @celery_app.task decorator by copying this code:

def sign_payload(payload: dict, secret: str) -> str:
    # Serialize to a deterministic JSON string
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    # Compute HMAC-SHA256 using the tenant's webhook secret
    signature = hmac.new(
        secret.encode("utf-8"),
        body.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    return f"sha256={signature}"

What does this code do?

json.dumps with separators=(",", ":") and sort_keys=True produces a compact, deterministic JSON string. Both sender and receiver must serialize identically for the signature to match.

hmac.new computes a keyed hash using SHA-256. The result is prefixed with sha256= so the receiver knows which algorithm to use for verification. This is the same pattern used by GitHub, Stripe, and Shopify webhooks.

• Now replace the deleted pass inside deliver_webhook with the signing and delivery logic. Copy this code into the function body:

    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = sign_payload(payload, webhook_secret)

    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Signature": signature,
        "X-Delivery-Id": delivery_id,
        "X-Event-Type": event_type,
    }

    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(url, content=body, headers=headers)

        if response.status_code < 300:
            return {"status": "delivered", "delivery_id": delivery_id}

    except httpx.TimeoutException:
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except self.MaxRetriesExceededError:
        _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, "Max retries exceeded", self.request.retries)

What does this code do?

The function serializes the payload to JSON, computes the HMAC signature, then fires an HTTP POST to the destination URL with a 5-second timeout.

Three custom headers ride along with every delivery: X-Webhook-Signature for verification, X-Delivery-Id for deduplication on the receiver side, and X-Event-Type so the receiver can route by event without parsing the body.

If the response status is below 300, delivery succeeded. The TimeoutException and MaxRetriesExceededError handlers are placeholders you will flesh out in the next substeps.

Do not save yet. You will add error handling in the next substep before testing.

Add retry logic with exponential backoff and jitter

A single failed delivery should not be the end of the story. Different HTTP status codes mean different things, and your delivery engine needs to react accordingly.

• Inside the try block of deliver_webhook, find the if response.status_code < 300: check. Add the following code directly below its return statement:

        if response.status_code == 429:
            # Rate limited by destination - respect their Retry-After header
            retry_after = int(response.headers.get("Retry-After", "10"))
            redis_client.delete(f"delivery:{delivery_id}")
            raise self.retry(countdown=retry_after)

        if 400 <= response.status_code < 500:
            # Client error (not rate limit) - permanent failure, no retry
            _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, f"HTTP {response.status_code}", self.request.retries)
            return {"status": "permanent_failure", "delivery_id": delivery_id}

        # 5xx server error - retry with backoff
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

Why different handling for each status code?

HTTP 429 means the receiver is rate limiting you. The Retry-After header tells you exactly how long to wait before trying again.

HTTP 4xx (other than 429) indicates a permanent problem. The URL might be wrong or the receiver rejected the payload. Retrying will not help, so the delivery goes straight to the dead-letter queue.

HTTP 5xx means the receiver's server had a transient error. Exponential backoff gives it time to recover without hammering it with retries.

• Now add the _backoff_with_jitter helper function after the deliver_webhook function definition (outside the function, at the module level):

def _backoff_with_jitter(attempt: int) -> float:
    # Exponential backoff: 5s, 10s, 20s, 40s, 80s... capped at 300s
    base = min(300, (2**attempt) * 5)
    # Add up to 10% random jitter to prevent thundering herd
    jitter = random.uniform(0, base * 0.1)
    return base + jitter

What is thundering herd?

If a destination goes down and 1000 deliveries all retry at exactly the same interval, they all hit the recovering server simultaneously. That surge can knock it down again.

Adding random jitter spreads the retries across a time window, giving the destination breathing room. The formula min(300, 2^attempt * 5) + jitter starts at 5 seconds and caps at 5 minutes.

Do not save yet. The code references _write_to_dlq which does not exist. You will define it next.

Write failed deliveries to the dead-letter queue

When a delivery has exhausted all retries or hit a permanent 4xx error, you need a record of what happened. A dead-letter queue captures the full context so an operator can inspect failures, fix the issue, and replay if needed.

• Add the _write_to_dlq helper function below _backoff_with_jitter at the module level:

def _write_to_dlq(
    tenant_id: str,
    delivery_id: str,
    url: str,
    payload: dict,
    event_type: str,
    error: str,
    attempts: int,
):
    # Write full delivery context to MongoDB for inspection and replay
    db.dead_letter_deliveries.insert_one({
        "tenant_id": tenant_id,
        "delivery_id": delivery_id,
        "url": url,
        "event_type": event_type,
        "payload": payload,
        "error": error,
        "attempts": attempts,
        "failed_at": datetime.now(timezone.utc),
    })
    logger.error(f"Delivery {delivery_id} moved to DLQ: {error}")

What does the dead-letter queue store?

Every field an operator needs to diagnose and replay the delivery: which tenant sent it, the destination URL that failed, the original payload, the error reason, and a timestamp. The dead_letter_deliveries collection in MongoDB serves as your inspection and replay source.

• Save app/worker.py.
• Start the Celery worker by running this command in a new terminal:

celery -A app.worker.celery_app worker --loglevel=info

• In another terminal, start the FastAPI server:

uvicorn app.main:app --reload

• Test a successful delivery by registering a tenant, creating a subscription pointing to https://httpbin.org/post, and publishing an event. In a third terminal, run:

curl -X POST http://localhost:8000/tenants -H "Content-Type: application/json" -d '{"name": "test-tenant"}'

What should I see?

The response includes a tenant_id, api_key, and webhook_secret. Copy the tenant_id and api_key values for the next commands.

• Create a subscription using the tenant ID from the previous response:

curl -X POST http://localhost:8000/tenants/<your-tenant-id>/subscriptions -H "Content-Type: application/json" -d '{"url": "https://httpbin.org/post", "event_types": ["order.created"]}'

• Publish an event using the API key from the tenant creation response:

curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -H "X-Api-Key: <your-api-key>" -d '{"event_type": "order.created", "payload": {"order_id": "12345", "amount": 99.99}}'

• Check your Celery worker terminal. You should see a log entry confirming the task completed successfully.
• Test a failing delivery by creating a subscription pointing to a non-existent URL (e.g. http://localhost:9999/webhook) and publishing another event. Watch the Celery terminal for retry attempts with increasing backoff delays.

What should I see in the Celery logs?

For the successful delivery, you will see Task deliver_webhook succeeded in the Celery output.

For the failing delivery, you will see retry attempts spaced out with increasing delays (5s, 10s, 20s, etc.) until all 5 retries are exhausted and the delivery lands in the dead-letter queue.

• Confirm the DLQ entry exists by checking MongoDB. Run this in a terminal:

docker exec -it $(docker ps -qf "ancestor=mongo:7") mongosh webhook_platform --eval "db.dead_letter_deliveries.find().pretty()"

• You should see a document containing the tenant_id, delivery_id, url, payload, and error fields confirming the delivery was routed to the dead-letter queue.

Not seeing retry attempts?

Make sure your Celery worker is running and connected to RabbitMQ. Check for connection errors in the worker startup output.

If tasks are queued but not executing, confirm the Docker containers are healthy by running docker compose ps. All three services (rabbitmq, redis, mongodb) should show "healthy" status.

help me debug why my Celery worker is not processing webhook delivery tasks

✔️ Awesome, I've got everything!

Great. Double check you have saved app/worker.py with the full delivery engine.

ⓧ I'd like to double check the full code

import hashlib
import hmac
import json
import logging
import random
import time
from datetime import datetime, timezone

import httpx
from celery import Celery

from app.config import settings
from app.database import db, redis_client

logger = logging.getLogger(__name__)

celery_app = Celery("webhook_worker", broker=settings.rabbitmq_url)
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,
    worker_prefetch_multiplier=1,
)


def sign_payload(payload: dict, secret: str) -> str:
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = hmac.new(
        secret.encode("utf-8"),
        body.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    return f"sha256={signature}"


@celery_app.task(bind=True, max_retries=5, name="deliver_webhook")
def deliver_webhook(
    self,
    tenant_id: str,
    delivery_id: str,
    url: str,
    event_type: str,
    payload: dict,
    webhook_secret: str,
):
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = sign_payload(payload, webhook_secret)

    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Signature": signature,
        "X-Delivery-Id": delivery_id,
        "X-Event-Type": event_type,
    }

    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(url, content=body, headers=headers)

        if response.status_code < 300:
            return {"status": "delivered", "delivery_id": delivery_id}

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", "10"))
            redis_client.delete(f"delivery:{delivery_id}")
            raise self.retry(countdown=retry_after)

        if 400 <= response.status_code < 500:
            _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, f"HTTP {response.status_code}", self.request.retries)
            return {"status": "permanent_failure", "delivery_id": delivery_id}

        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except httpx.TimeoutException:
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except self.MaxRetriesExceededError:
        _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, "Max retries exceeded", self.request.retries)


def _backoff_with_jitter(attempt: int) -> float:
    base = min(300, (2**attempt) * 5)
    jitter = random.uniform(0, base * 0.1)
    return base + jitter


def _write_to_dlq(
    tenant_id: str,
    delivery_id: str,
    url: str,
    payload: dict,
    event_type: str,
    error: str,
    attempts: int,
):
    db.dead_letter_deliveries.insert_one({
        "tenant_id": tenant_id,
        "delivery_id": delivery_id,
        "url": url,
        "event_type": event_type,
        "payload": payload,
        "error": error,
        "attempts": attempts,
        "failed_at": datetime.now(timezone.utc),
    })
    logger.error(f"Delivery {delivery_id} moved to DLQ: {error}")

Your delivery engine is now fully operational. It signs, delivers, retries intelligently, and captures failures for later inspection. Next up, you will add circuit breakers to protect unhealthy destinations and idempotency keys to prevent duplicate deliveries.

Add Reliability — Circuit Breaker and Idempotency

Your delivery engine can now sign payloads and retry failed attempts. But what happens when a destination URL goes completely down?

Without protection, your workers will waste resources hammering a dead endpoint over and over. In this step, you will build two critical reliability patterns: a circuit breaker that detects unhealthy destinations and stops sending to them, and an idempotency guard that prevents duplicate deliveries when tasks are retried.

In this step, get ready to:

• Build a Redis-backed circuit breaker with three states (CLOSED, OPEN, HALF_OPEN) that trips after 5 consecutive failures.
• Integrate the circuit breaker into the delivery task so unhealthy URLs are skipped automatically.
• Add idempotency via Redis SET NX to prevent duplicate webhook deliveries.

Implement the Redis-backed circuit breaker class

A circuit breaker monitors delivery attempts to each destination URL. After a threshold of consecutive failures, it "opens" and blocks further requests to that URL. This gives the failing endpoint time to recover instead of burying it under retries.

• Create a new file called circuit_breaker.py inside the app/services/ directory.
• Add the class definition, constants, and helper methods by pasting this code:

import hashlib
import json
import time


class CircuitBreaker:
    STATE_CLOSED = "closed"
    STATE_OPEN = "open"
    STATE_HALF_OPEN = "half_open"

    def __init__(self, redis_client, url: str, settings):
        self.redis = redis_client
        # Hash the URL to create a short, safe Redis key
        url_hash = hashlib.md5(url.encode()).hexdigest()[:12]
        self.key_prefix = f"circuit:{url_hash}"
        self.failure_threshold = settings.circuit_breaker_failure_threshold
        self.recovery_timeout = settings.circuit_breaker_recovery_timeout

    def _state_key(self) -> str:
        return f"{self.key_prefix}:state"

    def _failures_key(self) -> str:
        return f"{self.key_prefix}:failures"

    def _last_failure_key(self) -> str:
        return f"{self.key_prefix}:last_failure"

    def get_state(self) -> str:
        # Default to CLOSED if no state exists yet
        state = self.redis.get(self._state_key())
        return state if state else self.STATE_CLOSED

What does this code do?

Each destination URL gets its own circuit breaker, identified by a hash of the URL. The class stores three pieces of state in Redis: the current state (closed/open/half_open), the consecutive failure count, and the timestamp of the last failure.

The thresholds come from your config.py settings: 5 failures to trip the breaker, and 30 seconds before attempting recovery.

• Add the allow_request() method below get_state():

    def allow_request(self) -> bool:
        state = self.get_state()

        if state == self.STATE_CLOSED:
            return True

        if state == self.STATE_OPEN:
            # Check if enough time has passed to attempt recovery
            last_failure = self.redis.get(self._last_failure_key())
            if last_failure and (time.time() - float(last_failure)) >= self.recovery_timeout:
                self.redis.set(self._state_key(), self.STATE_HALF_OPEN)
                return True
            return False

        if state == self.STATE_HALF_OPEN:
            # Allow one probe request through
            return True

        return True

How do the three states work?

• CLOSED: Normal operation. All requests pass through.
• OPEN: The destination is considered down. All requests are blocked. After 30 seconds, the state transitions to HALF_OPEN.
• HALF_OPEN: One probe request is allowed through. If it succeeds, the circuit closes. If it fails, the circuit reopens.

• Add the state transition methods below allow_request():

    def record_success(self):
        state = self.get_state()
        if state == self.STATE_HALF_OPEN:
            # Probe succeeded - close the circuit
            pipe = self.redis.pipeline()
            pipe.set(self._state_key(), self.STATE_CLOSED)
            pipe.set(self._failures_key(), 0)
            pipe.execute()
        elif state == self.STATE_CLOSED:
            # Reset failure counter on success
            self.redis.set(self._failures_key(), 0)

    def record_failure(self):
        state = self.get_state()

        if state == self.STATE_HALF_OPEN:
            # Probe failed - reopen immediately
            pipe = self.redis.pipeline()
            pipe.set(self._state_key(), self.STATE_OPEN)
            pipe.set(self._last_failure_key(), str(time.time()))
            pipe.execute()
            return

        # In CLOSED state, increment failures and check threshold
        failures = self.redis.incr(self._failures_key())
        self.redis.set(self._last_failure_key(), str(time.time()))

        if failures >= self.failure_threshold:
            self.redis.set(self._state_key(), self.STATE_OPEN)

Why use Redis pipelines here?

When transitioning states, multiple Redis keys need to update together. A pipeline groups these commands into one round trip, reducing latency and keeping state consistent across concurrent workers.

• Save app/services/circuit_breaker.py.

Seeing an import error?

Make sure the file is saved inside app/services/ and not in the root app/ directory. The __init__.py file must also exist in app/services/ for Python to recognise it as a package.

help me debug a module import error for circuit_breaker

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/services/circuit_breaker.py before moving on.

ⓧ I'd like to double check the full code

import hashlib
import json
import time


class CircuitBreaker:
    STATE_CLOSED = "closed"
    STATE_OPEN = "open"
    STATE_HALF_OPEN = "half_open"

    def __init__(self, redis_client, url: str, settings):
        self.redis = redis_client
        url_hash = hashlib.md5(url.encode()).hexdigest()[:12]
        self.key_prefix = f"circuit:{url_hash}"
        self.failure_threshold = settings.circuit_breaker_failure_threshold
        self.recovery_timeout = settings.circuit_breaker_recovery_timeout

    def _state_key(self) -> str:
        return f"{self.key_prefix}:state"

    def _failures_key(self) -> str:
        return f"{self.key_prefix}:failures"

    def _last_failure_key(self) -> str:
        return f"{self.key_prefix}:last_failure"

    def get_state(self) -> str:
        state = self.redis.get(self._state_key())
        return state if state else self.STATE_CLOSED

    def allow_request(self) -> bool:
        state = self.get_state()

        if state == self.STATE_CLOSED:
            return True

        if state == self.STATE_OPEN:
            last_failure = self.redis.get(self._last_failure_key())
            if last_failure and (time.time() - float(last_failure)) >= self.recovery_timeout:
                self.redis.set(self._state_key(), self.STATE_HALF_OPEN)
                return True
            return False

        if state == self.STATE_HALF_OPEN:
            return True

        return True

    def record_success(self):
        state = self.get_state()
        if state == self.STATE_HALF_OPEN:
            pipe = self.redis.pipeline()
            pipe.set(self._state_key(), self.STATE_CLOSED)
            pipe.set(self._failures_key(), 0)
            pipe.execute()
        elif state == self.STATE_CLOSED:
            self.redis.set(self._failures_key(), 0)

    def record_failure(self):
        state = self.get_state()

        if state == self.STATE_HALF_OPEN:
            pipe = self.redis.pipeline()
            pipe.set(self._state_key(), self.STATE_OPEN)
            pipe.set(self._last_failure_key(), str(time.time()))
            pipe.execute()
            return

        failures = self.redis.incr(self._failures_key())
        self.redis.set(self._last_failure_key(), str(time.time()))

        if failures >= self.failure_threshold:
            self.redis.set(self._state_key(), self.STATE_OPEN)

Integrate the circuit breaker into the delivery task

The circuit breaker class is ready. Now you need to wire it into the delivery task so it actually protects your workers from hammering dead endpoints.

• In app/worker.py, find this import line:

from app.database import db, redis_client

• Add the circuit breaker import directly below it, so both lines look like this:

from app.database import db, redis_client
from app.services.circuit_breaker import CircuitBreaker

• Now replace the entire deliver_webhook function body (everything inside the function, starting after the function signature) with the following code that adds circuit breaker checks:

@celery_app.task(bind=True, max_retries=5, name="deliver_webhook")
def deliver_webhook(
    self,
    tenant_id: str,
    delivery_id: str,
    url: str,
    event_type: str,
    payload: dict,
    webhook_secret: str,
):
    # Circuit breaker
    circuit = CircuitBreaker(redis_client, url, settings)
    if not circuit.allow_request():
        raise self.retry(countdown=settings.circuit_breaker_recovery_timeout)

    # Deliver
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = sign_payload(payload, webhook_secret)

    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Signature": signature,
        "X-Delivery-Id": delivery_id,
        "X-Event-Type": event_type,
    }

    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(url, content=body, headers=headers)

        if response.status_code < 300:
            circuit.record_success()
            return {"status": "delivered", "delivery_id": delivery_id}

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", "10"))
            circuit.record_failure()
            redis_client.delete(f"delivery:{delivery_id}")
            raise self.retry(countdown=retry_after)

        if 400 <= response.status_code < 500:
            circuit.record_failure()
            _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, f"HTTP {response.status_code}", self.request.retries)
            return {"status": "permanent_failure", "delivery_id": delivery_id}

        # 5xx - retry
        circuit.record_failure()
        redis_client.delete(f"delivery:{delivery_id}")
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except httpx.TimeoutException:
        circuit.record_failure()
        redis_client.delete(f"delivery:{delivery_id}")
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except self.MaxRetriesExceededError:
        _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, "Max retries exceeded", self.request.retries)

What changed in the delivery task?

• A CircuitBreaker instance is created for the destination URL at the start of every delivery attempt.
• If the circuit is OPEN, the task immediately re-queues itself with a delay instead of attempting delivery.
• On a successful delivery, circuit.record_success() resets the failure counter (or closes the circuit if in HALF_OPEN).
• On any failure (4xx, 5xx, timeout), circuit.record_failure() increments the counter toward the threshold.

• Save app/worker.py.

Getting a NameError for CircuitBreaker?

Confirm the import line reads from app.services.circuit_breaker import CircuitBreaker and that the file app/services/circuit_breaker.py exists with the class name spelled exactly as CircuitBreaker.

help me fix my CircuitBreaker import

Add idempotency to prevent duplicate deliveries

When a Celery worker crashes mid-delivery or a task is retried, the same webhook could be sent twice. Idempotency prevents this by checking whether a delivery has already been attempted before proceeding.

What is idempotency?

An idempotent operation produces the same result no matter how many times it runs. Here, you use Redis SET NX (set if not exists) to claim a delivery ID. If the key already exists, another worker already handled it, so you skip.

The 24-hour TTL ensures keys expire automatically so Redis does not grow unbounded.

• In app/worker.py, find the circuit breaker block at the top of the deliver_webhook function:

    # Circuit breaker
    circuit = CircuitBreaker(redis_client, url, settings)
    if not circuit.allow_request():

• Add the idempotency check above the circuit breaker block, so the function body starts with this:

    # Idempotency check
    if not redis_client.set(f"delivery:{delivery_id}", "1", nx=True, ex=86400):
        logger.info(f"Duplicate delivery skipped: {delivery_id}")
        return {"status": "deduplicated", "delivery_id": delivery_id}

    # Circuit breaker
    circuit = CircuitBreaker(redis_client, url, settings)
    if not circuit.allow_request():
        redis_client.delete(f"delivery:{delivery_id}")
        raise self.retry(countdown=settings.circuit_breaker_recovery_timeout)

What does this code do?

• redis_client.set(..., nx=True, ex=86400) atomically sets the key only if it does not already exist, with a 24-hour expiry. If the key already exists, it returns None (falsy), triggering the early return.
• The redis_client.delete(f"delivery:{delivery_id}") in the circuit breaker denial clears the idempotency key so the task can be retried later when the circuit recovers.

• Save app/worker.py.
• Restart your Celery worker to pick up the changes.

Why delete the idempotency key on retries?

The idempotency key is claimed before delivery starts. If the delivery fails and needs to be retried, you must delete the key first. Otherwise the retry attempt would see the key and skip, thinking the delivery already succeeded.

Verify both patterns are working by testing with your existing tenant and subscription.

• Trigger 5 or more failures to the same URL (point a subscription at a non-existent endpoint like http://localhost:9999/fail) and check the Celery logs. After 5 consecutive failures, subsequent deliveries to that URL should be skipped immediately with a re-queue message.
• Wait 30 seconds and publish another event. The circuit enters HALF_OPEN and allows one probe request through.
• Publish the same event twice in quick succession. Check Celery logs to confirm the second delivery is logged as Duplicate delivery skipped.

Circuit not opening after 5 failures?

Open a Redis CLI (redis-cli) and run KEYS circuit:* to inspect the circuit breaker keys. Check that the :failures key is incrementing. If failures are not accumulating, verify that each failed delivery calls circuit.record_failure() in every error branch (429, 4xx, 5xx, and timeout).

help me debug my circuit breaker not opening

✔️ Awesome, I've got everything!

Great. Double check you have saved both app/services/circuit_breaker.py and app/worker.py.

ⓧ I'd like to double check the full code

import hashlib
import json
import time


class CircuitBreaker:
    STATE_CLOSED = "closed"
    STATE_OPEN = "open"
    STATE_HALF_OPEN = "half_open"

    def __init__(self, redis_client, url: str, settings):
        self.redis = redis_client
        url_hash = hashlib.md5(url.encode()).hexdigest()[:12]
        self.key_prefix = f"circuit:{url_hash}"
        self.failure_threshold = settings.circuit_breaker_failure_threshold
        self.recovery_timeout = settings.circuit_breaker_recovery_timeout

    def _state_key(self) -> str:
        return f"{self.key_prefix}:state"

    def _failures_key(self) -> str:
        return f"{self.key_prefix}:failures"

    def _last_failure_key(self) -> str:
        return f"{self.key_prefix}:last_failure"

    def get_state(self) -> str:
        state = self.redis.get(self._state_key())
        return state if state else self.STATE_CLOSED

    def allow_request(self) -> bool:
        state = self.get_state()

        if state == self.STATE_CLOSED:
            return True

        if state == self.STATE_OPEN:
            last_failure = self.redis.get(self._last_failure_key())
            if last_failure and (time.time() - float(last_failure)) >= self.recovery_timeout:
                self.redis.set(self._state_key(), self.STATE_HALF_OPEN)
                return True
            return False

        if state == self.STATE_HALF_OPEN:
            return True

        return True

    def record_success(self):
        state = self.get_state()
        if state == self.STATE_HALF_OPEN:
            pipe = self.redis.pipeline()
            pipe.set(self._state_key(), self.STATE_CLOSED)
            pipe.set(self._failures_key(), 0)
            pipe.execute()
        elif state == self.STATE_CLOSED:
            self.redis.set(self._failures_key(), 0)

    def record_failure(self):
        state = self.get_state()

        if state == self.STATE_HALF_OPEN:
            pipe = self.redis.pipeline()
            pipe.set(self._state_key(), self.STATE_OPEN)
            pipe.set(self._last_failure_key(), str(time.time()))
            pipe.execute()
            return

        failures = self.redis.incr(self._failures_key())
        self.redis.set(self._last_failure_key(), str(time.time()))

        if failures >= self.failure_threshold:
            self.redis.set(self._state_key(), self.STATE_OPEN)

import hashlib
import hmac
import json
import logging
import random
import time
from datetime import datetime, timezone

import httpx
from celery import Celery

from app.config import settings
from app.database import db, redis_client
from app.services.circuit_breaker import CircuitBreaker

logger = logging.getLogger(__name__)

celery_app = Celery("webhook_worker", broker=settings.rabbitmq_url)
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,
    worker_prefetch_multiplier=1,
)


def sign_payload(payload: dict, secret: str) -> str:
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = hmac.new(
        secret.encode("utf-8"),
        body.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    return f"sha256={signature}"


@celery_app.task(bind=True, max_retries=5, name="deliver_webhook")
def deliver_webhook(
    self,
    tenant_id: str,
    delivery_id: str,
    url: str,
    event_type: str,
    payload: dict,
    webhook_secret: str,
):
    # Idempotency check
    if not redis_client.set(f"delivery:{delivery_id}", "1", nx=True, ex=86400):
        logger.info(f"Duplicate delivery skipped: {delivery_id}")
        return {"status": "deduplicated", "delivery_id": delivery_id}

    # Circuit breaker
    circuit = CircuitBreaker(redis_client, url, settings)
    if not circuit.allow_request():
        redis_client.delete(f"delivery:{delivery_id}")
        raise self.retry(countdown=settings.circuit_breaker_recovery_timeout)

    # Deliver
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = sign_payload(payload, webhook_secret)

    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Signature": signature,
        "X-Delivery-Id": delivery_id,
        "X-Event-Type": event_type,
    }

    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(url, content=body, headers=headers)

        if response.status_code < 300:
            circuit.record_success()
            return {"status": "delivered", "delivery_id": delivery_id}

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", "10"))
            circuit.record_failure()
            redis_client.delete(f"delivery:{delivery_id}")
            raise self.retry(countdown=retry_after)

        if 400 <= response.status_code < 500:
            circuit.record_failure()
            _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, f"HTTP {response.status_code}", self.request.retries)
            return {"status": "permanent_failure", "delivery_id": delivery_id}

        # 5xx - retry
        circuit.record_failure()
        redis_client.delete(f"delivery:{delivery_id}")
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except httpx.TimeoutException:
        circuit.record_failure()
        redis_client.delete(f"delivery:{delivery_id}")
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except self.MaxRetriesExceededError:
        _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, "Max retries exceeded", self.request.retries)


def _backoff_with_jitter(attempt: int) -> float:
    base = min(300, (2**attempt) * 5)
    jitter = random.uniform(0, base * 0.1)
    return base + jitter


def _write_to_dlq(
    tenant_id: str,
    delivery_id: str,
    url: str,
    payload: dict,
    event_type: str,
    error: str,
    attempts: int,
):
    db.dead_letter_deliveries.insert_one({
        "tenant_id": tenant_id,
        "delivery_id": delivery_id,
        "url": url,
        "event_type": event_type,
        "payload": payload,
        "error": error,
        "attempts": attempts,
        "failed_at": datetime.now(timezone.utc),
    })
    logger.error(f"Delivery {delivery_id} moved to DLQ: {error}")

Your delivery engine now protects itself from cascading failures and guarantees exactly-once delivery semantics. Next up, you will build a Redis Lua token bucket to enforce per-tenant rate limits.

Enforce Per-Tenant Rate Limiting with a Redis Lua Token Bucket

Your delivery engine now handles failures gracefully with circuit breakers and prevents duplicates with idempotency keys. But there is still one gap: a single tenant could flood the system with thousands of events per second and starve every other tenant of delivery capacity.

In this step, you will build a per-tenant rate limiter using a Redis Lua script that implements the token bucket algorithm. The entire check-and-consume operation runs atomically inside Redis, so multiple Celery workers can share the same bucket without race conditions.

In this step, get ready to:

• Write a Lua script that atomically refills and consumes tokens from a Redis HASH.
• Create a RateLimiter class that loads and executes the script per tenant.
• Integrate rate limiting into the delivery task and expose a status endpoint.

Write the Lua token bucket script

A token bucket works by granting a fixed number of tokens to each tenant. Every delivery consumes one token, and tokens refill at a steady rate over time. If the bucket is empty, the delivery is denied until tokens replenish.

The critical requirement is atomicity. With multiple Celery workers checking the same bucket simultaneously, a plain Python check-then-decrement would create race conditions. A Lua script runs as a single indivisible operation inside Redis, eliminating this problem entirely.

Why a Lua script instead of application-level limiting?

Application-level rate limiting (checking a counter in Python) fails in distributed deployments because two workers can read the same counter value, both decide to allow, and both decrement. The bucket is overdrawn.

Lua scripts execute atomically inside Redis. The entire refill-check-consume cycle happens as one indivisible operation, so no matter how many workers call it concurrently, the bucket state is always consistent.

• Create a new file app/services/rate_limiter.py inside the app/services/ folder.
• Add the imports and the Lua script string by pasting this code:

import time
from pathlib import Path

from app.database import redis_client
from app.config import settings

LUA_SCRIPT = """
local key = KEYS[1]
local max_tokens = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

local data = redis.call('HGETALL', key)
local tokens = max_tokens
local last_refill = now

if #data > 0 then
    local fields = {}
    for i = 1, #data, 2 do
        fields[data[i]] = data[i + 1]
    end
    tokens = tonumber(fields['tokens']) or max_tokens
    last_refill = tonumber(fields['last_refill']) or now
end

local elapsed = now - last_refill
local new_tokens = elapsed * refill_rate
tokens = math.min(max_tokens, tokens + new_tokens)

local allowed = 0
local remaining = tokens
local retry_after_ms = 0

if tokens >= 1 then
    tokens = tokens - 1
    remaining = tokens
    allowed = 1
else
    retry_after_ms = math.ceil((1 - tokens) / refill_rate * 1000)
end

redis.call('HSET', key, 'tokens', tostring(tokens), 'last_refill', tostring(now))
redis.call('EXPIRE', key, math.ceil(max_tokens / refill_rate) + 10)

return {allowed, math.floor(remaining), retry_after_ms}
"""

What does this Lua script do?

• HGETALL reads the current bucket state (tokens remaining and last refill timestamp) from a Redis HASH keyed by tenant ID.
• It calculates how much time has passed since the last refill and adds new tokens proportionally, capping at max_tokens.
• If at least 1 token is available, it consumes one and returns allowed=1. Otherwise it returns allowed=0 with a retry_after_ms value telling the caller how long to wait.
• HSET writes the updated token count and timestamp back, and EXPIRE sets a TTL so inactive buckets clean themselves up.

• Save app/services/rate_limiter.py.

You won't see a change yet. The next chunk adds the Python class that calls this script.

Create the RateLimiter class

The Lua script handles the atomic logic, but you need a Python wrapper to load it into Redis, pass the right arguments, and parse the result into a dictionary the rest of your app can use.

• In app/services/rate_limiter.py, add the following class and singleton below the LUA_SCRIPT string:

class RateLimiter:
    def __init__(self, redis_client, max_tokens: int, refill_rate: float):
        self.redis = redis_client
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate
        self._script_sha = None

    def _load_script(self):
        if self._script_sha is None:
            self._script_sha = self.redis.script_load(LUA_SCRIPT)

    def check(self, tenant_id: str) -> dict:
        self._load_script()
        key = f"bucket:{tenant_id}"
        now = time.time()

        try:
            result = self.redis.evalsha(
                self._script_sha, 1, key,
                str(self.max_tokens), str(self.refill_rate), str(now),
            )
        except Exception:
            self._script_sha = None
            self._load_script()
            result = self.redis.evalsha(
                self._script_sha, 1, key,
                str(self.max_tokens), str(self.refill_rate), str(now),
            )

        return {
            "allowed": result[0] == 1,
            "remaining": result[1],
            "retry_after_ms": result[2] if result[0] == 0 else None,
        }


rate_limiter = RateLimiter(
    redis_client=redis_client,
    max_tokens=settings.rate_limit_tokens,
    refill_rate=settings.rate_limit_refill_rate,
)

What does this class do?

• _load_script() uploads the Lua source to Redis once and caches its SHA hash. Subsequent calls use evalsha() to avoid resending the full script text on every request.
• check() calls the Lua script with the tenant's bucket key, max tokens, refill rate, and current timestamp. It returns a dictionary with allowed, remaining, and retry_after_ms.
• The except block handles the case where Redis restarts and loses the cached script. It reloads the script and retries.
• The module-level rate_limiter singleton is configured with 100 tokens and a 1.67 tokens/second refill rate (about 100 per minute).

• Save app/services/rate_limiter.py.

You won't see a visible change yet. The next chunk wires this into the delivery task.

Seeing an import error on save?

If your editor flags from app.database import redis_client, make sure you are running PyCharm with the project root as the working directory so the app package resolves correctly.

help me fix an import error in rate_limiter.py

✔️ Awesome, I've got everything!

Great. Double check you have saved app/services/rate_limiter.py.

ⓧ I'd like to double check the full code

import time
from pathlib import Path

from app.database import redis_client
from app.config import settings

LUA_SCRIPT = """
local key = KEYS[1]
local max_tokens = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

local data = redis.call('HGETALL', key)
local tokens = max_tokens
local last_refill = now

if #data > 0 then
    local fields = {}
    for i = 1, #data, 2 do
        fields[data[i]] = data[i + 1]
    end
    tokens = tonumber(fields['tokens']) or max_tokens
    last_refill = tonumber(fields['last_refill']) or now
end

local elapsed = now - last_refill
local new_tokens = elapsed * refill_rate
tokens = math.min(max_tokens, tokens + new_tokens)

local allowed = 0
local remaining = tokens
local retry_after_ms = 0

if tokens >= 1 then
    tokens = tokens - 1
    remaining = tokens
    allowed = 1
else
    retry_after_ms = math.ceil((1 - tokens) / refill_rate * 1000)
end

redis.call('HSET', key, 'tokens', tostring(tokens), 'last_refill', tostring(now))
redis.call('EXPIRE', key, math.ceil(max_tokens / refill_rate) + 10)

return {allowed, math.floor(remaining), retry_after_ms}
"""


class RateLimiter:
    def __init__(self, redis_client, max_tokens: int, refill_rate: float):
        self.redis = redis_client
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate
        self._script_sha = None

    def _load_script(self):
        if self._script_sha is None:
            self._script_sha = self.redis.script_load(LUA_SCRIPT)

    def check(self, tenant_id: str) -> dict:
        self._load_script()
        key = f"bucket:{tenant_id}"
        now = time.time()

        try:
            result = self.redis.evalsha(
                self._script_sha, 1, key,
                str(self.max_tokens), str(self.refill_rate), str(now),
            )
        except Exception:
            self._script_sha = None
            self._load_script()
            result = self.redis.evalsha(
                self._script_sha, 1, key,
                str(self.max_tokens), str(self.refill_rate), str(now),
            )

        return {
            "allowed": result[0] == 1,
            "remaining": result[1],
            "retry_after_ms": result[2] if result[0] == 0 else None,
        }


rate_limiter = RateLimiter(
    redis_client=redis_client,
    max_tokens=settings.rate_limit_tokens,
    refill_rate=settings.rate_limit_refill_rate,
)

Integrate the rate limiter into the delivery task

The rate limiter module is ready. Now you need to call it from the Celery worker before each delivery attempt so that a tenant cannot exceed their allocated throughput.

• In app/worker.py, find this import line near the top:

from app.services.circuit_breaker import CircuitBreaker

• Add this import directly below it:

from app.services.rate_limiter import rate_limiter

• Next, scroll to the deliver_webhook function. Find the idempotency check block:

    # Idempotency check
    if not redis_client.set(f"delivery:{delivery_id}", "1", nx=True, ex=86400):
        logger.info(f"Duplicate delivery skipped: {delivery_id}")
        return {"status": "deduplicated", "delivery_id": delivery_id}

    # Circuit breaker
    circuit = CircuitBreaker(redis_client, url, settings)

• Replace the gap between the idempotency check and the circuit breaker section with the rate limiting block, so the code looks like this:

    # Idempotency check
    if not redis_client.set(f"delivery:{delivery_id}", "1", nx=True, ex=86400):
        logger.info(f"Duplicate delivery skipped: {delivery_id}")
        return {"status": "deduplicated", "delivery_id": delivery_id}

    # Rate limiting
    rate_result = rate_limiter.check(tenant_id)
    if not rate_result["allowed"]:
        redis_client.delete(f"delivery:{delivery_id}")
        retry_after = rate_result.get("retry_after_ms", 5000) / 1000
        raise self.retry(countdown=retry_after)

    # Circuit breaker
    circuit = CircuitBreaker(redis_client, url, settings)
    if not circuit.allow_request():
        redis_client.delete(f"delivery:{delivery_id}")
        raise self.retry(countdown=settings.circuit_breaker_recovery_timeout)

    # Deliver
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = sign_payload(payload, webhook_secret)

What does the rate limiting block do?

• rate_limiter.check(tenant_id) calls the Lua script to attempt consuming one token from this tenant's bucket.
• If denied, it deletes the idempotency key (so the delivery can retry later) and re-queues the task with a delay based on retry_after_ms.
• The rate check runs after idempotency (no point rate-limiting a duplicate) and before the circuit breaker (no point checking the circuit if you have no tokens).

• Save app/worker.py.

You won't see a change yet. The next chunk adds the API endpoint so you can inspect rate limit state.

Seeing a NameError for rate_limiter?

Make sure the import from app.services.rate_limiter import rate_limiter is at the top of app/worker.py, not inside the function body.

help me fix a NameError for rate_limiter in my worker

✔️ Awesome, I've got everything!

Great. Confirm you saved app/worker.py with both the new import and the rate limiting block.

ⓧ I'd like to double check the full code

import hashlib
import hmac
import json
import logging
import random
import time
from datetime import datetime, timezone

import httpx
from celery import Celery

from app.config import settings
from app.database import db, redis_client
from app.services.circuit_breaker import CircuitBreaker
from app.services.rate_limiter import rate_limiter

logger = logging.getLogger(__name__)

celery_app = Celery("webhook_worker", broker=settings.rabbitmq_url)
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,
    worker_prefetch_multiplier=1,
)


def sign_payload(payload: dict, secret: str) -> str:
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = hmac.new(
        secret.encode("utf-8"),
        body.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    return f"sha256={signature}"


@celery_app.task(bind=True, max_retries=5, name="deliver_webhook")
def deliver_webhook(
    self,
    tenant_id: str,
    delivery_id: str,
    url: str,
    event_type: str,
    payload: dict,
    webhook_secret: str,
):
    # Idempotency check
    if not redis_client.set(f"delivery:{delivery_id}", "1", nx=True, ex=86400):
        logger.info(f"Duplicate delivery skipped: {delivery_id}")
        return {"status": "deduplicated", "delivery_id": delivery_id}

    # Rate limiting
    rate_result = rate_limiter.check(tenant_id)
    if not rate_result["allowed"]:
        redis_client.delete(f"delivery:{delivery_id}")
        retry_after = rate_result.get("retry_after_ms", 5000) / 1000
        raise self.retry(countdown=retry_after)

    # Circuit breaker
    circuit = CircuitBreaker(redis_client, url, settings)
    if not circuit.allow_request():
        redis_client.delete(f"delivery:{delivery_id}")
        raise self.retry(countdown=settings.circuit_breaker_recovery_timeout)

    # Deliver
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = sign_payload(payload, webhook_secret)

    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Signature": signature,
        "X-Delivery-Id": delivery_id,
        "X-Event-Type": event_type,
    }

    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(url, content=body, headers=headers)

        if response.status_code < 300:
            circuit.record_success()
            return {"status": "delivered", "delivery_id": delivery_id}

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", "10"))
            circuit.record_failure()
            redis_client.delete(f"delivery:{delivery_id}")
            raise self.retry(countdown=retry_after)

        if 400 <= response.status_code < 500:
            circuit.record_failure()
            _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, f"HTTP {response.status_code}", self.request.retries)
            return {"status": "permanent_failure", "delivery_id": delivery_id}

        # 5xx - retry
        circuit.record_failure()
        redis_client.delete(f"delivery:{delivery_id}")
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except httpx.TimeoutException:
        circuit.record_failure()
        redis_client.delete(f"delivery:{delivery_id}")
        countdown = _backoff_with_jitter(self.request.retries)
        raise self.retry(countdown=countdown)

    except self.MaxRetriesExceededError:
        _write_to_dlq(tenant_id, delivery_id, url, payload, event_type, "Max retries exceeded", self.request.retries)


def _backoff_with_jitter(attempt: int) -> float:
    base = min(300, (2**attempt) * 5)
    jitter = random.uniform(0, base * 0.1)
    return base + jitter


def _write_to_dlq(
    tenant_id: str,
    delivery_id: str,
    url: str,
    payload: dict,
    event_type: str,
    error: str,
    attempts: int,
):
    db.dead_letter_deliveries.insert_one({
        "tenant_id": tenant_id,
        "delivery_id": delivery_id,
        "url": url,
        "event_type": event_type,
        "payload": payload,
        "error": error,
        "attempts": attempts,
        "failed_at": datetime.now(timezone.utc),
    })
    logger.error(f"Delivery {delivery_id} moved to DLQ: {error}")

Add the rate limit status endpoint

A good rate limiter is observable. You will add an endpoint that lets tenants (or your operations team) check the current bucket state without consuming a token.

• In app/main.py, scroll to the bottom of the file (after the ingest_event function).
• Add this new endpoint at the very end:

@app.get("/tenants/{tenant_id}/rate-limit", response_model=RateLimitStatus)
def get_rate_limit_status(tenant_id: str):
    from app.services.rate_limiter import rate_limiter

    result = rate_limiter.check(tenant_id)
    return RateLimitStatus(
        tenant_id=tenant_id,
        tokens_remaining=result["remaining"],
        allowed=result["allowed"],
        retry_after_ms=result.get("retry_after_ms"),
    )

What does this endpoint do?

It calls rate_limiter.check() to read the current bucket state and returns how many tokens remain, whether the next request would be allowed, and how long to wait if denied. The import is inside the function to avoid circular imports with the worker module.

• Save app/main.py.
• Restart your FastAPI server and your Celery worker so both pick up the new code.
• Test the rate limit endpoint by running:

curl http://localhost:8000/tenants/test-tenant/rate-limit

You should see a JSON response with tokens_remaining showing 99 (one token consumed by the check), allowed: true, and retry_after_ms: null.

Getting a connection error?

Make sure your Docker containers (Redis, RabbitMQ, MongoDB) are still running. Run docker compose ps to check their status. If any show as unhealthy, restart them with docker compose up -d.

help me debug a connection error when calling the rate limit endpoint

✔️ Awesome, I've got everything!

Great. Confirm you saved app/main.py with the new endpoint.

ⓧ I'd like to double check the full code

import secrets
from datetime import datetime, timezone

from bson import ObjectId
from fastapi import FastAPI, Header, HTTPException

from app.config import settings
from app.database import db, redis_client
from app.models import (
    EventCreate,
    RateLimitStatus,
    SubscriptionCreate,
    SubscriptionResponse,
    TenantCreate,
    TenantResponse,
)
from app.worker import deliver_webhook

app = FastAPI(title="Webhook Delivery Platform")


@app.get("/health")
def health_check():
    try:
        redis_client.ping()
        redis_status = "connected"
    except Exception:
        redis_status = "disconnected"

    try:
        db.command("ping")
        mongo_status = "connected"
    except Exception:
        mongo_status = "disconnected"

    return {"status": "ok", "redis": redis_status, "mongodb": mongo_status}


@app.post("/tenants", response_model=TenantResponse)
def create_tenant(tenant: TenantCreate):
    tenant_doc = {
        "name": tenant.name,
        "api_key": secrets.token_hex(16),
        "webhook_secret": secrets.token_hex(32),
        "created_at": datetime.now(timezone.utc),
    }
    result = db.tenants.insert_one(tenant_doc)
    return TenantResponse(
        tenant_id=str(result.inserted_id),
        name=tenant_doc["name"],
        api_key=tenant_doc["api_key"],
        webhook_secret=tenant_doc["webhook_secret"],
    )


@app.post("/tenants/{tenant_id}/subscriptions", response_model=SubscriptionResponse)
def create_subscription(tenant_id: str, subscription: SubscriptionCreate):
    tenant = db.tenants.find_one({"_id": ObjectId(tenant_id)})
    if not tenant:
        raise HTTPException(status_code=404, detail="Tenant not found")

    sub_doc = {
        "tenant_id": tenant_id,
        "url": subscription.url,
        "event_types": subscription.event_types,
        "created_at": datetime.now(timezone.utc),
    }
    result = db.subscriptions.insert_one(sub_doc)
    return SubscriptionResponse(
        subscription_id=str(result.inserted_id),
        tenant_id=tenant_id,
        url=sub_doc["url"],
        event_types=sub_doc["event_types"],
    )


@app.post("/events", status_code=202)
def ingest_event(event: EventCreate, x_api_key: str = Header()):
    tenant = db.tenants.find_one({"api_key": x_api_key})
    if not tenant:
        raise HTTPException(status_code=401, detail="Invalid API key")

    tenant_id = str(tenant["_id"])
    subscriptions = db.subscriptions.find(
        {"tenant_id": tenant_id, "event_types": event.event_type}
    )

    event_doc = {
        "tenant_id": tenant_id,
        "event_type": event.event_type,
        "payload": event.payload,
        "created_at": datetime.now(timezone.utc),
    }
    event_result = db.events.insert_one(event_doc)
    event_id = str(event_result.inserted_id)

    delivery_count = 0
    for sub in subscriptions:
        delivery_id = f"{event_id}:{sub['_id']}"
        deliver_webhook.delay(
            tenant_id=tenant_id,
            delivery_id=delivery_id,
            url=sub["url"],
            event_type=event.event_type,
            payload=event.payload,
            webhook_secret=tenant["webhook_secret"],
        )
        delivery_count += 1

    return {"event_id": event_id, "deliveries_queued": delivery_count}


@app.get("/tenants/{tenant_id}/rate-limit", response_model=RateLimitStatus)
def get_rate_limit_status(tenant_id: str):
    from app.services.rate_limiter import rate_limiter

    result = rate_limiter.check(tenant_id)
    return RateLimitStatus(
        tenant_id=tenant_id,
        tokens_remaining=result["remaining"],
        allowed=result["allowed"],
        retry_after_ms=result.get("retry_after_ms"),
    )

How do I test the rate limiter under load?

To see rate limiting in action, temporarily lower the limit. In app/config.py, set rate_limit_tokens to 5 and rate_limit_refill_rate to 0.083 (about 5 per minute). Fire 10 events rapidly and watch the Celery logs. Only 5 deliveries go through immediately. The remaining 5 are re-queued with delays.

Call GET /tenants/{tenant_id}/rate-limit between bursts to watch the bucket drain to zero and then slowly refill. Remember to restore the defaults (100 tokens, 1.67 rate) when you are done testing.

Your webhook platform now enforces per-tenant throughput limits with zero race conditions. Each tenant gets their own token bucket, bursts are absorbed gracefully, and denied deliveries are re-queued automatically. You have built a production-grade multi-tenant webhook delivery system with signing, retries, circuit breakers, idempotency, rate limiting, and dead-letter routing.

Secret mission

Delivery Analytics Dashboard API

Your webhook platform delivers events reliably, but how do you know which tenants are healthy and which endpoints are struggling? Build a real-time analytics layer using Redis sorted sets that tracks per-tenant success rates, failure counts, and p95 delivery latency.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally via Docker, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Docker containers (RabbitMQ, Redis, MongoDB) running via Docker Compose.
• Python virtual environment with project dependencies.
• Project folder on your Desktop containing all application code.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the webhook platform.

• Your Docker containers will continue running in the background and using system memory.
• Your Python virtual environment and project files remain on disk.
• You can access the RabbitMQ management UI at http://localhost:15672 and hit your API at http://localhost:8000 any time.

Pause - I'll come back to this later

Shut down running containers to free up memory and CPU, but keep your files intact so you can resume later.

• Stop the Celery worker and FastAPI server if they are still running by pressing Ctrl+C in their terminal windows.
• Stop Docker containers by running this command from your project folder:

docker compose stop

• Deactivate your Python virtual environment by running:

deactivate

To resume later, navigate back to your project folder, run docker compose start, reactivate your virtual environment, and start the FastAPI and Celery processes again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is permanent.

• Stop the Celery worker and FastAPI server if they are still running by pressing Ctrl+C in their terminal windows.
• Remove Docker containers and their associated volumes by running this from your project folder:

docker compose down -v

What does this command do?

The -v flag removes named volumes declared in your Compose file alongside the containers. This ensures all MongoDB data, Redis keys, and RabbitMQ queues are wiped clean.

• Remove the downloaded Docker images to free disk space by running:

docker rmi rabbitmq:3-management redis:7-alpine mongo:7

• Delete the project folder from your Desktop. On macOS or Linux, run:

cd ~/Desktop
rm -rf [[PROJECT_FOLDER="your project folder name"]]

• On Windows, run:

cd ~\Desktop
rmdir /s /q [[PROJECT_FOLDER="your project folder name"]]

This removes your virtual environment, application code, and all project files.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade multi-tenant webhook delivery platform from scratch.

This is the exact system that interviewers ask you to design on a whiteboard. Except you have actually built it, tested it, and observed it handling failures gracefully.

Here's what you accomplished:

• Built a multi-tenant webhook delivery platform with FastAPI, Celery, RabbitMQ, Redis, and MongoDB that signs payloads with HMAC-SHA256 and delivers them asynchronously.
• Implemented production-grade reliability patterns including a Redis-backed per-destination circuit breaker, idempotency via SET NX, and retry with exponential backoff and dead-letter routing.
• Built an atomic Redis Lua token bucket for per-tenant rate limiting that prevents race conditions across distributed workers.
• Secret Mission: Added delivery analytics with time-series metrics stored in Redis sorted sets and p95 latency tracking per destination.

Ready to quiz yourself?