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

@@ -0,0 +1,797 @@
+"""Платёжный роутер МЕРЫ (Т-Банк эквайринг) — 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
+
+_ORDER_ID_PREFIX = "mera-"
+# 32 байта энтропии (43 символа base64url) — перебор capability-ссылки
+# неосуществим, а сама ссылка остаётся кликабельной в мессенджере.
+_REPORT_TOKEN_BYTES = 32
+
+
+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:
+        # Запись остаётся в БД со статусом NEW и текстом ошибки — иначе факт
+        # попытки (и возможного холда, если обрыв случился после приёма запроса
+        # банком) не остался бы нигде. Слепой повтор Init по тому же order_id
+        # запрещён (см. докстринг init_payment) — это работа реконсиляции.
+        db.execute(
+            text(
+                """
+                UPDATE payments
+                SET error_code = :code, error_message = :message, updated_at = NOW()
+                WHERE order_id = :order_id
+                """
+            ),
+            {"code": exc.error_code[:64], "message": str(exc)[:500], "order_id": order_id},
+        )
+        db.commit()
+        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 — контракт банка нарушен; выдумывать
+        # ссылку нечем.
+        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
+    )

tradein-mvp/backend/app/api/v1/trade_in.py

@@ -627,6 +627,31 @@ def get_estimate(
 
     Возвращает 404 если оценка не найдена или TTL истёк.
     """
