Вебхуки в Django: приём и идемпотентность на практике

Ситуация из практики: интеграция с платёжным шлюзом, эндпоинт принимает уведомление об успешной оплате и начисляет пользователю баланс. В тестах всё работает. В проде через неделю всплывает жалоба — у части клиентов баланс задвоился, а у одного утроился. Причина не в коде начисления, а в том, что провайдер повторил доставку одного и того же события: первый ответ пришёл к нему с таймаутом, и он ретраил, пока не получил 200. Обработчик отработал три раза.

Вебхук — это HTTP-запрос, который внешний сервис шлёт вам при наступлении события. Ключевое свойство почти всех провайдеров: доставка at-least-once, «хотя бы один раз». Один и тот же запрос может прийти дважды, трижды, с разницей в миллисекунды или в минуты. Значит, обработчик обязан быть идемпотентным — повторный вызов с теми же данными не должен менять состояние системы сверх первого вызова. Ниже — как это делается в Django 5.x с PostgreSQL 16, с рабочим кодом и разбором мест, где решение ломается.

Почему очевидная схема даёт дубликаты

Наивный обработчик выглядит так: приняли POST, распарсили JSON, начислили, вернули 200. Проблема не в одном месте, а в трёх сразу.

• Ретраи провайдера. Если ваш ответ не дошёл или пришёл с задержкой больше таймаута провайдера (обычно 5–30 секунд), он повторит запрос. Ваш код уже начислил, но провайдер об этом не знает.
• Долгая обработка внутри запроса. Пока вы синхронно ходите в три внешних сервиса и пишете в базу, провайдер успевает отвалиться по таймауту и ретраить. Теперь два запроса выполняются параллельно.
• Отсутствие проверки на уникальность события. Даже поймав дубликат, без записи «это событие уже обработано» вы не отличите его от нового.

Идемпотентность вебхука — это не «постараться не задваивать». Это гарантия на уровне базы: два запроса с одним идентификатором события физически не могут создать две записи.

Три подхода к дедупликации

1. Уникальный индекс на идентификаторе события

Большинство провайдеров кладут в тело или заголовок уникальный ID события (у Stripe это поле id вида evt_... и заголовок Stripe-Signature; у других — event_id, X-Request-Id). Заводим таблицу принятых событий с уникальным индексом и вставляем туда ID до обработки. Повторная вставка падает на IntegrityError — это сигнал «уже видели».

2. Блокировка строки на время обработки

Уникального индекса мало, если два дубля пришли параллельно: оба прошли проверку «нет такой записи» до того, как первый успел вставить. Здесь помогает атомарная вставка плюс select_for_update на связанной сущности (например, на счёте пользователя), чтобы начисление не выполнялось одновременно.

3. Вынос тяжёлой работы в очередь

Эндпоинт должен отвечать провайдеру быстро — за сотни миллисекунд, а не за секунды. Приняли, проверили подпись, записали событие, поставили задачу в Celery, вернули 200. Вся долгая логика — в воркере. Это резко снижает шанс ретрая по таймауту.

Рабочая реализация

Собираем всё вместе. Модель для журнала событий:

# models.py — Django 5.x
from django.db import models

class WebhookEvent(models.Model):
    # ID события от провайдера — источник идемпотентности
    external_id = models.CharField(max_length=255, unique=True)
    provider = models.CharField(max_length=50)
    payload = models.JSONField()
    processed = models.BooleanField(default=False)
    received_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        indexes = [
            models.Index(fields=["provider", "received_at"]),
        ]

Уникальность по external_id — это и есть замок. База не даст вставить второе событие с тем же ID.

Вьюха приёма. Проверяем подпись, атомарно фиксируем событие, ставим задачу:

# views.py — Django 5.x
import json
from django.db import IntegrityError, transaction
from django.http import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from .models import WebhookEvent
from .signatures import verify_signature  # ваша проверка HMAC
from .tasks import process_payment_event

