15 changed files with 7 additions and 2233 deletions
--- a/.forgejo/workflows/deploy-tradein.yml
+++ b/.forgejo/workflows/deploy-tradein.yml
@ -483,11 +483,7 @@ jobs:
            #      рискует ложно отменить НЕ относящийся к этому recreate run (напр.
            #      admin-triggered scrape внутри backend, если backend в этом деплое
            #      не пересоздавался — его heartbeat продолжит расти после checkpoint'а).
-            # tgbot: тот же backend-образ (rebuild уже покрыт filters.backend —
+            SERVICES="browser backend frontend"
            # tradein-mvp/backend/** включает app/tgbot_main.py), никакого
            # in-flight state вроде scrape_runs → пересоздаётся безусловно вместе
            # с browser/backend/frontend, отдельного graceful-drain не требует.
            SERVICES="browser backend frontend tgbot"
            SCRAPER_STOP_TS=""
            if [ "${SCRAPER_CHANGED:-true}" = "true" ]; then
              echo "→ scraper paths changed — waiting for in-flight scrape_runs to drain (up to 5 min)"
--- a/tradein-mvp/.env.example
+++ b/tradein-mvp/.env.example
@ -40,14 +40,3 @@ DADATA_API_SECRET=
 POSTGRES_USER=tradein
 POSTGRES_PASSWORD=tradein
 POSTGRES_DB=tradein
 # === Telegram support-bot bridge (tgbot service, docker-compose.prod.yml) ===
 # Long-polling worker: пересылает support-обращения в Telegram-топик. Не FastAPI,
 # отдельный процесс (app/tgbot_main.py), env читается из backend/.env.runtime на VPS.
 #
 # BotFather token. Пусто = бот не стартует (выключен).
 TELEGRAM_BOT_TOKEN=
 # ID супергруппы-форума с включёнными топиками (вида -100XXXXXXXXXX).
 TELEGRAM_SUPPORT_CHAT_ID=
 # ID топика (thread) внутри супергруппы, куда падают support-сообщения.
 TELEGRAM_SUPPORT_TOPIC_ID=
--- a/tradein-mvp/DEPLOY.md
+++ b/tradein-mvp/DEPLOY.md
@ -89,47 +89,10 @@ GLITCHTIP_DSN=<dsn или пусто>
 # Регистрация ключей: https://dadata.ru/api/clean/
 DADATA_API_TOKEN=<token или пусто>
 DADATA_API_SECRET=<secret или пусто>
 # Telegram support-bot bridge (сервис tgbot, docker-compose.prod.yml).
 # Long-polling воркер (app/tgbot_main.py), тот же образ что backend/scraper,
 # отдельный контейнер tradein-tgbot. Пусто TELEGRAM_BOT_TOKEN = бот НЕ падает
 # и НЕ рестарт-лупится — процесс стартует, уходит в idle-блокировку и просто
 # висит (это норма для окружений без токена, не сбой; см. tgbot_main.py).
 #
 # 1. TELEGRAM_BOT_TOKEN — токен от @BotFather (/newbot). Пусто = бот выключен
 #    (idle, не polling).
 TELEGRAM_BOT_TOKEN=<token или пусто>
 # 2. TELEGRAM_SUPPORT_CHAT_ID — id супергруппы-форума (Topics включены в
 #    настройках группы), вида -100XXXXXXXXXX. Получить: добавить бота в группу,
 #    отправить любое сообщение в любой топик, дернуть
 #    https://api.telegram.org/bot<token>/getUpdates — в ответе
 #    message.chat.id (для супергруппы всегда отрицательный, начинается с -100).
 TELEGRAM_SUPPORT_CHAT_ID=<-100... или пусто>
 # 3. TELEGRAM_SUPPORT_TOPIC_ID — id конкретного топика (thread) внутри группы,
 #    куда падают support-обращения. Открыть нужный топик в Telegram Desktop/Web →
 #    в URL топика (t.me/c/<chat>/<topic_id>) последнее число — это topic_id.
 #    Либо взять message_thread_id из того же getUpdates-ответа (п.2), отправив
 #    тестовое сообщение именно в целевой топик.
 TELEGRAM_SUPPORT_TOPIC_ID=<topic_id или пусто>
 ```
 Оба файла создаются вручную при первом деплое.
 ### Деплой / рестарт `tgbot`
 ```bash
 # .env.runtime читается на старте container — `compose restart` НЕ перечитывает.
 docker compose -p gendesign-tradein -f docker-compose.prod.yml \
  up -d --force-recreate --no-deps tgbot
 ```
 `restart: unless-stopped` + `stop_grace_period: 120s` в compose (см. `docker-compose.prod.yml`)
 — автозапуск после ребута VPS гарантирован (`on-failure` сюда не годится: код выхода
 контейнера при ребуте — гонка с long-poll таймаутом 30с, `unless-stopped`/`always`
 не зависят от exit-кода). 120s grace даёт time докрутить long-poll + отработать
 кооперативный drain (`_DRAIN_TIMEOUT_S=100s` в `tgbot_main.py`) до docker SIGKILL —
 паттерн скопирован с `scraper` (см. комментарий там же).
 ### После изменения `backend/.env.runtime`
 ```bash
--- a/tradein-mvp/backend/app/core/config.py
+++ b/tradein-mvp/backend/app/core/config.py
@ -664,19 +664,5 @@ class Settings(BaseSettings):
    # честная маркировка.
    sell_time_sensitivity_min_n_lots: int = 10
    # ── Telegram support bridge (@MERAsupport_bot) ───────────────────────────
    # Клиент пишет боту в личку → зеркалится в топик support-группы → оператор
    # отвечает реплаем в топике → бот доставляет ответ клиенту. Standalone
    # long-polling воркер (app.tgbot_main), НЕ webhook — см. app/services/tgbot/.
    # Пусто/0 = бот выключен: tgbot_main логирует «disabled» и выходит с кодом 0
    # (чтобы контейнер без секрета не крутил рестарт-луп). ENV: TELEGRAM_BOT_TOKEN,
    # TELEGRAM_SUPPORT_CHAT_ID, TELEGRAM_SUPPORT_TOPIC_ID.
    telegram_bot_token: str = Field(default="", validation_alias="TELEGRAM_BOT_TOKEN")
    # Telegram id форум-группы (супергруппы с включёнными топиками), куда
    # зеркалятся обращения клиентов. Отрицательный для supergroup id (напр. -100...).
    telegram_support_chat_id: int = Field(default=0, validation_alias="TELEGRAM_SUPPORT_CHAT_ID")
    # message_thread_id топика внутри support-группы, в который идут зеркала.
    telegram_support_topic_id: int = Field(default=0, validation_alias="TELEGRAM_SUPPORT_TOPIC_ID")
 settings = Settings()
