Build 3 — Async Webhook Handler

# 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 ```bash uv add fastapi 'uvicorn[standard]' arq redis pydantic-settings ``` ## Step 2: Webhook endpoint (ACK fast) ```python 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 ```python # 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: ```bash arq tasks.WorkerSettings ``` ## Step 4: Idempotency table ```python 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: ```python 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: ```python 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: ```python @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 ```python 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 se