@csrf_exempt          # вебхук приходит без CSRF-токена
@require_POST
def payment_webhook(request):
    raw_body = request.body

    # 1. Подпись проверяем ДО парсинга — иначе примем мусор
    if not verify_signature(raw_body, request.headers.get("X-Signature", "")):
        return HttpResponseBadRequest("invalid signature")

    try:
        payload = json.loads(raw_body)
    except json.JSONDecodeError:
        return HttpResponseBadRequest("invalid json")

    external_id = payload.get("id")
    if not external_id:
        return HttpResponseBadRequest("no event id")

    # 2. Атомарная вставка — дедупликация на уровне БД
    try:
        with transaction.atomic():
            event = WebhookEvent.objects.create(
                external_id=external_id,
                provider="payments",
                payload=payload,
            )
    except IntegrityError:
        # Событие уже принимали — отвечаем 200, чтобы провайдер
        # прекратил ретраи, но ничего не делаем повторно
        return HttpResponse(status=200)

    # 3. Тяжёлую работу — в очередь, эндпоинт отвечает быстро
    process_payment_event.delay(event.id)
    return HttpResponse(status=200)

Обратите внимание на порядок: подпись до JSON, вставка в отдельной transaction.atomic(), задача — только после успешной вставки. Если сначала поставить задачу, а потом упасть на вставке, воркер обработает то, чего нет в журнале.

Воркер с защитой от параллельного начисления:

# tasks.py — Celery 5.x
from celery import shared_task
from django.db import transaction

from .models import WebhookEvent, Account

@shared_task
def process_payment_event(event_pk):
    with transaction.atomic():
        # Блокируем строку события, чтобы два воркера
        # не взяли одно и то же (например, при ретрае задачи)
        event = (
            WebhookEvent.objects
            .select_for_update()
            .get(pk=event_pk)
        )
        if event.processed:
            return  # уже обработано — выходим тихо

        account = (
            Account.objects
            .select_for_update()
            .get(user_id=event.payload["user_id"])
        )
        amount = event.payload["amount"]
        account.balance += amount
        account.save(update_fields=["balance"])

        event.processed = True
        event.save(update_fields=["processed"])

Флаг processed под блокировкой закрывает случай, когда Celery ретраит саму задачу после сбоя воркера. select_for_update на счёте — от гонки, если по одному пользователю пришло несколько разных событий разом.

Где эта схема ломается

• Провайдер не даёт стабильный ID. Тогда индекс строить не на чем. Придётся собирать ключ идемпотентности из полей тела — тип события, сумма, ID заказа, метка времени — и это хрупко. Проверьте документацию провайдера: у большинства ID есть, просто лежит не там, где ждёте.
• ID стабилен, но меняется payload. Некоторые провайдеры при ретрае шлют тот же ID с обновлёнными данными. Слепая дедупликация по ID проглотит обновление. Решайте по смыслу события: для «платёж прошёл» первое состояние финально, для «статус заказа изменился» — нет.
• Долгая транзакция и блокировки. select_for_update держит строку до конца транзакции. Если внутри воркера ходить во внешний API под блокировкой, счёт залочен на время сетевого запроса — растут ожидания и риск дедлоков. Все внешние вызовы — вне блокировки.
• Проверка подписи после чтения тела. Django кеширует request.body, но если где-то в мидлваре уже прочитали поток, повторное чтение может дать пусто. Читайте request.body целиком в самом начале вьюхи, до любых обращений к request.POST.

Что делать дальше

Схема выше закрывает дубликаты, но не заменяет наблюдаемость. Поставьте на прод три вещи: метрику расхождения между числом принятых событий и обработанных (растёт — застряла очередь), алерт на всплеск IntegrityError (провайдер начал массово ретраить — где-то отвалилась обработка) и ручку для повторного проигрывания события из журнала WebhookEvent.payload, чтобы чинить последствия без запроса к провайдеру.

Отдельная тема — очистка журнала событий: он растёт бесконечно, и через полгода в нём миллионы строк. Здесь её не разбираем, но заложите TTL или партиционирование по received_at сразу. Порядок захвата блокировок при select_for_update на нескольких таблицах, чтобы не поймать дедлок, — тоже отдельный разговор, всегда блокируйте сущности в одном и том же порядке во всех обработчиках.