--- a/tradein-mvp/backend/app/observability/sentry_scrub.py
+++ b/tradein-mvp/backend/app/observability/sentry_scrub.py
@ -1,51 +1,22 @@
-"""Хуки before_send для GlitchTip/Sentry SDK (tradein-local, #396, #tgsupport).
+"""Хук before_send для GlitchTip/Sentry SDK (tradein-local, #396).
 Redact-ит consumer-PII (client_name / client_phone / client_email и пр.)
 из error events до отправки в GlitchTip — estimator/trade-in flow таскает
 эти поля, а send_default_pii=False их не покрывает (это user-data в
 request.data / extra / contexts, не PII-заголовки).
 `redact_telegram_bot_token` — отдельный хук (#tgsupport review): Telegram Bot
 API токен живёт в URL-пути (`https://api.telegram.org/bot<id>:<secret>/...`),
 а не в query/userinfo, поэтому НЕ покрывается sentry_sdk `sanitize_url` (тот
 режет только `user:pass@` и query-параметры). Токен утекает ДВУМЯ путями,
 которые `_scrub`/`scrub_pii_event` (ключ-based, PII-словарь) не ловят:
  1. `include_local_variables=True` (sentry_sdk default) кладёт locals
     stack-фрейма (`self._base`, `url` в `TelegramClient._request`) в
     traceback → полный токен открытым текстом.
  2. `HttpxIntegration` кладёт полный request URL в span `data` (виден при
     любом ненулевом `traces_sample_rate`), а не только в traceback.
 Поэтому редактор — НЕ ключ-based, а regex full-text по КАЖДОЙ строке во всём
 event (глубокий обход dict/list/tuple) — токен может всплыть в любом поле.
 """
 from __future__ import annotations
 import re
 from typing import Any
 from sentry_sdk.types import Event
 _REDACTED = "[REDACTED]"
 # Ключи consumer-PII (нижний регистр; сверка case-insensitive).
-_PII_KEYS = frozenset({"client_name", "client_phone", "client_email", "phone", "email", "name"})
+_PII_KEYS = frozenset(
-
+    {"client_name", "client_phone", "client_email", "phone", "email", "name"}
-# Telegram Bot API токен в пути URL: /bot<numeric_id>:<secret-part>/<method>.
+)
 # Матчим ровно этот сегмент (не весь URL) — сохраняет остальной путь/query
 # читаемым для диагностики (метод API, error code и т.п.).
 _TG_BOT_TOKEN_RE = re.compile(r"/bot\d+:[A-Za-z0-9_-]+")
 _TG_BOT_TOKEN_REPLACEMENT = "/bot[REDACTED]"
 # Тот же токен БЕЗ префикса `/bot` — форма `<numeric_id>:<secret>` сама по себе
 # (напр. локаль `token` в конструкторе TelegramClient, или если его кто-то
 # засунет в log-сообщение). Сейчас единственный путь такой формы в event —
 # locals стек-фрейма, а они выключены через include_local_variables=False в
 # tgbot_main. Но именно на отказ того флага этот редактор и страхует: без этой
 # ветки рубеж был бы один, а не два. Формат токена BotFather: 8-12 цифр `:` 35
 # символов base64url — нижние границы взяты с запасом, чтобы не промахнуться
 # на нестандартных id, но остаться уже, чем `\d+:\S+` (тот бил бы по любым
 # `id:value` в логах, напр. `chat_id:12345`).
 _TG_BOT_TOKEN_BARE_RE = re.compile(r"\b\d{6,12}:[A-Za-z0-9_-]{30,}\b")
 def _scrub(obj: Any) -> None:
@ -71,34 +42,3 @@ def scrub_pii_event(event: Event, _hint: dict[str, Any]) -> Event | None:
    _scrub(event.get("extra"))
    _scrub(event.get("contexts"))
    return event
 def _redact_strings(obj: Any) -> Any:
    """Рекурсивно проходит dict/list/tuple и прогоняет обе токен-регулярки по КАЖДОЙ
    строке (не только по конкретным ключам) — токен может оказаться в locals
    stack-фрейма, span data, breadcrumb message, request.url и т.д. Возвращает
    НОВУЮ структуру (не мутирует `obj` — в отличие от `_scrub`, чтобы не зависеть
    от того, какие контейнеры sentry_sdk считает mutable в своём event dict)."""
    if isinstance(obj, str):
        redacted = _TG_BOT_TOKEN_RE.sub(_TG_BOT_TOKEN_REPLACEMENT, obj)
        return _TG_BOT_TOKEN_BARE_RE.sub(_REDACTED, redacted)
    if isinstance(obj, dict):
        return {k: _redact_strings(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_redact_strings(v) for v in obj]
    if isinstance(obj, tuple):
        return tuple(_redact_strings(v) for v in obj)
    return obj
 def redact_telegram_bot_token(event: Event, _hint: dict[str, Any]) -> Event | None:
    """Full-text regex redaction Telegram Bot API токена по ВСЕМУ event (#tgsupport).
    Ловит оба вектора утечки токена в GlitchTip, которые ключ-based `scrub_pii_event`
    не покрывает: locals stack-фреймов (`include_local_variables=True`) и httpx-span
    `data` (полный request URL). Композировать с `scrub_pii_event`, не вместо него —
    разные классы секретов (PII полей формы vs bot-токен в URL).
    """
    if not isinstance(event, dict):
        return event
    return _redact_strings(event)  # type: ignore[return-value]
--- a/tradein-mvp/backend/app/services/tgbot/init.py
+++ b/tradein-mvp/backend/app/services/tgbot/init.py
@ -1,9 +0,0 @@
 """Telegram support bridge (@MERAsupport_bot) — long-polling мост клиент↔оператор.
 Клиент пишет боту в личку → зеркалится в топик support-группы (`client.py` —
 тонкая HTTP-обёртка над Bot API; `bridge.py` — маршрутизация апдейтов и
 персистентность через tg_support_* таблицы). Standalone entrypoint —
 `app.tgbot_main` (long-polling воркер, НЕ webhook, отдельный контейнер/процесс).
 """
 from __future__ import annotations
--- a/tradein-mvp/backend/app/services/tgbot/bridge.py
+++ b/tradein-mvp/backend/app/services/tgbot/bridge.py
@ -1,537 +0,0 @@
 """Маршрутизация Telegram-апдейтов для support-моста (#tgsupport).
 Поток:
  A) Клиент пишет боту в личку (chat.type == 'private') →
     upsert tg_support_users → (если первое сообщение за последний час — шапка
     с идентификацией клиента в топик) → copyMessage контента в support-топик →
     запись в tg_support_messages (direction='in', topic_message_id — ключ
     маршрутизации ответа).
  B) Оператор отвечает РЕПЛАЕМ в support-группе на зеркало клиента →
     находим chat_id по topic_message_id → copyMessage ответа в личку клиента →
     запись (direction='out'). Реплай не на зеркало (или не реплай вообще) —
     обычная болтовня в топике, тихий игнор. Telegram 403 (клиент заблокировал
     бота) → is_blocked=true + уведомление в топике.
  C) Дедуп: update_id <= сохранённого offset — skip. Offset сохраняется И
     коммитится в той же транзакции, что и запись сообщения (см. `process_update`
     `finally`), после КАЖДОГО апдейта — рестарт воркера не переигрывает уже
     обработанные апдейты и не подвисает вечно на «ядовитом» апдейте.
  D) TELEGRAM_BOT_TOKEN пуст → бот выключен — проверяется в `app.tgbot_main`
     (entrypoint), не здесь.
  E) /start клиенту → короткое приветствие МЕРЫ, без зеркалирования в топик
     (команда — не содержательное обращение, не должна засорять топик).
 Персистентность вынесена за `BridgeStorage`-протокол — маршрутизирующая логика
 (`process_update` и приватные `_handle_*`) не завязана на реальную БД, тестируется
 на in-memory fake storage + mock httpx (см. tests/services/tgbot/).
 """
 from __future__ import annotations
 import asyncio
 import logging
 from collections.abc import Callable
 from typing import Any, Protocol
 from sqlalchemy import text
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 from app.core.config import settings
 from app.core.shutdown import shutdown_requested
 from app.services.tgbot.client import TelegramApiError, TelegramClient
 logger = logging.getLogger(__name__)
 # Ключ в tg_support_state под который сохраняется last processed update_id
 # (см. data/sql/186_tg_support.sql — комментарий на колонке .key).
 _OFFSET_KEY = "last_update_id"
 # Окно, за которое повторное сообщение клиента НЕ дублирует шапку-идентификацию
 # в топике (одна шапка на "сессию" обращения).
 _HEADER_THROTTLE_WINDOW_S = 3600
 GREETING_TEXT = (
    "Здравствуйте! Это служба поддержки МЕРА (сервис trade-in квартир). "
    "Опишите ваш вопрос — оператор ответит вам в этом чате в ближайшее время."
 )
 # Отправляется клиенту вместо тихой потери сообщения, если TELEGRAM_SUPPORT_CHAT_ID
 # не сконфигурирован (иначе клиент ждёт ответа, которого никогда не будет — #5 review).
 SERVICE_UNAVAILABLE_TEXT = (
    "Служба поддержки временно недоступна. Пожалуйста, попробуйте написать позже."
 )
 # Kinds, задокументированные в data/sql/186_tg_support.sql (COMMENT ON COLUMN
 # tg_support_messages.kind): "text | photo | document | video | voice | other".
 _KNOWN_KINDS = ("text", "photo", "document", "video", "voice")
 # ── Storage abstraction (testable без реальной БД) ──────────────────────────
 class BridgeStorage(Protocol):
    """Persistence-контракт моста. `SqlBridgeStorage` — прод-реализация поверх
    tg_support_* (см. data/sql/186_tg_support.sql). Тесты используют in-memory fake."""
    def get_offset(self) -> int: ...
    def save_offset(self, update_id: int) -> None: ...
    def commit(self) -> None: ...
    def rollback(self) -> None: ...
    def upsert_user(
        self,
        *,
        chat_id: int,
        username: str | None,
        first_name: str | None,
        last_name: str | None,
        language_code: str | None,
    ) -> None: ...
    def had_recent_inbound(self, chat_id: int, window_seconds: int) -> bool: ...
    def record_message(
        self,
        *,
        chat_id: int,
        direction: str,
        tg_message_id: int | None,
        topic_message_id: int | None,
        kind: str,
        text_body: str | None,
        operator_tg_id: int | None,
    ) -> int | None: ...
    def find_chat_by_topic_message(self, topic_message_id: int) -> int | None: ...
    def mark_blocked(self, chat_id: int) -> None: ...
 class SqlBridgeStorage:
    """`BridgeStorage` поверх SQLAlchemy Session (psycopg v3), tg_support_* таблицы.
    Методы исполняют SQL немедленно, но НЕ коммитят по отдельности — коммит
    один раз в конце `process_update` (после записи сообщения И offset'а), чтобы
    оба изменения фиксировались атомарно в одной транзакции (требование C).
    """
    def __init__(self, db: Session) -> None:
        self._db = db
    def get_offset(self) -> int:
        row = self._db.execute(
            text("SELECT value FROM tg_support_state WHERE key = CAST(:key AS text)"),
            {"key": _OFFSET_KEY},
        ).fetchone()
        if row is None or row[0] is None:
            return 0
        try:
            return int(row[0])
        except (TypeError, ValueError):
            logger.warning("tgbot storage: невалидный offset в БД (%r) — считаем 0", row[0])
            return 0
    def save_offset(self, update_id: int) -> None:
        self._db.execute(
            text(
                """
                INSERT INTO tg_support_state (key, value, updated_at)
                VALUES (CAST(:key AS text), CAST(:value AS text), NOW())
                ON CONFLICT (key) DO UPDATE
                    SET value = EXCLUDED.value, updated_at = NOW()
                """
            ),
            {"key": _OFFSET_KEY, "value": str(update_id)},
        )
    def commit(self) -> None:
        self._db.commit()
    def rollback(self) -> None:
        """Откатывает текущую (возможно failed-transaction) сессию перед save_offset.
        Нужно, когда исключение пришло от самой БД (напр. обрыв коннекта к
        postgres при деплое) — SQLAlchemy Session после такого исключения
        переходит в failed-transaction state, и ЛЮБОЙ следующий `execute()`
        (включая `save_offset`) кидает `PendingRollbackError` без явного
        rollback() (#3 review — иначе update_id не сдвигается, апдейт
        переигрывается на следующей итерации, copyMessage дублирует зеркало
        клиента в топик на каждый повтор).
        """
        self._db.rollback()
    def upsert_user(
        self,
        *,
        chat_id: int,
        username: str | None,
        first_name: str | None,
        last_name: str | None,
        language_code: str | None,
    ) -> None:
        self._db.execute(
            text(
                """
                INSERT INTO tg_support_users
                    (chat_id, username, first_name, last_name, language_code,
                     created_at, last_seen_at, is_blocked)
                VALUES
                    (CAST(:chat_id AS bigint), :username, :first_name, :last_name,
                     :language_code, NOW(), NOW(), FALSE)
                ON CONFLICT (chat_id) DO UPDATE
                    SET username = EXCLUDED.username,
                        first_name = EXCLUDED.first_name,
                        last_name = EXCLUDED.last_name,
                        language_code = EXCLUDED.language_code,
                        last_seen_at = NOW(),
                        is_blocked = FALSE
                """
            ),
            {
                "chat_id": chat_id,
                "username": username,
                "first_name": first_name,
                "last_name": last_name,
                "language_code": language_code,
            },
        )
    def had_recent_inbound(self, chat_id: int, window_seconds: int) -> bool:
        row = self._db.execute(
            text(
                """
                SELECT 1
                FROM tg_support_messages
                WHERE chat_id = CAST(:chat_id AS bigint)
                  AND direction = 'in'
                  AND created_at > NOW() - make_interval(secs => CAST(:window_seconds AS integer))
                LIMIT 1
                """
            ),
            {"chat_id": chat_id, "window_seconds": window_seconds},
        ).fetchone()
        return row is not None
    def record_message(
        self,
        *,
        chat_id: int,
        direction: str,
        tg_message_id: int | None,
        topic_message_id: int | None,
        kind: str,
        text_body: str | None,
        operator_tg_id: int | None,
    ) -> int | None:
        row = self._db.execute(
            text(
                """
                INSERT INTO tg_support_messages
                    (chat_id, direction, tg_message_id, topic_message_id, kind,
                     text_body, operator_tg_id, created_at)
                VALUES
                    (CAST(:chat_id AS bigint), CAST(:direction AS text),
                     CAST(:tg_message_id AS bigint), CAST(:topic_message_id AS bigint),
                     CAST(:kind AS text), :text_body, CAST(:operator_tg_id AS bigint), NOW())
                RETURNING id
                """
            ),
            {
                "chat_id": chat_id,
                "direction": direction,
                "tg_message_id": tg_message_id,
                "topic_message_id": topic_message_id,
                "kind": kind,
                "text_body": text_body,
                "operator_tg_id": operator_tg_id,
            },
        ).fetchone()
        return int(row[0]) if row is not None else None
    def find_chat_by_topic_message(self, topic_message_id: int) -> int | None:
        row = self._db.execute(
            text(
                """
                SELECT chat_id
                FROM tg_support_messages
                WHERE topic_message_id = CAST(:topic_message_id AS bigint)
                  AND direction = 'in'
                ORDER BY created_at DESC
                LIMIT 1
                """
            ),
            {"topic_message_id": topic_message_id},
        ).fetchone()
        return int(row[0]) if row is not None else None
    def mark_blocked(self, chat_id: int) -> None:
        self._db.execute(
            text(
                "UPDATE tg_support_users SET is_blocked = TRUE "
                "WHERE chat_id = CAST(:chat_id AS bigint)"
            ),
            {"chat_id": chat_id},
        )
 # ── Pure helpers ──────────────────────────────────────────────────────────────
 def _infer_kind(message: dict[str, Any]) -> str:
    """Content-type сообщения → kind-строка. Неизвестные типы (voice/sticker/location/
    etc.) сворачиваются в 'other' — см. документированный набор в COMMENT ON COLUMN."""
    for field in _KNOWN_KINDS:
        if field in message:
            return field
    return "other"
 def _format_topic_header(
    chat_id: int, username: str | None, first_name: str | None, last_name: str | None
 ) -> str:
    """Короткая шапка-идентификация клиента для support-топика."""
    display_name = " ".join(p for p in (first_name, last_name) if p) or "Без имени"
    username_part = f", @{username}" if username else ""
    return f"Новое обращение от {display_name}{username_part} (chat_id={chat_id})"
 # ── Update routing ────────────────────────────────────────────────────────────
 async def _handle_private_message(
    message: dict[str, Any], client: TelegramClient, storage: BridgeStorage
 ) -> None:
    """A) Личка клиента → бот. upsert user → (опц. шапка) → зеркало в топик."""
    chat = message.get("chat") or {}
    chat_id = chat.get("id")
    if not isinstance(chat_id, int):
        logger.warning("tgbot bridge: приватное сообщение без валидного chat.id — игнор")
        return
    from_user = message.get("from") or {}
    username = from_user.get("username")
    first_name = from_user.get("first_name")
    last_name = from_user.get("last_name")
    language_code = from_user.get("language_code")
    storage.upsert_user(
        chat_id=chat_id,
        username=username,
        first_name=first_name,
        last_name=last_name,
        language_code=language_code,
    )
    text_body = message.get("text")
    if text_body == "/start":
        # E) команда — не содержательное обращение, топик не засоряем.
        await client.send_message(chat_id=chat_id, text=GREETING_TEXT)
        return
    if not settings.telegram_support_chat_id:
        logger.warning(
            "tgbot bridge: TELEGRAM_SUPPORT_CHAT_ID не задан — сообщение от chat_id=%d "
            "не может быть зеркалировано; отвечаем клиенту вместо тихой потери (#5 review)",
            chat_id,
        )
        # Не молчим клиенту (#5 review) — иначе он ждёт ответа, которого никогда не будет.
        await client.send_message(chat_id=chat_id, text=SERVICE_UNAVAILABLE_TEXT)
        return
    message_id = message.get("message_id")
    if not isinstance(message_id, int):
        logger.warning("tgbot bridge: приватное сообщение без message_id — игнор")
        return
    # Шапка — только на первое сообщение клиента за окно, иначе топик засоряется.
    if not storage.had_recent_inbound(chat_id, window_seconds=_HEADER_THROTTLE_WINDOW_S):
        header = _format_topic_header(chat_id, username, first_name, last_name)
        await client.send_message(
            chat_id=settings.telegram_support_chat_id,
            text=header,
            message_thread_id=settings.telegram_support_topic_id or None,
        )
    mirrored = await client.copy_message(
        chat_id=settings.telegram_support_chat_id,
        from_chat_id=chat_id,
        message_id=message_id,
        message_thread_id=settings.telegram_support_topic_id or None,
    )
    topic_message_id = mirrored.get("message_id") if isinstance(mirrored, dict) else None
    storage.record_message(
        chat_id=chat_id,
        direction="in",
        tg_message_id=message_id,
        topic_message_id=topic_message_id,
        kind=_infer_kind(message),
        text_body=text_body or message.get("caption"),
        operator_tg_id=None,
    )
 async def _handle_group_reply(
    message: dict[str, Any], client: TelegramClient, storage: BridgeStorage
 ) -> None:
    """B) Реплай оператора в support-группе → доставка ответа клиенту."""
    reply_to = message.get("reply_to_message")
    if not isinstance(reply_to, dict):
        return  # не реплай вообще — обычная болтовня в топике, тихий игнор
    mirror_message_id = reply_to.get("message_id")
    if not isinstance(mirror_message_id, int):
        return
    target_chat_id = storage.find_chat_by_topic_message(mirror_message_id)
    if target_chat_id is None:
        # Обычная болтовня в топике (реплай на чьё-то ещё сообщение) — не логируем,
        # это ожидаемый шум. НО реплай на сообщение, отправленное САМИМ БОТОМ
        # (is_bot=True) и при этом отсутствующее в tg_support_messages — подозрительно:
        # вероятная причина — осиротевшее зеркало (воркер упал МЕЖДУ copyMessage и
        # storage.commit() в `_handle_private_message` — зеркало ушло в Telegram, а
        # запись в БД потерялась). Дискриминатор неидеальный (шапка-идентификация
        # тоже от бота, но не routing-ключ — тоже даст этот WARNING), но лучше редкий
        # ложный WARNING, чем оператор молча решает, что ответ клиенту доставлен,
        # хотя реплай тихо утонул (#4 review — двухфазный протокол НЕ делаем, overkill).
        reply_from = reply_to.get("from") or {}
        if reply_from.get("is_bot"):
            logger.warning(
                "tgbot bridge: реплай на сообщение бота (message_id=%d) не найден в "
                "tg_support_messages как зеркало клиента — возможно, осиротевшее "
                "зеркало (крах между copyMessage и commit) или шапка-идентификация; "
                "ответ оператора НЕ доставлен клиенту",
                mirror_message_id,
            )
        return
    message_id = message.get("message_id")
    if not isinstance(message_id, int):
        return
    operator = message.get("from") or {}
    operator_id = operator.get("id")
    try:
        delivered = await client.copy_message(
            chat_id=target_chat_id,
            from_chat_id=settings.telegram_support_chat_id,
            message_id=message_id,
        )
    except TelegramApiError as exc:
        if exc.error_code == 403:
            # Клиент заблокировал бота — фиксируем и уведомляем оператора в топике.
            storage.mark_blocked(target_chat_id)
            await client.send_message(
                chat_id=settings.telegram_support_chat_id,
                text=(
                    f"Не удалось доставить сообщение клиенту (chat_id={target_chat_id}) — "
                    "бот заблокирован."
                ),
                message_thread_id=settings.telegram_support_topic_id or None,
                reply_to_message_id=message_id,
            )
            return
        raise
    tg_message_id = delivered.get("message_id") if isinstance(delivered, dict) else None
    storage.record_message(
        chat_id=target_chat_id,
        direction="out",
        tg_message_id=tg_message_id,
        topic_message_id=None,
        kind=_infer_kind(message),
        text_body=message.get("text") or message.get("caption"),
        operator_tg_id=operator_id,
    )
 async def process_update(
    update: dict[str, Any], client: TelegramClient, storage: BridgeStorage
 ) -> None:
    """Маршрутизирует один Telegram update. Дедуп (C) + атомарный offset-commit.
    Дедуп: update_id <= сохранённого offset — skip без side-effects. Offset
    сохраняется и коммитится ПОСЛЕ обработки (в т.ч. если обработка упала —
    иначе «ядовитый» апдейт блокировал бы весь поток навсегда).
    Различаем сбой БД (`SQLAlchemyError`) от прочих (Telegram API и т.п.):
    сбой БД оставляет сессию в failed-transaction state — `rollback()` ОБЯЗАН
    отработать ПЕРЕД `save_offset`, иначе тот сам кинет `PendingRollbackError`,
    `process_update` вылетит без сохранения offset'а, следующая итерация
    `run_poll_loop` получит СТАРЫЙ offset от `get_offset()` и переиграет тот же
    апдейт заново — copyMessage задублирует зеркало клиента в топике на
    каждый повтор поллинга (#3 review, воспроизведено).
    """
    update_id = update.get("update_id")
    if not isinstance(update_id, int):
        logger.warning("tgbot bridge: update без валидного update_id — игнор")
        return
    current_offset = storage.get_offset()
    if update_id <= current_offset:
        logger.debug(
            "tgbot bridge: update_id=%d уже обработан (offset=%d) — skip",
            update_id,
            current_offset,
        )
        return
    message = update.get("message")
    try:
        if isinstance(message, dict):
            chat = message.get("chat") or {}
            chat_type = chat.get("type")
            chat_id = chat.get("id")
            if chat_type == "private":
                await _handle_private_message(message, client, storage)
            elif settings.telegram_support_chat_id and chat_id == settings.telegram_support_chat_id:
                await _handle_group_reply(message, client, storage)
            # иначе — необрабатываемый тип чата/апдейта (edited_message, канал и
            # т.п.) — тихий игнор, но offset всё равно сдвигаем ниже.
    except SQLAlchemyError:
        logger.exception(
            "tgbot bridge: DB-ошибка на update_id=%d — rollback перед сохранением "
            "offset (иначе save_offset сам упадёт на failed-transaction state)",
            update_id,
        )
        storage.rollback()
    except Exception:
        logger.exception(
            "tgbot bridge: обработка update_id=%d упала — offset всё равно сдвигаем "
            "(не блокируем поток на 'ядовитом' апдейте)",
            update_id,
        )
    finally:
        storage.save_offset(update_id)
        storage.commit()
 # ── Long-polling loop ─────────────────────────────────────────────────────────
 async def run_poll_loop(
    client: TelegramClient,
    session_factory: Callable[[], Session],
    poll_timeout_s: int = 30,
 ) -> None:
    """Бесконечный long-polling цикл до `shutdown_requested()`.
    Свежая DB-сессия на каждую итерацию (одна итерация = один getUpdates-вызов +
    обработка полученной пачки апдейтов) — не держим соединение открытым на
    неопределённый срок между итерациями.
    """
    logger.info("tgbot bridge: старт poll loop (timeout=%ds)", poll_timeout_s)
    consecutive_errors = 0
    while not shutdown_requested():
        try:
            with session_factory() as db:
                storage = SqlBridgeStorage(db)
                offset = storage.get_offset()
                updates = await client.get_updates(
                    offset=offset + 1, timeout=poll_timeout_s, allowed_updates=["message"]
                )
                for update in updates:
                    await process_update(update, client, storage)
            consecutive_errors = 0
        except Exception:
            consecutive_errors += 1
            backoff = min(5 * consecutive_errors, 60)
            logger.exception("tgbot bridge: итерация poll loop упала — retry через %ds", backoff)
            await asyncio.sleep(backoff)
    logger.info("tgbot bridge: poll loop остановлен (shutdown)")
--- a/tradein-mvp/backend/app/services/tgbot/client.py
+++ b/tradein-mvp/backend/app/services/tgbot/client.py
@ -1,249 +0,0 @@
 """Тонкая httpx-обёртка над Telegram Bot API (#tgsupport).
 Зачем свой клиент, а не aiogram: единственные нужные методы — `getUpdates`
 (long-polling), `copyMessage` (зеркалирование ЛЮБОГО типа контента без ре-аплоада)
 и `sendMessage` (заголовки/приветствия/уведомления). aiogram — избыточная
 зависимость (webhook-framework, dispatcher, FSM) ради трёх HTTP-вызовов; в стеке
 уже есть httpx (см. `app.services.dadata`, `app.services.geocoder` — тот же паттерн
 retry/timeout).
 Docs: https://core.telegram.org/bots/api
 Ретраи:
  - HTTP 429 (Too Many Requests) — уважаем `parameters.retry_after` из тела ответа
    (Telegram сам говорит сколько ждать), fallback на `_DEFAULT_RETRY_AFTER_S`.
  - HTTP 5xx / сетевые ошибки (timeout/connect) — экспоненциальный backoff,
    `capped` на `_MAX_BACKOFF_S`.
  - Любая другая 4xx (400/401/403/404) — НЕ ретраится, сразу `TelegramApiError`
    (запрос некорректен или прав нет — повтор не поможет).
 БЕЗОПАСНОСТЬ: наши `logger.*`-вызовы здесь содержат только имя метода API,
 HTTP-статус и `description` из ответа Telegram — токен туда не пишем.
 Это НЕ гарантирует, что токен не утечёт по другим стокам: он живёт в
 `self._base`/`url` (локальные переменные stack-фрейма `_request`), а GlitchTip
 (sentry_sdk) по умолчанию прикладывает locals к traceback и Httpx-интеграция
 кладёт полный URL в span data. Эти стоки закрываются НЕ здесь, а в
 `app.tgbot_main` (`include_local_variables=False`, `before_send`-редактор,
 `traces_sample_rate=0.0`) и подавлением INFO-логов самого `httpx`-логгера
 (который печатает полный request URL, включая токен, на уровне INFO).
 """
 from __future__ import annotations
 import asyncio
 import logging
 from typing import Any
 import httpx
 logger = logging.getLogger(__name__)
 _DEFAULT_TIMEOUT_S = 15.0
 _DEFAULT_RETRY_AFTER_S = 5.0
 _MAX_BACKOFF_S = 30.0
 _DEFAULT_MAX_RETRIES = 5
 class TelegramApiError(Exception):
    """Telegram Bot API ответил `ok: false` (после исчерпания ретраев, если применимо)."""
    def __init__(self, method: str, error_code: int, description: str) -> None:
        self.method = method
        self.error_code = error_code
        self.description = description
        super().__init__(f"Telegram API {method} failed: {error_code} {description}")
 def _extract_retry_after(
    response: httpx.Response, default: float = _DEFAULT_RETRY_AFTER_S
 ) -> float:
    """Достаёт `parameters.retry_after` из тела 429-ответа. Fallback — `default`."""
    try:
        data = response.json()
    except ValueError:
        return default
    if not isinstance(data, dict):
        return default
    params = data.get("parameters")
    if isinstance(params, dict):
        retry_after = params.get("retry_after")
        if isinstance(retry_after, int | float):
            return float(retry_after)
    return default
 def _error_from_body(response: httpx.Response) -> tuple[int, str]:
    """Парсит (error_code, description) из тела ответа Telegram; fallback на HTTP-статус."""
    try:
        data = response.json()
    except ValueError:
        return response.status_code, (response.text or "")[:200]
    if not isinstance(data, dict):
        return response.status_code, str(data)[:200]
    error_code = data.get("error_code", response.status_code)
    description = data.get("description", "")
    code = int(error_code) if isinstance(error_code, int | float) else response.status_code
    return code, str(description)
 class TelegramClient:
    """Bot API клиент на httpx.AsyncClient. Каждый вызов — отдельное короткоживущее соединение."""
    def __init__(
        self,
        token: str,
        base_url: str = "https://api.telegram.org",
        timeout: float = _DEFAULT_TIMEOUT_S,
    ) -> None:
        self._base = f"{base_url}/bot{token}"
        self._timeout = timeout
    async def _request(
        self,
        method: str,
        payload: dict[str, Any],
        *,
        timeout: float | None = None,
        max_retries: int = _DEFAULT_MAX_RETRIES,
    ) -> Any:
        """POST `method` с JSON-телом `payload`. Ретраит 429/5xx/network, иначе raise сразу."""
        url = f"{self._base}/{method}"
        effective_timeout = timeout if timeout is not None else self._timeout
        attempt = 0
        while True:
            attempt += 1
            try:
                async with httpx.AsyncClient(timeout=effective_timeout) as client:
                    response = await client.post(url, json=payload)
            except (httpx.TimeoutException, httpx.NetworkError) as exc:
                if attempt > max_retries:
                    logger.error(
                        "tg client: %s — network error после %d попыток: %s", method, attempt, exc
                    )
                    raise
                backoff = min(2.0**attempt, _MAX_BACKOFF_S)
                logger.warning(
                    "tg client: %s — network error (попытка %d/%d): %s — retry через %.0fs",
                    method,
                    attempt,
                    max_retries,
                    exc,
                    backoff,
                )
                await asyncio.sleep(backoff)
                continue
            if response.status_code == 429:
                retry_after = _extract_retry_after(response)
                if attempt > max_retries:
                    error_code, description = _error_from_body(response)
                    logger.error("tg client: %s — 429 после %d попыток, сдаёмся", method, attempt)
                    raise TelegramApiError(method, error_code, description)
                logger.warning(
                    "tg client: %s — HTTP 429 (попытка %d/%d), retry_after=%.0fs",
                    method,
                    attempt,
                    max_retries,
                    retry_after,
                )
                await asyncio.sleep(retry_after)
                continue
            if response.status_code >= 500:
                if attempt > max_retries:
                    error_code, description = _error_from_body(response)
                    logger.error(
                        "tg client: %s — HTTP %d после %d попыток, сдаёмся",
                        method,
                        response.status_code,
                        attempt,
                    )
                    raise TelegramApiError(method, error_code, description)
                backoff = min(2.0**attempt, _MAX_BACKOFF_S)
                logger.warning(
                    "tg client: %s — HTTP %d (попытка %d/%d) — retry через %.0fs",
                    method,
                    response.status_code,
                    attempt,
                    max_retries,
                    backoff,
                )
                await asyncio.sleep(backoff)
                continue
            if response.status_code >= 400:
                # 4xx кроме 429 — запрос некорректен/прав нет, повтор не поможет.
                error_code, description = _error_from_body(response)
                raise TelegramApiError(method, error_code, description)
            try:
                data = response.json()
            except ValueError as exc:
                raise TelegramApiError(
                    method, response.status_code, f"invalid json: {exc}"
                ) from exc
            if not isinstance(data, dict) or not data.get("ok"):
                error_code, description = _error_from_body(response)
                raise TelegramApiError(method, error_code, description)
            return data.get("result")
    async def get_updates(
        self,
        offset: int,
        timeout: int = 30,
        allowed_updates: list[str] | None = None,
    ) -> list[dict[str, Any]]:
        """Long-polling getUpdates. `timeout` — сколько Telegram держит запрос открытым (сек).
        HTTP-таймаут запроса берётся с запасом (`timeout + 10s`), чтобы не обрывать
        соединение раньше, чем ответит сам Telegram long-poll.
        """
        payload: dict[str, Any] = {"offset": offset, "timeout": timeout}
        if allowed_updates is not None:
            payload["allowed_updates"] = allowed_updates
        result = await self._request(
            "getUpdates", payload, timeout=float(timeout) + 10.0, max_retries=3
        )
        return result if isinstance(result, list) else []
    async def copy_message(
        self,
        *,
        chat_id: int,
        from_chat_id: int,
        message_id: int,
        message_thread_id: int | None = None,
        reply_to_message_id: int | None = None,
    ) -> dict[str, Any]:
        """copyMessage — зеркалит ЛЮБОЙ тип контента без ре-аплоада файла."""
        payload: dict[str, Any] = {
            "chat_id": chat_id,
            "from_chat_id": from_chat_id,
            "message_id": message_id,
        }
        if message_thread_id:
            payload["message_thread_id"] = message_thread_id
        if reply_to_message_id:
            payload["reply_to_message_id"] = reply_to_message_id
        result = await self._request("copyMessage", payload)
        return result if isinstance(result, dict) else {}
    async def send_message(
        self,
        *,
        chat_id: int,
        text: str,
        message_thread_id: int | None = None,
        reply_to_message_id: int | None = None,
    ) -> dict[str, Any]:
        """sendMessage — текстовое сообщение (заголовки, приветствия, уведомления об ошибке)."""
        payload: dict[str, Any] = {"chat_id": chat_id, "text": text}
        if message_thread_id:
            payload["message_thread_id"] = message_thread_id
        if reply_to_message_id:
            payload["reply_to_message_id"] = reply_to_message_id
        result = await self._request("sendMessage", payload)
        return result if isinstance(result, dict) else {}
--- a/tradein-mvp/backend/app/tgbot_main.py
+++ b/tradein-mvp/backend/app/tgbot_main.py
@ -1,180 +0,0 @@
 """Standalone entrypoint для Telegram support-bridge воркера (#tgsupport).
 Зачем отдельный процесс/контейнер: `getUpdates` long-polling держит открытый
 HTTP-запрос к Telegram до 30с за раз в бесконечном цикле — деплой основного API
 (docker restart tradein-backend) не должен обрывать эту петлю на середине, как и
 API не должен блокироваться долгим poll'ом. Тот же паттерн, что и
 `scheduler_main.py` (#1182) для scraper'ов — отдельный контейнер с тем же образом,
 другая команда.
 Запуск: python -m app.tgbot_main
 Kill-switch: TELEGRAM_BOT_TOKEN пуст (дефолт) → воркер логирует «disabled» и
 блокируется на `wait_for_shutdown()` (idle, ~0 CPU) — НЕ `sys.exit(0)`. Сервис в
 compose поднят с `restart: unless-stopped`, который рестартует контейнер
 независимо от кода выхода — чистый exit(0) без токена дал бы бесконечный
 рестарт-луп. Idle-блокировка держит процесс живым (автозапуск после ребута VPS
 работает штатно через restart-policy) без CPU-луп и без спама рестартов;
 SIGTERM просто убивает процесс — восстанавливать здесь нечего (bridge-задача
 не запущена).
 """
 from __future__ import annotations
 import asyncio
 import logging
 import os
 import signal
 from contextlib import suppress
 from typing import Any
 from app.core.config import settings
 from app.core.db import SessionLocal
 from app.core.shutdown import request_shutdown, shutdown_requested, wait_for_shutdown
 from app.services.tgbot.bridge import run_poll_loop
 from app.services.tgbot.client import TelegramClient
 logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
 )
 # httpx INFO-логи печатают ПОЛНЫЙ request URL, включая Telegram Bot API токен
 # в пути (https://api.telegram.org/bot<id>:<secret>/...) — `httpx: HTTP Request:
 # POST https://api.telegram.org/bot<TOKEN>/getMe "HTTP/1.1 401 Unauthorized"`.
 # При бесконечном long-polling'e это боевой токен в `docker logs` каждые ~30с.
 # WARNING+ у httpx не логирует URL запроса (#tgsupport review, воспроизведено).
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logger = logging.getLogger(__name__)
 # Тот же safety-net паттерн, что scheduler_main.py — ниже docker stop_grace_period.
 _DRAIN_TIMEOUT_S = 100.0
 # Мониторинг ошибок — GlitchTip (Sentry-совместимый, #396). Только integrations
 # без Starlette/FastAPI — здесь нет ASGI-приложения (тот же выбор что scheduler_main).
 if settings.glitchtip_dsn:
    import sentry_sdk
    from sentry_sdk.integrations.httpx import HttpxIntegration
    from sentry_sdk.integrations.logging import LoggingIntegration
    from app.observability.sentry_scrub import redact_telegram_bot_token, scrub_pii_event
    def _before_send(event: Any, hint: dict[str, Any]) -> Any:
        """Композиция PII-scrub (form-данные) + Telegram bot-токен redaction
        (#tgsupport review). Токен утекает ДВУМЯ независимыми векторами, которые
        `include_local_variables=False` ниже и этот хук закрывают вместе:
          1. `include_local_variables=True` (sentry_sdk default) кладёт stack-frame
             locals (`self._base`/`url` в `TelegramClient._request`) в traceback —
             закрыто через `include_local_variables=False` в `sentry_sdk.init`.
          2. `HttpxIntegration` кладёт полный request URL в span `data` (не только
             traceback) — `traces_sample_rate=0.0` спасает СЕЙЧАС, но молча
             перестанет спасать, если трейсинг когда-нибудь включат. Regex-редактор
             — belt-and-suspenders на случай #1 (если include_local_variables
             случайно вернут) И на span data.
        """
        scrubbed = scrub_pii_event(event, hint)
        if scrubbed is None:
            return None
        return redact_telegram_bot_token(scrubbed, hint)
    sentry_sdk.init(
        dsn=settings.glitchtip_dsn,
        environment=settings.environment,
        release=os.getenv("GIT_SHA") or os.getenv("SENTRY_RELEASE") or "unknown",
        traces_sample_rate=0.0,
        send_default_pii=False,
        include_local_variables=False,
        before_send=_before_send,
        integrations=[
            HttpxIntegration(),
            LoggingIntegration(level=logging.INFO, event_level=logging.ERROR),
        ],
    )
    logger.info("GlitchTip monitoring enabled (tgbot_main)")
 def _should_run() -> bool:
    """Kill-switch: TELEGRAM_BOT_TOKEN не задан → бот выключен (dev/staging без секрета)."""
    return bool(settings.telegram_bot_token)
 async def _run_bridge() -> None:
    client = TelegramClient(settings.telegram_bot_token)
    await run_poll_loop(client, SessionLocal)
 async def _await_bridge(task: asyncio.Task[None]) -> None:
    """Кооперативный SIGTERM-drain — идентичная семантика scheduler_main._await_scheduler.
    `run_poll_loop` сам проверяет `shutdown_requested()` между итерациями (между
    getUpdates-вызовами) — long-polling запрос к Telegram (до 30с) докручивается,
    затем цикл выходит сам. Safety-net здесь на случай зависшего HTTP-вызова.
    """
    shutdown_waiter = asyncio.create_task(wait_for_shutdown())
    try:
        await asyncio.wait({task, shutdown_waiter}, return_when=asyncio.FIRST_COMPLETED)
    finally:
        shutdown_waiter.cancel()
        with suppress(asyncio.CancelledError):
            await shutdown_waiter
    if task.done():
        task.result()
        logger.info("tgbot_main: bridge task exited cleanly")
        return
    logger.info(
        "tgbot_main: SIGTERM-drain — waiting up to %.0fs for current poll iteration to finish",
        _DRAIN_TIMEOUT_S,
    )
    try:
        await asyncio.wait_for(task, timeout=_DRAIN_TIMEOUT_S)
        logger.info("tgbot_main: bridge drained and exited cleanly")
    except TimeoutError:
        logger.warning(
            "tgbot_main: drain exceeded %.0fs grace — hard-cancelling bridge task",
            _DRAIN_TIMEOUT_S,
        )
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task
 async def _run() -> None:
    task = asyncio.create_task(_run_bridge())
    loop = asyncio.get_running_loop()
    def _on_signal(signum: int) -> None:
        logger.info("tgbot_main: signal %d received — requesting cooperative drain", signum)
        request_shutdown()
    try:
        loop.add_signal_handler(signal.SIGTERM, lambda: _on_signal(signal.SIGTERM))
        loop.add_signal_handler(signal.SIGINT, lambda: _on_signal(signal.SIGINT))
    except NotImplementedError:
        # Windows dev: signal handlers через loop не поддерживаются
        logger.warning("tgbot_main: loop.add_signal_handler not supported (Windows dev)")
    await _await_bridge(task)
    if shutdown_requested():
        logger.info("tgbot_main: bridge drained cleanly (SIGTERM)")
    else:
        logger.info("tgbot_main: bridge task exited")
 if __name__ == "__main__":
    if not _should_run():
        # NOT sys.exit(0): compose service has `restart: unless-stopped`, который
        # рестартует контейнер независимо от кода выхода — чистый exit(0) без
        # токена дал бы бесконечный рестарт-луп. Idle-блокировка вместо этого:
        # ~0 CPU, SIGTERM просто убивает процесс (нечего дренировать).
        logger.warning(
            "tgbot_main: TELEGRAM_BOT_TOKEN не задан — бот выключен, "
            "блокируемся на idle (не exit, чтобы не было рестарт-лупа с restart:unless-stopped)"
        )
        asyncio.run(wait_for_shutdown())
    else:
        asyncio.run(_run())
--- a/tradein-mvp/backend/data/sql/186_tg_support.sql
+++ b/tradein-mvp/backend/data/sql/186_tg_support.sql
@ -1,93 +0,0 @@
 -- 186_tg_support.sql
 -- Telegram support bridge: @MERAsupport_bot mirrors client DMs into a
 -- support-group topic; an operator replies in-thread; the bot relays the
 -- reply back to the client's private chat.
 --
 -- WHY:
 --   No durable state existed for this flow. Two things are required to make
 --   it work reliably:
 --     1. A mapping from "message mirrored into the support topic" back to
 --        "which client chat_id it came from" — this is how an operator's
 --        reply (a Telegram reply-to a topic message) gets routed to the
 --        right client. `topic_message_id` on tg_support_messages is that
 --        routing key.
 --     2. A durable long-polling offset (`tg_support_state`) so a worker
 --        restart does not replay already-processed Telegram updates.
 --
 -- WHAT:
 --   - tg_support_users     — one row per client Telegram private chat
 --                             (chat_id is the Telegram chat id, stable per
 --                             client, used directly as PK — no surrogate key
 --                             needed).
 --   - tg_support_messages  — full conversation log, both directions.
 --   - tg_support_state     — singleton key/value store for worker offsets
 --                             (e.g. key='last_update_id').
 --
 -- 152-FZ:
 --   tg_support_users / tg_support_messages hold personal data (Telegram
 --   username/name + free-text conversation content). ON DELETE CASCADE from
 --   tg_support_users -> tg_support_messages makes client erasure a single
 --   `DELETE FROM tg_support_users WHERE chat_id = :chat_id` statement, no
 --   separate cleanup pass needed.
 --
 -- IDEMPOTENCY / SAFETY:
 --   CREATE TABLE IF NOT EXISTS + CREATE INDEX IF NOT EXISTS throughout —
 --   safe re-run. Purely additive: no existing table/view/column touched.
 --
 -- Dependencies: none (new standalone tables). Auto-applied on deploy via
 -- _schema_migrations tracking (tradein-mvp/backend/data/sql convention).
 BEGIN;
 CREATE TABLE IF NOT EXISTS tg_support_users (
    chat_id       bigint PRIMARY KEY,
    username      text,
    first_name    text,
    last_name     text,
    language_code text,
    created_at    timestamptz NOT NULL DEFAULT now(),
    last_seen_at  timestamptz NOT NULL DEFAULT now(),
    is_blocked    boolean NOT NULL DEFAULT false
 );
 COMMENT ON TABLE tg_support_users IS '152-ФЗ: ПДн клиентов Telegram-поддержки (@MERAsupport_bot). Удаление клиента — DELETE FROM tg_support_users WHERE chat_id=...; ON DELETE CASCADE в tg_support_messages подчищает переписку одной операцией.';
 COMMENT ON COLUMN tg_support_users.chat_id IS 'Telegram private chat id клиента (стабильный, используется как PK напрямую).';
 COMMENT ON COLUMN tg_support_users.is_blocked IS 'true, если клиент заблокировал бота (Telegram 403 на отправку) — бот перестаёт пытаться слать сообщения.';
 CREATE TABLE IF NOT EXISTS tg_support_messages (
    id               bigserial PRIMARY KEY,
    chat_id          bigint NOT NULL REFERENCES tg_support_users (chat_id) ON DELETE CASCADE,
    direction        text NOT NULL CHECK (direction IN ('in', 'out')),
    tg_message_id    bigint,
    topic_message_id bigint,
    kind             text NOT NULL,
    text_body        text,
    operator_tg_id   bigint,
    created_at       timestamptz NOT NULL DEFAULT now()
 );
 COMMENT ON TABLE tg_support_messages IS '152-ФЗ: полный лог переписки Telegram-поддержки (ПДн, содержимое сообщений). Каскадно удаляется вместе с tg_support_users по chat_id.';
 COMMENT ON COLUMN tg_support_messages.direction IS '''in'' — сообщение от клиента боту; ''out'' — ответ бота/оператора клиенту.';
 COMMENT ON COLUMN tg_support_messages.tg_message_id IS 'id сообщения в личном чате с клиентом (Telegram message_id в chat_id).';
 COMMENT ON COLUMN tg_support_messages.topic_message_id IS 'id зеркала сообщения в support-топике группы — ключ маршрутизации: реплай оператора на это сообщение адресуется данному chat_id.';
 COMMENT ON COLUMN tg_support_messages.kind IS 'text | photo | document | video | voice | other.';
 COMMENT ON COLUMN tg_support_messages.text_body IS 'Текст сообщения или caption медиа; NULL для медиа без подписи.';
 COMMENT ON COLUMN tg_support_messages.operator_tg_id IS 'Telegram user id оператора, ответившего в топике; заполняется только для direction=''out''.';
 CREATE UNIQUE INDEX IF NOT EXISTS tg_support_messages_topic_message_id_uq
    ON tg_support_messages (topic_message_id)
    WHERE topic_message_id IS NOT NULL;
 CREATE INDEX IF NOT EXISTS tg_support_messages_chat_id_created_at_idx
    ON tg_support_messages (chat_id, created_at DESC);
 CREATE TABLE IF NOT EXISTS tg_support_state (
    key        text PRIMARY KEY,
    value      text NOT NULL,
    updated_at timestamptz NOT NULL DEFAULT now()
 );
 COMMENT ON TABLE tg_support_state IS 'Singleton key/value store для состояния Telegram-поддержки (например last_update_id для long-polling), переживает рестарт воркера.';
 COMMENT ON COLUMN tg_support_state.key IS 'e.g. ''last_update_id''.';
 COMMIT;
--- a/tradein-mvp/backend/tests/services/tgbot/init.py
+++ b/tradein-mvp/backend/tests/services/tgbot/init.py
--- a/tradein-mvp/backend/tests/services/tgbot/test_bridge.py
+++ b/tradein-mvp/backend/tests/services/tgbot/test_bridge.py
@ -1,654 +0,0 @@
 """Unit tests for `app.services.tgbot.bridge` — чистая логика роутинга.
 Coverage (per task spec + review follow-up):
 - user → topic (личка клиента зеркалится в support-топик, с шапкой на первое
  сообщение за throttle-окно, без шапки на повторное В окне, и снова с шапкой
  после истечения окна — #6 review)
 - реплай оператора → user (доставка ответа клиенту + запись direction='out')
 - реплай не на зеркало (или не реплай вообще) — тихий игнор, не мусорим в чат;
  реплай на СООБЩЕНИЕ БОТА без записи в БД — WARNING про осиротевшее зеркало
  (#4 review)
 - дедуп update_id (<=offset — skip без side-effects; poison-pill апдейт всё
  равно сдвигает offset, чтобы не подвесить весь поток)
 - сбой БД (SQLAlchemyError) во время обработки → rollback() ПЕРЕД save_offset,
  offset всё равно сдвигается — без этого следующий поллинг переиграл бы тот
  же апдейт и задублировал зеркало клиента в топике (#3 review)
 - /start → приветствие без зеркалирования
 - Telegram 403 на доставку оператору → is_blocked + уведомление в топике
 - TELEGRAM_SUPPORT_CHAT_ID не задан → клиенту уходит "сервис недоступен"
  вместо тихой потери сообщения (#5 review)
 NEVER calls real Telegram API — все HTTP-запросы mock'аются через
 httpx.MockTransport (consistent с tests/services/test_dadata.py).
 """
 from __future__ import annotations
 import logging
 import os
 from typing import Any
 import httpx
 import pytest
 from sqlalchemy.exc import SQLAlchemyError
 # DATABASE_URL required by app.core.config before any app import (см. test_dadata.py).
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 from app.services.tgbot import bridge
 from app.services.tgbot.client import TelegramClient
 SUPPORT_CHAT_ID = -100123456789
 SUPPORT_TOPIC_ID = 42
 # ── Fake in-memory storage (БД не нужна) ─────────────────────────────────────
 class FakeBridgeStorage:
    """In-memory `BridgeStorage` — никакой реальной БД, чистая логика роутинга.
    `clock_s` — управляемые тестом "фейковые часы" (просто float, продвигается
    вручную через `storage.clock_s += ...`), чтобы честно проверить throttle-окно
    в `had_recent_inbound` (#6 review) без реального `time.sleep`/datetime-моков.
    `fail_next_record_message` — если True, следующий вызов `record_message`
    кидает `SQLAlchemyError` (симулирует обрыв коннекта к БД) и сбрасывается в
    False — для теста rollback-пути в `process_update` (#3 review).
    """
    def __init__(self, offset: int = 0) -> None:
        self._offset = offset
        self.users: dict[int, dict[str, Any]] = {}
        self.messages: list[dict[str, Any]] = []
        self.blocked: set[int] = set()
        self.commits = 0
        self.rollbacks = 0
        self._next_id = 1
        self.clock_s: float = 0.0
        self.fail_next_record_message = False
    def get_offset(self) -> int:
        return self._offset
    def save_offset(self, update_id: int) -> None:
        self._offset = update_id
    def commit(self) -> None:
        self.commits += 1
    def rollback(self) -> None:
        self.rollbacks += 1
    def upsert_user(
        self,
        *,
        chat_id: int,
        username: str | None,
        first_name: str | None,
        last_name: str | None,
        language_code: str | None,
    ) -> None:
        self.users[chat_id] = {
            "username": username,
            "first_name": first_name,
            "last_name": last_name,
            "language_code": language_code,
        }
    def had_recent_inbound(self, chat_id: int, window_seconds: int) -> bool:
        return any(
            m["chat_id"] == chat_id
            and m["direction"] == "in"
            and (self.clock_s - m["recorded_at_s"]) < window_seconds
            for m in self.messages
        )
    def record_message(
        self,
        *,
        chat_id: int,
        direction: str,
        tg_message_id: int | None,
        topic_message_id: int | None,
        kind: str,
        text_body: str | None,
        operator_tg_id: int | None,
    ) -> int:
        if self.fail_next_record_message:
            self.fail_next_record_message = False
            raise SQLAlchemyError("simulated DB failure (deploy connection reset)")
        row_id = self._next_id
        self._next_id += 1
        self.messages.append(
            {
                "id": row_id,
                "chat_id": chat_id,
                "direction": direction,
                "tg_message_id": tg_message_id,
                "topic_message_id": topic_message_id,
                "kind": kind,
                "text_body": text_body,
                "operator_tg_id": operator_tg_id,
                "recorded_at_s": self.clock_s,
            }
        )
        return row_id
    def find_chat_by_topic_message(self, topic_message_id: int) -> int | None:
        for m in reversed(self.messages):
            if m["direction"] == "in" and m["topic_message_id"] == topic_message_id:
                return m["chat_id"]
        return None
    def mark_blocked(self, chat_id: int) -> None:
        self.blocked.add(chat_id)
 # ── httpx mocking helpers (mirrors tests/services/test_dadata.py) ───────────
 _REAL_ASYNC_CLIENT = httpx.AsyncClient
 def _method_from_url(url: httpx.URL) -> str:
    return str(url).rsplit("/", 1)[-1]
 def _make_client(
    responses: dict[str, Any], calls: list[tuple[str, dict[str, Any]]]
 ) -> TelegramClient:
    """TelegramClient wired to a MockTransport. `responses[method]` may be a dict
    (returned as Bot API `result`), an int (HTTP error status), or a callable
    `(payload) -> dict`. Every request is recorded into `calls`."""
    def handler(request: httpx.Request) -> httpx.Response:
        import json as _json
        method = _method_from_url(request.url)
        payload = _json.loads(request.content.decode("utf-8")) if request.content else {}
        calls.append((method, payload))
        canned = responses.get(method)
        if isinstance(canned, int):
            return httpx.Response(
                canned, json={"ok": False, "error_code": canned, "description": "mocked error"}
            )
        if callable(canned):
            canned = canned(payload)
        result = canned if canned is not None else {"message_id": 999}
        return httpx.Response(200, json={"ok": True, "result": result})
    transport = httpx.MockTransport(handler)
    def factory(*_: object, **__: object) -> httpx.AsyncClient:
        return _REAL_ASYNC_CLIENT(transport=transport)
    client = TelegramClient(token="fake-token")
    import unittest.mock as mock
    # Патчим httpx.AsyncClient ТОЛЬКО внутри client-модуля — не трогаем глобальный httpx.
    patcher = mock.patch("app.services.tgbot.client.httpx.AsyncClient", factory)
    patcher.start()
    return client
@pytest.fixture(autouse=True)
 def _support_chat_settings(monkeypatch: pytest.MonkeyPatch) -> None:
    """Все тесты по умолчанию считают support-группу/топик настроенными."""
    monkeypatch.setattr(bridge.settings, "telegram_support_chat_id", SUPPORT_CHAT_ID)
    monkeypatch.setattr(bridge.settings, "telegram_support_topic_id", SUPPORT_TOPIC_ID)
@pytest.fixture(autouse=True)
 def _stop_patches():
    """Останавливает httpx.AsyncClient monkeypatch после каждого теста (unittest.mock.patch.start()
    без контекст-менеджера требует явного stop, чтобы не утекать в соседние тесты)."""
    import unittest.mock as mock
    yield
    mock.patch.stopall()
 def _private_message(
    *,
    message_id: int = 1,
    chat_id: int = 555,
    text: str | None = "Здравствуйте, вопрос по trade-in",
    username: str | None = "client_ivan",
    first_name: str | None = "Иван",
    last_name: str | None = "Петров",
 ) -> dict[str, Any]:
    msg: dict[str, Any] = {
        "message_id": message_id,
        "chat": {"id": chat_id, "type": "private"},
        "from": {
            "id": chat_id,
            "username": username,
            "first_name": first_name,
            "last_name": last_name,
            "language_code": "ru",
        },
    }
    if text is not None:
        msg["text"] = text
    return msg
 def _group_reply_message(
    *,
    message_id: int = 200,
    reply_to_message_id: int | None = 100,
    text: str = "Ответ оператора",
    operator_id: int = 777,
 ) -> dict[str, Any]:
    msg: dict[str, Any] = {
        "message_id": message_id,
        "chat": {"id": SUPPORT_CHAT_ID, "type": "supergroup"},
        "from": {"id": operator_id, "username": "operator1"},
        "text": text,
    }
    if reply_to_message_id is not None:
        msg["reply_to_message"] = {"message_id": reply_to_message_id}
    return msg
 # ── A) user → topic ──────────────────────────────────────────────────────────
 async def test_private_message_mirrors_to_topic_with_header_on_first_contact() -> None:
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({"copyMessage": {"message_id": 555}}, calls)
    storage = FakeBridgeStorage()
    update = {"update_id": 10, "message": _private_message()}
    await bridge.process_update(update, client, storage)
    methods = [m for m, _ in calls]
    # Первое сообщение за окно → шапка ПЕРЕД зеркалом контента.
    assert methods == ["sendMessage", "copyMessage"]
    header_call = calls[0][1]
    assert header_call["chat_id"] == SUPPORT_CHAT_ID
    assert header_call["message_thread_id"] == SUPPORT_TOPIC_ID
    assert "Иван Петров" in header_call["text"]
    assert "@client_ivan" in header_call["text"]
    mirror_call = calls[1][1]
    assert mirror_call["from_chat_id"] == 555
    assert mirror_call["chat_id"] == SUPPORT_CHAT_ID
    assert mirror_call["message_id"] == 1
    assert mirror_call["message_thread_id"] == SUPPORT_TOPIC_ID
    assert len(storage.messages) == 1
    rec = storage.messages[0]
    assert rec["direction"] == "in"
    assert rec["chat_id"] == 555
    assert rec["topic_message_id"] == 555  # copyMessage result.message_id
    assert rec["kind"] == "text"
    assert rec["text_body"] == "Здравствуйте, вопрос по trade-in"
    assert storage.get_offset() == 10
    assert storage.commits == 1
    assert 555 in storage.users
 async def test_private_message_second_message_within_window_skips_header() -> None:
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({"copyMessage": {"message_id": 556}}, calls)
    storage = FakeBridgeStorage()
    # Симулируем уже существующее inbound-сообщение за последний час.
    storage.record_message(
        chat_id=555,
        direction="in",
        tg_message_id=0,
        topic_message_id=100,
        kind="text",
        text_body="первое сообщение",
        operator_tg_id=None,
    )
    update = {"update_id": 11, "message": _private_message(message_id=2)}
    await bridge.process_update(update, client, storage)
    methods = [m for m, _ in calls]
    # Шапка НЕ отправляется повторно — только зеркало.
    assert methods == ["copyMessage"]
    assert len(storage.messages) == 2
 async def test_private_message_header_resent_after_window_expires() -> None:
    """#6 review: throttle-окно (3600с) реально проверяется по времени — после
    истечения окна шапка отправляется заново (не одна на весь чат навсегда)."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({"copyMessage": {"message_id": 557}}, calls)
    storage = FakeBridgeStorage()
    storage.record_message(
        chat_id=555,
        direction="in",
        tg_message_id=0,
        topic_message_id=100,
        kind="text",
        text_body="первое сообщение (час назад)",
        operator_tg_id=None,
    )
    # Продвигаем фейковые часы за throttle-окно (3600с).
    storage.clock_s += bridge._HEADER_THROTTLE_WINDOW_S + 1
    update = {"update_id": 13, "message": _private_message(message_id=3)}
    await bridge.process_update(update, client, storage)
    methods = [m for m, _ in calls]
    assert methods == ["sendMessage", "copyMessage"]  # шапка снова отправлена
    assert len(storage.messages) == 2
 async def test_private_message_start_sends_greeting_without_mirroring() -> None:
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    update = {"update_id": 12, "message": _private_message(text="/start")}
    await bridge.process_update(update, client, storage)
    methods = [m for m, _ in calls]
    assert methods == ["sendMessage"]
    greeting_call = calls[0][1]
    assert greeting_call["chat_id"] == 555
    assert "МЕРА" in greeting_call["text"]
    # /start не зеркалируется и не попадает в лог переписки.
    assert storage.messages == []
    assert storage.get_offset() == 12
 async def test_private_message_notifies_client_when_support_chat_unset(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """#5 review: TELEGRAM_SUPPORT_CHAT_ID не задан → клиент получает "сервис
    недоступен" вместо того, чтобы молча ждать ответа, который никогда не придёт."""
    monkeypatch.setattr(bridge.settings, "telegram_support_chat_id", 0)
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    update = {"update_id": 14, "message": _private_message()}
    await bridge.process_update(update, client, storage)
    methods = [m for m, _ in calls]
    assert methods == ["sendMessage"]
    notify_call = calls[0][1]
    assert notify_call["chat_id"] == 555
    assert notify_call["text"] == bridge.SERVICE_UNAVAILABLE_TEXT
    # Ничего не зеркалируется и не пишется в лог переписки — support-группа не настроена.
    assert storage.messages == []
    assert storage.get_offset() == 14
 # ── B) реплай оператора → user ──────────────────────────────────────────────
 async def test_group_reply_delivers_to_client_and_records_outbound() -> None:
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({"copyMessage": {"message_id": 42}}, calls)
    storage = FakeBridgeStorage()
    storage.record_message(
        chat_id=555,
        direction="in",
        tg_message_id=1,
        topic_message_id=100,
        kind="text",
        text_body="вопрос клиента",
        operator_tg_id=None,
    )
    update = {
        "update_id": 20,
        "message": _group_reply_message(reply_to_message_id=100),
    }
    await bridge.process_update(update, client, storage)
    methods = [m for m, _ in calls]
    assert methods == ["copyMessage"]
    delivery = calls[0][1]
    assert delivery["chat_id"] == 555
    assert delivery["from_chat_id"] == SUPPORT_CHAT_ID
    assert len(storage.messages) == 2
    out_rec = storage.messages[-1]
    assert out_rec["direction"] == "out"
    assert out_rec["chat_id"] == 555
    assert out_rec["operator_tg_id"] == 777
    assert out_rec["text_body"] == "Ответ оператора"
    assert storage.get_offset() == 20
 async def test_group_reply_not_a_reply_is_ignored() -> None:
    """Обычное сообщение в топике (не реплай) — тихий игнор, никаких Telegram-вызовов."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    update = {
        "update_id": 21,
        "message": _group_reply_message(reply_to_message_id=None),
    }
    await bridge.process_update(update, client, storage)
    assert calls == []
    assert storage.messages == []
    # offset всё равно сдвигается — апдейт "обработан" (даже если ничего не сделано).
    assert storage.get_offset() == 21
 async def test_group_reply_to_unknown_message_is_ignored() -> None:
    """Реплай на сообщение, которого нет в tg_support_messages как зеркало клиента —
    тихий игнор (обычная болтовня в топике на постороннее сообщение, не от бота)."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    update = {
        "update_id": 22,
        "message": _group_reply_message(reply_to_message_id=999999),
    }
    await bridge.process_update(update, client, storage)
    assert calls == []
    assert storage.messages == []
    assert storage.get_offset() == 22
 async def test_group_reply_to_bot_message_without_record_logs_orphaned_mirror_warning(
    caplog: pytest.LogCaptureFixture,
 ) -> None:
    """#4 review: реплай на сообщение БОТА, которого нет в tg_support_messages, —
    вероятное осиротевшее зеркало (крах между copyMessage и commit). WARNING, не
    тихий игнор — оператор иначе решит, что ответ клиенту доставлен."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    message = _group_reply_message(reply_to_message_id=100)
    message["reply_to_message"]["from"] = {"id": 999, "is_bot": True, "username": "MERAsupport_bot"}
    update = {"update_id": 24, "message": message}
    with caplog.at_level(logging.WARNING, logger="app.services.tgbot.bridge"):
        await bridge.process_update(update, client, storage)
    assert calls == []  # ответ НЕ доставлен — routing-ключ потерян
    assert "осиротевшее" in caplog.text
    assert storage.get_offset() == 24
 async def test_group_reply_to_non_bot_message_without_record_stays_silent(
    caplog: pytest.LogCaptureFixture,
 ) -> None:
    """Обычный реплай на сообщение ДРУГОГО ЧЕЛОВЕКА (не бота) в топике — реальная
    болтовня, никакого WARNING (дискриминатор `is_bot` работает в обе стороны)."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    message = _group_reply_message(reply_to_message_id=101)
    message["reply_to_message"]["from"] = {"id": 42, "is_bot": False, "username": "colleague"}
    update = {"update_id": 25, "message": message}
    with caplog.at_level(logging.WARNING, logger="app.services.tgbot.bridge"):
        await bridge.process_update(update, client, storage)
    assert calls == []
    assert caplog.text == ""
 async def test_group_reply_403_marks_blocked_and_notifies_topic() -> None:
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({"copyMessage": 403}, calls)
    storage = FakeBridgeStorage()
    storage.record_message(
        chat_id=555,
        direction="in",
        tg_message_id=1,
        topic_message_id=100,
        kind="text",
        text_body="вопрос клиента",
        operator_tg_id=None,
    )
    update = {
        "update_id": 23,
        "message": _group_reply_message(reply_to_message_id=100),
    }
    await bridge.process_update(update, client, storage)
    methods = [m for m, _ in calls]
    assert methods == ["copyMessage", "sendMessage"]
    notify_call = calls[1][1]
    assert notify_call["chat_id"] == SUPPORT_CHAT_ID
    assert "заблокирован" in notify_call["text"]
    assert 555 in storage.blocked
    # Неудачная доставка НЕ должна создавать фейковую запись 'out'.
    assert len(storage.messages) == 1
    assert storage.get_offset() == 23
 # ── C) дедуп ──────────────────────────────────────────────────────────────────
 async def test_dedup_update_id_leq_offset_is_skipped_without_side_effects() -> None:
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage(offset=50)
    update = {"update_id": 50, "message": _private_message()}
    await bridge.process_update(update, client, storage)
    assert calls == []
    assert storage.messages == []
    assert storage.commits == 0  # ранний return — offset уже актуален, коммитить нечего
    assert storage.get_offset() == 50
    update_older = {"update_id": 10, "message": _private_message()}
    await bridge.process_update(update_older, client, storage)
    assert calls == []
    assert storage.get_offset() == 50
 async def test_poison_pill_update_still_advances_offset() -> None:
    """Апдейт, на котором обработчик упал (например, message без chat), не должен
    подвесить весь поток — offset сдвигается даже при исключении внутри handler'а."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    malformed_message = {"message_id": 1, "chat": {"type": "private"}}  # нет chat.id
    update = {"update_id": 30, "message": malformed_message}
    await bridge.process_update(update, client, storage)
    assert storage.get_offset() == 30
    assert storage.commits == 1
    assert storage.messages == []
 async def test_db_error_during_processing_rolls_back_and_still_advances_offset() -> None:
    """#3 review: SQLAlchemyError (напр. обрыв коннекта к БД при деплое) во время
    `record_message` → storage.rollback() ПЕРЕД save_offset, offset всё равно
    сдвигается. Без rollback() save_offset сам кинул бы PendingRollbackError →
    process_update вылетел бы без сохранения offset'а → следующая итерация
    переиграла бы тот же апдейт → copyMessage задублировал бы зеркало в топике
    на каждый повтор поллинга (воспроизведено ревьюером)."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({"copyMessage": {"message_id": 558}}, calls)
    storage = FakeBridgeStorage()
    storage.fail_next_record_message = True
    update = {"update_id": 15, "message": _private_message()}
    await bridge.process_update(update, client, storage)
    # copyMessage успел уйти в Telegram (реальная утечка мирроринга при DB-сбое —
    # известное ограничение атомарности между внешним API и БД, вне scope этого фикса),
    # но rollback() отработал, offset сдвинут, commit вызван РОВНО один раз (в finally).
    assert storage.rollbacks == 1
    assert storage.commits == 1
    assert storage.get_offset() == 15
    # Запись сообщения НЕ попала в storage (record_message упал до append).
    assert storage.messages == []
    # Повторный вызов с тем же update_id теперь корректно дедупится — НЕ переигрывается.
    calls.clear()
    await bridge.process_update(update, client, storage)
    assert calls == []
    assert storage.get_offset() == 15
 async def test_update_without_update_id_is_ignored() -> None:
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    await bridge.process_update({"message": _private_message()}, client, storage)
    assert calls == []
    assert storage.commits == 0
    assert storage.get_offset() == 0
 # ── kind inference ────────────────────────────────────────────────────────────
@pytest.mark.parametrize(
    ("message_extra", "expected_kind"),
    [
        ({"text": "hi"}, "text"),
        ({"photo": [{"file_id": "x"}]}, "photo"),
        ({"document": {"file_id": "x"}}, "document"),
        ({"video": {"file_id": "x"}}, "video"),
        ({"voice": {"file_id": "x"}}, "voice"),
        ({"sticker": {"file_id": "x"}}, "other"),
        ({"location": {"latitude": 1, "longitude": 2}}, "other"),
        ({}, "other"),
    ],
 )
 def test_infer_kind(message_extra: dict[str, Any], expected_kind: str) -> None:
    message = {"message_id": 1, "chat": {"id": 1, "type": "private"}, **message_extra}
    assert bridge._infer_kind(message) == expected_kind
 # ── unrelated chat types ─────────────────────────────────────────────────────
 async def test_update_from_unrelated_chat_is_ignored_but_offset_advances() -> None:
    """Апдейт не из личного чата и не из support-группы (например, другой чат/канал)
    — молча игнорируется, offset всё равно сдвигается."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    message = {
        "message_id": 1,
        "chat": {"id": -999, "type": "group"},
        "from": {"id": 1},
        "text": "болтовня в постороннем чате",
    }
    update = {"update_id": 40, "message": message}
    await bridge.process_update(update, client, storage)
    assert calls == []
    assert storage.messages == []
    assert storage.get_offset() == 40
--- a/tradein-mvp/backend/tests/services/tgbot/test_client.py
+++ b/tradein-mvp/backend/tests/services/tgbot/test_client.py
@ -1,164 +0,0 @@
 """Unit tests for `app.services.tgbot.client.TelegramClient` retry/backoff logic.
 NEVER calls real Telegram API — httpx.MockTransport only (consistent с
 tests/services/test_dadata.py). `asyncio.sleep` is patched to a no-op so retry
 tests run instantly regardless of configured backoff/retry_after durations.
 """
 from __future__ import annotations
 import os
 from typing import Any
 from unittest import mock
 import httpx
 import pytest
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 from app.services.tgbot.client import TelegramApiError, TelegramClient
 _REAL_ASYNC_CLIENT = httpx.AsyncClient
 def _install_transport(handler) -> None:
    transport = httpx.MockTransport(handler)
    def factory(*_: object, **__: object) -> httpx.AsyncClient:
        return _REAL_ASYNC_CLIENT(transport=transport)
    mock.patch("app.services.tgbot.client.httpx.AsyncClient", factory).start()
@pytest.fixture(autouse=True)
 def _stop_patches_and_noop_sleep():
    sleep_patcher = mock.patch("app.services.tgbot.client.asyncio.sleep", return_value=None)
    sleep_patcher.start()
    yield
    mock.patch.stopall()
 async def test_get_updates_happy_path_returns_list() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path.endswith("/getUpdates")
        return httpx.Response(200, json={"ok": True, "result": [{"update_id": 1}]})
    _install_transport(handler)
    client = TelegramClient(token="fake-token")
    updates = await client.get_updates(offset=1)
    assert updates == [{"update_id": 1}]
 async def test_never_logs_or_leaks_token_in_request_url_host() -> None:
    """Sanity: token lives only in the path, base host stays api.telegram.org."""
    captured: dict[str, str] = {}
    def handler(request: httpx.Request) -> httpx.Response:
        captured["url"] = str(request.url)
        return httpx.Response(200, json={"ok": True, "result": {}})
    _install_transport(handler)
    client = TelegramClient(token="super-secret-token")
    await client.send_message(chat_id=1, text="hi")
    assert "bot" + "super-secret-token" in captured["url"]  # goes over the wire, not logged
 async def test_copy_message_retries_on_429_then_succeeds() -> None:
    calls = {"n": 0}
    def handler(request: httpx.Request) -> httpx.Response:
        calls["n"] += 1
        if calls["n"] == 1:
            return httpx.Response(
                429,
                json={
                    "ok": False,
                    "error_code": 429,
                    "description": "Too Many Requests",
                    "parameters": {"retry_after": 3},
                },
            )
        return httpx.Response(200, json={"ok": True, "result": {"message_id": 5}})
    _install_transport(handler)
    client = TelegramClient(token="fake-token")
    result = await client.copy_message(chat_id=1, from_chat_id=2, message_id=3)
    assert result == {"message_id": 5}
    assert calls["n"] == 2
 async def test_send_message_retries_on_5xx_then_succeeds() -> None:
    calls = {"n": 0}
    def handler(request: httpx.Request) -> httpx.Response:
        calls["n"] += 1
        if calls["n"] < 3:
            return httpx.Response(
                502, json={"ok": False, "error_code": 502, "description": "bad gw"}
            )
        return httpx.Response(200, json={"ok": True, "result": {"message_id": 9}})
    _install_transport(handler)
    client = TelegramClient(token="fake-token")
    result = await client.send_message(chat_id=1, text="retrying")
    assert result == {"message_id": 9}
    assert calls["n"] == 3
 async def test_send_message_raises_immediately_on_non_retryable_4xx() -> None:
    calls = {"n": 0}
    def handler(request: httpx.Request) -> httpx.Response:
        calls["n"] += 1
        return httpx.Response(
            403, json={"ok": False, "error_code": 403, "description": "Forbidden: bot blocked"}
        )
    _install_transport(handler)
    client = TelegramClient(token="fake-token")
    with pytest.raises(TelegramApiError) as exc_info:
        await client.send_message(chat_id=1, text="hi")
    assert exc_info.value.error_code == 403
    assert calls["n"] == 1  # НЕ ретраится
 async def test_copy_message_gives_up_after_max_retries_on_persistent_5xx() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(500, json={"ok": False, "error_code": 500, "description": "boom"})
    _install_transport(handler)
    client = TelegramClient(token="fake-token")
    with pytest.raises(TelegramApiError) as exc_info:
        await client.copy_message(chat_id=1, from_chat_id=2, message_id=3)
    assert exc_info.value.error_code == 500
 async def test_get_updates_returns_empty_list_on_malformed_result() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"ok": True, "result": "not-a-list"})
    _install_transport(handler)
    client = TelegramClient(token="fake-token")
    assert await client.get_updates(offset=1) == []
 async def test_optional_thread_and_reply_params_omitted_when_falsy() -> None:
    captured: dict[str, Any] = {}
    def handler(request: httpx.Request) -> httpx.Response:
        import json as _json
        captured["body"] = _json.loads(request.content.decode("utf-8"))
        return httpx.Response(200, json={"ok": True, "result": {"message_id": 1}})
    _install_transport(handler)
    client = TelegramClient(token="fake-token")
    await client.copy_message(chat_id=1, from_chat_id=2, message_id=3, message_thread_id=0)
    assert "message_thread_id" not in captured["body"]
--- a/tradein-mvp/backend/tests/test_sentry_scrub.py
+++ b/tradein-mvp/backend/tests/test_sentry_scrub.py
@ -1,21 +1,9 @@
-"""Unit-тесты consumer-PII scrubber для GlitchTip (#396) + Telegram bot-токен
+"""Unit-тесты consumer-PII scrubber для GlitchTip (#396)."""
 redaction (#tgsupport review — CRITICAL: токен утекал в GlitchTip двумя
 независимыми векторами, воспроизведёнными ревьюером на реальном событии
 sentry_sdk: stack-frame locals (`include_local_variables=True`) и httpx-span
 `data` (`HttpxIntegration`). Тесты ниже конструируют event-словари ровно в той
 форме, в которой их реально отдаёт sentry_sdk (frames[].vars, spans[].data),
 чтобы regression на этот конкретный CRITICAL не проскочил молча."""
 import os
 import pytest
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
-from app.observability.sentry_scrub import (
+from app.observability.sentry_scrub import scrub_pii_event
    redact_telegram_bot_token,
    scrub_pii_event,
 )
 def test_redacts_pii_in_request_data() -> None:
@ -100,162 +88,3 @@ def test_returns_event_not_none() -> None:
    out = scrub_pii_event(event, {})
    assert out is not None
    assert out is event
 # ── Telegram bot-token redaction (#tgsupport review, CRITICAL) ──────────────
 _LEAKED_TOKEN_URL = "https://api.telegram.org/bot8663867262:AAExampleSecretPartAbCdEf123/getMe"
 def test_redacts_token_in_stack_frame_locals() -> None:
    """Вектор #1 (ревьюер): `include_local_variables=True` кладёт locals
    `TelegramClient._request` (`self`, `url`) в traceback frame `vars`."""
    event = {
        "exception": {
            "values": [
                {
                    "type": "NetworkError",
                    "stacktrace": {
                        "frames": [
                            {
                                "function": "_request",
                                "vars": {
                                    "url": _LEAKED_TOKEN_URL,
                                    "self": {"_base": _LEAKED_TOKEN_URL.rsplit("/", 1)[0]},
                                    "method": "getMe",
                                },
                            }
                        ]
                    },
                }
            ]
        }
    }
    out = redact_telegram_bot_token(event, {})
    assert out is not None
    frame_vars = out["exception"]["values"][0]["stacktrace"]["frames"][0]["vars"]
    assert "8663867262:AAExampleSecretPartAbCdEf123" not in frame_vars["url"]
    assert "8663867262:AAExampleSecretPartAbCdEf123" not in frame_vars["self"]["_base"]
    assert frame_vars["url"] == "https://api.telegram.org/bot[REDACTED]/getMe"
    # Не PII/секрет — остаётся как есть.
    assert frame_vars["method"] == "getMe"
 def test_redacts_token_in_httpx_span_data() -> None:
    """Вектор #2 (ревьюер): `HttpxIntegration` кладёт полный request URL в span
    `data` независимо от traceback — `traces_sample_rate=0.0` спасает сейчас, но
    редактор — belt-and-suspenders на случай если трейсинг когда-нибудь включат."""
    event = {
        "spans": [
            {
                "op": "http.client",
                "description": "POST " + _LEAKED_TOKEN_URL,
                "data": {"url": _LEAKED_TOKEN_URL, "http.method": "POST"},
            }
        ]
    }
    out = redact_telegram_bot_token(event, {})
    assert out is not None
    span = out["spans"][0]
    assert "8663867262:AAExampleSecretPartAbCdEf123" not in span["data"]["url"]
    assert "8663867262:AAExampleSecretPartAbCdEf123" not in span["description"]
    assert span["data"]["http.method"] == "POST"
 def test_redacts_token_in_breadcrumb_message() -> None:
    """httpx-логгер (INFO, до нашего getLogger('httpx').setLevel(WARNING) в
    tgbot_main) может всплыть breadcrumb'ом с полным URL через LoggingIntegration."""
    event = {
        "breadcrumbs": {
            "values": [
                {
                    "category": "httpx",
                    "message": f'HTTP Request: POST {_LEAKED_TOKEN_URL} "HTTP/1.1 401"',
                }
            ]
        }
    }
    out = redact_telegram_bot_token(event, {})
    assert out is not None
    msg = out["breadcrumbs"]["values"][0]["message"]
    assert "8663867262:AAExampleSecretPartAbCdEf123" not in msg
    assert "/bot[REDACTED]/getMe" in msg
 def test_redact_token_leaves_unrelated_urls_and_strings_untouched() -> None:
    event = {"extra": {"public_url": "https://gendsgn.ru/trade-in", "region": "66"}}
    out = redact_telegram_bot_token(event, {})
    assert out is not None
    assert out["extra"]["public_url"] == "https://gendsgn.ru/trade-in"
    assert out["extra"]["region"] == "66"
 def test_redact_token_handles_non_dict_event() -> None:
    assert redact_telegram_bot_token(None, {}) is None  # type: ignore[arg-type]
 # Голая форма `<id>:<secret>` без префикса `/bot` — так токен выглядит в локали
 # `token` конструктора TelegramClient. Штатно в event не попадает
 # (include_local_variables=False в tgbot_main), но редактор обязан покрывать и
 # этот случай: иначе рубеж против утечки токена всего один — флаг SDK.
 _LEAKED_TOKEN_BARE = "8663867262:AAFakeSecretPartForTestsOnly1234567"
 def test_redacts_bare_token_without_bot_prefix() -> None:
    event = {
        "logentry": {"message": f"init failed for {_LEAKED_TOKEN_BARE}"},
        "exception": {
            "values": [{"stacktrace": {"frames": [{"vars": {"token": _LEAKED_TOKEN_BARE}}]}}]
        },
    }
    out = redact_telegram_bot_token(event, {})
    assert out is not None
    assert "AAFakeSecretPartForTestsOnly1234567" not in repr(out)
    frame = out["exception"]["values"][0]["stacktrace"]["frames"][0]
    assert frame["vars"]["token"] == "[REDACTED]"
@pytest.mark.parametrize(
    "benign",
    [
        "chat_id:12345",
        "ratio 3:4",
        "2026-07-16T10:00:00",
        "postgresql://user:pass@postgres:5432/tradein",
    ],
 )
 def test_bare_token_redaction_leaves_benign_colon_strings_untouched(benign: str) -> None:
    """Голая регулярка не должна бить по любым `x:y` — иначе диагностика ослепнет."""
    out = redact_telegram_bot_token({"logentry": {"message": benign}}, {})
    assert out is not None
    assert out["logentry"]["message"] == benign
 def test_composed_before_send_scrubs_pii_and_token_together() -> None:
    """Композиция, реально используемая в `app.tgbot_main._before_send`: PII-scrub
    (ключ-based) И token-redaction (regex full-text) применяются оба, не заменяя
    друг друга — разные классы секретов, разные механизмы обнаружения."""
    event = {
        "request": {"data": {"client_phone": "+79991234567"}},
        "exception": {
            "values": [
                {
                    "stacktrace": {
                        "frames": [{"vars": {"url": _LEAKED_TOKEN_URL}}],
                    }
                }
            ]
        },
    }
    def composed_before_send(evt, hint):
        scrubbed = scrub_pii_event(evt, hint)
        if scrubbed is None:
            return None
        return redact_telegram_bot_token(scrubbed, hint)
    out = composed_before_send(event, {})
    assert out is not None
    assert out["request"]["data"]["client_phone"] == "[REDACTED]"
    frame_url = out["exception"]["values"][0]["stacktrace"]["frames"][0]["vars"]["url"]
    assert "8663867262:AAExampleSecretPartAbCdEf123" not in frame_url
--- a/tradein-mvp/docker-compose.prod.yml
+++ b/tradein-mvp/docker-compose.prod.yml
@ -201,49 +201,6 @@ services:
    networks:
      - tradein-net
  # Telegram support-bot bridge: long-polling worker (asyncio/httpx), тот же
  # backend-образ что и scraper/backend (один Dockerfile, другая команда).
  # Ничего не слушает (long-polling исходящий к Telegram API) → без expose/ports.
  # НЕ подписан на gendesign_shared — не проксируется Caddy, только tradein-net
  # (нужен доступ к postgres). TELEGRAM_* переменные — ТОЛЬКО из backend/.env.runtime
  # (env_file ниже); пусто/нет токена = бот молча не стартует (см. tgbot_main.py).
  # ⚠️ НЕ дублировать TELEGRAM_* в блоке environment: (см. предупреждение у backend
  # выше про environment: перекрывающий env_file при пустой host-env).
  tgbot:
    image: ghcr.io/lekss361/gendesign-tradein-backend:${IMAGE_TAG:-latest}
    container_name: tradein-tgbot
    # mem: нет прод-замера (новый сервис, 2026-07-16) — консервативный старт по
    # аналогии с профилем scraper (лёгкая asyncio-оркестрация, без headless-браузера
    # и без больших payload'ов). Пересмотреть после первого docker stats на проде.
    mem_limit: 256m
    memswap_limit: 256m
    logging: *default-logging
    command: ["python", "-m", "app.tgbot_main"]
    env_file:
      - path: ./backend/.env.runtime
        required: false
    environment:
      DATABASE_URL: "postgresql+psycopg://${TRADEIN_POSTGRES_USER:-tradein}:${TRADEIN_POSTGRES_PASSWORD}@postgres:5432/tradein"
      ENVIRONMENT: "production"
      GLITCHTIP_DSN: "${GLITCHTIP_DSN:-}"
      # Этот процесс — не scheduler_main; false на всякий случай, если общий
      # Settings()-объект образа где-то читает флаг при импорте (defense-in-depth,
      # аналогично backend). tgbot_main.py не должен зависеть от этого значения.
      SCHEDULER_ENABLE: "false"
    depends_on:
      postgres:
        condition: service_healthy
    # unless-stopped (как и весь остальной стек): только always/unless-stopped
    # гарантируют автозапуск после ребута VPS. Kill-switch при пустом
    # TELEGRAM_BOT_TOKEN реализован idle-блокировкой в tgbot_main.py, а НЕ exit(0) —
    # иначе эта политика дала бы рестарт-луп (перезапускает независимо от кода).
    restart: unless-stopped
    # >_DRAIN_TIMEOUT_S (100s) в tgbot_main.py, иначе docker убьёт по дефолтным 10с
    # раньше, чем докрутится long-poll (до 30с), и кооперативный drain не сработает.
    stop_grace_period: 120s
    networks:
      - tradein-net
  frontend:
    image: ghcr.io/lekss361/gendesign-tradein-frontend:${IMAGE_TAG:-latest}
    container_name: tradein-frontend