+    return load_estimate(db, estimate_id, x_authenticated_user=x_authenticated_user)
+
+
+def load_estimate(
+    db: Session,
+    estimate_id: UUID,
+    *,
+    x_authenticated_user: str | None,
+    capability_granted: bool = False,
+) -> AggregatedEstimate:
+    """Тело GET /estimate/{id} без FastAPI-обвязки — чтобы у ВТОРОГО права
+    доступа был тот же самый загрузчик, а не его копия.
+
+    `capability_granted=True` — вызывающая сторона уже доказала право доступа
+    ДРУГИМ способом, чем `X-Authenticated-User` + roles.yaml: capability-ссылка
+    `/r/<token>` (app/api/v1/payments.py) отдаёт оплаченный отчёт анониму,
+    у которого идентичности нет и не будет — там правом является сам
+    непредсказуемый токен, сверенный по `payment_entitlements`.
+
+    Флаг — ИМЕННО параметр обычной функции, а не поле запроса: у route-хендлера
+    `get_estimate` выше его нет, поэтому подобрать его снаружи (query/заголовком)
+    невозможно — включить его может только код в этом процессе. Обратное
+    (добавить параметр в сам хендлер с default=False) сделало бы обход IDOR-
+    гварда #690 доступным любому клиенту через `?capability_granted=true`.
+    """
     row = db.execute(
         text(
             f"""
@@ -654,7 +679,8 @@ def get_estimate(
     if row is None:
         raise HTTPException(status_code=404, detail="estimate not found or expired")
 
-    _assert_estimate_access(row.created_by, x_authenticated_user)
+    if not capability_granted:
+        _assert_estimate_access(row.created_by, x_authenticated_user)
 
     # #incident-2026-08-10: строка «мертва» (median_price<=0/NULL) — посчитана
     # ДО фикса оценщика (#oblast-E/#oblast-F, PR #2823/#2825). Пробуем

tradein-mvp/backend/app/core/rbac.py

@@ -129,8 +129,27 @@ _PUBLIC_PATHS = frozenset(
         # Обе ручки дополнительно закрыты флагом settings.public_estimate_enabled.
         "/api/public/mera/estimate",
         "/api/public/mera/estimate/read",
+        # Вебхук Т-Банка (app/api/v1/payments.py::notify): сервер-к-серверу,
+        # X-Authenticated-User/сессии у банка нет и быть не может. Путь не
+        # секрет — аутентификацией здесь работает подпись `Token` тела,
+        # которую проверяет сам хендлер (services/payments/token.py), плюс
+        # отдельный узкий rate-limit (core/ratelimit.py). Всё остальное
+        # платёжное (checkout, статус заказа) остаётся ЗАКРЫТЫМ — открыт ровно
+        # тот путь, который иначе получил бы 401 у банка.
+        "/api/v1/trade-in/payments/notify",
     }
 )
+# Capability-ссылка на оплаченный отчёт: /api/v1/trade-in/r/<token>. Точной
+# строкой её в множество выше не положить — токен переменный, а множество
+# проверяется как `path in`. Отдельный кортеж префиксов, а не превращение
+# `_PUBLIC_PATHS` в префиксный матчер: тот механизм держит auth-гейт всего
+# бэкенда, расширять его семантику ради одного роута нельзя.
+#
+# Открывается ИМЕННО подпуть /r/ и ничего выше: правом доступа служит сам
+# непредсказуемый токен (32 байта энтропии), сверяемый по payment_entitlements
+# в app/api/v1/payments.py. У покупателя-физлица на meraocenka.ru идентичности
+# нет и не будет, поэтому иного способа отдать ему купленное не существует.
+_PUBLIC_PATH_PREFIXES = ("/api/v1/trade-in/r/",)
 # #R2-H3: Caddy срезает внешний префикс /trade-in (uri strip_prefix) перед
 # tradein-backend, а globs в roles.yaml — ВНЕШНИЕ (/trade-in/api/v1/**). Для
 # scope-проверки восстанавливаем внешний путь.
@@ -217,7 +236,7 @@ async def rbac_guard(
     call_next: Callable[[Request], Awaitable[Response]],
 ) -> Response:
     path = request.url.path
-    if path in _PUBLIC_PATHS:
+    if path in _PUBLIC_PATHS or path.startswith(_PUBLIC_PATH_PREFIXES):
         return await call_next(request)
 
     username: str | None = None

tradein-mvp/backend/app/main.py

@@ -31,6 +31,7 @@ from app.api.v1 import (
     glitchtip,
     lead,
     me,
+    payments,
     privacy_admin,
     search,
     support,
@@ -286,6 +287,11 @@ app.include_router(version.router, prefix="/api/v1/trade-in", tags=["trade-in-ve
 app.include_router(lead.router, prefix="/api/v1/trade-in", tags=["trade-in"])
 app.include_router(support.router, prefix="/api/v1/trade-in", tags=["trade-in-support"])
 app.include_router(glitchtip.router, prefix="/api/v1/trade-in", tags=["trade-in-ops"])
+# Платёжный контур — весь за settings.payments_enabled (дефолт False): роутер
+# подключён всегда, но каждая его ручка отвечает 503, пока контур выключен.
+# Подключать по флагу было бы хуже: путь /payments/notify обязан существовать
+# и отвечать предсказуемо, а не менять форму ответа вместе с конфигом.
+app.include_router(payments.router, prefix="/api/v1/trade-in", tags=["trade-in-payments"])
 app.include_router(buildings.router, prefix="/api/v1/buildings", tags=["buildings"])
 app.include_router(search.router, prefix="/api/v1", tags=["search"])
 app.include_router(me.router, prefix="/api/v1", tags=["me"])

tradein-mvp/backend/app/observability/sentry_scrub.py

@@ -131,6 +131,20 @@ _URL_SECRET_QUERY_REPLACEMENT = r"\g<1>" + _REDACTED
 _HTTPX_ERROR_URL_QUERY_RE = re.compile(r"(for url '[^'?]*)\?[^']*(')")
 _HTTPX_ERROR_URL_QUERY_REPLACEMENT = r"\g<1>?" + _REDACTED + r"\g<2>"
 
+# Capability-токен оплаченного отчёта живёт В ПУТИ (`/api/v1/trade-in/r/<token>`,
+# app/api/v1/payments.py), а не в query и не в теле — значит ни `_PII_KEYS`
+# (ключ-based), ни `scrub_payment_request_body` (режет request.data по сегменту
+# `/payments/`), ни sentry_sdk `sanitize_url` (режет только userinfo и query)
+# его не касаются. А путь попадает в событие несколькими путями сразу:
+# `event.request.url`, `transaction`, breadcrumb'ы, текст исключения. Токен —
+# это ПРАВО ДОСТУПА целиком: утёкший в GlitchTip путь равен выданному отчёту.
+# Поэтому — full-text regex по всему событию, как у TG-токена.
+# `[\w-]` покрывает алфавит `secrets.token_urlsafe` (base64url), хвост
+# `(?=[/?#]|$)` оставляет нетронутым остаток URL (query/фрагмент) — он полезен
+# для диагностики и секретом не является.
+_REPORT_LINK_TOKEN_RE = re.compile(r"(/api/v1/trade-in/r/)[\w-]+(?=[/?#]|$)")
+_REPORT_LINK_TOKEN_REPLACEMENT = r"\g<1>" + _REDACTED
+
 
 def _scrub(obj: Any) -> None:
     """Рекурсивно заменить значения PII-ключей в dict на [REDACTED] (in-place)."""
@@ -205,6 +219,7 @@ def scrub_pii_event(event: Event, _hint: dict[str, Any]) -> Event | None:
     _scrub(event.get("contexts"))
     _regex_redact_inplace(event, _URL_SECRET_QUERY_RE, _URL_SECRET_QUERY_REPLACEMENT)
     _regex_redact_inplace(event, _HTTPX_ERROR_URL_QUERY_RE, _HTTPX_ERROR_URL_QUERY_REPLACEMENT)
+    _regex_redact_inplace(event, _REPORT_LINK_TOKEN_RE, _REPORT_LINK_TOKEN_REPLACEMENT)
     return event
 
 

tradein-mvp/backend/data/sql/279_payments_live_checkout_uidx.sql

@@ -0,0 +1,65 @@
+-- 279_payments_live_checkout_uidx.sql
+-- Идемпотентность checkout на уровне БД: не больше одного ЖИВОГО платежа на
+-- пару (estimate_id, product_code).
+--
+-- ── WHY ──────────────────────────────────────────────────────────────────────
+-- До этой миграции переиспользование живого платежа в
+-- app/api/v1/payments.py::checkout держалось на «SELECT, потом INSERT» — ровно
+-- на том, что шапка того же модуля называет дефектом. Обычный двойной клик по
+-- кнопке оплаты даёт два параллельных запроса: оба проходят SELECT до того, как
+-- любой из них вставит строку, оба делают INSERT, оба зовут Init — и на карте
+-- покупателя появляются ДВА ХОЛДА. Порядок выполнения тут ничего не решает,
+-- решает уникальный ключ: второй INSERT обязан отбиться самой БД.
+--
+-- ── Почему частичный, а не полный UNIQUE(estimate_id, product_code) ──────────
+-- Полный запретил бы повторную покупку навсегда: после CANCELED/REJECTED
+-- человек вправе попробовать оплатить заново, а после CONFIRMED — купить
+-- второй раз. Поэтому в предикате только «живые» статусы — те, при которых
+-- платёжная сессия ещё не завершилась. Список = _REUSABLE_STATUSES
+-- (= _PRE_CONFIRM_STATUSES) в app/api/v1/payments.py; расхождение ловит
+-- tests/test_payments_router.py::test_live_status_predicate_matches_code —
+-- индекс, который шире кода, запрещает легитимный повтор, а который уже —
+-- пропускает второй холд.
+--
+-- Предикат ссылается только на неизменяемые выражения (сравнение колонок с
+-- литералами), поэтому индекс легален как частичный, а строка входит в него и
+-- выходит автоматически при UPDATE статуса.
+--
+-- estimate_id IS NOT NULL — обязательная часть предиката: колонка nullable
+-- (FK ON DELETE SET NULL, 233_payments.sql), а обычный UNIQUE считает NULL
+-- отличным от NULL, так что без этого условия строки с NULL просто копились бы
+-- в индексе без пользы.
+--
+-- ── Безопасность применения ─────────────────────────────────────────────────
+-- Контур выключен (PAYMENTS_ENABLED=false), таблица на проде пуста
+-- (проверено 2026-08-29: SELECT count(*) FROM payments → 0), поэтому обычный
+-- CREATE UNIQUE INDEX не может упасть на существующих дублях и не блокирует
+-- ничью запись. IF NOT EXISTS — повторное применение безвредно.
+
+BEGIN;
+-- Конвенция проекта (#2752): CREATE UNIQUE INDEX на существующей таблице берёт
+-- блокировку и без lock_timeout встанет в очередь за чужой сессией.
+SET LOCAL lock_timeout = '5s';
+
+CREATE UNIQUE INDEX IF NOT EXISTS payments_live_estimate_product_uidx
+    ON payments (estimate_id, product_code)
+    WHERE estimate_id IS NOT NULL
+      AND status IN (
+        'NEW',
+        'FORM_SHOWED',
+        'PREAUTHORIZING',
+        'AUTHORIZING',
+        'AUTHORIZED',
+        '3DS_CHECKING',
+        '3DS_CHECKED',
+        'CHECKING',
+        'CHECKED',
+        'PROCESSING',
+        'CONFIRMING'
+      );
+
+COMMENT ON INDEX payments_live_estimate_product_uidx IS
+    'Не больше одного живого платежа на (estimate_id, product_code): двойной '
+    'клик по кнопке оплаты не создаёт второй холд на карте покупателя.';
+
+COMMIT;

tradein-mvp/backend/tests/test_estimate_idor.py

@@ -635,12 +635,18 @@ def test_estimate_readable_sql_uses_disjunction() -> None:
 def test_get_estimate_sql_built_from_shared_constant() -> None:
     """GET /estimate/{id} SQL filter is built FROM ESTIMATE_READABLE_SQL, not a
     hand-copied literal — regression guard against the two gates drifting apart
-    again (that's exactly what happened before this PR: 404 here, 410 in /pdf)."""
+    again (that's exactly what happened before this PR: 404 here, 410 in /pdf).
+
+    Inspects `load_estimate`, not the `get_estimate` route: the payments PR moved
+    the body there so the paid capability link (`/r/<token>`, app/api/v1/
+    payments.py) reuses the SAME loader instead of growing a third copy of the
+    readability gate — which is precisely what this guard exists to prevent.
+    """
     import inspect
 
-    from app.api.v1.trade_in import get_estimate
+    from app.api.v1.trade_in import load_estimate
 
-    src = inspect.getsource(get_estimate)
+    src = inspect.getsource(load_estimate)
     assert "ESTIMATE_READABLE_SQL" in src
     assert "expires_at > NOW()" not in src, "hand-copied predicate, not the shared constant"
     assert "retain_until" in src, "SELECT must also fetch retain_until"

tradein-mvp/backend/tests/test_payments_router.py

@@ -0,0 +1,800 @@
+"""Роутер платежей (app/api/v1/payments.py): идемпотентность выдачи, подпись,
+kill-switch, capability-ссылка.
+
+ПОЧЕМУ здесь свой мини-эмулятор БД, а не MagicMock. Проверяемое свойство —
+«повторная нотификация НЕ создаёт вторую выдачу» — целиком держится на UNIQUE
+из миграции 233 плюс `ON CONFLICT DO NOTHING`. MagicMock отдаёт то, что ему
+скажут, поэтому такой тест был бы зелёным по построению: он не покраснел бы,
+если убрать `ON CONFLICT` или заменить его на «SELECT, потом INSERT».
+`_FakeDb` ниже объявляет UNIQUE-ключи ОТДЕЛЬНО от проверяемого SQL — ровно
+теми колонками, что записаны в миграции, — и ведёт себя как Postgres: дубль
+без `ON CONFLICT` падает ошибкой, дубль с `ON CONFLICT DO NOTHING` не
+возвращает строку.
+
+Фальсификация проверена руками (каждый раз краснеет ИМЕННО тот тест, который
+про это свойство, и по значению, а не по ImportError):
+- убрать `ON CONFLICT DO NOTHING` из INSERT в `payment_entitlements` →
+  `test_retry_after_crash_between_issue_and_processed_does_not_double_issue`
+  падает 500 вместо "OK";
+- убрать его же из INSERT в `payment_notifications` → падают все три теста
+  про идемпотентность;
+- отключить проверку подписи → `test_notification_with_invalid_token_is_rejected`;
+- отключить проверку срока → `test_report_link_rejects_expired_token`;
+- убрать `ON CONFLICT DO NOTHING` из INSERT в `payments` →
+  `test_parallel_checkout_does_not_create_second_payment` краснеет
+  _UniqueViolationError вместо 409;
+- растянуть `_ABANDONED_AFTER_MINUTES` до бесконечности (= убрать границу по
+  времени) → `test_abandoned_checkout_does_not_lock_the_buyer_out` получает в
+  ответе мёртвую ссылку — красное ПО ЗНАЧЕНИЮ;
+- убрать `_assert_estimate_access` из checkout → `test_checkout_rejects_foreign_estimate`
+  видит 200 и созданный платёж вместо 404.
+
+SQLite вместо этого не годится: NULLS NOT DISTINCT, jsonb, make_interval и
+CAST(:x AS uuid) там не существуют, а настоящий Postgres в юнит-тестах этого
+репозитория не поднимается (см. tests/conftest.py — DATABASE_URL заглушка).
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import sys
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+sys.modules.setdefault("weasyprint.CSS", _wp_mock)
+sys.modules.setdefault("weasyprint.HTML", _wp_mock)
+
+import pytest  # noqa: E402
+from fastapi import FastAPI  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+from pydantic import SecretStr  # noqa: E402
+
+_PASSWORD = "test-terminal-password"
+_ORDER_ID = "mera-0123456789abcdef0123456789abcdef"
+_PAYMENT_ID = "3000000001"
+_PAYMENT_UUID = "22222222-2222-2222-2222-222222222222"
+_ESTIMATE_UUID = "33333333-3333-3333-3333-333333333333"
+_AMOUNT = 15_000
+
+_REPO_ROOT = Path(__file__).resolve().parents[1].parent
+_MIGRATION = _REPO_ROOT / "backend" / "data" / "sql" / "233_payments.sql"
+_CONTENT_TS = _REPO_ROOT / "frontend" / "src" / "app" / "mera-public" / "content.ts"
+
+
+class _UniqueViolationError(RuntimeError):
+    """Стенд-in для psycopg UniqueViolation — INSERT без ON CONFLICT в дубль."""
+
+
+class _FakeDb:
+    """Мини-Postgres на словарях: только те statement'ы, что шлёт роутер.
+
+    UNIQUE-ключи заданы ЗДЕСЬ, по миграции 233, а не выведены из проверяемого
+    SQL — иначе тест поедет вслед за дефектом вместо того, чтобы его поймать.
+    NULL считается равным NULL (NULLS NOT DISTINCT), как в миграции.
+    """
+
+    _NOTIFICATION_KEY = ("tbank_payment_id", "status", "amount_kopecks", "token")
+    _ENTITLEMENT_KEY = ("payment_id", "kind", "ref_id")
+    # Частичный UNIQUE миграции 279: ключ (estimate_id, product_code), предикат —
+    # «живые» статусы. Переписан здесь ПО МИГРАЦИИ, а не импортирован из
+    # payments.py: иначе тест поехал бы вслед за дефектом. Совпадение списка с
+    # кодом отдельно гейтит test_live_status_predicate_matches_code.
+    _LIVE_PAYMENT_KEY = ("estimate_id", "product_code")
+    _LIVE_STATUSES = frozenset(
+        {
+            "NEW",
+            "FORM_SHOWED",
+            "PREAUTHORIZING",
+            "AUTHORIZING",
+            "AUTHORIZED",
+            "3DS_CHECKING",
+            "3DS_CHECKED",
+            "CHECKING",
+            "CHECKED",
+            "PROCESSING",
+            "CONFIRMING",
+        }
+    )
+
+    def __init__(self) -> None:
+        self.notifications: list[SimpleNamespace] = []
+        self.entitlements: list[SimpleNamespace] = []
+        self.payments: list[SimpleNamespace] = [
+            SimpleNamespace(
+                id=_PAYMENT_UUID,
+                order_id=_ORDER_ID,
+                status="NEW",
+                amount_kopecks=_AMOUNT,
+                estimate_id=_ESTIMATE_UUID,
+                created_by=None,
+                # product_code=None — эта строка обслуживает тесты нотификаций и
+                # не должна попадать в ключ живого платежа checkout-тестов.
+                product_code=None,
+                payment_url=None,
+                created_at=datetime.now(tz=UTC),
+            )
+        ]
+        self.estimate_created_by: str | None = None
+        self.retain_until_updates: list[str] = []
+        self.commits = 0
+
+    # -- SQLAlchemy-совместимая поверхность -------------------------------
+    def execute(self, statement: Any, params: dict[str, Any] | None = None) -> Any:
+        sql = " ".join(str(statement).split())
+        params = params or {}
+        if "INSERT INTO payment_notifications" in sql:
+            return self._insert_notification(sql, params)
+        if "FROM payment_notifications" in sql and sql.startswith("SELECT"):
+            return _Result(self._find_notification(params))
+        if "UPDATE payment_notifications" in sql:
+            for row in self.notifications:
+                if row.id == params["id"] and row.processed_at is None:
+                    row.processed_at = datetime.now(tz=UTC)
+            return _Result(None)
+        if "INSERT INTO payment_entitlements" in sql:
+            return self._insert_entitlement(sql, params)
+        if "FROM payment_entitlements" in sql and sql.startswith("SELECT"):
+            return _Result(
+                next(
+                    (
+                        row
+                        for row in self.entitlements
+                        if row.kind == params["kind"] and row.subject == params["token"]
+                    ),
+                    None,
+                )
+            )
+        if "INSERT INTO payments" in sql:
+            return self._insert_payment(sql, params)
+        if "FROM payments" in sql and sql.startswith("SELECT"):
+            if "estimate_id" in params:
+                return _Result(self._find_live_payment(params))
+            return _Result(
+                next((p for p in self.payments if p.order_id == params.get("order_id")), None)
+            )
+        if "UPDATE payments" in sql and "DEADLINE_EXPIRED" in sql:
+            return _Result(self._expire_abandoned(params))
+        if "UPDATE payments" in sql:
+            for payment in self.payments:
+                if payment.order_id == params["order_id"] and "status" in params:
+                    payment.status = params["status"]
+                    if "payment_url" in params:
+                        payment.payment_url = params["payment_url"]
+            return _Result(None)
+        if "FROM trade_in_estimates" in sql and sql.startswith("SELECT"):
+            if params.get("id") != _ESTIMATE_UUID:
+                return _Result(None)
+            return _Result(SimpleNamespace(id=_ESTIMATE_UUID, created_by=self.estimate_created_by))
+        if "UPDATE trade_in_estimates" in sql:
+            self.retain_until_updates.append(params["id"])
+            return _Result(None)
+        raise AssertionError(f"неожиданный SQL в тесте: {sql[:120]}")
+
+    def commit(self) -> None:
+        self.commits += 1
+
+    def close(self) -> None:
+        pass
+
+    # -- эмуляция UNIQUE ---------------------------------------------------
+    def _insert_notification(self, sql: str, params: dict[str, Any]) -> _Result:
+        key = tuple(
+            params[name]
+            for name in ("payment_id", "status", "amount", "token")  # порядок = _NOTIFICATION_KEY
+        )
+        if any(self._key_of(row, self._NOTIFICATION_KEY) == key for row in self.notifications):
+            return self._conflict(sql)
+        row = SimpleNamespace(
+            id=len(self.notifications) + 1,
+            order_id=params["order_id"],
+            tbank_payment_id=params["payment_id"],
+            status=params["status"],
+            amount_kopecks=params["amount"],
+            token=params["token"],
+            token_valid=params["token_valid"],
+            processed_at=None,
+        )
+        self.notifications.append(row)
+        return _Result(row)
+
+    def _insert_payment(self, sql: str, params: dict[str, Any]) -> _Result:
+        """Ведёт себя как Postgres с частичным UNIQUE миграции 279.
+
+        Конфликт наступает только когда УЖЕ есть строка с той же парой
+        (estimate_id, product_code) И статусом из предиката — ровно как у
+        частичного индекса. Без `ON CONFLICT DO NOTHING` в SQL — исключение.
+        """
+        key = (params["estimate_id"], params["product_code"])
+        if key[0] is not None and any(
+            self._key_of(row, self._LIVE_PAYMENT_KEY) == key and row.status in self._LIVE_STATUSES
+            for row in self.payments
+        ):
+            return self._conflict(sql)
+        row = SimpleNamespace(
+            id=f"pay-{len(self.payments) + 1}",
+            order_id=params["order_id"],
+            status="NEW",
+            amount_kopecks=params["amount"],
+            estimate_id=params["estimate_id"],
+            created_by=params["created_by"],
+            product_code=params["product_code"],
+            payment_url=None,
+            created_at=datetime.now(tz=UTC),
+        )
+        self.payments.append(row)
+        return _Result(row)
+
+    def _find_live_payment(self, params: dict[str, Any]) -> SimpleNamespace | None:
+        return next(
+            (
+                row
+                for row in self.payments
+                if self._key_of(row, self._LIVE_PAYMENT_KEY)
+                == (params["estimate_id"], params["product_code"])
+                and row.payment_url is not None
+                and row.status in set(params["reusable"])
+            ),
+            None,
+        )
+
+    def _expire_abandoned(self, params: dict[str, Any]) -> None:
+        deadline = datetime.now(tz=UTC) - timedelta(minutes=int(params["mins"]))
+        for row in self.payments:
+            if (
+                row.estimate_id == params["estimate_id"]
+                and row.product_code == params["product_code"]
+                and row.status in set(params["abandonable"])
+                and row.created_at < deadline
+            ):
+                row.status = "DEADLINE_EXPIRED"
+        return None
+
+    def _insert_entitlement(self, sql: str, params: dict[str, Any]) -> _Result:
+        key = (params["payment_id"], params["kind"], params["ref_id"])
+        if any(self._key_of(row, self._ENTITLEMENT_KEY) == key for row in self.entitlements):
+            return self._conflict(sql)
+        row = SimpleNamespace(
+            id=f"ent-{len(self.entitlements) + 1}",
+            payment_id=params["payment_id"],
+            subject=params["subject"],
+            kind=params["kind"],
+            ref_id=params["ref_id"],
+            expires_at=datetime.now(tz=UTC) + timedelta(days=int(params["days"])),
+        )
+        self.entitlements.append(row)
+        return _Result(row)
+
+    @staticmethod
+    def _key_of(row: SimpleNamespace, names: tuple[str, ...]) -> tuple[Any, ...]:
+        return tuple(getattr(row, name) for name in names)
+
+    @staticmethod
+    def _conflict(sql: str) -> _Result:
+        if "ON CONFLICT DO NOTHING" not in sql:
+            raise _UniqueViolationError("duplicate key value violates unique constraint")
+        return _Result(None)
+
+    def _find_notification(self, params: dict[str, Any]) -> SimpleNamespace | None:
+        key = (params["payment_id"], params["status"], params["amount"], params["token"])
+        return next(
+            (row for row in self.notifications if self._key_of(row, self._NOTIFICATION_KEY) == key),
+            None,
+        )
+
+
+class _Result:
+    def __init__(self, row: Any) -> None:
+        self._row = row
+
+    def fetchone(self) -> Any:
+        return self._row
+
+
+@pytest.fixture()
+def db() -> _FakeDb:
+    return _FakeDb()
+
+
+@pytest.fixture()
+def client(db: _FakeDb, monkeypatch: pytest.MonkeyPatch) -> TestClient:
+    from app.api.v1 import payments as payments_module
+    from app.core.config import settings
+    from app.core.db import get_db
+
+    monkeypatch.setattr(settings, "payments_enabled", True)
+    monkeypatch.setattr(settings, "tbank_password", SecretStr(_PASSWORD))
+    monkeypatch.setattr(settings, "tbank_terminal_key", "TERM-TEST")
+
+    app = FastAPI()
+    app.include_router(payments_module.router, prefix="/api/v1/trade-in")
+    app.dependency_overrides[get_db] = lambda: db
+    return TestClient(app)
+
+
+def _signed_notification(**overrides: Any) -> dict[str, Any]:
+    from app.services.payments.token import sign
+
+    body: dict[str, Any] = {
+        "TerminalKey": "TERM-TEST",
+        "OrderId": _ORDER_ID,
+        "PaymentId": _PAYMENT_ID,
+        "Status": "CONFIRMED",
+        "Success": True,
+        "Amount": _AMOUNT,
+    }
+    body.update(overrides)
+    body["Token"] = sign(body, _PASSWORD)
+    return body
+
+
+# ── идемпотентность выдачи ───────────────────────────────────────────────────
+
+
+def test_repeat_notification_issues_only_one_entitlement(client: TestClient, db: _FakeDb) -> None:
+    """Ретрай банка (тот же Token) не выдаёт второй отчёт и не рвёт ответ "OK".
+
+    Фальсификация (проверено руками): убрать `ON CONFLICT DO NOTHING` из
+    INSERT в payment_entitlements → второй запрос падает _UniqueViolationError и
+    тест краснеет на status_code 500; заменить дедуп на «SELECT потом INSERT»
+    и снять UNIQUE → len(entitlements) == 2.
+    """
+    body = _signed_notification()
+
+    first = client.post("/api/v1/trade-in/payments/notify", json=body)
+    second = client.post("/api/v1/trade-in/payments/notify", json=body)
+
+    assert first.status_code == 200, first.text
+    assert first.text == "OK"
+    assert second.status_code == 200, second.text
+    assert second.text == "OK"
+    assert len(db.entitlements) == 1, "повторная нотификация выдала второй отчёт"
+    assert len(db.notifications) == 1, "дубль нотификации записался второй строкой"
+    assert db.notifications[0].processed_at is not None
+    assert db.retain_until_updates == [_ESTIMATE_UUID]
+
+
+def test_unprocessed_duplicate_is_fulfilled_on_retry(client: TestClient, db: _FakeDb) -> None:
+    """Строка нотификации есть, а processed_at пуст → выдача ОБЯЗАНА состояться.
+
+    Это контракт processed_at из миграции 233: падение процесса между записью
+    нотификации и выдачей не должно оставить клиента без товара при списанных
+    деньгах. Красный вариант — трактовать существование строки как «уже
+    обработано» (тогда entitlements пуст).
+    """
+    body = _signed_notification()
+    db.notifications.append(
+        SimpleNamespace(
+            id=1,
+            order_id=_ORDER_ID,
+            tbank_payment_id=_PAYMENT_ID,
+            status="CONFIRMED",
+            amount_kopecks=_AMOUNT,
+            token=body["Token"],
+            token_valid=True,
+            processed_at=None,
+        )
+    )
+
+    response = client.post("/api/v1/trade-in/payments/notify", json=body)
+
+    assert response.status_code == 200, response.text
+    assert len(db.entitlements) == 1
+    assert db.notifications[0].processed_at is not None
+
+
+def test_retry_after_crash_between_issue_and_processed_does_not_double_issue(
+    client: TestClient, db: _FakeDb
+) -> None:
+    """Худший реальный случай: выдача прошла, а processed_at проставить не успели.
+
+    Ретрай банка ОБЯЗАН дойти до конца (иначе processed_at не проставится
+    никогда и так будет каждый час сутки) — и при этом не выдать второй отчёт.
+    Единственное, что здесь работает, — UNIQUE (payment_id, kind, ref_id) +
+    ON CONFLICT DO NOTHING: `_FakeDb` ведёт себя как Postgres и на INSERT без
+    ON CONFLICT кидает _UniqueViolationError (проверено руками: убрать ON CONFLICT
+    из INSERT в payment_entitlements → 500 вместо "OK", тест краснеет).
+    """
+    body = _signed_notification()
+    db.notifications.append(
+        SimpleNamespace(
+            id=1,
+            order_id=_ORDER_ID,
+            tbank_payment_id=_PAYMENT_ID,
+            status="CONFIRMED",
+            amount_kopecks=_AMOUNT,
+            token=body["Token"],
+            token_valid=True,
+            processed_at=None,
+        )
+    )
+    _issue_entitlement(db, "already-issued-token", expires_at=None)
+
+    response = client.post("/api/v1/trade-in/payments/notify", json=body)
+
+    assert response.status_code == 200, response.text
+    assert response.text == "OK"
+    assert len(db.entitlements) == 1, "выдан второй отчёт по тому же платежу"
+    assert db.entitlements[0].subject == "already-issued-token", "токен подменён на новый"
+    assert db.notifications[0].processed_at is not None
+
+
+# ── подпись и сумма ──────────────────────────────────────────────────────────
+
+
+def test_notification_with_invalid_token_is_rejected(client: TestClient, db: _FakeDb) -> None:
+    body = _signed_notification()
+    body["Token"] = "deadbeef" * 8  # подпись не от нашего пароля
+
+    response = client.post("/api/v1/trade-in/payments/notify", json=body)
+
+    assert response.status_code == 403
+    assert db.entitlements == [], "выдача по неподписанной нотификации"
+    assert len(db.notifications) == 1, "факт попытки должен остаться в append-only логе"
+    assert db.notifications[0].token_valid is False
+
+
+def test_amount_mismatch_is_rejected(client: TestClient, db: _FakeDb) -> None:
+    """Подпись Т-Банка не гарантирует сумму (см. docstring notification.py) —
+    расхождение с суммой заказа обязано быть отказом, а не выдачей."""
+    response = client.post(
+        "/api/v1/trade-in/payments/notify", json=_signed_notification(Amount=_AMOUNT + 1)
+    )
+
+    assert response.status_code == 400
+    assert db.entitlements == []
+
+
+def test_non_confirmed_status_does_not_fulfill(client: TestClient, db: _FakeDb) -> None:
+    response = client.post(
+        "/api/v1/trade-in/payments/notify", json=_signed_notification(Status="AUTHORIZED")
+    )
+
+    assert response.status_code == 200
+    assert db.entitlements == []
+    assert db.payments[0].status == "AUTHORIZED"
+
+
+def test_pre_confirm_notification_does_not_downgrade_confirmed(
+    client: TestClient, db: _FakeDb
+) -> None:
+    """Отставший AUTHORIZED не имеет права откатить уже подтверждённый платёж."""
+    db.payments[0].status = "CONFIRMED"
+
+    client.post("/api/v1/trade-in/payments/notify", json=_signed_notification(Status="AUTHORIZED"))
+
+    assert db.payments[0].status == "CONFIRMED"
+
+
+# ── capability-ссылка ────────────────────────────────────────────────────────
+
+
+def _issue_entitlement(db: _FakeDb, token: str, *, expires_at: datetime | None) -> None:
+    db.entitlements.append(
+        SimpleNamespace(
+            id="ent-1",
+            payment_id=_PAYMENT_UUID,
+            subject=token,
+            kind="report_link",
+            ref_id=_ESTIMATE_UUID,
+            expires_at=expires_at,
+        )
+    )
+
+
+def test_report_link_serves_estimate_for_valid_token(
+    client: TestClient, db: _FakeDb, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from app.api.v1 import payments as payments_module
+    from app.schemas.trade_in import AggregatedEstimate
+
+    seen: dict[str, Any] = {}
+
+    def _fake_loader(_db: Any, estimate_id: Any, **kwargs: Any) -> AggregatedEstimate:
+        seen["estimate_id"] = str(estimate_id)
+        seen.update(kwargs)
+        return AggregatedEstimate(
+            estimate_id=_ESTIMATE_UUID,
+            median_price_rub=5_000_000,
+            range_low_rub=4_500_000,
+            range_high_rub=5_500_000,
+            median_price_per_m2=100_000,
+            confidence="medium",
+            n_analogs=7,
+            period_months=6,
+            analogs=[],
+            actual_deals=[],
+            expires_at=datetime.now(tz=UTC) + timedelta(hours=12),
+        )
+
+    monkeypatch.setattr(payments_module, "load_estimate", _fake_loader)
+    _issue_entitlement(db, "good-token", expires_at=datetime.now(tz=UTC) + timedelta(days=1))
+
+    response = client.get("/api/v1/trade-in/r/good-token")
+
+    assert response.status_code == 200, response.text
+    assert seen["estimate_id"] == _ESTIMATE_UUID
+    assert seen["capability_granted"] is True
+    assert seen["x_authenticated_user"] is None
+
+
+def test_report_link_rejects_foreign_token(client: TestClient, db: _FakeDb) -> None:
+    _issue_entitlement(db, "good-token", expires_at=datetime.now(tz=UTC) + timedelta(days=1))
+
+    response = client.get("/api/v1/trade-in/r/someone-elses-token")
+
+    assert response.status_code == 404
+
+
+def test_report_link_rejects_expired_token(client: TestClient, db: _FakeDb) -> None:
+    _issue_entitlement(db, "stale-token", expires_at=datetime.now(tz=UTC) - timedelta(seconds=1))
+
+    response = client.get("/api/v1/trade-in/r/stale-token")
+
+    assert response.status_code == 404
+
+
+def test_issued_token_is_unpredictable(client: TestClient, db: _FakeDb) -> None:
+    """Токен — это всё право доступа: он обязан быть случайным, а не производной
+    от order_id/payment_id (иначе выводится по данным, которые видит покупатель)."""
+    client.post("/api/v1/trade-in/payments/notify", json=_signed_notification())
+
+    token = db.entitlements[0].subject
+    assert len(token) >= 40
+    assert _ORDER_ID not in token
+    assert _PAYMENT_ID not in token
+
+
+# ── kill-switch ──────────────────────────────────────────────────────────────
+
+
+def test_endpoints_are_closed_when_payments_disabled(
+    db: _FakeDb, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """PAYMENTS_ENABLED=false — 503 на всех ручках, без падений и без записи в БД."""
+    from app.api.v1 import payments as payments_module
+    from app.core.config import settings
+    from app.core.db import get_db
+
+    monkeypatch.setattr(settings, "payments_enabled", False)
+    app = FastAPI()
+    app.include_router(payments_module.router, prefix="/api/v1/trade-in")
+    app.dependency_overrides[get_db] = lambda: db
+    disabled = TestClient(app)
+
+    assert disabled.get("/api/v1/trade-in/r/any-token").status_code == 503
+    assert disabled.get(f"/api/v1/trade-in/payments/status/{_ORDER_ID}").status_code == 503
+    assert disabled.post("/api/v1/trade-in/payments/notify", json={}).status_code == 503
+    assert (
+        disabled.post(
+            "/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID}
+        ).status_code
+        == 503
+    )
+    assert db.notifications == []
+    assert db.entitlements == []
+
+
+# ── периметр и синхронизация констант ────────────────────────────────────────
+
+
+def test_notify_is_public_and_checkout_is_not() -> None:
+    from app.core.rbac import _PUBLIC_PATH_PREFIXES, _PUBLIC_PATHS
+
+    assert "/api/v1/trade-in/payments/notify" in _PUBLIC_PATHS
+    assert "/api/v1/trade-in/payments/checkout" not in _PUBLIC_PATHS
+    assert "/api/v1/trade-in/r/some-token".startswith(_PUBLIC_PATH_PREFIXES)
+    assert not "/api/v1/trade-in/history".startswith(_PUBLIC_PATH_PREFIXES)
+
+
+def test_report_link_token_is_redacted_from_sentry_events() -> None:
+    from app.observability.sentry_scrub import scrub_pii_event
+
+    event = {
+        "request": {"url": "https://meraocenka.ru/api/v1/trade-in/r/s3cr3t-token-value"},
+        "transaction": "GET /api/v1/trade-in/r/s3cr3t-token-value",
+    }
+    scrubbed = scrub_pii_event(event, {})  # type: ignore[arg-type]
+
+    assert "s3cr3t-token-value" not in str(scrubbed)
+    assert "/api/v1/trade-in/r/[REDACTED]" in scrubbed["transaction"]  # type: ignore[index]
+
+
+def test_status_whitelist_matches_migration() -> None:
+    """Свой список статусов не должен разъезжаться с CHECK миграции 233:
+    статус, которого нет в CHECK, уронит INSERT уже на проде."""
+    from app.api.v1.payments import _KNOWN_STATUSES
+
+    source = _MIGRATION.read_text(encoding="utf-8")
+    check = re.search(
+        r"ADD CONSTRAINT\s+payments_status_check\s+CHECK \(status IN \((.*?)\)\)", source, re.S
+    )
+    assert check is not None, "CHECK payments_status_check не найден в миграции 233"
+    migration_statuses = set(re.findall(r"'([^']+)'", check.group(1)))
+
+    assert _KNOWN_STATUSES == migration_statuses
+
+
+def test_price_matches_published_offer() -> None:
+    """Цена в коде обязана совпадать с ценой, названной в оферте (п. 4.1) —
+    единственный её источник для юр-текстов, mera-public/content.ts."""
+    from app.api.v1.payments import _PRODUCTS
+
+    match = re.search(r"SERVICE_PRICE_RUB\s*=\s*(\d+)", _CONTENT_TS.read_text(encoding="utf-8"))
+    assert match is not None, "SERVICE_PRICE_RUB не найден в mera-public/content.ts"
+
+    _name, price_kopecks = _PRODUCTS["paid_report"]
+    assert price_kopecks == int(match.group(1)) * 100
+
+
+def test_live_status_predicate_matches_code() -> None:
+    """Предикат частичного UNIQUE (279) и `_REUSABLE_STATUSES` — один список.
+
+    Индекс шире кода запрещает легитимную повторную попытку оплаты; индекс уже
+    кода пропускает второй холд на карте. И то и другое — про деньги, поэтому
+    дрейф ловит тест, а не внимательность читателя.
+    """
+    from app.api.v1.payments import _REUSABLE_STATUSES
+
+    path = _REPO_ROOT / "backend" / "data" / "sql" / "279_payments_live_checkout_uidx.sql"
+    source = path.read_text(encoding="utf-8")
+    predicate = re.search(r"AND status IN \((.*?)\)", source, re.S)
+    assert predicate is not None, "предикат по статусам не найден в миграции 279"
+
+    assert set(re.findall(r"'([^']+)'", predicate.group(1))) == set(_REUSABLE_STATUSES)
+
+
+# ── checkout: один живой платёж на оценку ────────────────────────────────────
+
+
+@pytest.fixture()
+def bank(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
+    """Подменяет Т-Банк: собирает вызовы Init и отдаёт предсказуемую ссылку.
+
+    Список вызовов — не украшение: «второй Init» и есть «второй холд на карте»,
+    поэтому проверяется именно его длина, а не только тело ответа.
+    """
+    from app.api.v1 import payments as payments_module
+
+    calls: list[dict[str, Any]] = []
+
+    class _FakeClient:
+        async def init_payment(self, **kwargs: Any) -> dict[str, Any]:
+            calls.append(kwargs)
+            return {
+                "Success": True,
+                "Status": "NEW",
+                "PaymentId": f"300000000{len(calls)}",
+                "PaymentURL": f"https://securepayments.tinkoff.ru/{len(calls)}",
+            }
+
+    monkeypatch.setattr(payments_module, "_client", _FakeClient)
+    return calls
+
+
+def _seed_live_payment(db: _FakeDb, *, payment_url: str | None, age: timedelta) -> SimpleNamespace:
+    row = SimpleNamespace(
+        id="pay-seed",
+        order_id="mera-seed",
+        status="NEW",
+        amount_kopecks=_AMOUNT,
+        estimate_id=_ESTIMATE_UUID,
+        created_by=None,
+        product_code="paid_report",
+        payment_url=payment_url,
+        created_at=datetime.now(tz=UTC) - age,
+    )
+    db.payments.append(row)
+    return row
+
+
+def _paid_report_rows(db: _FakeDb) -> list[SimpleNamespace]:
+    return [p for p in db.payments if p.product_code == "paid_report"]
+
+
+def test_parallel_checkout_does_not_create_second_payment(
+    client: TestClient, db: _FakeDb, bank: list[dict[str, Any]]
+) -> None:
+    """Двойной клик: соперник уже вставил строку, но ссылки от банка ещё нет.
+
+    Второй запрос обязан отбиться о частичный UNIQUE (миграция 279) и не пойти
+    в банк — второй Init это второй холд на карте покупателя. Ответ 409, а не
+    выдуманная ссылка.
+
+    Фальсификация (проверено): убрать `ON CONFLICT DO NOTHING` из INSERT в
+    payments → _FakeDb кидает _UniqueViolationError, как настоящий Postgres, и
+    тест краснеет вместо ответа 409; вернуть «SELECT, потом INSERT» без
+    обработки конфликта → вторая строка payments и второй вызов Init.
+    """
+    _seed_live_payment(db, payment_url=None, age=timedelta(seconds=1))
+
+    response = client.post(
+        "/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID}
+    )
+
+    assert response.status_code == 409, response.text
+    assert len(_paid_report_rows(db)) == 1, "параллельный checkout создал второй платёж"
+    assert bank == [], "второй Init = второй холд на карте покупателя"
+
+
+def test_repeat_checkout_reuses_live_link(
+    client: TestClient, db: _FakeDb, bank: list[dict[str, Any]]
+) -> None:
+    """Честный повтор по живому платежу возвращает ТУ ЖЕ ссылку и не зовёт Init."""
+    seeded = _seed_live_payment(
+        db, payment_url="https://securepayments/live", age=timedelta(minutes=5)
+    )
+
+    response = client.post(
+        "/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID}
+    )
+
+    assert response.status_code == 200, response.text
+    assert response.json()["payment_url"] == "https://securepayments/live"
+    assert response.json()["order_id"] == seeded.order_id
+    assert len(_paid_report_rows(db)) == 1
+    assert bank == []
+
+
+def test_abandoned_checkout_does_not_lock_the_buyer_out(
+    client: TestClient, db: _FakeDb, bank: list[dict[str, Any]]
+) -> None:
+    """Брошенный NEW старше окна не переиспользуется: ссылка у банка протухла.
+
+    Без границы по времени покупатель навсегда получал бы одну и ту же мёртвую
+    ссылку и не мог начать оплату заново — нотификации по зависшему NEW может
+    не прийти вовсе, сам из этого статуса платёж не выйдет.
+
+    Фальсификация (проверено руками): убрать UPDATE ... DEADLINE_EXPIRED (или
+    условие по created_at в нём) → в ответе старая мёртвая ссылка
+    https://securepayments/dead, Init не вызывается: тест краснеет по значению,
+    а не по исключению.
+    """
+    stale = _seed_live_payment(
+        db, payment_url="https://securepayments/dead", age=timedelta(hours=2)
+    )
+
+    response = client.post(
+        "/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID}
+    )
+
+    assert response.status_code == 200, response.text
+    assert response.json()["payment_url"] == "https://securepayments.tinkoff.ru/1"
+    assert stale.status == "DEADLINE_EXPIRED"
+    assert len(bank) == 1, "новая попытка оплаты обязана получить свежую ссылку банка"
+    assert len(_paid_report_rows(db)) == 2
+
+
+def test_checkout_rejects_foreign_estimate(
+    client: TestClient, db: _FakeDb, bank: list[dict[str, Any]], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """IDOR: checkout по чужой оценке — 404, платёж не создаётся.
+
+    Дверь здесь не «показать чужой отчёт», а order_id: он же право доступа для
+    /payments/status/<order_id>, который отдаёт capability-ссылку на отчёт,
+    как только владелец заплатит.
+
+    Фальсификация (проверено руками): убрать вызов `_assert_estimate_access` в
+    checkout → 200 и строка в payments, тест краснеет по значению.
+    """
+    from app.core import auth
+
+    monkeypatch.setattr(auth, "get_role", lambda username: "pilot")
+    db.estimate_created_by = "victim"
+
+    response = client.post(
+        "/api/v1/trade-in/payments/checkout",
+        json={"estimate_id": _ESTIMATE_UUID},
+        headers={"X-Authenticated-User": "attacker"},
+    )
+
+    assert response.status_code == 404, response.text
+    assert _paid_report_rows(db) == []
+    assert bank == []