Build a Webhook Ingestion API

Build a local webhook relay that ingests, routes, and delivers events.

Introduction

30 Second Summary

Every time you check out online, update a subscription, or receive a shipping notification, webhooks are firing behind the scenes. These small HTTP messages connect services together, but something has to receive them, decide where they go, and deliver them reliably.

In this project, you will build HookRelay, a local webhook relay service that accepts any HTTP POST, stores it, evaluates it against routing rules, and delivers it to the right destination.

What You'll Build

A webhook ingestion API with a rule engine and async delivery worker, all running locally with FastAPI, PostgreSQL, and Redis.

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

  • A FastAPI ingestion server with a /ingest/{endpoint_id} route that accepts any JSON payload and persists it.
  • A PostgreSQL schema for events, targets, and routing rules that drives configurable fan-out delivery.
  • An arq-based async worker that evaluates rules against incoming payloads and delivers matched events to target URLs.
  • Secret Mission: Add a retry mechanism with exponential backoff for failed deliveries.

Are there any prerequisites?

This guide includes setup instructions for all required tools. You will need Python, PostgreSQL, and Redis installed on your machine. Step 1 walks you through verifying each one.

Set Up the Project

Before you can build the webhook relay, you need a project environment with the right libraries, a database to store events, and a message queue to coordinate async work.

In this step, you will create a Python project, install the four core dependencies, set up a PostgreSQL database, and confirm Redis is ready to act as your job queue.

In this step, get ready to:
  • Create a Python virtual environment and install project dependencies.
  • Create the PostgreSQL database for storing events.
  • Verify Redis is running and accepting connections.
Create a Python virtual environment and install dependencies

A virtual environment keeps this project's packages isolated from the rest of your system. You will install FastAPI for the web server, arq for async task processing, asyncpg for connecting to PostgreSQL, and httpx for making HTTP requests.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), typing Terminal, and pressing Enter.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir hookrelay
cd hookrelay
  • Create the virtual environment and activate it by running the commands for your operating system:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you are inside the hookrelay folder before running the activation command.

On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell first, then try the activate command again.

Help me activate my Python virtual environment

  • Install the four project dependencies by running:
pip install "fastapi[standard]==0.139.0" arq==0.28.0 asyncpg==0.31.0 httpx==0.28.1

What are these packages?

  • fastapi[standard] is the web framework for building your API. The [standard] extra includes uvicorn (the server) and other essentials.
  • arq is a lightweight async task queue backed by Redis. It processes jobs like event delivery in the background.
  • asyncpg is a fast async PostgreSQL driver that lets your app talk to the database without blocking.
  • httpx is an async HTTP client for sending webhook deliveries to target URLs.
  • Verify the installation succeeded by running:
python -c "import fastapi; print(fastapi.__version__)"

You should see 0.139.0 printed in your terminal.

Seeing a ModuleNotFoundError?

This means your virtual environment is not active. Check that (venv) appears in your prompt. If not, re-run the activation command from above.

If you see a different version number, the install may have resolved a cached package. Run the install command again with --force-reinstall added after pip install to pin the correct version.

Help me fix my FastAPI installation

Create the PostgreSQL database

HookRelay needs a database to persist incoming events, targets, and routing rules. You will create an empty database now and add tables in the next step.

  • Create the database by running:
createdb hookrelay
  • Verify the database is accessible by running:
psql -d hookrelay -c "SELECT 1"

You should see a table with a single column and the value 1. This confirms PostgreSQL accepted the connection to your new database.

Seeing a connection refused error?

Make sure your PostgreSQL server is running. On macOS, start it with brew services start postgresql. On Linux, use sudo systemctl start postgresql.

If createdb says the database already exists, that is fine. The psql verification command will still work.

Help me connect to my PostgreSQL database

Verify Redis is running

Redis acts as the message broker between your API and the background worker. The API enqueues jobs into Redis, and the worker picks them up.

  • Check that Redis is accepting connections by running:
redis-cli ping

You should see PONG printed in the terminal. This confirms Redis is running and ready.

Seeing a connection refused error from Redis?

Your Redis server is not running. On macOS, start it with brew services start redis. On Linux, use sudo systemctl start redis.

