Idempotent Webhook Ingestion With a Postgres Dedupe Store in FastAPI

import datetime as dt
from dataclasses import dataclass

from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

from .models import WebhookDelivery


@dataclass
class ClaimResult:
    delivery: WebhookDelivery
    is_new: bool


def claim_delivery(db: Session, dedupe_key: str, provider: str, event_id: str) -> ClaimResult:
    stmt = (
        pg_insert(WebhookDelivery)
        .values(dedupe_key=dedupe_key, provider=provider, event_id=event_id, status="processing")
        .on_conflict_do_nothing(index_elements=["dedupe_key"])
        .returning(WebhookDelivery.id)
    )
    inserted_id = db.execute(stmt).scalar_one_or_none()
    db.commit()

    if inserted_id is not None:
        row = db.get(WebhookDelivery, inserted_id)
        return ClaimResult(delivery=row, is_new=True)

    existing = db.execute(
        select(WebhookDelivery).where(WebhookDelivery.dedupe_key == dedupe_key)
    ).scalar_one()
    return ClaimResult(delivery=existing, is_new=False)


def mark_succeeded(db: Session, delivery: WebhookDelivery) -> None:
    delivery.status = "succeeded"
    delivery.completed_at = dt.datetime.utcnow()
    db.commit()

import hashlib
import hmac

from fastapi import APIRouter, BackgroundTasks, Depends, Header, HTTPException, Request
from sqlalchemy.orm import Session

from .db import get_db
from .idempotency import claim_delivery, mark_succeeded
from .processing import handle_event

router = APIRouter()
WEBHOOK_SECRET = b"whsec_replace_me"


def verify_signature(body: bytes, signature: str) -> None:
    expected = hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()
    if not signature or not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="invalid signature")


@router.post("/webhooks/stripe")
async def receive_stripe(
    request: Request,
    background: BackgroundTasks,
    stripe_signature: str = Header(default="", alias="Stripe-Signature"),
    db: Session = Depends(get_db),
):
    body = await request.body()
    verify_signature(body, stripe_signature)

    payload = await request.json()
    event_id = payload.get("id")
    if not event_id:
        raise HTTPException(status_code=400, detail="missing event id")

    dedupe_key = f"stripe:{event_id}"
    result = claim_delivery(db, dedupe_key, provider="stripe", event_id=event_id)

    if not result.is_new:
        if result.delivery.status == "succeeded":
            return {"status": "duplicate", "event_id": event_id}
        return {"status": "in_progress", "event_id": event_id}

    def _process():
        handle_event(payload)
        mark_succeeded(db, result.delivery)

    background.add_task(_process)
    return {"status": "accepted", "event_id": event_id}