gendesign/tradein-mvp/backend/app/api/v1/payments.py

"""Платёжный роутер МЕРЫ (Т-Банк эквайринг) — PR-D3: checkout, нотификация, выдача.

Проводка уже готовых слоёв: `app/services/payments/*` (подпись, разбор, httpx-
клиент — PR-C, БД и конфига не знают), схема `data/sql/233_payments.sql` (PR-B),
периметр (`ratelimit`/`request_audit`/`sentry_scrub` — PR-D2). Здесь — только
то, чего не было: HTTP-ручки и статус-машина.

ВСЁ за kill-switch `settings.payments_enabled` (дефолт False): при выключенном
контуре каждая ручка отвечает 503 и не ходит ни в банк, ни в платёжные таблицы.
Merge безопасен на выключенном контуре — на проде это ровно ноль изменений
поведения, пока владелец не выставит PAYMENTS_ENABLED=true вместе с ключами
терминала (`app/main.py` роняет старт, если включить без ключей).

── Идемпотентность держится на БД, а не на «проверить-потом-вставить» ────────
Все три гонки, которые здесь реальны (двойной клик по кнопке оплаты; банк шлёт
AUTHORIZED и CONFIRMED одновременно; банк ретраит нотификацию почасово сутки),
закрыты UNIQUE-ключами миграций 233 и 279 + `ON CONFLICT DO NOTHING`. Пара
«SELECT, потом INSERT» здесь была бы дефектом: между ними успевает пройти
параллельный запрос, и выдача (или холд на карте) происходит дважды. SELECT
живого платежа в `checkout` остался, но только как быстрый путь для честного
повтора — гонку ловит не он, а `payments_live_estimate_product_uidx`.

`payment_notifications.processed_at` — единственный признак «выдача
состоялась». Наличие строки нотификации таким признаком НЕ является: процесс
мог упасть между INSERT нотификации и выдачей, и тогда ретрай банка обязан
довести выдачу до конца, а не ответить "OK" на полпути (см. блок про
processed_at в шапке 233_payments.sql).

── Никаких исходящих HTTP внутри notify ─────────────────────────────────────
Бюджет одного вызова `TBankClient` — до ~74 с (см. его докстринг), окно ответа
банку — порядка 10 с. Поэтому обработчик нотификации только валидирует,
сохраняет и отвечает `"OK"` текстом (банк ждёт именно эту строку); любая сверка
с банком (GetState/CheckOrder) — задача реконсиляции, вне HTTP-цикла.

── Доставка купленного: capability-ссылка ───────────────────────────────────
`/r/<token>` — непредсказуемый `secrets.token_urlsafe(32)`, лежит в
`payment_entitlements.subject` (колонка документирована как «username или
anon-token, кому выдано»). Право доступа — сам токен: у покупателя-физлица на
meraocenka.ru идентичности нет и не будет. `ref_id` при этом остаётся
`estimate_id` — как задокументировано в миграции: именно на `(payment_id, kind,
ref_id)` держится UNIQUE «выдали один раз», и подстановка туда случайного
токена молча отменила бы эту гарантию (каждый повтор дал бы новый ref_id →
новую строку).

Токен НЕ логируется: ни в `logger.*` здесь, ни в GlitchTip — путь `/r/<token>`
режет `redact_report_link_token` в `app/observability/sentry_scrub.py`.
"""

from __future__ import annotations

import asyncio
import json
import logging
import secrets
from datetime import UTC, datetime
from typing import Annotated, Any
from uuid import UUID, uuid4

from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response
from pydantic import BaseModel, Field
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.api.v1.trade_in import (
    ESTIMATE_READABLE_SQL,
    _assert_estimate_access,
    load_estimate,
)
from app.core.config import settings
from app.core.db import get_db
from app.schemas.trade_in import AggregatedEstimate
from app.services.payments.notification import NotificationParseError, parse_notification
from app.services.payments.receipt import ReceiptItem, build_receipt, receipt_total_kopecks
from app.services.payments.tbank_client import TBankApiError, TBankClient
from app.services.payments.token import verify_notification_token

logger = logging.getLogger(__name__)

router = APIRouter()

