Merge pull request 'Поддержка: доставленное сообщение не теряется при сбое БД, отказы Telegram расходуют бюджет' (#3459) from fix/tg-support-db-and-ratelimit into main

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

@@ -62,10 +62,12 @@ import hashlib
 import logging
 import re
 import secrets
+from datetime import UTC, datetime
 from typing import Annotated, Literal
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
 from pydantic import BaseModel, Field, field_validator
+from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
@@ -123,6 +125,46 @@ _INTERACTIVE_SEND_TIMEOUT_S = 5.0
 _INTERACTIVE_SEND_MAX_RETRIES = 3
 _INTERACTIVE_SEND_MAX_BACKOFF_S = 1.0
 
+# Счётчик ОТКАЗОВ отправки — отдельный от бюджетов выше (#tgsupport-fail-cooldown).
+#
+# Зачем вообще второй счётчик. `_send_limiter`/`_anon_ip_limiter` расходуются
+# ТОЛЬКО на успехе (review L3, non-destructive peek выше) — это верно для
+# «не наказывать за чужую аварию», но имеет обратную сторону: пока Telegram
+# недоступен, лимита нет ВООБЩЕ. Каждый повтор пользователя при этом стоит до
+# `1 + _INTERACTIVE_SEND_MAX_RETRIES` = 4 попыток к api.telegram.org и не
+# расходует ни один бюджет. Двух-трёх вкладок с авто-ретраем хватает, чтобы
+# выесть лимиты группы ровно в тот момент, когда канал и так еле жив.
+#
+# Отсюда — дешёвый gate ПЕРЕД походом в Telegram, на своих ключах (тех же, что
+# у основных лимитеров: username / anon-thread-key и client IP).
+#
+# N=5. Отказ одной отправки — не событие: по замеру #tgsupport-retry доля отказов
+# на попытку 15-38%, но после 4 попыток до пользователя доходит ~0.8-2% отказов.
+# Пять подряд на живом канале — вероятность порядка 1e-10, то есть cooldown
+# физически не может сработать на «просто не повезло»; он срабатывает только на
+# настоящей недоступности. Плюс `reset()` на успехе: считаем именно ПОДРЯД, одна
+# успешная отправка стирает историю.
+#
+# 30с окна (оно же длительность cooldown: блок держится, пока самый старый из
+# N отказов не выпадет из окна). Верхняя граница осмысленности — реальная
+# недоступность Telegram по тому же замеру длится минутами, так что 30с заведомо
+# короче и пользователя после восстановления канала не наказывают. Нижняя —
+# один отказавший интерактивный запрос сам по себе занимает до 23с
+# (4 попытки × 5с + 3 паузы × 1с), cooldown короче этого просто не имел бы смысла.
+_SEND_FAILURE_LIMIT = 5
+_SEND_FAILURE_WINDOW_S = 30.0
+_send_failure_limiter = SlidingWindowLimiter(
+    limit=_SEND_FAILURE_LIMIT, window_s=_SEND_FAILURE_WINDOW_S
+)
+# Отдельный экземпляр для IP-ключей — ровно как `_anon_ip_limiter` отдельный от
+# `_send_limiter`: ключи разных пространств (IP vs thread-key) в одном словаре
+# смешивать нельзя.
+_anon_ip_failure_limiter = SlidingWindowLimiter(
+    limit=_SEND_FAILURE_LIMIT, window_s=_SEND_FAILURE_WINDOW_S
+)
+
+_SEND_UNAVAILABLE_DETAIL = "Telegram сейчас недоступен. Попробуйте через полминуты."
+
 # #tgsupport-web review M5: без LIMIT каждое монтирование виджета на старом
 # треде отдавало бы ВЕСЬ лог переписки. См. `web_support_storage.list_messages`.
 _LIST_MESSAGES_LIMIT = 200
@@ -166,6 +208,18 @@ class SupportMessageOut(BaseModel):
     text_body: str
     operator_tg_id: int | None = None
     created_at: str
+    # Дошло ли сообщение до БД. `True` по умолчанию — все существующие пути
+    # (`GET /support/messages`, успешный POST) строят модель из storage-строки и
+    # ничего про флаг не знают, контракт для них не меняется.
+    #
+    # `False` ставит ТОЛЬКО `_unpersisted_message_out`: доставлено оператору, но
+    # не записано. Флаг нужен потому, что без него деградация неотличима от
+    # тишины: фронт выбрасывает тело POST и рендерит транскрипт исключительно из
+    # `GET /support/messages` (`useSupportChat.ts`), где этого сообщения нет —
+    # поле ввода очищается, сообщение не появляется, ошибки нет. Пользователь
+    # решает, что не отправилось, и шлёт снова — ровно тот дубль в топике,
+    # против которого вся эта ветка и сделана.
+    persisted: bool = True
 
     @field_validator("created_at", mode="before")
     @classmethod
@@ -191,6 +245,105 @@ def _format_mirror_text(username: str, message_text: str) -> str:
     return f"[С САЙТА] {username}:\n{message_text}"
 
 
+def _too_many_failures_error(retry_after: float) -> HTTPException:
+    """429 вместо похода в Telegram — канал только что отказал N раз подряд."""
+    return HTTPException(
+        status_code=429,
+        detail=_SEND_UNAVAILABLE_DETAIL,
+        headers={"Retry-After": str(int(retry_after) + 1)},
+    )
+
+
+def _unpersisted_message_out(text_body: str) -> SupportMessageOut:
+    """Синтетический ответ для случая «в топик доставлено, а в БД не записано».
+
+    Почему ответ вообще УСПЕШНЫЙ. Сообщение оператору реально доставлено —
+    отдать на это ошибку значит соврать: пользователь повторит, и в топике
+    окажется дубль (плюс второе предупреждение оператору). Успех здесь честнее
+    отказа, но он неполный, и об этом клиенту надо сказать явно.
+
+    Что сообщает `persisted=False`: «доставлено, но ответить тебе через чат не
+    смогут; повторять не надо». Именно флаг — контракт для клиента, а НЕ `id=0`:
+    по идентификатору клиент отличить деградацию не обязан и не будет.
+
+    `id` при этом всё равно нужен модели, и 0 — сознательный сентинел, а не
+    выдумка: реальные id в `web_support_messages` начинаются с 1 (serial),
+    поэтому 0 ни с чем не столкнётся, а `GET /support/messages` принимает
+    `since>=0`. Даже если фронт когда-нибудь начнёт курсорить по возвращённому
+    id (сейчас он перечитывает тред целиком с `since=0`, `useSupportChat.ts`),
+    нулевой курсор не перепрыгнет ни одного реального сообщения: занижение
+    курсора безопасно, завышение — нет.
+    """
+    return SupportMessageOut(
+        id=0,
+        direction="in",
+        text_body=text_body,
+        operator_tg_id=None,
+        created_at=datetime.now(UTC),
+        persisted=False,
+    )
+
+
+async def _warn_operator_message_not_recorded(*, label: str, topic_message_id: int | None) -> None:
+    """Реплаем к только что доставленному зеркалу предупреждает оператора, что
+    ответ на ЭТО сообщение не смаршрутизируется: треда в БД нет, на реплай к
+    осиротевшему зеркалу `bridge._handle_group_reply` напишет только WARNING, а
+    клиент не увидит ничего. Без предупреждения оператор отвечает в пустоту и
+    считает, что помог.
+
+    Текст обращения сюда НЕ дублируется (ПДн) — оператор видит его в сообщении,
+    к которому это реплай.
+
+    Формулировка учитывает, что клиенту ответили успехом (`persisted=False`,
+    «принято, повторять не надо»): рассчитывать на «клиент напишет снова»
+    оператору нельзя, поднимать тред придётся иначе.
+
+    Любой отказ гасится логом: основное сообщение УЖЕ доставлено, превращать
+    неудачу служебного уведомления в 500 поверх успеха нельзя.
+    """
+    text = (
+        f"ВНИМАНИЕ · {label}: сообщение доставлено в топик, но НЕ записано в базу "
+        "(сбой БД). Треда у этого обращения нет — ответить клиенту через бота "
+        "НЕЛЬЗЯ: реплай на это сообщение никуда не уйдёт. Повтора тоже не ждите, "
+        "клиенту показано, что сообщение принято и отправлять его снова не нужно. "
+        "Если в обращении есть контакт — свяжитесь напрямую; иначе передайте "
+        "дежурному и сообщите об отказе БД."
+    )
+    try:
+        client = get_telegram_client()
+        await client.send_message(
+            chat_id=settings.telegram_support_chat_id,
+            text=text,
+            message_thread_id=settings.telegram_support_topic_id or None,
+            # reply_to может отсутствовать (sendMessage не вернул message_id) —
+            # тогда уведомление уходит отдельным сообщением в топик: хуже, чем
+            # реплай, но несравнимо лучше тишины.
+            reply_to_message_id=topic_message_id,
+            # Тот же узкий интерактивный бюджет (review H1): клиент ждёт ответа
+            # ручки, а не доставки служебного уведомления.
+            timeout=_INTERACTIVE_SEND_TIMEOUT_S,
+            max_retries=_INTERACTIVE_SEND_MAX_RETRIES,
+            max_backoff=_INTERACTIVE_SEND_MAX_BACKOFF_S,
+        )
+    except Exception:
+        # Шире `TelegramError` намеренно: это best-effort хвост уже успешного
+        # запроса, любой отказ здесь обязан остаться в логе, а не у клиента.
+        logger.exception(
+            "web support: не удалось предупредить оператора о несохранённом сообщении (%s)",
+            label,
+        )
+
+
+def _rollback_quietly(db: Session) -> None:
+    """Откат после `SQLAlchemyError`. Сам rollback на мёртвом соединении тоже
+    может бросить, а мы уже решили отдать клиенту успех — `get_db` в любом
+    случае закроет сессию в `finally`."""
+    try:
+        db.rollback()
+    except SQLAlchemyError:
+        logger.warning("web support: rollback после сбоя БД тоже не удался", exc_info=True)
+
+
 @router.post("/support/messages", response_model=SupportMessageOut)
 async def send_support_message(
     payload: SupportMessageInput,
@@ -220,6 +373,12 @@ async def send_support_message(
             headers={"Retry-After": str(int(retry_after) + 1)},
         )
 
+    # Cooldown по ОТКАЗАМ (см. `_SEND_FAILURE_LIMIT`): канал только что отказал
+    # N раз подряд — не тратим на этот запрос ещё четыре попытки к Telegram.
+    cooldown = _send_failure_limiter.retry_after(username)
+    if cooldown is not None:
+        raise _too_many_failures_error(cooldown)
+
     # Общий клиент приложения (#tg-connection-resilience): на каждый запрос
     # свой создавать нельзя — это ноль keep-alive и полный TCP+TLS-хендшейк
     # до api.telegram.org перед каждой отправкой. Живёт в lifespan.
@@ -242,10 +401,13 @@ async def send_support_message(
         logger.exception(
             "web support: не удалось отправить зеркало в топик (username=%s)", username
         )
+        _send_failure_limiter.record(username)
         raise HTTPException(status_code=502, detail=SERVICE_UNAVAILABLE_TEXT) from None
 
     # Отправка удалась — теперь и только теперь расходуем rate-limit бюджет.
     _send_limiter.record(username)
+    # Канал жив — счётчик отказов считает именно ПОДРЯД идущие отказы.
+    _send_failure_limiter.reset(username)
 
     topic_message_id = mirrored.get("message_id") if isinstance(mirrored, dict) else None
     if topic_message_id is None:
@@ -259,15 +421,34 @@ async def send_support_message(
         )
 
     # review H1: БД-операция ПОСЛЕ успешной отправки — см. docstring модуля.
-    thread_id = storage.get_or_create_thread(db, username)
-    row = storage.record_inbound(
-        db,
-        thread_id=thread_id,
-        text_body=payload.text,
-        topic_message_id=topic_message_id,
-        support_chat_id=settings.telegram_support_chat_id,
-    )
-    db.commit()
+    #
+    # Оборотная сторона этого порядка: отказ БД здесь означает, что сообщение
+    # оператору УЖЕ доставлено. Отдавать на это 500 (как было) — худший из
+    # вариантов: клиент видит ошибку, шлёт повторно, в топике дубль, а на
+    # осиротевшее зеркало оператор отвечает в пустоту. Поэтому отвечаем успехом
+    # (доставка правда состоялась) и отдельным сообщением предупреждаем
+    # оператора, что отвечать на это зеркало бесполезно.
+    try:
+        thread_id = storage.get_or_create_thread(db, username)
+        row = storage.record_inbound(
+            db,
+            thread_id=thread_id,
+            text_body=payload.text,
+            topic_message_id=topic_message_id,
+            support_chat_id=settings.telegram_support_chat_id,
+        )
+        db.commit()
+    except SQLAlchemyError:
+        # Ни текста обращения (ПДн), ни токена в логе — только кто и что сломалось.
+        logger.exception(
+            "web support: сообщение доставлено в топик, но не записано в БД (username=%s)",
+            username,
+        )
+        _rollback_quietly(db)
+        await _warn_operator_message_not_recorded(
+            label=f"[С САЙТА] {username}", topic_message_id=topic_message_id
+        )
+        return _unpersisted_message_out(payload.text)
 
     logger.info("web support: message sent username=%s thread_id=%d", username, thread_id)
     return SupportMessageOut(**row)
@@ -414,6 +595,16 @@ async def send_anon_support_message(
                 headers={"Retry-After": str(int(retry_after) + 1)},
             )
 
+    # Cooldown по ОТКАЗАМ (см. `_SEND_FAILURE_LIMIT`) — оба ключа, как и у
+    # бюджетов выше: per-token ловит одну вкладку с авто-ретраем, per-IP — ту же
+    # петлю после сброса куки.
+    for cooldown in (
+        _send_failure_limiter.retry_after(thread_key),
+        _anon_ip_failure_limiter.retry_after(ip),
+    ):
+        if cooldown is not None:
+            raise _too_many_failures_error(cooldown)
+
     display_id = _anon_display_id(token)
     # Общий клиент приложения (#tg-connection-resilience): на каждый запрос
     # свой создавать нельзя — это ноль keep-alive и полный TCP+TLS-хендшейк
@@ -434,10 +625,15 @@ async def send_anon_support_message(
         logger.exception(
             "web support (anon): не удалось отправить зеркало в топик (%s)", display_id
         )
+        _send_failure_limiter.record(thread_key)
+        _anon_ip_failure_limiter.record(ip)
         raise HTTPException(status_code=502, detail=SERVICE_UNAVAILABLE_TEXT) from None
 
     _send_limiter.record(thread_key)
     _anon_ip_limiter.record(ip)
+    # Канал жив — счётчики отказов считают именно ПОДРЯД идущие отказы.
+    _send_failure_limiter.reset(thread_key)
+    _anon_ip_failure_limiter.reset(ip)
 
     topic_message_id = mirrored.get("message_id") if isinstance(mirrored, dict) else None
     if topic_message_id is None:
@@ -447,15 +643,36 @@ async def send_anon_support_message(
             display_id,
         )
 
-    thread_id = storage.get_or_create_thread(db, thread_key)
-    row = storage.record_inbound(
-        db,
-        thread_id=thread_id,
-        text_body=payload.text,
-        topic_message_id=topic_message_id,
-        support_chat_id=settings.telegram_support_chat_id,
-    )
-    db.commit()
+    # Отказ БД после успешной отправки — см. развёрнутый комментарий в парной
+    # (авторизованной) ручке: сообщение оператору уже доставлено, 500 тут создаёт
+    # дубли в топике и «осиротевшее» зеркало, на которое оператор отвечает зря.
+    try:
+        thread_id = storage.get_or_create_thread(db, thread_key)
+        row = storage.record_inbound(
+            db,
+            thread_id=thread_id,
+            text_body=payload.text,
+            topic_message_id=topic_message_id,
+            support_chat_id=settings.telegram_support_chat_id,
+        )
+        db.commit()
+    except SQLAlchemyError:
+        # Ни текста обращения (ПДн), ни токена (bearer треда) в логе — только
+        # несекретный ярлык треда.
+        logger.exception(
+            "web support (anon): сообщение доставлено в топик, но не записано в БД (%s)",
+            display_id,
+        )
+        _rollback_quietly(db)
+        await _warn_operator_message_not_recorded(
+            label=f"[С САЙТА · БЕЗ ВХОДА] {display_id}", topic_message_id=topic_message_id
+        )
+        # Куку ставим ВСЁ РАВНО: тред в БД не создан, но идентичность посетителя
+        # обязана пережить этот сбой — иначе следующее сообщение (когда база
+        # поднимется) заведёт ВТОРОЙ тред, и переписка разъедется на два.
+        if is_new_token:
+            _set_anon_cookie(response, token)
+        return _unpersisted_message_out(payload.text)
 
     if is_new_token:
         _set_anon_cookie(response, token)

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

@@ -161,6 +161,18 @@ class SlidingWindowLimiter:
                 del self._hits[k]
         return len(bucket)
 
+    def reset(self, key: str) -> None:
+        """Обнуляет окно под *key*.
+
+        Нужен вызывающим, которые считают не «сколько сделано», а «сколько ПОДРЯД»
+        — например счётчик отказов отправки (`support.py:_send_failure_limiter`):
+        успешная отправка означает, что канал жив, и предыдущие отказы больше не
+        должны приближать cooldown. Для лимитеров-бюджетов (`_send_limiter`,
+        `_anon_ip_limiter`, `_notify_limiter`) не используется — там обнуление
+        по запросу было бы дырой в самом лимите.
+        """
+        self._hits.pop(key, None)
+
     def check(self, key: str) -> float | None:
         """Комбинированная проверка+регистрация (peek+record за один вызов) —
         для вызывающих, которым не нужно различать "попытка"/"успех" (см.

tradein-mvp/backend/tests/test_support.py

@@ -15,6 +15,7 @@ NEVER touches real DB / real Telegram API.
 from __future__ import annotations
 
 import os
+import time
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
@@ -24,6 +25,7 @@ from unittest.mock import MagicMock
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
+from sqlalchemy.exc import SQLAlchemyError
 
 from app.api.v1 import support as support_module
 from app.core.db import get_db
@@ -49,6 +51,15 @@ def _fresh_rate_limiter(monkeypatch: pytest.MonkeyPatch) -> None:
     )
 
 
+@pytest.fixture(autouse=True)
+def _fresh_failure_limiters(monkeypatch: pytest.MonkeyPatch) -> None:
+    """То же, что `_fresh_rate_limiter`, но для счётчиков ОТКАЗОВ отправки: они
+    модульные, и тесты, специально роняющие Telegram, иначе оставляли бы cooldown
+    следующим тестам в том же процессе."""
+    for name in ("_send_failure_limiter", "_anon_ip_failure_limiter"):
+        monkeypatch.setattr(support_module, name, SlidingWindowLimiter(limit=1000, window_s=30.0))
+
+
 class _FakeTelegramClient:
     """Подменяет `TelegramClient` внутри `support` модуля — никакого httpx/сети."""
 
@@ -254,6 +265,8 @@ def test_send_message_happy_path_mirrors_with_website_marker(
     body = r.json()
     assert body["id"] == 100
     assert body["direction"] == "in"
+    # Успешный путь ничего не знает про флаг — дефолт `True` (#persisted).
+    assert body["persisted"] is True
 
     # Зеркало помечено "С САЙТА" + username — оператор не путает с TG-клиентом.
     assert len(_fake_telegram_client.calls) == 1
@@ -468,6 +481,8 @@ def test_list_messages_returns_thread_scoped_rows(
     body = r.json()
     assert len(body) == 1
     assert body[0]["text_body"] == "hi"
+    # GET строит модель из storage-строки и про флаг не знает — дефолт `True`.
+    assert body[0]["persisted"] is True
 
 
 def test_list_messages_passes_bounded_limit_to_storage(
@@ -615,6 +630,7 @@ def test_anon_send_without_any_auth_succeeds_and_sets_cookie(
     r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "не могу войти"})
     assert r.status_code == 200, r.text
     assert r.json()["text_body"] == "не могу войти"
+    assert r.json()["persisted"] is True
 
     token = client.cookies.get(support_module._ANON_COOKIE_NAME)
     assert token is not None
@@ -773,3 +789,266 @@ def test_anon_thread_key_cannot_collide_with_real_username() -> None:
     key = support_module._anon_thread_key(support_module.secrets.token_urlsafe(18))
     assert key.startswith("anon:")
     assert _USERNAME_RE.match(key) is None
+
+
+# ── отказ БД ПОСЛЕ успешной доставки в топик ─────────────────────────────────
+#
+# Порядок «сначала Telegram, потом БД» (H1) означает, что SQLAlchemyError здесь
+# = сообщение оператору уже доставлено. 500 на это — худший вариант: клиент
+# повторяет, в топике дубль, а на осиротевшее зеркало оператор отвечает в пустоту.
+
+
+def _patch_storage_ok(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(support_module.storage, "get_or_create_thread", lambda db, username: 7)
+    monkeypatch.setattr(
+        support_module.storage,
+        "record_inbound",
+        lambda *a, **kw: {
+            "id": 1,
+            "direction": "in",
+            "text_body": kw["text_body"],
+            "operator_tg_id": None,
+            "created_at": "2026-09-12T00:00:00+00:00",
+        },
+    )
+
+
+def test_send_message_db_failure_after_delivery_returns_success_and_warns_operator(
+    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
+) -> None:
+    _patch_storage_ok(monkeypatch)
+    db.commit.side_effect = SQLAlchemyError("connection lost")
+
+    r = client.post(
+        "/api/v1/trade-in/support/messages", json={"text": "секретный вопрос"}, headers=_auth()
+    )
+
+    # Клиенту НЕ врём про отказ: сообщение оператору правда доставлено, повтор
+    # создал бы дубль в топике.
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["text_body"] == "секретный вопрос"
+    assert body["id"] == 0  # сентинел «не персистировано», курсор не завышаем
+    # Контракт для клиента — именно флаг: без него 200 неотличим от тишины
+    # (фронт рендерит транскрипт только из GET, где сообщения нет) и клиент
+    # шлёт повтор → дубль в топике, ровно то, против чего вся правка.
+    assert body["persisted"] is False
+    assert db.rollback.called
+
+    # Оператор предупреждён реплаем к только что доставленному зеркалу.
+    assert len(_fake_telegram_client.calls) == 2
+    warning = _fake_telegram_client.calls[-1]
+    assert warning["reply_to_message_id"] == 555
+    assert "НЕ записано в базу" in warning["text"]
+    # Оператор не должен ждать повтора: клиенту сказано, что сообщение принято.
+    assert "Повтора тоже не ждите" in warning["text"]
+    # Служебное уведомление не дублирует текст обращения (ПДн).
+    assert "секретный вопрос" not in warning["text"]
+
+
+def test_send_message_db_failure_survives_failed_operator_warning(
+    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Отказ уведомления оператору поверх отказа БД не должен становиться 500
+    поверх уже доставленного сообщения."""
+    _patch_storage_ok(monkeypatch)
+    db.commit.side_effect = SQLAlchemyError("connection lost")
+
+    class _FlakyClient:
+        calls = 0
+
+        async def send_message(self, **kwargs: Any) -> dict[str, Any]:
+            _FlakyClient.calls += 1
+            if _FlakyClient.calls == 1:
+                return {"message_id": 555}  # зеркало доставлено
+            raise TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
+
+    monkeypatch.setattr(support_module, "get_telegram_client", _FlakyClient)
+
+    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())
+
+    assert r.status_code == 200, r.text
+    assert _FlakyClient.calls == 2
+
+
+def test_anon_db_failure_after_delivery_returns_success_and_sets_cookie(
+    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
+) -> None:
+    """Аноним: кука обязана пережить сбой БД — иначе следующее сообщение заведёт
+    ВТОРОЙ тред и переписка разъедется на два."""
+    _patch_anon_storage(monkeypatch)
+    db.commit.side_effect = SQLAlchemyError("connection lost")
+
+    r = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "не могу войти"})
+
+    assert r.status_code == 200, r.text
+    assert r.json()["id"] == 0
+    assert r.json()["persisted"] is False
+    token = client.cookies.get(support_module._ANON_COOKIE_NAME)
+    assert token is not None and support_module._ANON_TOKEN_RE.match(token)
+
+    warning = _fake_telegram_client.calls[-1]
+    assert warning["reply_to_message_id"] == 555
+    assert "НЕ записано в базу" in warning["text"]
+    assert token not in warning["text"]  # bearer треда в топик не уходит
+
+
+def test_send_message_db_failure_without_topic_message_id_still_warns(
+    client: TestClient, db: MagicMock, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
+) -> None:
+    """Нет message_id → реплай невозможен, но уведомление всё равно уходит
+    отдельным сообщением в топик: хуже реплая, но лучше тишины."""
+    _patch_storage_ok(monkeypatch)
+    _fake_telegram_client._response = {}  # sendMessage без message_id
+    db.commit.side_effect = SQLAlchemyError("connection lost")
+
+    r = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())
+
+    assert r.status_code == 200, r.text
+    assert _fake_telegram_client.calls[-1]["reply_to_message_id"] is None
+
+
+# ── cooldown по отказам отправки ─────────────────────────────────────────────
+#
+# Дефект: `retry_after()` — peek, `record()` только на успехе, значит пока
+# Telegram лежит, лимита нет ВООБЩЕ, и каждый повтор стоит до 4 попыток к
+# api.telegram.org, не расходуя ни один бюджет.
+
+
+def test_send_failures_trigger_cooldown_and_stop_reaching_telegram(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
+) -> None:
+    monkeypatch.setattr(
+        support_module, "_send_failure_limiter", SlidingWindowLimiter(limit=2, window_s=30.0)
+    )
+    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
+
+    for _ in range(2):
+        assert (
+            client.post(
+                "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
+            ).status_code
+            == 502
+        )
+    assert len(_fake_telegram_client.calls) == 2
+
+    blocked = client.post("/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth())
+    assert blocked.status_code == 429
+    assert int(blocked.headers["Retry-After"]) >= 1
+    # Главное: до Telegram запрос не дошёл вообще.
+    assert len(_fake_telegram_client.calls) == 2
+
+
+def test_send_failure_cooldown_is_per_key(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
+) -> None:
+    """Cooldown одного пользователя не должен задевать другого."""
+    monkeypatch.setattr(
+        support_module, "_send_failure_limiter", SlidingWindowLimiter(limit=1, window_s=30.0)
+    )
+    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
+    assert (
+        client.post(
+            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth("alice")
+        ).status_code
+        == 502
+    )
+
+    assert (
+        client.post(
+            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth("alice")
+        ).status_code
+        == 429
+    )
+    assert (
+        client.post(
+            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth("bob")
+        ).status_code
+        == 502
+    )
+
+
+def test_send_failure_cooldown_expires_with_window(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
+) -> None:
+    """После окончания окна cooldown снимается сам — пользователя не наказывают
+    после восстановления Telegram."""
+    monkeypatch.setattr(
+        support_module, "_send_failure_limiter", SlidingWindowLimiter(limit=1, window_s=0.2)
+    )
+    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
+    assert (
+        client.post(
+            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
+        ).status_code
+        == 502
+    )
+    assert (
+        client.post(
+            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
+        ).status_code
+        == 429
+    )
+
+    time.sleep(0.25)
+    _fake_telegram_client._response = {"message_id": 555}
+    _patch_storage_ok(monkeypatch)
+    ok = client.post("/api/v1/trade-in/support/messages", json={"text": "ok"}, headers=_auth())
+    assert ok.status_code == 200, ok.text
+
+
+def test_successful_send_resets_failure_counter(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
+) -> None:
+    """Счётчик считает ПОДРЯД идущие отказы: успех стирает историю, иначе редкие
+    транзиентные отказы копились бы в cooldown на живом канале."""
+    monkeypatch.setattr(
+        support_module, "_send_failure_limiter", SlidingWindowLimiter(limit=2, window_s=30.0)
+    )
+    _patch_storage_ok(monkeypatch)
+
+    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
+    assert (
+        client.post(
+            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
+        ).status_code
+        == 502
+    )
+    _fake_telegram_client._response = {"message_id": 555}
+    assert (
+        client.post(
+            "/api/v1/trade-in/support/messages", json={"text": "ok"}, headers=_auth()
+        ).status_code
+        == 200
+    )
+
+    # Ещё один отказ — всего второй в окне, но ПЕРВЫЙ подряд: cooldown не встаёт.
+    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
+    assert (
+        client.post(
+            "/api/v1/trade-in/support/messages", json={"text": "hi"}, headers=_auth()
+        ).status_code
+        == 502
+    )
+
+
+def test_anon_send_failures_trigger_cooldown_per_ip(
+    client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
+) -> None:
+    """Обход ротацией куки: новый токен — новый per-token счётчик, но per-IP
+    счётчик отказов тот же."""
+    _patch_anon_storage(monkeypatch)
+    monkeypatch.setattr(
+        support_module, "_anon_ip_failure_limiter", SlidingWindowLimiter(limit=1, window_s=30.0)
+    )
+    _fake_telegram_client._response = TelegramNetworkError("sendMessage", "ConnectTimeout", 4)
+
+    assert (
+        client.post("/api/v1/trade-in/support/anon/messages", json={"text": "1"}).status_code == 502
+    )
+    client.cookies.clear()  # «сбросил куку» — по токену бюджет снова пуст
+    calls_before = len(_fake_telegram_client.calls)
+
+    blocked = client.post("/api/v1/trade-in/support/anon/messages", json={"text": "2"})
+    assert blocked.status_code == 429
+    assert len(_fake_telegram_client.calls) == calls_before  # до Telegram не дошло

tradein-mvp/frontend/src/components/trade-in/v2/SupportChatPanel.tsx

@@ -69,6 +69,12 @@ export function SupportChatPanel({
   onCloseRef.current = onClose;
 
   const [draft, setDraft] = useState("");
+  // Последняя отправка доставлена оператору, но не попала в переписку
+  // (`persisted: false`). Без этого флага такой ответ визуально неотличим от
+  // полной тишины: поле очистилось, в списке пусто, ошибки нет — и посетитель
+  // шлёт то же самое второй раз, дублируя сообщение в топике оператора.
+  // Снимается следующей успешной отправкой, которая записалась нормально.
+  const [notPersisted, setNotPersisted] = useState(false);
 
   const messagesQuery = useSupportMessages(open, scope);
   const sendMessage = useSendSupportMessage(scope);
@@ -161,7 +167,10 @@ export function SupportChatPanel({
   function handleSend() {
     if (!canSend) return;
     sendMessage.mutate(draft.trim(), {
-      onSuccess: () => setDraft(""),
+      onSuccess: (message) => {
+        setDraft("");
+        setNotPersisted(message.persisted === false);
+      },
     });
   }
 
@@ -352,6 +361,28 @@ export function SupportChatPanel({
           </div>
         )}
 
+        {/* delivered-but-not-saved notice — предупреждение, не отказ: отправка
+            УДАЛАСЬ, просто сообщения не будет видно в списке выше */}
+        {notPersisted && (
+          <div
+            role="status"
+            style={{
+              flex: "0 0 auto",
+              margin: "0 14px 8px",
+              fontSize: 11.5,
+              lineHeight: 1.45,
+              color: tokens.warn,
+              background: "rgba(181,138,62,.1)",
+              border: "1px solid rgba(181,138,62,.35)",
+              borderRadius: 7,
+              padding: "7px 10px",
+            }}
+          >
+            Сообщение получено оператором, но не сохранилось в переписке —
+            поэтому его нет в списке выше. Отправлять ещё раз не нужно.
+          </div>
+        )}
+
         {/* composer */}
         <div
           style={{

tradein-mvp/frontend/src/lib/useSupportChat.ts

@@ -26,6 +26,15 @@
  *     any extra visibilitychange plumbing. Only the panel-open/closed gate
  *     above is bespoke.
  *
+ * `SupportMessage.persisted` is the one non-obvious bit of the response shape:
+ * the backend delivers to the Telegram topic FIRST and writes the row second,
+ * so there is a path where the operator has the message but the transcript
+ * does not (write failed). That answer is a success — reporting failure would
+ * make the visitor resend and duplicate the message in the topic — and carries
+ * `persisted: false`. Optional here because it is absent on older responses and
+ * the safe default is "everything is fine"; only an explicit `false` means the
+ * message will never show up in the transcript (see SupportChatPanel's notice).
+ *
  * Always re-fetches the FULL thread (`since=0`) rather than incrementally
  * merging `since=<lastId>` pages into the query cache: at MVP support-chat
  * volume (a handful to a few dozen messages per thread) the bandwidth cost of
@@ -66,6 +75,13 @@ export interface SupportMessage {
   text_body: string;
   operator_tg_id: number | null;
   created_at: string;
+  /**
+   * `false` — сообщение доставлено оператору, но НЕ записано в переписку, т.е.
+   * в списке оно не появится никогда. Отсутствует/`true` — всё штатно (в т.ч.
+   * во всех ответах `GET /support/messages`). Необязательное: старый бэкенд
+   * поля не присылает, и default-поведение должно быть «всё хорошо».
+   */
+  persisted?: boolean;
 }
 
 export interface SupportUnread {