Build 3 — Async Webhook Handler

The brief

Build a webhook receiver: validates signature, ACKs quickly (<2s), processes async with retries. Pattern for Stripe, GitHub, Slack, etc.

Step 1: Setup

uv add fastapi 'uvicorn[standard]' arq redis pydantic-settings

Step 2: Webhook endpoint (ACK fast)

import hmac, hashlib
from fastapi import FastAPI, Request, HTTPException, Header, Depends
from arq import create_pool
from arq.connections import RedisSettings

app = FastAPI()

async def get_arq_pool():
    return await create_pool(RedisSettings())

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

@app.post("/webhook/stripe")
async def stripe_webhook(
    request: Request,
    stripe_signature: str = Header(None),
    pool = Depends(get_arq_pool),
):
    payload = await request.body()

    if not verify_signature(payload, stripe_signature, settings.stripe_secret):
        raise HTTPException(401, "invalid signature")

    event = await request.json()

    # Quick ACK; process async
    await pool.enqueue_job(
        "process_stripe_event",
        event["id"],
        event["type"],
        event["data"],
    )

    return {"received": True}

ACK <100ms; webhook sender happy.

Step 3: Async processor

# tasks.py
from arq.connections import RedisSettings

async def process_stripe_event(ctx, event_id: str, event_type: str, data: dict):
    # Idempotency: check if already processed
    if await already_processed(event_id):
        return

    if event_type == "payment_intent.succeeded":
        await handle_payment_success(data)
    elif event_type == "customer.subscription.deleted":
        await handle_subscription_cancel(data)
    # ...

    await mark_processed(event_id)

class WorkerSettings:
    functions = [process_stripe_event]
    redis_settings = RedisSettings()
    max_tries = 5                # retries on failure

Run worker:

arq tasks.WorkerSettings

Step 4: Idempotency table

class ProcessedEvent(Base):
    __tablename__ = "processed_events"
    event_id: Mapped[str] = mapped_column(primary_key=True)
    processed_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

async def already_processed(event_id: str) -> bool:
    return await db.scalar(select(ProcessedEvent).where(ProcessedEvent.event_id == event_id)) is not None

async def mark_processed(event_id: str):
    db.add(ProcessedEvent(event_id=event_id))
    await db.commit()

Critical: webhooks at-least-once delivered. Without idempotency, you'd charge customers twice on duplicates.

Step 5: Retry behavior

ARQ retries automatically (max_tries). For controlled retries:

async def process_stripe_event(ctx, event_id, event_type, data):
    try:
        # work
        pass
    except TransientError:
        # ARQ will retry
        raise
    except FatalError:
        # don't retry; mark as failed
        await mark_failed(event_id)
        return

Distinguish transient (retry) from fatal (don't retry).

Step 6: Dead letter queue

After max retries, log + alert:

async def process_stripe_event(ctx, event_id, event_type, data):
    if ctx["job_try"] >= 5:
        await send_alert(f"DLQ: {event_id}")
        # don't raise; ARQ stops retrying
        return
    # normal work

Step 7: Replay

For investigating issues, allow replaying events:

@app.post("/admin/replay/{event_id}")
async def replay(event_id: str, admin: User = Depends(require_admin), pool = Depends(get_arq_pool)):
    event = await fetch_stored_event(event_id)
    if not event:
        raise HTTPException(404)
    # Mark not processed (or use different idempotency key for replay)
    await pool.enqueue_job("process_stripe_event", event["id"], event["type"], event["data"])
    return {"queued": True}

For replay, you might want to skip idempotency, OR keep it (replays will be no-op if already processed).

Step 8: Testing

def test_webhook_ack_fast(client):
    payload = {"id": "evt_123", "type": "test", "data": {}}
    body = json.dumps(payload).encode()
    sig = compute_signature(body, "test-secret")
    response = client.post("/webhook/stripe", data=body, headers={"stripe-signature": sig})
    assert response.status_code == 200

def test_invalid_signature(client):
    response = client.post("/webhook/stripe", json={"id": "x"}, headers={"stripe-signature": "fake"})
    assert response.status_code == 401

Deliverables

Webhook endpoint with signature verification.
Async task processing via ARQ.
Idempotency table; replay-safe.
DLQ behavior on max retries.
Tests for happy + error paths.

What you'll learn

Webhook signature verification (HMAC).
Fast ACK pattern (quick endpoint + async processing).
Idempotency for at-least-once delivery.
Retry semantics with ARQ.
DLQ pattern.

Stretch

Stripe SDK integration for typed events.
Webhook delivery retries from your side (call third-party webhook URL with backoff).
Webhook subscription management UI.
Per-event-type handlers via registry pattern.

📊 100-Point Evaluation Rubric (Case 3)

Evaluation Dimension	Weight	Target Standard (Pass ≥ 80%)
1. Fast ACK + Verify	40 pts	HMAC-SHA256 signature check with `compare_digest`; 401 on bad signature; ACK returned in <2s without waiting on work.
2. Async Processing	30 pts	ARQ worker with retries (max_tries ≥ 3); idempotency table prevents double-processing on redelivery.
3. Docs + OpenAPI	15 pts	README documents the signed-payload contract and replay procedure; `/docs` shows the webhook route schema.
4. Code Quality	15 pts	Raw-body read before JSON parse, constant-time compare, no secret logging.

📤 Submission Instructions & Guidelines

Deliverable Formats:
- Option A (Recommended): Upload a single .zip archive containing your project directory: src/, tests/, Dockerfile (where applicable), and README.md with run instructions.
- Option B: Upload your README.md and include your GitHub repository link in the Submission Notes box below.
Pre-Submission Verification:
- Run your full test suite (pytest) green; include the passing output in your README.
- Exercise every endpoint once via /docs (Swagger UI) and paste one sample request/response per endpoint into your README.
- Confirm no secrets are committed (.env git-ignored; secrets via environment only).

Assignment Instructions