# ── Каталог ───────────────────────────────────────────────────────────────────
# Цена — НЕ из тела запроса (иначе клиент назначает её сам), а из этой таблицы
# по product_code. 15 000 копеек = 150 ₽ = `SERVICE_PRICE_RUB` в
# frontend/src/app/mera-public/content.ts, где цена названа в оферте (п. 4.1) и
# в политике возврата. Рассинхронизацию ловит тест
# tests/test_payments_router.py::test_price_matches_published_offer — цена,
# отличающаяся от опубликованной в оферте, это не баг рендера, а неисполнение
# договора.
PRODUCT_PAID_REPORT = "paid_report"
_PRODUCTS: dict[str, tuple[str, int]] = {
    # product_code: (наименование позиции чека — <=128 символов, цена в копейках)
    PRODUCT_PAID_REPORT: ("Оценка стоимости квартиры (онлайн-отчёт)", 15_000),
}

ENTITLEMENT_REPORT_LINK = "report_link"

# Полный список из CHECK payments_status_check (233_payments.sql). Любой статус
# вне списка пишется как 'UNKNOWN' — контракт миграции: тихо исказить статус
# хуже, чем громко упасть, но и падать на нотификации нельзя (банк ретраит
# сутки). Дрейф относительно миграции ловит
# tests/test_payments_router.py::test_status_whitelist_matches_migration.
_KNOWN_STATUSES = frozenset(
    {
        "NEW",
        "FORM_SHOWED",
        "DEADLINE_EXPIRED",
        "CANCELED",
        "PREAUTHORIZING",
        "AUTHORIZING",
        "AUTHORIZED",
        "AUTH_FAIL",
        "REJECTED",
        "3DS_CHECKING",
        "3DS_CHECKED",
        "CHECKING",
        "CHECKED",
        "PROCESSING",
        "CONFIRMING",
        "CONFIRMED",
        "COMPLETING",
        "COMPLETED",
        "REVERSING",
        "PARTIAL_REVERSED",
        "REVERSED",
        "REFUNDING",
        "PARTIAL_REFUNDED",
        "REFUNDED",
        "REFUND_FAILED",
        "UNKNOWN",
    }
)

# Статусы ДО подтверждения: нотификация с таким статусом не имеет права
# затереть уже проставленный CONFIRMED. Банк не гарантирует порядок доставки
# (AUTHORIZED и CONFIRMED уходят одновременно при одностадийной оплате), а
# ретрай «отставшей» нотификации может прийти через час — без этой проверки
# оплаченный платёж откатился бы в AUTHORIZED и отчёт перестал бы выдаваться.
_PRE_CONFIRM_STATUSES = frozenset(
    {
        "NEW",
        "FORM_SHOWED",
        "PREAUTHORIZING",
        "AUTHORIZING",
        "AUTHORIZED",
        "3DS_CHECKING",
        "3DS_CHECKED",
        "CHECKING",
        "CHECKED",
        "PROCESSING",
        "CONFIRMING",
    }
)

# Платёж в одном из этих статусов ещё «живой»: повторный checkout по той же
# оценке обязан вернуть ту же ссылку, а не создавать второй холд на карте
# покупателя. Терминальные (CANCELED/REJECTED/REFUNDED/...) сюда не входят —
# после отказа человек вправе попробовать оплатить заново.
#
# Этот же список — предикат частичного UNIQUE(estimate_id, product_code)
# миграции 279, который и делает «один живой платёж» свойством БД, а не
# порядка выполнения. Расхождение кода и миграции ловит
# tests/test_payments_router.py::test_live_status_predicate_matches_code.
_REUSABLE_STATUSES = _PRE_CONFIRM_STATUSES

# Статусы, из которых платёж НИКОГДА не выйдет сам: банк шлёт нотификации по
# исходу платежа, а не по факту «форма открыта», поэтому брошенный checkout
# остаётся в NEW/FORM_SHOWED навсегда. Без границы по времени покупатель
# получал бы на каждый повторный checkout одну и ту же ссылку — а она у банка
# уже протухла, и начать оплату заново становилось бы невозможно.
#
# Границу двигаем ТОЛЬКО по этим двум статусам. Всё, что дальше по цепочке
# (AUTHORIZING/AUTHORIZED/3DS_*/CHECKING/PROCESSING/CONFIRMING), означает, что
# карточный поток уже начался и деньги, возможно, захолдированы: пометить такое
# «просроченным» и выдать вторую ссылку — это и есть второй холд. Разгребать
# зависшие карточные статусы — работа реконсиляции (PR-E, GetState), а не
# checkout'а.
_ABANDONABLE_STATUSES = frozenset({"NEW", "FORM_SHOWED"})