Help me get Redis running on my machine

Your project environment is ready. Next up, you will build the database schema and the ingestion API that accepts incoming webhooks.

Build the Database Schema and Ingestion API

Your database exists but has no structure yet. Before you can accept webhooks, you need tables to store incoming events, define where they should go, and specify what rules determine the routing.

In this step, you will design the PostgreSQL schema for your relay, build the FastAPI ingestion endpoint, and seed the database with test data so you can verify the pipeline end to end.

In this step, get ready to:
  • Create a database schema with tables for endpoints, events, targets, and rules.
  • Build an ingestion API that accepts any JSON POST and enqueues a job.
  • Seed the database with a test endpoint, target, and routing rule.
Create the database schema

The schema defines four tables that model the webhook relay lifecycle: where events come from, what they contain, where they should go, and what rules govern routing.

  • Create a file called schema.sql in your hookrelay/ project folder.
  • Add the endpoints and events tables by pasting this code:
-- Represents a source that sends webhooks to this relay
CREATE TABLE endpoints (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Stores every raw incoming webhook payload and its headers
CREATE TABLE events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    endpoint_id UUID NOT NULL REFERENCES endpoints(id),
    headers JSONB NOT NULL DEFAULT '{}',
    payload JSONB NOT NULL,
    received_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Speeds up lookups when fetching recent events for an endpoint
CREATE INDEX idx_events_endpoint_received ON events (endpoint_id, received_at DESC);

What do these tables do?

The endpoints table represents each webhook source (e.g. a payment provider or CI system). Every incoming POST maps to one endpoint.

The events table stores the raw payload and HTTP headers for every webhook received. The index on (endpoint_id, received_at DESC) makes it fast to query recent events for a given endpoint.

  • Below the index, add the targets and rules tables:
-- Defines a destination URL where matched events will be delivered
CREATE TABLE targets (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    endpoint_id UUID NOT NULL REFERENCES endpoints(id),
    url TEXT NOT NULL,
    name TEXT NOT NULL,
    is_active BOOLEAN NOT NULL DEFAULT TRUE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Each rule defines one condition a payload must match for its target
CREATE TABLE rules (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    target_id UUID NOT NULL REFERENCES targets(id),
    field_path TEXT NOT NULL,
    operator TEXT NOT NULL DEFAULT 'equals',
    value TEXT NOT NULL
);

What do these tables do?

The targets table holds the URLs where matched events get delivered. Each target belongs to an endpoint and can be toggled on/off with is_active.

The rules table stores matching conditions. A rule says "if the value at field_path in the JSON payload equals the expected value, deliver to this target."

  • Save schema.sql.
  • Apply the schema to your database by running:
psql -d hookrelay -f schema.sql
  • You should see output like CREATE TABLE repeated for each table and CREATE INDEX for the index.

Seeing a "relation already exists" error?

If you ran the schema before and need to start fresh, drop all tables first with psql -d hookrelay -c "DROP TABLE IF EXISTS rules, targets, events, endpoints CASCADE;" then run the schema file again.

help me fix a PostgreSQL schema error when creating tables

✔️ Awesome, I've got everything!

Great. Double check you saved schema.sql and saw the CREATE TABLE confirmations in your terminal.

ⓧ I'd like to double check the full code

CREATE TABLE endpoints (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    endpoint_id UUID NOT NULL REFERENCES endpoints(id),
    headers JSONB NOT NULL DEFAULT '{}',
    payload JSONB NOT NULL,
    received_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_events_endpoint_received ON events (endpoint_id, received_at DESC);

CREATE TABLE targets (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    endpoint_id UUID NOT NULL REFERENCES endpoints(id),
    url TEXT NOT NULL,
    name TEXT NOT NULL,
    is_active BOOLEAN NOT NULL DEFAULT TRUE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE rules (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    target_id UUID NOT NULL REFERENCES targets(id),
    field_path TEXT NOT NULL,
    operator TEXT NOT NULL DEFAULT 'equals',
    value TEXT NOT NULL
);
Create the ingestion API

The ingestion API is the front door of your webhook relay. It accepts any HTTP POST, persists the raw payload to the database, and enqueues an async job for the worker to process later.

  • Create a file called main.py in your hookrelay/ project folder.
  • Add the imports, configuration, and lifespan handler by pasting this code:
import uuid
from contextlib import asynccontextmanager

import asyncpg
from arq import create_pool
from arq.connections import RedisSettings
from fastapi import FastAPI, Request, HTTPException

# Connection settings for Redis (arq job queue) and PostgreSQL
REDIS_SETTINGS = RedisSettings()
DATABASE_URL = "postgresql://localhost/hookrelay"


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create connection pools on startup, close them on shutdown
    app.state.db = await asyncpg.create_pool(DATABASE_URL)
    app.state.redis = await create_pool(REDIS_SETTINGS)
    yield
    await app.state.db.close()
    await app.state.redis.close()


app = FastAPI(lifespan=lifespan)

What does this code do?

The lifespan function runs once when the server starts and once when it shuts down. On startup, it creates an asyncpg connection pool (so multiple requests share database connections efficiently) and an arq Redis pool (for enqueuing background jobs).

Everything before the yield runs at startup. Everything after runs at shutdown. This is FastAPI's recommended pattern for managing resources that outlive a single request.

  • Below the app = FastAPI(lifespan=lifespan) line, add the ingestion route:
@app.post("/ingest/{endpoint_id}", status_code=202)
async def ingest(endpoint_id: uuid.UUID, request: Request):
    body = await request.json()
    headers = dict(request.headers)

    # Verify the endpoint exists
    pool = request.app.state.db
    row = await pool.fetchrow(
        "SELECT id FROM endpoints WHERE id = $1", endpoint_id
    )
    if not row:
        raise HTTPException(status_code=404, detail="Endpoint not found")

    # Store the event with its headers and payload
    event_id = await pool.fetchval(
        """
        INSERT INTO events (endpoint_id, headers, payload)
        VALUES ($1, $2::jsonb, $3::jsonb)
        RETURNING id
        """,
        endpoint_id,
        __import__("json").dumps(headers),
        __import__("json").dumps(body),
    )

    # Enqueue a background job to evaluate rules and deliver
    redis = request.app.state.redis
    await redis.enqueue_job("deliver_event", str(event_id))

    return {"event_id": str(event_id), "status": "accepted"}

What does this route do?

The route accepts any JSON POST to /ingest/{endpoint_id}. It first checks that the endpoint exists in the database. If it does, it inserts the raw payload and headers into the events table.

After storing the event, it enqueues a job called deliver_event onto the Redis queue. The worker (which you will build in the next step) picks up this job and handles rule evaluation and delivery. This separation means the API responds instantly with a 202 status instead of making the caller wait.

Why return 202 instead of 200?

HTTP 202 means "accepted for processing." It signals to the caller that the webhook was received but will be processed asynchronously. This is the standard pattern used by production webhook platforms because it keeps ingestion fast and decoupled from downstream delivery.

  • Save main.py.

Seeing import errors in your editor?

Make sure your virtual environment is activated. Your editor needs to use the Python interpreter from the virtual environment to resolve fastapi, asyncpg, and arq imports.

help me fix import errors for fastapi, asyncpg, or arq in my editor

✔️ Awesome, I've got everything!

Great. Double check you saved main.py with both the lifespan handler and the ingest route.

ⓧ I'd like to double check the full code

import uuid
from contextlib import asynccontextmanager

import asyncpg
from arq import create_pool
from arq.connections import RedisSettings
from fastapi import FastAPI, Request, HTTPException

REDIS_SETTINGS = RedisSettings()
DATABASE_URL = "postgresql://localhost/hookrelay"


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db = await asyncpg.create_pool(DATABASE_URL)
    app.state.redis = await create_pool(REDIS_SETTINGS)
    yield
    await app.state.db.close()
    await app.state.redis.close()


app = FastAPI(lifespan=lifespan)


@app.post("/ingest/{endpoint_id}", status_code=202)
async def ingest(endpoint_id: uuid.UUID, request: Request):
    body = await request.json()
    headers = dict(request.headers)

    pool = request.app.state.db
    row = await pool.fetchrow(
        "SELECT id FROM endpoints WHERE id = $1", endpoint_id
    )
    if not row:
        raise HTTPException(status_code=404, detail="Endpoint not found")

    event_id = await pool.fetchval(
        """
        INSERT INTO events (endpoint_id, headers, payload)
        VALUES ($1, $2::jsonb, $3::jsonb)
        RETURNING id
        """,
        endpoint_id,
        __import__("json").dumps(headers),
        __import__("json").dumps(body),
    )

    redis = request.app.state.redis
    await redis.enqueue_job("deliver_event", str(event_id))

    return {"event_id": str(event_id), "status": "accepted"}
Seed the database with test data

To test the ingestion API, you need at least one endpoint, one target, and one routing rule in the database. The seed file inserts these with fixed UUIDs so you can reference them in curl commands later.

  • Create a file called seed.sql in your hookrelay/ project folder and paste this code:
-- A test endpoint representing an external webhook source
INSERT INTO endpoints (id, name) VALUES
    ('a1b2c3d4-e5f6-7890-abcd-ef1234567890', 'test-endpoint');

-- A target that will receive matched events at port 9000
INSERT INTO targets (id, endpoint_id, url, name) VALUES
    ('b2c3d4e5-f6a7-8901-bcde-f12345678901', 'a1b2c3d4-e5f6-7890-abcd-ef1234567890', 'http://localhost:9000/hook', 'My Test Target');

-- A rule: only deliver events where event_type equals order.created
INSERT INTO rules (target_id, field_path, operator, value) VALUES
    ('b2c3d4e5-f6a7-8901-bcde-f12345678901', 'event_type', 'equals', 'order.created');

What does this seed data set up?

The seed creates a test endpoint called test-endpoint. Webhooks POSTed to its UUID will be stored in the events table.

The target points to http://localhost:9000/hook (a mock server you will build in the next step). The rule says: only deliver events where the event_type field in the JSON payload equals order.created.

  • Save seed.sql.
  • Apply the seed data by running:
psql -d hookrelay -f seed.sql
  • You should see INSERT 0 1 printed three times (one for each insert).
  • Verify the seed data loaded correctly by running:
psql -d hookrelay -c "SELECT name FROM endpoints;"
  • You should see test-endpoint in the output.

Seeing a foreign key violation?

The seed inserts must run in order because targets references endpoints and rules references targets. If you see a foreign key error, make sure you applied schema.sql first and that it completed without errors.

help me fix a foreign key violation when seeding my hookrelay database

✔️ Awesome, I've got everything!

Great. Double check you saved seed.sql and saw three INSERT 0 1 confirmations.

ⓧ I'd like to double check the full code

INSERT INTO endpoints (id, name) VALUES
    ('a1b2c3d4-e5f6-7890-abcd-ef1234567890', 'test-endpoint');

INSERT INTO targets (id, endpoint_id, url, name) VALUES
    ('b2c3d4e5-f6a7-8901-bcde-f12345678901', 'a1b2c3d4-e5f6-7890-abcd-ef1234567890', 'http://localhost:9000/hook', 'My Test Target');

INSERT INTO rules (target_id, field_path, operator, value) VALUES
    ('b2c3d4e5-f6a7-8901-bcde-f12345678901', 'event_type', 'equals', 'order.created');

Your database is structured and seeded. Next up, you will build the rule engine and async worker that pick up enqueued jobs and deliver matched events to their targets.

Build the Rule Engine and Delivery Worker

Your ingestion API is accepting webhooks and storing them in PostgreSQL. Now you need the logic that decides where each event should go.

In this step, you will build a rule engine that evaluates JSON payloads against your stored rules, an async worker that fetches events and delivers them to matching targets, and a mock server to observe deliveries landing.

In this step, get ready to:
  • Create a rule engine that matches JSON fields using dot-notation paths.
  • Create an async worker that evaluates rules and delivers webhooks via HTTP POST.
  • Create a mock target server to receive and log delivered events.
Create the rule engine

The rule engine is the decision-maker in your pipeline. It takes an incoming JSON payload and a set of rules, then returns whether the payload matches all of them.

  • Create a new file called rule_engine.py in your hookrelay/ folder.
  • Add the get_nested_value function by pasting this code:
def get_nested_value(payload: dict, field_path: str):
    # Split dot-notation path into individual keys
    keys = field_path.split(".")
    current = payload
    # Walk down the nested dict one key at a time
    for key in keys:
        if isinstance(current, dict) and key in current:
            current = current[key]
        else:
            return None
    return current

What does this function do?

This function extracts a value from a nested JSON payload using a dot-notation path like payment.amount. It splits the path on dots and walks down the dictionary one key at a time.

If any key along the path is missing, it returns None instead of crashing. This makes rule evaluation safe against payloads that do not contain the expected structure.

  • Below get_nested_value, add the evaluate_rules function:

def evaluate_rules(payload: dict, rules: list[dict]) -> bool:
    for rule in rules:
        # Extract what we need from each rule
        field_path = rule["field_path"]
        operator = rule["operator"]
        expected = rule["value"]

        # Get the actual value from the payload at this path
        actual = get_nested_value(payload, field_path)

        if operator == "equals":
            # Compare as strings for consistent matching
            if str(actual) != expected:
                return False
        else:
            # Unknown operator means the rule cannot pass
            return False

    return True

What does this function do?

This function takes a payload and a list of rules, then checks whether ALL rules match. It uses an AND logic approach. If any single rule fails, the function returns False immediately.

The equals operator compares the extracted value (converted to a string) against the rule's expected value. This is the same pattern used by production webhook platforms like Convoy for payload matching.

  • Save rule_engine.py.
  • Verify the module loads without errors by running:
python -c "from rule_engine import evaluate_rules; print('OK')"

You should see OK printed to the terminal.

Seeing an import error?

  • Make sure your virtual environment is still activated. You should see (venv) or similar at the start of your terminal prompt.
  • Check that rule_engine.py is saved in the same hookrelay/ directory where your terminal is running.

Help me debug an import error when loading rule_engine.py

✔️ Awesome, I've got everything!

Great. Double check you have saved rule_engine.py before moving on.

ⓧ I'd like to double check the full code

def get_nested_value(payload: dict, field_path: str):
    keys = field_path.split(".")
    current = payload
    for key in keys:
        if isinstance(current, dict) and key in current:
            current = current[key]
        else:
            return None
    return current


def evaluate_rules(payload: dict, rules: list[dict]) -> bool:
    for rule in rules:
        field_path = rule["field_path"]
        operator = rule["operator"]
        expected = rule["value"]

        actual = get_nested_value(payload, field_path)

        if operator == "equals":
            if str(actual) != expected:
                return False
        else:
            return False

    return True
Create the delivery worker

The worker is a background process that picks up jobs from Redis and delivers events to their matching targets. It uses arq to manage the job queue.

  • Create a new file called worker.py in your hookrelay/ folder.
  • Add the imports, database URL, and lifecycle functions by pasting this code:
import json

import asyncpg
import httpx
from arq.connections import RedisSettings

from rule_engine import evaluate_rules

DATABASE_URL = "postgresql://localhost/hookrelay"


# Called once when the worker starts up
async def startup(ctx):
    ctx["db"] = await asyncpg.create_pool(DATABASE_URL)
    ctx["http"] = httpx.AsyncClient(timeout=10.0)


# Called once when the worker shuts down
async def shutdown(ctx):
    await ctx["http"].aclose()
    await ctx["db"].close()

What does this code do?

The startup function runs once when the worker process begins. It creates an asyncpg connection pool and an httpx async HTTP client, storing both in ctx (a shared dictionary that every job function receives).

The shutdown function cleans up those resources when the worker stops. Creating the pool once in startup (instead of per-job) avoids the overhead of reconnecting to the database on every delivery.

  • Below the shutdown function, add the deliver_event function:

async def deliver_event(ctx, event_id: str):
    pool = ctx["db"]
    client = ctx["http"]

    # Fetch the event from the database
    event = await pool.fetchrow(
        "SELECT id, endpoint_id, payload FROM events WHERE id = $1",
        __import__("uuid").UUID(event_id),
    )
    if not event:
        print(f"Event {event_id} not found, skipping.")
        return

    payload = json.loads(event["payload"])
    endpoint_id = event["endpoint_id"]

    # Get all active targets for this endpoint
    targets = await pool.fetch(
        """
        SELECT t.id, t.url, t.name
        FROM targets t
        WHERE t.endpoint_id = $1 AND t.is_active = TRUE
        """,
        endpoint_id,
    )

    for target in targets:
        # Get rules for this specific target
        rules = await pool.fetch(
            "SELECT field_path, operator, value FROM rules WHERE target_id = $1",
            target["id"],
        )

        rule_list = [dict(r) for r in rules]

        # Deliver if no rules exist OR if all rules match
        if not rule_list or evaluate_rules(payload, rule_list):
            print(f"Delivering event {event_id} to {target['name']} ({target['url']})")
            response = await client.post(target["url"], json=payload)
            print(f"  -> Response: {response.status_code}")

What does this function do?

This is the core job function that arq calls for each queued event. It fetches the stored event from PostgreSQL, finds all active targets for that endpoint, and evaluates each target's rules against the payload.

If a target has no rules, the event is delivered unconditionally. If rules exist, delivery only happens when ALL rules match. This is the fan-out pattern: one event can be delivered to multiple targets simultaneously.

  • Below deliver_event, add the WorkerSettings class that registers everything with arq:


class WorkerSettings:
    functions = [deliver_event]
    on_startup = startup
    on_shutdown = shutdown
    redis_settings = RedisSettings()

What does WorkerSettings do?

This class is how arq discovers your worker configuration. It declares which job functions are available, which lifecycle hooks to call, and where to find Redis.

When you run arq worker.WorkerSettings later, arq imports this class and uses it to start the worker process with all the right settings.

  • Save worker.py.
  • Verify the module loads correctly by running:
python -c "from worker import WorkerSettings; print('OK')"

You should see OK printed with no errors.

Seeing a ModuleNotFoundError?

  • If it says rule_engine is missing, confirm rule_engine.py is saved in the same directory as worker.py.
  • If it says asyncpg or httpx is missing, make sure your virtual environment is activated and run pip install asyncpg httpx again.

Help me fix a ModuleNotFoundError in worker.py

✔️ Awesome, I've got everything!

Great. Double check you have saved worker.py before moving on.

ⓧ I'd like to double check the full code

import json

import asyncpg
import httpx
from arq.connections import RedisSettings

from rule_engine import evaluate_rules

DATABASE_URL = "postgresql://localhost/hookrelay"


async def startup(ctx):
    ctx["db"] = await asyncpg.create_pool(DATABASE_URL)
    ctx["http"] = httpx.AsyncClient(timeout=10.0)


async def shutdown(ctx):
    await ctx["http"].aclose()
    await ctx["db"].close()


async def deliver_event(ctx, event_id: str):
    pool = ctx["db"]
    client = ctx["http"]

    event = await pool.fetchrow(
        "SELECT id, endpoint_id, payload FROM events WHERE id = $1",
        __import__("uuid").UUID(event_id),
    )
    if not event:
        print(f"Event {event_id} not found, skipping.")
        return

    payload = json.loads(event["payload"])
    endpoint_id = event["endpoint_id"]

    targets = await pool.fetch(
        """
        SELECT t.id, t.url, t.name
        FROM targets t
        WHERE t.endpoint_id = $1 AND t.is_active = TRUE
        """,
        endpoint_id,
    )

    for target in targets:
        rules = await pool.fetch(
            "SELECT field_path, operator, value FROM rules WHERE target_id = $1",
            target["id"],
        )

        rule_list = [dict(r) for r in rules]

        if not rule_list or evaluate_rules(payload, rule_list):
            print(f"Delivering event {event_id} to {target['name']} ({target['url']})")
            response = await client.post(target["url"], json=payload)
            print(f"  -> Response: {response.status_code}")


class WorkerSettings:
    functions = [deliver_event]
    on_startup = startup
    on_shutdown = shutdown
    redis_settings = RedisSettings()
Create the mock target server

You need somewhere for the worker to deliver events. A mock target server gives you a visible destination so you can confirm deliveries are landing correctly.

  • Create a new file called mock_target.py in your hookrelay/ folder.
  • Paste the following code to set up a minimal FastAPI server on port 9000:
from fastapi import FastAPI, Request

app = FastAPI()


@app.post("/hook")
async def receive_hook(request: Request):
    # Log the incoming payload to the terminal
    body = await request.json()
    print(f"\n--- Webhook Received ---")
    print(f"Payload: {body}")
    print(f"------------------------\n")
    return {"status": "received"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=9000)

What does this server do?

This is a simple HTTP server with a single POST /hook route. When the worker delivers a webhook here, the server prints the payload to your terminal so you can see exactly what arrived.

The if __name__ == "__main__" block lets you run this file directly with python mock_target.py. It starts on port 9000, which matches the target URL you seeded into the database earlier.

  • Save mock_target.py.
  • Confirm all three new files exist by running:
ls rule_engine.py worker.py mock_target.py

You should see all three filenames listed in the output.

A file is missing from the output?

  • Make sure you saved each file in the hookrelay/ directory (the same folder where main.py lives).
  • Check for typos in the filename. Python files must end with .py.

Help me find why my files are missing from the ls output

✔️ Awesome, I've got everything!

Great. Double check you have saved mock_target.py before moving on.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, Request

app = FastAPI()


@app.post("/hook")
async def receive_hook(request: Request):
    body = await request.json()
    print(f"\n--- Webhook Received ---")
    print(f"Payload: {body}")
    print(f"------------------------\n")
    return {"status": "received"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=9000)

All three pieces of the delivery pipeline are in place. Next up, you will start all services together and watch a webhook flow from ingestion through rule matching to delivery.

Test the Full Pipeline

In the last step, you built the rule engine and delivery worker. Now it's time to see the full pipeline in action.

You'll start three services in separate terminals and send test webhooks to prove that routing works. A matching event should flow from ingestion through rule evaluation to delivery. A non-matching event should stop at storage.

In this step, get ready to:
  • Start the mock target, arq worker, and FastAPI server.
  • Send a webhook that matches the routing rule and observe delivery.
  • Confirm that a non-matching webhook is stored but not delivered.
Start all three services

The pipeline has three moving parts that need to run simultaneously. Each one runs in its own terminal tab so their logs stay separate.

  • Open three new terminal tabs (or windows). In each one, navigate to your hookrelay/ folder and activate your virtual environment.

Why three separate terminals?

Each service blocks its terminal while running. The mock target listens for deliveries, the arq worker polls Redis for jobs, and the FastAPI server accepts incoming webhooks. Keeping them in separate tabs lets you watch each one's logs independently.

  • In Terminal 1, start the mock target server by running:
python mock_target.py
  • You should see a message containing Uvicorn running on http://0.0.0.0:9000 confirming the mock target is listening.
  • In Terminal 2, start the arq worker by running:
arq worker.WorkerSettings
  • You should see output indicating the worker has started and is polling for jobs.
  • In Terminal 3, start the ingestion API by running:
fastapi dev main.py
  • You should see Uvicorn running on http://127.0.0.1:8000 confirming the API is ready to accept requests.

Service won't start?

  • If you see a "port already in use" error, another process is occupying that port. Stop it or choose a different port.
  • If arq worker.WorkerSettings fails with a connection error, confirm Redis is running with redis-cli ping.
  • Make sure your virtual environment is activated in each terminal before running the command.

Still stuck? Help me debug why my services won't start.

Send a matching webhook

Your seed data includes a rule that matches events where event_type equals order.created. Time to send a payload that triggers this rule and watch it flow through the entire system.

  • Open a fourth terminal tab (your virtual environment is not required here since you're using curl).
  • Send a matching webhook by running:
curl -X POST http://localhost:8000/ingest/a1b2c3d4-e5f6-7890-abcd-ef1234567890 \
  -H "Content-Type: application/json" \
  -d '{"event_type": "order.created", "order_id": "abc123"}'

What does this command do?

  • curl -X POST sends an HTTP POST request to your ingestion API.
  • The UUID in the URL (a1b2c3d4-e5f6-7890-abcd-ef1234567890) is the test endpoint ID you seeded earlier.
  • -d sends a JSON body with event_type set to order.created. This matches the rule in your database.

Now check the output across all your terminals:

  • Terminal 4 (curl): you should see a JSON response with "status": "accepted" and an event_id.
  • Terminal 2 (arq worker): you should see log output showing the event being delivered to My Test Target with a 200 response code.
  • Terminal 1 (mock target): you should see --- Webhook Received --- followed by the payload {'event_type': 'order.created', 'order_id': 'abc123'}.

The event traveled from ingestion to storage, through rule evaluation, and out to the target URL. That is the full fan-out pipeline working end to end.

Mock target didn't receive anything?

  • Check Terminal 2 (arq worker) for errors. If the worker cannot reach http://localhost:9000/hook, make sure the mock target in Terminal 1 is still running.
  • If the worker shows no activity at all, the job may not have been enqueued. Confirm that Redis is running and that Terminal 3 (FastAPI) shows no errors.
  • If curl returned a 404, double-check the endpoint UUID matches a1b2c3d4-e5f6-7890-abcd-ef1234567890 from your seed data.

Still stuck? Help me figure out why my webhook didn't get delivered to the mock target.

Send a non-matching webhook

The rule engine should reject events that do not match. Sending a payload with a different event_type proves that routing is selective.

  • In Terminal 4, send a non-matching webhook by running:
curl -X POST http://localhost:8000/ingest/a1b2c3d4-e5f6-7890-abcd-ef1234567890 \
  -H "Content-Type: application/json" \
  -d '{"event_type": "user.updated", "user_id": "xyz"}'
  • Terminal 4 (curl): you should again see "status": "accepted" with a new event_id. The API always accepts and stores the event.
  • Terminal 2 (arq worker): the worker picks up the job but does NOT log a delivery line because the rule does not match.
  • Terminal 1 (mock target): no new output appears. The event was stored in PostgreSQL but never delivered.

Why does the API still return 202?

Webhook platforms always accept first and evaluate later. The 202 status means "received and queued for processing." Whether the event matches a rule is determined asynchronously by the worker, not at ingestion time.

This pattern separates acceptance from processing. The sender gets a fast response regardless of how complex your routing logic becomes.

Your webhook relay is fully operational. Events flow in, get stored, evaluated against rules, and selectively delivered. Next up, the Secret Mission challenges you to handle what happens when delivery fails.

Secret mission

Add Retry with Exponential Backoff

Right now, if a target is down when your worker tries to deliver, the event is lost forever. In production webhook platforms, failed deliveries are retried with increasing delays. Your mission: add exponential backoff so transient failures recover automatically.

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

  • PostgreSQL database hookrelay (stores events, endpoints, targets, and rules).
  • Redis queue data (jobs processed by the arq worker).
  • Project directory hookrelay/ on your Desktop (source files and virtual environment).

Keep everything running

No action needed. Choose this if you are still actively building on HookRelay or plan to extend it soon.

  • Your PostgreSQL hookrelay database stays in place for future queries.
  • Redis continues running as a system service.
  • Your project files remain on the Desktop.

Pause - I'll come back to this later

Stop the running processes to free up memory. Your files and database stay intact for when you return.

  • Stop the FastAPI dev server, arq worker, and mock target by pressing Ctrl+C in each of their terminal windows.
  • PostgreSQL and Redis will continue running as background services but consume minimal resources when idle.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • Stop all running processes by pressing Ctrl+C in each terminal window (FastAPI server, arq worker, mock target).
  • Drop the PostgreSQL database by running:
dropdb hookrelay
  • Clear the Redis queue data by running:
redis-cli FLUSHALL
  • Delete the project directory by running:
rm -rf ~/Desktop/hookrelay

On Windows, run this instead:

rmdir /s /q %USERPROFILE%\Desktop\hookrelay

Nice Work!

Nice Work!

You just built a fully functional webhook relay service from scratch. HookRelay ingests arbitrary JSON, routes it through configurable rules, and delivers it to target URLs with retry support.

You've learned how to:

  • Build a FastAPI webhook ingestion API that accepts any HTTP POST, persists events to PostgreSQL, and enqueues async jobs.
  • Implement a JSON rule engine with dot-notation field path matching and fan-out delivery to multiple targets.
  • Build an async arq worker that evaluates routing rules and delivers matched events via HTTP POST.
  • Secret Mission: Added retry with exponential backoff using arq's built-in Retry mechanism.

Ready to quiz yourself?