diff --git a/tradein-mvp/backend/app/api/v1/support.py b/tradein-mvp/backend/app/api/v1/support.py index bdaf5268..07406a23 100644 --- a/tradein-mvp/backend/app/api/v1/support.py +++ b/tradein-mvp/backend/app/api/v1/support.py @@ -62,6 +62,7 @@ import hashlib import logging import re import secrets +import time from datetime import UTC, datetime from typing import Annotated, Literal @@ -170,6 +171,14 @@ _SEND_UNAVAILABLE_DETAIL = "Telegram сейчас недоступен. Попр # треде отдавало бы ВЕСЬ лог переписки. См. `web_support_storage.list_messages`. _LIST_MESSAGES_LIMIT = 200 +# Идемпотентность отправки (#3471 retry-storm) — см. `_resolve_idempotency_key`. +_IDEMPOTENCY_HEADER = "idempotency-key" +# Форма клиентского ключа — как у anon-токена (`_ANON_TOKEN_RE`): произвольная +# opaque-строка клиента, без пробелов/спецсимволов, которые попали бы в SQL-параметр +# как есть. Не матчится — считаем заголовок отсутствующим и уходим на fallback, +# а не пытаемся его "починить" (тот же принцип, что у `_read_anon_token`). +_CLIENT_IDEMPOTENCY_KEY_RE = re.compile(r"^[A-Za-z0-9_.-]{8,128}\Z") + def _require_username(request: Request) -> str: """Достаёт X-Authenticated-User. rbac_guard (app/main.py) уже гарантирует его @@ -345,8 +354,41 @@ def _rollback_quietly(db: Session) -> None: logger.warning("web support: rollback после сбоя БД тоже не удался", exc_info=True) +def _resolve_idempotency_key(request: Request, *, identity_key: str, text: str) -> str: + """Ключ идемпотентности inbound-отправки (#3471, миграция 301). + + Почему заголовок + fallback, а не что-то одно. Явный `Idempotency-Key` — + предпочтительный путь: клиент генерирует ключ ОДИН раз на "намерение + отправить" и переиспользует его на любом ретрае (fetch retry / переотправка + после таймаута) независимо от того, что именно менялось в UI между + попытками. Без заголовка (старые клиенты, п.5 требования — они не должны + сломаться) используем детерминированный отпечаток sha256(identity, текст, + минутное окно): типичный повтор (двойной клик, ретрай браузера) укладывается + в секунды, минутное окно с запасом это покрывает, а любые два РАЗНЫХ по + смыслу сообщения с одинаковым текстом, отправленные намеренно с разницей в + пару минут, в одно не схлопнутся. + + Хранит и потенциально логирует (см. вызовы в этом файле) только САМ ключ — + он либо непрозрачный клиентский токен, либо хэш. Текст сообщения сюда + попадает ТОЛЬКО как вход в sha256, в открытом виде не сохраняется и не + возвращается — жёсткое правило проекта "текст не в логах" (ПДн) при этом + не нарушается, даже если бы этот ключ где-то залогировали. + + Префиксы `client:`/`auto:` разводят два namespace'а ключей (защита от + случайного совпадения клиентского токена с fallback-хэшем) — дёшево и не + требует отдельной колонки. + """ + header = (request.headers.get(_IDEMPOTENCY_HEADER) or "").strip() + if header and _CLIENT_IDEMPOTENCY_KEY_RE.match(header): + return f"client:{header}" + window = int(time.time() // 60) + fingerprint = f"{identity_key}|{text}|{window}" + return "auto:" + hashlib.sha256(fingerprint.encode("utf-8")).hexdigest() + + @router.post("/support/messages", response_model=SupportMessageOut) async def send_support_message( + request: Request, payload: SupportMessageInput, username: Annotated[str, Depends(_require_username)], db: Annotated[Session, Depends(get_db)], @@ -380,6 +422,28 @@ async def send_support_message( if cooldown is not None: raise _too_many_failures_error(cooldown) + # Идемпотентность (#3471): pre-check ДО похода в Telegram — повтор с тем же + # ключом не должен слать второе зеркало в топик, не только не писать вторую + # строку в БД. Тред может ещё не существовать (это первая отправка этого + # ключа) — тогда сравнивать не с чем, идём в Telegram как обычно; сам факт + # "нет треда" здесь безопасен, т.к. тред создаётся ТОЛЬКО в этой же ручке + # (см. H1) — если бы он уже был записан под этим ключом, тред уже был бы. + idempotency_key = _resolve_idempotency_key(request, identity_key=username, text=payload.text) + existing_thread_id = storage.find_thread_id(db, username) + if existing_thread_id is not None: + existing = storage.find_inbound_by_idempotency_key( + db, thread_id=existing_thread_id, idempotency_key=idempotency_key + ) + if existing is not None: + logger.info( + "web support: repeat send (idempotency key already recorded) " + "username=%s thread_id=%d message_id=%s", + username, + existing_thread_id, + existing["id"], + ) + return SupportMessageOut(**existing) + # Общий клиент приложения (#tg-connection-resilience): на каждый запрос # свой создавать нельзя — это ноль keep-alive и полный TCP+TLS-хендшейк # до api.telegram.org перед каждой отправкой. Живёт в lifespan. @@ -438,6 +502,7 @@ async def send_support_message( text_body=payload.text, topic_message_id=topic_message_id, support_chat_id=settings.telegram_support_chat_id, + idempotency_key=idempotency_key, ) db.commit() except SQLAlchemyError: @@ -608,6 +673,31 @@ async def send_anon_support_message( raise _too_many_failures_error(cooldown) display_id = _anon_display_id(token) + + # Идемпотентность (#3471) — см. развёрнутый комментарий в авторизованной + # ветке. Ограничение специфично для анонимной ветки: если предыдущая + # попытка сама провалилась ДО выдачи куки (`is_new_token=True` тогда и + # сейчас), identity_key каждый раз новый (случайный токен) и pre-check + # структурно не может найти прошлую попытку — это тот же класс проблемы, + # что и потеря куки в сети, вне scope этой правки. + idempotency_key = _resolve_idempotency_key(request, identity_key=thread_key, text=payload.text) + existing_thread_id = storage.find_thread_id(db, thread_key) + if existing_thread_id is not None: + existing = storage.find_inbound_by_idempotency_key( + db, thread_id=existing_thread_id, idempotency_key=idempotency_key + ) + if existing is not None: + logger.info( + "web support (anon): repeat send (idempotency key already recorded) " + "%s thread_id=%d message_id=%s", + display_id, + existing_thread_id, + existing["id"], + ) + if is_new_token: + _set_anon_cookie(response, token) + return SupportMessageOut(**existing) + # Общий клиент приложения (#tg-connection-resilience): на каждый запрос # свой создавать нельзя — это ноль keep-alive и полный TCP+TLS-хендшейк # до api.telegram.org перед каждой отправкой. Живёт в lifespan. @@ -657,6 +747,7 @@ async def send_anon_support_message( text_body=payload.text, topic_message_id=topic_message_id, support_chat_id=settings.telegram_support_chat_id, + idempotency_key=idempotency_key, ) db.commit() except SQLAlchemyError: diff --git a/tradein-mvp/backend/app/services/tgbot/web_support_storage.py b/tradein-mvp/backend/app/services/tgbot/web_support_storage.py index e1b87a23..feb9ec2b 100644 --- a/tradein-mvp/backend/app/services/tgbot/web_support_storage.py +++ b/tradein-mvp/backend/app/services/tgbot/web_support_storage.py @@ -61,6 +61,36 @@ def get_or_create_thread(db: Session, username: str) -> int: return int(row[0]) +def find_inbound_by_idempotency_key( + db: Session, *, thread_id: int, idempotency_key: str +) -> dict[str, Any] | None: + """Уже записанное inbound-сообщение с этим ключом идемпотентности в треде, + если есть (#3471). Вызывается ИЗ `app.api.v1.support` ДО похода в Telegram + (`send_support_message` / `send_anon_support_message`) — повтор с тем же + ключом не должен создавать второе зеркало в топике, а не только вторую + строку в БД. `thread_id`, а не username/anon-token: таблица не хранит + identity напрямую, а тред уже гарантированно существует к моменту, когда + этот ключ мог быть записан (тред создаётся ДО `record_inbound`, см. H1 в + докстринге `app.api.v1.support`).""" + row = ( + db.execute( + text( + """ + SELECT id, direction, text_body, operator_tg_id, created_at + FROM web_support_messages + WHERE thread_id = CAST(:thread_id AS bigint) + AND direction = 'in' + AND idempotency_key = :idempotency_key + """ + ), + {"thread_id": thread_id, "idempotency_key": idempotency_key}, + ) + .mappings() + .one_or_none() + ) + return dict(row) if row is not None else None + + def record_inbound( db: Session, *, @@ -68,6 +98,7 @@ def record_inbound( text_body: str, topic_message_id: int | None, support_chat_id: int | None, + idempotency_key: str | None = None, ) -> dict[str, Any]: """Записывает сообщение пользователя сайта (direction='in'). `topic_message_id` — id зеркала (sendMessage) в support-топике, ключ маршрутизации ответа оператора. @@ -75,18 +106,33 @@ def record_inbound( review M1): скоупит будущий резолв `find_thread_by_topic_message` к ТЕКУЩЕЙ support-группе — если группу когда-нибудь сменят/пересоздадут, Telegram message_id стартует заново с 1 в новом чате и может совпасть с числом из - старого — без этого поля коллизия была бы ТИХОЙ (см. миграцию 187/188).""" + старого — без этого поля коллизия была бы ТИХОЙ (см. миграцию 187/188). + + `idempotency_key` (#3471, миграция 301) — вызывающая сторона (`app.api.v1.support`) + делает SELECT-затем-действие pre-check ДО Telegram-похода (см. + `find_inbound_by_idempotency_key`), но этот pre-check САМ ПО СЕБЕ гонку не + закрывает (TOCTOU): два запроса с одним ключом могут пройти его одновременно + и оба уйти в Telegram. Последняя линия защиты — здесь: `INSERT ... ON + CONFLICT (thread_id, idempotency_key) DO NOTHING` на partial unique индексе + (мигр. 301, тот же predicate). Если конфликт всё же случился, проигравший + НЕ создаёт вторую строку — читает уже вставленную и возвращает её, так что + оба запроса-конкурента получают ОДИН и тот же id. `idempotency_key=None` + (дефолт) ведёт себя как раньше: NULL никогда не конфликтует сам с собой в + partial-индексе (WHERE idempotency_key IS NOT NULL), INSERT всегда проходит.""" row = ( db.execute( text( """ INSERT INTO web_support_messages (thread_id, direction, text_body, topic_message_id, - support_chat_id, operator_tg_id, created_at) + support_chat_id, operator_tg_id, idempotency_key, created_at) VALUES (CAST(:thread_id AS bigint), 'in', :text_body, CAST(:topic_message_id AS bigint), - CAST(:support_chat_id AS bigint), NULL, NOW()) + CAST(:support_chat_id AS bigint), NULL, :idempotency_key, NOW()) + ON CONFLICT (thread_id, idempotency_key) + WHERE idempotency_key IS NOT NULL AND direction = 'in' + DO NOTHING RETURNING id, direction, text_body, operator_tg_id, created_at """ ), @@ -95,12 +141,22 @@ def record_inbound( "text_body": text_body, "topic_message_id": topic_message_id, "support_chat_id": support_chat_id, + "idempotency_key": idempotency_key, }, ) .mappings() - .one() + .one_or_none() ) - return dict(row) + if row is not None: + return dict(row) + # Конфликт пойман индексом (см. докстринг выше) — idempotency_key точно не + # NULL здесь: при NULL partial-индекс в конфликт не участвует вообще. + assert idempotency_key is not None + existing = find_inbound_by_idempotency_key( + db, thread_id=thread_id, idempotency_key=idempotency_key + ) + assert existing is not None # конфликт по индексу гарантирует наличие строки + return existing def find_thread_by_topic_message( diff --git a/tradein-mvp/backend/data/sql/301_web_support_message_idempotency_key.sql b/tradein-mvp/backend/data/sql/301_web_support_message_idempotency_key.sql new file mode 100644 index 00000000..40b6771b --- /dev/null +++ b/tradein-mvp/backend/data/sql/301_web_support_message_idempotency_key.sql @@ -0,0 +1,53 @@ +-- 301_web_support_message_idempotency_key.sql +-- Идемпотентность отправки веб-сообщения в поддержку (#3471 retry-storm): +-- канал Selectel -> api.telegram.org теряет заметную долю коротких запросов +-- (см. app/api/v1/support.py, _INTERACTIVE_SEND_TIMEOUT_S), поэтому повтор +-- клиента (fetch-ретрай/двойной клик/переотправка по таймауту) — обычное +-- дело. Раньше повтор создавал ВТОРУЮ строку в web_support_messages и ВТОРОЕ +-- зеркало в support-топике. +-- +-- idempotency_key — client-provided (заголовок Idempotency-Key) ИЛИ +-- детерминированный fallback-отпечаток sha256(identity|текст|минутное окно) +-- для клиентов без заголовка (app/api/v1/support.py::_resolve_idempotency_key). +-- Всегда opaque-строка (клиентский токен или hex-хэш) — ТЕЛО СООБЩЕНИЯ сюда +-- никогда не попадает в открытом виде, только его отпечаток. +-- +-- Scoped к (thread_id, direction='in'): идемпотентность осмысленна только для +-- inbound (клиентских) сообщений — direction='out' (ответ оператора) её не +-- требует, там уже есть своя partial-уникальность на topic_message_id (187). +-- thread_id, а не username/anon-token: таблица не хранит identity напрямую, +-- а тред уже гарантированно создан (`get_or_create_thread`) к моменту INSERT +-- (см. H1 в app/api/v1/support.py — тред создаётся ДО record_inbound). +-- +-- Гонку двух одновременных запросов с одним ключом закрывает САМ уникальный +-- индекс + `INSERT ... ON CONFLICT DO NOTHING` в +-- web_support_storage.record_inbound (не read-then-write) — на конфликте +-- вызывающая сторона дочитывает уже вставленную строку и возвращает её тем +-- же ответом (тот же id), а не создаёт вторую строку. +-- +-- IDEMPOTENCY: ADD COLUMN/CREATE INDEX IF NOT EXISTS — безопасный re-run. +-- Зависимости: 187_web_support_chat.sql (таблица web_support_messages). + +BEGIN; + +ALTER TABLE web_support_messages + ADD COLUMN IF NOT EXISTS idempotency_key text; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint WHERE conname = 'web_support_messages_idempotency_key_len_chk' + ) THEN + ALTER TABLE web_support_messages + ADD CONSTRAINT web_support_messages_idempotency_key_len_chk + CHECK (idempotency_key IS NULL OR char_length(idempotency_key) BETWEEN 1 AND 160); + END IF; +END $$; + +COMMENT ON COLUMN web_support_messages.idempotency_key IS 'Ключ идемпотентности inbound-отправки (#3471) — client-provided заголовок Idempotency-Key ("client:...") ИЛИ sha256-отпечаток thread+текст+минутное окно ("auto:...") для клиентов без заголовка. NULL для direction=''out'' и для строк до этой миграции.'; + +CREATE UNIQUE INDEX IF NOT EXISTS web_support_messages_thread_idempotency_uq + ON web_support_messages (thread_id, idempotency_key) + WHERE idempotency_key IS NOT NULL AND direction = 'in'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_support.py b/tradein-mvp/backend/tests/test_support.py index 604bcef5..3abb418b 100644 --- a/tradein-mvp/backend/tests/test_support.py +++ b/tradein-mvp/backend/tests/test_support.py @@ -76,6 +76,20 @@ class _FakeTelegramClient: return _FakeTelegramClient._response +@pytest.fixture(autouse=True) +def _no_existing_idempotent_message(monkeypatch: pytest.MonkeyPatch) -> None: + """Дефолт для тестов, которые не проверяют идемпотентность (#3471) напрямую: + без этого автостаба pre-check в `support.py` дёргал бы РЕАЛЬНУЮ + storage-функцию против MagicMock `db` в каждом тесте, где замокан + `find_thread_id` для своих целей (rate-limit / DB-failure / anon-сценарии + и т.д.) — падало бы на MagicMock-результате, никак не связанное с тем, что + тест на самом деле проверяет. Тесты идемпотентности переопределяют это + через `_install_fake_storage` ниже.""" + monkeypatch.setattr( + support_module.storage, "find_inbound_by_idempotency_key", lambda *a, **kw: None + ) + + @pytest.fixture(autouse=True) def _fake_telegram_client(monkeypatch: pytest.MonkeyPatch) -> Any: _FakeTelegramClient.calls = [] @@ -238,7 +252,9 @@ def test_send_message_happy_path_mirrors_with_website_marker( monkeypatch.setattr(support_module.storage, "get_or_create_thread", fake_get_or_create_thread) recorded = {} - def fake_record_inbound(db, *, thread_id, text_body, topic_message_id, support_chat_id): + def fake_record_inbound( + db, *, thread_id, text_body, topic_message_id, support_chat_id, idempotency_key=None + ): recorded.update( thread_id=thread_id, text_body=text_body, @@ -1052,3 +1068,156 @@ def test_anon_send_failures_trigger_cooldown_per_ip( 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 не дошло +# ── idempotency (#3471) ────────────────────────────────────────────────────── + + +class _FakeIdempotentStorage: + """In-memory stand-in для `web_support_storage`, достаточный чтобы точно + воспроизвести идемпотентный контракт (thread_id, idempotency_key) БЕЗ + настоящей БД: unique-конфликт на повторном ключе внутри `record_inbound` + (см. миграцию 301 и storage.record_inbound docstring).""" + + def __init__(self) -> None: + self.threads: dict[str, int] = {} + self.messages: list[dict[str, Any]] = [] + self._next_id = 1 + + def find_thread_id(self, db: Any, username: str) -> int | None: + return self.threads.get(username) + + def get_or_create_thread(self, db: Any, username: str) -> int: + if username not in self.threads: + self.threads[username] = len(self.threads) + 1 + return self.threads[username] + + def find_inbound_by_idempotency_key( + self, db: Any, *, thread_id: int, idempotency_key: str + ) -> dict[str, Any] | None: + for m in self.messages: + if m["_thread_id"] == thread_id and m["_idempotency_key"] == idempotency_key: + return {k: v for k, v in m.items() if not k.startswith("_")} + return None + + def record_inbound( + self, + db: Any, + *, + thread_id: int, + text_body: str, + topic_message_id: int | None, + support_chat_id: int | None, + idempotency_key: str | None = None, + ) -> dict[str, Any]: + # Тот же контракт, что и настоящий `INSERT ... ON CONFLICT DO NOTHING`: + # конфликт по (thread_id, idempotency_key) отдаёт УЖЕ существующую строку. + if idempotency_key is not None: + existing = self.find_inbound_by_idempotency_key( + db, thread_id=thread_id, idempotency_key=idempotency_key + ) + if existing is not None: + return existing + row = { + "id": self._next_id, + "direction": "in", + "text_body": text_body, + "operator_tg_id": None, + "created_at": "2026-09-12T00:00:00+00:00", + "_thread_id": thread_id, + "_idempotency_key": idempotency_key, + } + self._next_id += 1 + self.messages.append(row) + return {k: v for k, v in row.items() if not k.startswith("_")} + + +def _install_fake_storage(monkeypatch: pytest.MonkeyPatch) -> _FakeIdempotentStorage: + fake = _FakeIdempotentStorage() + monkeypatch.setattr(support_module.storage, "find_thread_id", fake.find_thread_id) + monkeypatch.setattr(support_module.storage, "get_or_create_thread", fake.get_or_create_thread) + monkeypatch.setattr( + support_module.storage, + "find_inbound_by_idempotency_key", + fake.find_inbound_by_idempotency_key, + ) + monkeypatch.setattr(support_module.storage, "record_inbound", fake.record_inbound) + return fake + + +def test_repeat_send_with_same_idempotency_header_returns_same_id_single_mirror( + client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any +) -> None: + """Повтор с тем же `Idempotency-Key` — тот же id, ОДНО зеркало в Telegram + (не только одна строка в БД, см. #3471 требование п.3).""" + _install_fake_storage(monkeypatch) + headers = {**_auth("kopylov"), "Idempotency-Key": "retry-abc123"} + + r1 = client.post( + "/api/v1/trade-in/support/messages", json={"text": "первое сообщение"}, headers=headers + ) + r2 = client.post( + "/api/v1/trade-in/support/messages", json={"text": "первое сообщение"}, headers=headers + ) + + assert r1.status_code == 200, r1.text + assert r2.status_code == 200, r2.text + assert r1.json()["id"] == r2.json()["id"] + # Ключевая проверка: повтор НЕ ушёл в Telegram второй раз. + assert len(_fake_telegram_client.calls) == 1 + + +def test_send_with_different_idempotency_keys_creates_different_messages( + client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any +) -> None: + _install_fake_storage(monkeypatch) + r1 = client.post( + "/api/v1/trade-in/support/messages", + json={"text": "вопрос один"}, + headers={**_auth("kopylov"), "Idempotency-Key": "key-one-111"}, + ) + r2 = client.post( + "/api/v1/trade-in/support/messages", + json={"text": "вопрос два"}, + headers={**_auth("kopylov"), "Idempotency-Key": "key-two-222"}, + ) + + assert r1.status_code == 200, r1.text + assert r2.status_code == 200, r2.text + assert r1.json()["id"] != r2.json()["id"] + assert len(_fake_telegram_client.calls) == 2 + + +def test_repeat_send_without_idempotency_header_still_dedupes_via_fallback_window( + client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any +) -> None: + """Требование п.5: отсутствие явного ключа не должно ломать старых клиентов — + сервер сам считает детерминированный fallback-отпечаток (thread+текст+минутное + окно), так что тот же повтор ТЕМ ЖЕ клиентом без заголовка тоже не создаёт + дубль. Время фиксируем, чтобы не зависеть от границы минутного окна в CI.""" + _install_fake_storage(monkeypatch) + monkeypatch.setattr(support_module.time, "time", lambda: 1_800_000_000.0) + + r1 = client.post( + "/api/v1/trade-in/support/messages", json={"text": "не могу войти"}, headers=_auth("bob") + ) + r2 = client.post( + "/api/v1/trade-in/support/messages", json={"text": "не могу войти"}, headers=_auth("bob") + ) + + assert r1.status_code == 200, r1.text + assert r2.status_code == 200, r2.text + assert r1.json()["id"] == r2.json()["id"] + assert len(_fake_telegram_client.calls) == 1 + + +def test_send_without_idempotency_header_first_call_still_succeeds( + client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any +) -> None: + """Требование п.5 (не ломает старых клиентов): один запрос без заголовка + отрабатывает как раньше — 200, тред создан, зеркало отправлено ровно раз.""" + _install_fake_storage(monkeypatch) + r = client.post( + "/api/v1/trade-in/support/messages", json={"text": "просто вопрос"}, headers=_auth("carol") + ) + assert r.status_code == 200, r.text + assert r.json()["persisted"] is True + assert len(_fake_telegram_client.calls) == 1