# 30 минут. Обоснование срока: за это окно нельзя «случайно» не дойти до карты —
# ввод карты и 3DS укладываются в минуты, а как только поток начался, статус
# уходит из _ABANDONABLE_STATUSES и окно к платежу вообще не применяется.
# Значит, к моменту истечения окна карточный поток провабельно не начинался, и
# освободить пару (estimate_id, product_code) под новую попытку безопасно.
# Остаточный риск честно называем: если покупатель через час всё-таки дооплатит
# СТАРУЮ форму, банк подтвердит её (выдача состоится по ней — статус-машина
# ниже это переживает), а новая попытка так и останется неоплаченной; два
# списания требуют, чтобы человек намеренно оплатил обе формы.
_ABANDONED_AFTER_MINUTES = 30

# Статус для строки, чей Init отвалился: своего кода вроде INIT_FAILED в
# CHECK миграции 233 нет, а заводить его ради этого случая значило бы менять
# схему ради ярлыка — «почему» и так лежит в error_code/error_message. Берём
# терминальный DEADLINE_EXPIRED, которым checkout уже помечает попытки, из
# которых платёж не выйдет сам. Обязательное свойство ровно одно: статус ВНЕ
# предиката 279 (= вне _REUSABLE_STATUSES), иначе мёртвая строка продолжит
# держать пару (estimate_id, product_code). Это стережёт
# tests/test_payments_router.py::test_init_failed_status_is_terminal.
_INIT_FAILED_STATUS = "DEADLINE_EXPIRED"

_ORDER_ID_PREFIX = "mera-"
# 32 байта энтропии (43 символа base64url) — перебор capability-ссылки
# неосуществим, а сама ссылка остаётся кликабельной в мессенджере.
_REPORT_TOKEN_BYTES = 32


def _mark_init_failed(db: Session, order_id: str, *, code: str, message: str) -> None:
    """Переводит строку провалившегося Init в терминальный статус + «почему».

    Общая для ВСЕХ исходов, после которых ссылки у строки не будет: отказ банка
    и Success:true без PaymentURL. Статус NEW тут оставлять нельзя — см.
    комментарий к `_INIT_FAILED_STATUS`: строка без `payment_url` невидима для
    `_find_live_payment`, но видима предикату UNIQUE миграции 279, и следующий
    checkout получил бы ложный 409 на все `_ABANDONED_AFTER_MINUTES`.
    """
    db.execute(
        text(
            """
            UPDATE payments
            SET status = :status,
                error_code = :code,
                error_message = :message,
                updated_at = NOW()
            WHERE order_id = :order_id
            """
        ),
        {
            "status": _INIT_FAILED_STATUS,
            "code": code[:64],
            "message": message[:500],
            "order_id": order_id,
        },
    )
    db.commit()


def _require_enabled() -> None:
    """Kill-switch контура. 503, а не 404: путь существует, приём оплаты выключен."""
    if not settings.payments_enabled:
        raise HTTPException(status_code=503, detail="payments are disabled")


def _client() -> TBankClient:
    return TBankClient(
        terminal_key=settings.tbank_terminal_key,
        password=settings.tbank_password.get_secret_value(),
        base_url=settings.tbank_api_base_url,
    )


class CheckoutInput(BaseModel):
    estimate_id: UUID
    product_code: str = Field(default=PRODUCT_PAID_REPORT, max_length=64)
    # Чек 54-ФЗ требует Email ИЛИ Phone. Оба опциональны здесь и проверяются
    # только когда чек включён (TBANK_RECEIPT_ENABLED) — до подключения ОФД
    # требовать контакт незачем.
    customer_email: str | None = Field(default=None, max_length=254)
    customer_phone: str | None = Field(default=None, max_length=32)


class CheckoutOut(BaseModel):
    order_id: str
    payment_url: str
    amount_kopecks: int
    status: str


class ReportLinkOut(BaseModel):
    """Статус оплаты для экрана «после оплаты».

    `report_url` — None, пока выдача не состоялась. Именно None, а не
    правдоподобная ссылка «которая скоро заработает»: пустой результат честнее
    ссылки, ведущей в 404.
    """

    order_id: str
    status: str
    report_url: str | None


