gendesign/tradein-mvp/backend/tests/services/tgbot/test_bridge.py

"""Unit tests for `app.services.tgbot.bridge` — чистая логика роутинга.

Coverage (per task spec + review follow-up):
- user → topic (личка клиента зеркалится в support-топик, с шапкой на первое
  сообщение за throttle-окно, без шапки на повторное В окне, и снова с шапкой
  после истечения окна — #6 review)
- реплай оператора → user (доставка ответа клиенту + запись direction='out')
- реплай оператора → веб-чат (#tgsupport-web): зеркало веб-сообщения резолвится
  в web-тред, ответ пишется direction='out' БЕЗ Telegram-доставки; медиа-реплай
  на веб-зеркало (нет text/caption) — тихий игнор (веб-чат текстовый MVP);
  tg-резолв имеет приоритет над веб-резолвом при (искусственной) коллизии
- реплай не на зеркало (или не реплай вообще) — тихий игнор, не мусорим в чат;
  реплай на СООБЩЕНИЕ БОТА без записи в БД — 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
        # #tgsupport-web: web_support_messages-эквивалент (topic_message_id ->
        # thread_id) + журнал outbound-записей, записанных через реплай оператора.
        self.web_topic_to_thread: dict[int, int] = {}
        self.web_out_messages: list[dict[str, Any]] = []

    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)

    # ── #tgsupport-web ────────────────────────────────────────────────────
    def find_web_thread_by_topic_message(self, topic_message_id: int) -> int | None:
        return self.web_topic_to_thread.get(topic_message_id)

    def record_web_out_message(
        self, *, thread_id: int, text_body: str, operator_tg_id: int | None
    ) -> None:
        self.web_out_messages.append(
            {
                "thread_id": thread_id,
                "text_body": text_body,
                "operator_tg_id": operator_tg_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


# ── B') реплай оператора → веб-чат (#tgsupport-web) ──────────────────────────


async def test_group_reply_to_web_mirror_records_outbound_web_message() -> None:
    """Реплай на зеркало веб-сообщения (не найдено в tg_support_messages, найдено
    среди web_support_messages) → записывается в веб-тред, БЕЗ Telegram-доставки
    (у веб-клиента нет личного чата с ботом)."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    storage.web_topic_to_thread[300] = 42  # topic_message_id=300 -> thread_id=42

    update = {
        "update_id": 60,
        "message": _group_reply_message(reply_to_message_id=300, text="Ответ по веб-чату"),
    }
    await bridge.process_update(update, client, storage)

    # Никакого Telegram API вызова — веб-клиент не имеет личного чата с ботом.
    assert calls == []
    assert len(storage.web_out_messages) == 1
    rec = storage.web_out_messages[0]
    assert rec["thread_id"] == 42
    assert rec["text_body"] == "Ответ по веб-чату"
    assert rec["operator_tg_id"] == 777
    # tg-путь тоже не тронут — ни одной записи в tg_support_messages.
    assert storage.messages == []
    assert storage.get_offset() == 60


async def test_group_reply_to_web_mirror_without_text_is_ignored(
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Веб-чат — текстовый MVP: реплай медиа-типом (нет text/caption) на веб-зеркало
    не может быть доставлен — тихий (WARNING, не error) игнор, ничего не пишем."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({}, calls)
    storage = FakeBridgeStorage()
    storage.web_topic_to_thread[301] = 43

    message = _group_reply_message(reply_to_message_id=301)
    del message["text"]  # медиа-реплай без текста/caption
    update = {"update_id": 61, "message": message}

    with caplog.at_level(logging.WARNING, logger="app.services.tgbot.bridge"):
        await bridge.process_update(update, client, storage)

    assert calls == []
    assert storage.web_out_messages == []
    assert "текстовый" in caplog.text
    assert storage.get_offset() == 61


async def test_group_reply_prefers_tg_thread_when_both_would_match() -> None:
    """Приоритет резолва — tg СНАЧАЛА: если topic_message_id найден среди
    tg_support_messages, веб-резолв даже не вызывается (существующий Telegram-путь
    работает как раньше, не деградирует из-за новой ветки)."""
    calls: list[tuple[str, dict[str, Any]]] = []
    client = _make_client({"copyMessage": {"message_id": 999}}, calls)
    storage = FakeBridgeStorage()
    storage.record_message(
        chat_id=555,
        direction="in",
        tg_message_id=1,
        topic_message_id=400,
        kind="text",
        text_body="вопрос клиента",
        operator_tg_id=None,
    )
    # Тот же topic_message_id "случайно" тоже был бы в web-мапе — не должен
    # переопределять tg-резолв (defensive, в реальности Telegram message_id не
    # повторяется в пределах чата).
    storage.web_topic_to_thread[400] = 99

    update = {
        "update_id": 62,
        "message": _group_reply_message(reply_to_message_id=400),
    }
    await bridge.process_update(update, client, storage)

    methods = [m for m, _ in calls]
    assert methods == ["copyMessage"]
    assert storage.web_out_messages == []
    out_rec = storage.messages[-1]
    assert out_rec["direction"] == "out"
    assert out_rec["chat_id"] == 555


# ── 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