@router.post("/payments/checkout", response_model=CheckoutOut)
def checkout(
    payload: CheckoutInput,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> CheckoutOut:
    """Создаёт платёж и возвращает `PaymentURL` формы Т-Банка.

    Идемпотентность — свойство БД, а не порядка выполнения: частичный UNIQUE
    (estimate_id, product_code) по живым статусам (миграция 279) физически не
    даёт существовать двум живым платежам по одной оценке, а `ON CONFLICT DO
    NOTHING` превращает проигрыш в гонке в ответ, а не во второй `Init` (и,
    значит, во второй холд на карте покупателя).

    Три исхода: живой платёж с готовой ссылкой — 200 с ТОЙ ЖЕ ссылкой; параллельный
    checkout ещё не дошёл до ответа банка — 409 (ретрай через секунду вернёт
    ссылку); иначе создаём новый платёж.
    """
    _require_enabled()

    product = _PRODUCTS.get(payload.product_code)
    if product is None:
        raise HTTPException(status_code=400, detail="unknown product_code")
    item_name, amount_kopecks = product

    estimate = db.execute(
        text(
            f"""
            SELECT id, created_by
            FROM trade_in_estimates
            WHERE id = CAST(:id AS uuid) AND {ESTIMATE_READABLE_SQL}
            """
        ),
        {"id": str(payload.estimate_id)},
    ).fetchone()
    if estimate is None:
        raise HTTPException(status_code=404, detail="estimate not found or expired")
    if estimate.created_by is not None:
        # Тот же IDOR-гвард (#690), что у остальных ручек по оценке. Он здесь не
        # про «показать чужой отчёт напрямую», а про две другие двери: checkout
        # по чужому estimate_id возвращает order_id чужого живого платежа (ветка
        # переиспользования ниже), а order_id — это право доступа для
        # /payments/status/<order_id>, который отдаёт capability-ссылку на отчёт,
        # как только владелец заплатит.
        #
        # Проверяем только оценки, у которых владелец ЕСТЬ. У анонимной покупки
        # на meraocenka.ru идентичности нет (см. блок про capability-ссылку в
        # шапке модуля), и правом там работает сам неугадываемый estimate_id —
        # требовать заголовок означало бы сделать анонимный checkout
        # невозможным, а не более безопасным.
        _assert_estimate_access(estimate.created_by, x_authenticated_user)

    # Освобождаем пару (estimate_id, product_code) от брошенных попыток ДО
    # проверки живого платежа: иначе и переиспользование вернуло бы мёртвую
    # ссылку, и UNIQUE миграции 279 не дал бы создать новую (см. комментарий у
    # _ABANDONED_AFTER_MINUTES).
    db.execute(
        text(
            """
            UPDATE payments
            SET status = 'DEADLINE_EXPIRED', updated_at = NOW()
            WHERE estimate_id = CAST(:estimate_id AS uuid)
              AND product_code = :product_code
              AND status = ANY(CAST(:abandonable AS text[]))
              AND created_at < NOW() - make_interval(mins => CAST(:mins AS int))
            """
        ),
        {
            "estimate_id": str(payload.estimate_id),
            "product_code": payload.product_code,
            "abandonable": sorted(_ABANDONABLE_STATUSES),
            "mins": _ABANDONED_AFTER_MINUTES,
        },
    )

    existing = _find_live_payment(db, payload)
    if existing is not None:
        return CheckoutOut(
            order_id=existing.order_id,
            payment_url=existing.payment_url,
            amount_kopecks=existing.amount_kopecks,
            status=existing.status,
        )

    receipt = _build_receipt_or_none(item_name, amount_kopecks, payload)

    # order_id генерируем СВОЙ и до похода в банк: он и есть ключ, по которому
    # нотификация найдёт платёж, а UNIQUE(order_id) — страховка от двойной
    # записи. 5 + 32 = 37 символов, влезает в CHECK(char_length <= 50).
    order_id = f"{_ORDER_ID_PREFIX}{uuid4().hex}"
    inserted = db.execute(  # fetchone() ДО commit(): курсор после коммита пуст
        text(
            """
            INSERT INTO payments (
                order_id, terminal_key, product_code, amount_kopecks, status,
                created_by, estimate_id, customer_email, customer_phone
            ) VALUES (
                :order_id, :terminal_key, :product_code, :amount, 'NEW',
                :created_by, CAST(:estimate_id AS uuid), :email, :phone
            )
            ON CONFLICT DO NOTHING
            RETURNING order_id
            """
        ),
        {
            "order_id": order_id,
            "terminal_key": settings.tbank_terminal_key,
            "product_code": payload.product_code,
            "amount": amount_kopecks,
            "created_by": x_authenticated_user,
            "estimate_id": str(payload.estimate_id),
            "email": payload.customer_email,
            "phone": payload.customer_phone,
        },
    ).fetchone()
    db.commit()

    if inserted is None:
        # Гонку выиграл параллельный checkout (двойной клик). Своего Init не
        # делаем ни при каких условиях — он и есть второй холд.
        rival = _find_live_payment(db, payload)
        if rival is not None:
            return CheckoutOut(
                order_id=rival.order_id,
                payment_url=rival.payment_url,
                amount_kopecks=rival.amount_kopecks,
                status=rival.status,
            )
        # Соперник вставил строку, но ответа банка ещё не получил: ссылки пока
        # нет ни у кого. Честный 409 — придумать ссылку нечем, а ждать чужого
        # Init внутри HTTP-цикла нельзя (его бюджет — до ~74 с).
        logger.info("checkout: параллельный checkout ещё в полёте, estimate_id известен клиенту")
        raise HTTPException(status_code=409, detail="checkout already in progress, retry shortly")

    try:
        # asyncio.run в синхронном хендлере — тот же мост, что и
        # trade_in._try_revive_dead_estimate: Starlette гоняет `def`-хендлер в
        # threadpool, поэтому свой event loop здесь никому не мешает, а
        # остальные db.execute() остаются синхронными.
        init = asyncio.run(
            _client().init_payment(
                order_id=order_id,
                amount_kopecks=amount_kopecks,
                description=item_name,
                notification_url=settings.tbank_notification_url or None,
                success_url=settings.tbank_success_url or None,
                fail_url=settings.tbank_fail_url or None,
                receipt=receipt,
                pay_type=settings.tbank_pay_type,
            )
        )
    except TBankApiError as exc:
        # Запись остаётся в БД с текстом ошибки — иначе факт попытки (и заказа,
        # который банк мог принять до обрыва) не остался бы нигде. Холда здесь
        # быть не может: Init только заводит заказ и отдаёт ссылку на форму, а
        # авторизация суммы происходит, когда покупатель платит по форме — её
        # ему не выдавали. Слепой повтор Init по тому же order_id всё равно
        # запрещён (см. докстринг init_payment) — это работа реконсиляции.
        #
        # Статус — терминальный (см. `_mark_init_failed`); если банк всё же
        # пришлёт по этой строке нотификацию, статус-машина notify доведёт её до
        # конца (терминальный статус не блокирует CONFIRMED).
        _mark_init_failed(db, order_id, code=exc.error_code, message=str(exc))
        logger.warning("checkout: Init отклонён банком, order_id=%s: %s", order_id, exc)
        raise HTTPException(status_code=502, detail="payment provider error") from exc

    payment_url = init.get("PaymentURL")
    tbank_payment_id = init.get("PaymentId")
    if not isinstance(payment_url, str) or not payment_url:
        # Success:true без PaymentURL — контракт банка нарушен; выдумывать
        # ссылку нечем. Замок тот же, что и у отказа выше, и даже вернее: банк
        # заказ ПРИНЯЛ, а ссылки у строки уже не будет — оставить её в NEW
        # значит отдать следующему checkout ложный 409 на 30 минут.
        _mark_init_failed(
            db,
            order_id,
            code="no_payment_url",
            message=f"Init: Success без PaymentURL (Status={init.get('Status')!r})",
        )
        logger.error("checkout: Init без PaymentURL, order_id=%s", order_id)
        raise HTTPException(status_code=502, detail="payment provider returned no payment url")

    status = init.get("Status")
    db.execute(
        text(
            """
            UPDATE payments
            SET tbank_payment_id = :payment_id,
                payment_url = :payment_url,
                status = :status,
                init_response = CAST(:init AS jsonb),
                updated_at = NOW()
            WHERE order_id = :order_id
            """
        ),
        {
            "payment_id": str(tbank_payment_id) if tbank_payment_id is not None else None,
            "payment_url": payment_url,
            "status": status if status in _KNOWN_STATUSES else "UNKNOWN",
            "init": json.dumps(init, ensure_ascii=False),
            "order_id": order_id,
        },
    )
    db.commit()

    return CheckoutOut(
        order_id=order_id,
        payment_url=payment_url,
        amount_kopecks=amount_kopecks,
        status=status if status in _KNOWN_STATUSES else "UNKNOWN",
    )


def _find_live_payment(db: Session, payload: CheckoutInput) -> Any:
    """Живой платёж по этой оценке, у которого уже есть ссылка на форму.

    `payment_url IS NOT NULL` обязателен: строка без ссылки — это чужой checkout
    в полёте (Init ещё не ответил), и отдавать по ней `payment_url=None` значило
    бы соврать в схеме ответа. Такой случай — 409 у вызывающей стороны.
    """
    return db.execute(
        text(
            """
            SELECT order_id, payment_url, amount_kopecks, status
            FROM payments
            WHERE estimate_id = CAST(:estimate_id AS uuid)
              AND product_code = :product_code
              AND payment_url IS NOT NULL
              AND status = ANY(CAST(:reusable AS text[]))
            ORDER BY created_at DESC
            LIMIT 1
            """
        ),
        {
            "estimate_id": str(payload.estimate_id),
            "product_code": payload.product_code,
            "reusable": sorted(_REUSABLE_STATUSES),
        },
    ).fetchone()


def _build_receipt_or_none(
    item_name: str, amount_kopecks: int, payload: CheckoutInput
) -> dict[str, Any] | None:
    """Чек 54-ФЗ — только когда владелец включил его и задал систему налогообложения.

    Сверка `receipt_total_kopecks == Init.Amount` здесь и есть тот инвариант,
    который `build_receipt` намеренно не проверяет у себя (см. его докстринг):
    чек на сумму, отличную от списанной, — это фискальное нарушение, а не
    косметика.
    """
    if not settings.tbank_receipt_enabled:
        return None
    if not settings.tbank_taxation:
        logger.error("checkout: TBANK_RECEIPT_ENABLED=true, но TBANK_TAXATION не задан")
        raise HTTPException(status_code=503, detail="receipt is not configured")
    if not payload.customer_email and not payload.customer_phone:
        raise HTTPException(status_code=400, detail="customer_email or customer_phone is required")
    receipt = build_receipt(
        items=[ReceiptItem(name=item_name, price_kopecks=amount_kopecks)],
        taxation=settings.tbank_taxation,  # type: ignore[arg-type]
        email=payload.customer_email,
        phone=payload.customer_phone,
    )
    if receipt_total_kopecks(receipt) != amount_kopecks:
        logger.error("checkout: сумма чека разошлась с суммой заказа")
        raise HTTPException(status_code=500, detail="receipt total mismatch")
    return receipt


@router.post("/payments/notify")
async def notify(request: Request, db: Annotated[Session, Depends(get_db)]) -> Response:
    """Вебхук Т-Банка. Отвечает `"OK"` текстом — банк ждёт именно эту строку.

    Любой другой ответ банк трактует как недоставленную нотификацию и ретраит.
    Поэтому "OK" отдаётся и на том, что мы обработать не можем, но что нашей
    проблемой не является (неизвестный OrderId); отказ (4xx) остаётся только
    там, где принять событие означало бы солгать про деньги: неверная подпись
    и расхождение суммы.
    """
    _require_enabled()

    try:
        body = await request.json()
    except Exception:
        raise HTTPException(status_code=400, detail="malformed body") from None
    if not isinstance(body, dict):
        raise HTTPException(status_code=400, detail="malformed body")

    token_valid = verify_notification_token(body, settings.tbank_password.get_secret_value())

    # Пишем сырой лог ДО любых выводов о содержимом (включая невалидную
    # подпись): append-only лог входящих — единственное место, где остаётся
    # факт попытки. Ключи читаем «как есть», без разбора: разбор может упасть,
    # запись факта — нет.
    notif = _log_notification(db, body, token_valid=token_valid)

    if not token_valid:
        db.commit()
        logger.warning("notify: подпись не сошлась — нотификация отвергнута")
        raise HTTPException(status_code=403, detail="invalid token")

    try:
        parsed = parse_notification(body)
    except NotificationParseError as exc:
        db.commit()
        logger.warning("notify: тело не разобралось: %s", exc)
        raise HTTPException(status_code=400, detail="malformed notification") from None

    if notif is not None and notif.processed_at is not None:
        # Выдача по этой нотификации уже состоялась — ретрай банка.
        db.commit()
        return Response(content="OK", media_type="text/plain")

    payment = db.execute(
        text(
            """
            SELECT id, order_id, status, amount_kopecks, estimate_id, created_by
            FROM payments
            WHERE order_id = :order_id
            """
        ),
        {"order_id": parsed.order_id},
    ).fetchone()
    if payment is None:
        # Чужой/устаревший OrderId. Ретраи не помогут — отвечаем "OK", факт
        # уже лежит в payment_notifications.
        db.commit()
        logger.warning("notify: нотификация по неизвестному order_id")
        return Response(content="OK", media_type="text/plain")

    if parsed.amount_kopecks != payment.amount_kopecks:
        # Подпись Т-Банка конкатенирует значения БЕЗ разделителя, поэтому сама
        # по себе не гарантирует сумму (см. докстринг notification.py). Сверка
        # с суммой, записанной при Init, — единственная реальная защита.
        db.commit()
        logger.error(
            "notify: сумма нотификации разошлась с суммой заказа (order_id=%s)", parsed.order_id
        )
        raise HTTPException(status_code=400, detail="amount mismatch")

    status = parsed.status if parsed.status in _KNOWN_STATUSES else "UNKNOWN"
    if not (payment.status == "CONFIRMED" and status in _PRE_CONFIRM_STATUSES):
        _apply_status(db, payment.order_id, status, parsed.payment_id)

    if status == "CONFIRMED" and parsed.success:
        _fulfill(db, payment_id=payment.id, estimate_id=payment.estimate_id)

    if notif is not None:
        # processed_at ставится ПОСЛЕ выдачи — иначе ретрай банка увидел бы
        # «обработано» по нотификации, выдача по которой не состоялась.
        db.execute(
            text(
                "UPDATE payment_notifications SET processed_at = NOW() "
                "WHERE id = :id AND processed_at IS NULL"
            ),
            {"id": notif.id},
        )
    db.commit()
    return Response(content="OK", media_type="text/plain")


def _log_notification(db: Session, body: dict[str, Any], *, token_valid: bool) -> Any:
    """INSERT ... ON CONFLICT DO NOTHING + добор уже существующей строки.

    Дедуп делает БД (UNIQUE NULLS NOT DISTINCT на (tbank_payment_id, status,
    amount_kopecks, token), миграция 233), а не «SELECT, потом INSERT»: между
    проверкой и вставкой проходит параллельный ретрай банка, и выдача
    происходит дважды.

    Если строка уже была — достаём её тем же ключом через IS NOT DISTINCT FROM
    (зеркало NULLS NOT DISTINCT), потому что решение «выдавать или нет»
    принимается по её `processed_at`, а не по факту существования.
    """
    key = {
        "order_id": _raw_str(body.get("OrderId")),
        "payment_id": _raw_str(body.get("PaymentId")),
        "status": _raw_str(body.get("Status")),
        "amount": body.get("Amount") if isinstance(body.get("Amount"), int) else None,
        "token": _raw_str(body.get("Token")),
    }
    inserted = db.execute(
        text(
            """
            INSERT INTO payment_notifications (
                order_id, tbank_payment_id, status, amount_kopecks, token, token_valid, body
            ) VALUES (
                :order_id, :payment_id, :status, :amount, :token, :token_valid,
                CAST(:body AS jsonb)
            )
            ON CONFLICT DO NOTHING
            RETURNING id, processed_at
            """
        ),
        {**key, "token_valid": token_valid, "body": json.dumps(body, ensure_ascii=False)},
    ).fetchone()
    if inserted is not None:
        return inserted
    return db.execute(
        text(
            """
            SELECT id, processed_at
            FROM payment_notifications
            WHERE tbank_payment_id IS NOT DISTINCT FROM :payment_id
              AND status IS NOT DISTINCT FROM :status
              AND amount_kopecks IS NOT DISTINCT FROM :amount
              AND token IS NOT DISTINCT FROM :token
            """
        ),
        key,
    ).fetchone()


def _raw_str(value: Any) -> str | None:
    """Значение для сырого лога: строка как есть, всё остальное — NULL.

    Приводить чужие типы к строке здесь нельзя: дедуп-ключ должен совпадать у
    оригинала и ретрая побайтово, а `str(1)` и `"1"` — уже разные истории.
    """
    return value if isinstance(value, str) else None


def _apply_status(db: Session, order_id: str, status: str, tbank_payment_id: str) -> None:
    """Двигает статус платежа и проставляет отметки времени переходов."""
    db.execute(
        text(
            """
            UPDATE payments
            SET status = :status,
                tbank_payment_id = COALESCE(tbank_payment_id, :payment_id),
                authorized_at = CASE WHEN :status = 'AUTHORIZED'
                                     THEN COALESCE(authorized_at, NOW()) ELSE authorized_at END,
                confirmed_at  = CASE WHEN :status = 'CONFIRMED'
                                     THEN COALESCE(confirmed_at, NOW()) ELSE confirmed_at END,
                refunded_at   = CASE WHEN :status IN ('REFUNDED', 'PARTIAL_REFUNDED',
                                                      'REVERSED', 'PARTIAL_REVERSED')
                                     THEN COALESCE(refunded_at, NOW()) ELSE refunded_at END,
                updated_at = NOW()
            WHERE order_id = :order_id
            """
        ),
        {"status": status, "payment_id": tbank_payment_id, "order_id": order_id},
    )


def _fulfill(db: Session, *, payment_id: Any, estimate_id: Any) -> None:
    """Выдача: capability-ссылка на оплаченный отчёт + продление хранения оценки.

    «Выдали один раз» гарантирует UNIQUE NULLS NOT DISTINCT (payment_id, kind,
    ref_id) миграции 233: повторная нотификация не вернёт строку из RETURNING и
    второй токен не родится. Проверять существование заранее нельзя — гонка.
    """
    if estimate_id is None:
        logger.error("fulfill: платёж без estimate_id — выдавать нечего")
        return
    row = db.execute(
        text(
            """
            INSERT INTO payment_entitlements (payment_id, subject, kind, ref_id, expires_at)
            VALUES (
                CAST(:payment_id AS uuid), :subject, :kind, CAST(:ref_id AS uuid),
                NOW() + make_interval(days => CAST(:days AS int))
            )
            ON CONFLICT DO NOTHING
            RETURNING id
            """
        ),
        {
            "payment_id": str(payment_id),
            "subject": secrets.token_urlsafe(_REPORT_TOKEN_BYTES),
            "kind": ENTITLEMENT_REPORT_LINK,
            "ref_id": str(estimate_id),
            "days": settings.trade_in_paid_retention_days,
        },
    ).fetchone()
    if row is None:
        logger.info("fulfill: выдача по этому платежу уже была — повтор не создаётся")
        return

    # Оплаченная оценка живёт год (retain_until, миграция 240) — иначе purge-
    # джоба удалит строку через 24 часа и capability-ссылка укажет в пустоту.
    # GREATEST — чтобы повторная покупка не УКОРАЧИВАЛА уже выданный срок.
    db.execute(
        text(
            """
            UPDATE trade_in_estimates
            SET retain_until = GREATEST(
                COALESCE(retain_until, NOW()),
                NOW() + make_interval(days => CAST(:days AS int))
            )
            WHERE id = CAST(:id AS uuid)
            """
        ),
        {"days": settings.trade_in_paid_retention_days, "id": str(estimate_id)},
    )


@router.get("/payments/status/{order_id}", response_model=ReportLinkOut)
def payment_status(
    order_id: str,
    db: Annotated[Session, Depends(get_db)],
) -> ReportLinkOut:
    """Экран «после оплаты»: статус заказа и ссылка на отчёт, когда выдача была.

    Правом здесь работает сам `order_id` — 128 бит случайности, известные
    только браузеру покупателя (он же уходит в банк как OrderId и возвращается
    на SuccessURL). Пока выдачи нет — `report_url: null`.
    """
    _require_enabled()
    row = db.execute(
        text(
            """
            SELECT p.order_id, p.status, pe.subject AS report_token
            FROM payments p
            LEFT JOIN payment_entitlements pe
                   ON pe.payment_id = p.id AND pe.kind = :kind
            WHERE p.order_id = :order_id
            """
        ),
        {"order_id": order_id, "kind": ENTITLEMENT_REPORT_LINK},
    ).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="order not found")
    return ReportLinkOut(
        order_id=row.order_id,
        status=row.status,
        report_url=f"/api/v1/trade-in/r/{row.report_token}" if row.report_token else None,
    )


@router.get("/r/{token}", response_model=AggregatedEstimate)
def report_by_link(
    token: str,
    db: Annotated[Session, Depends(get_db)],
) -> AggregatedEstimate:
    """Оплаченный отчёт по capability-ссылке — БЕЗ RBAC, право доступа = токен.

    Одна и та же 404 на «токена нет», «токен протух» и «оценка удалена»:
    различать их значило бы подтверждать существование чужих токенов
    перебирающему.
    """
    _require_enabled()
    row = db.execute(
        text(
            """
            SELECT ref_id, expires_at
            FROM payment_entitlements
            WHERE kind = :kind AND subject = :token
            """
        ),
        {"kind": ENTITLEMENT_REPORT_LINK, "token": token},
    ).fetchone()
    if row is None or row.ref_id is None:
        raise HTTPException(status_code=404, detail="report not found")
    if row.expires_at is not None and row.expires_at.replace(tzinfo=UTC) <= datetime.now(tz=UTC):
        raise HTTPException(status_code=404, detail="report not found")

    # capability_granted=True: право уже доказано токеном выше. Флаг не является
    # полем запроса — подобрать его снаружи нельзя (см. докстринг load_estimate).
    return load_estimate(
        db, UUID(str(row.ref_id)), x_authenticated_user=None, capability_granted=True
    )