"""Веб-чат поддержки (#tgsupport-web) — поверх уже существующего Telegram
support-моста (`app.services.tgbot.bridge`, data/sql/186_tg_support.sql).

Источник обращения — сайт (не Telegram-личка клиента): пользователь пишет через
это API, сообщение зеркалится `sendMessage`-ом в тот же support-топик, оператор
отвечает РЕПЛАЕМ ровно так же, как на Telegram-клиента — маршрутизация ответа
обратно реализована в `bridge._handle_group_reply` (ветка добавлена там же, без
изменения существующего Telegram-пути).

Изоляция тредов: все 4 ручки резолвят тред ИСКЛЮЧИТЕЛЬНО по `X-Authenticated-User`
(rbac_guard в app/main.py гарантирует его наличие и валидность для non-public
путей). thread_id НИКОГДА не принимается снаружи (ни в query, ни в body) — чужой
тред прочитать/отметить нельзя ни при каких параметрах запроса, потому что
параметра, которым можно было бы адресовать чужой тред, попросту не существует.

Копия зеркала в топике всегда помечена "[С САЙТА] <username>: ..." — оператор
не должен путать веб-обращение с Telegram-клиентом (#tgsupport-web AC).

КРИТИЧНО (review H1) — порядок операций в `send_support_message`:
  БД-запись (`get_or_create_thread`) идёт ПОСЛЕ успешного `send_message`, не до.
  Прод — один uvicorn-процесс БЕЗ `--workers` (docker-compose.prod.yml) с
  синхронным SQLAlchemy engine (пул 5+10 overflow) на ОДНОМ event loop. Если бы
  `INSERT ... ON CONFLICT DO UPDATE` уходил ДО Telegram-вызова, строка/row-lock
  держались бы всё время, пока `send_message` ждёт Telegram (секунды-минуты при
  429/5xx на воркерных ретраях) — второй параллельный запрос ТОГО ЖЕ юзера
  (двойной клик, вторая вкладка) упёрся бы в этот lock ВНУТРИ синхронного
  psycopg-вызова внутри `async def`, останавливая event loop целиком (весь API
  встаёт, не только этот эндпоинт). `_format_mirror_text` использует только
  `username` — thread_id для отправки не нужен вообще, поэтому эту БД-операцию
  можно безопасно отложить до после успешного sendMessage. Бонус: неудачная
  отправка больше не создаёт тред.

Анонимная ветка (`/support/anon/*`, инцидент 2026-07-31)
-------------------------------------------------------
Ровно те же 4 действия, но БЕЗ авторизации — доступны с экрана входа. Причина:
после cutover'а на свою авторизацию (#2558) единственным каналом в поддержку был
чат ЗА логином, а самая частая причина писать в поддержку — как раз «не могу
войти». 2026-07-31 «Практика» весь день билась в форму (5 login_failed, 0
успешных) и достучаться из продукта не могла ничем.

Идентичность анонима — opaque-токен в httpOnly-куке (`_ANON_COOKIE_NAME`),
тред живёт в тех же `web_support_threads` под ключом `anon:<token>`. Двоеточие
делает коллизию с реальным логином структурно невозможной: `tradein_users`
допускает только `^[A-Za-z0-9._-]{3,64}$` (CHECK из миграции 193 + Pydantic),
двоеточия там быть не может — аноним НИКОГДА не попадёт в чужой тред и не
«станет» существующим юзером.

Изоляция тредов та же, что у авторизованной ветки, и по той же причине:
thread_id не принимается снаружи ни в каком виде, тред резолвится
ИСКЛЮЧИТЕЛЬНО из куки. Кука здесь — bearer-токен своего треда, поэтому
httpOnly+Secure+SameSite=Lax (как session-cookie) и `token_urlsafe(18)`
(144 бита) вместо чего-то угадываемого.

В Telegram-топик уходит НЕ сам токен, а `anon-<6 hex от sha256(токен)>`
(`_anon_display_id`): оператору нужен стабильный ярлык треда, а не bearer —
зеркало топика читают люди и пересылают дальше.
"""

from __future__ import annotations

import hashlib
import logging
import re
import secrets
from typing import Annotated, Literal

from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
from pydantic import BaseModel, Field, field_validator
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.db import get_db
from app.core.ratelimit import SlidingWindowLimiter, _client_ip
from app.services.tgbot import web_support_storage as storage
from app.services.tgbot.bridge import SERVICE_UNAVAILABLE_TEXT
from app.services.tgbot.client import TelegramApiError, TelegramClient

logger = logging.getLogger(__name__)

router = APIRouter()

# Лимит Telegram sendMessage (4096) с запасом — см. #tgsupport-web AC ("~4000").
MAX_MESSAGE_LENGTH = 4000

# Жёстче общего RateLimitMiddleware (300 req/60с на пользователя, app/main.py):
# бот-токен общий на ВСЕХ клиентов веб-чата, флуд одного клиента иначе может
# упереться в Telegram-лимиты (`sendMessage` 429, лимит группы ~20 msg/min) и
# застопорить доставку всем остальным (см. задачу, п.6). 12 сообщений/минуту —
# щедро для живого диалога, но режет скрипт-флуд на порядок раньше общего API-лимита.
_SEND_RATE_LIMIT = 12
_SEND_RATE_WINDOW_S = 60.0
_send_limiter = SlidingWindowLimiter(limit=_SEND_RATE_LIMIT, window_s=_SEND_RATE_WINDOW_S)

# #tgsupport-web review H1: интерактивный HTTP-запрос НЕ МОЖЕТ наследовать
# воркерную политику ретраев `TelegramClient` (по умолчанию — до 5 попыток, на
# 429 спит `retry_after` Telegram'а — для группы штатно 30-60с, на 5xx backoff до
# 30с — легальный суммарный бюджет минуты). Узкий бюджет здесь: 1 повтор, короткий
# timeout — интерактивный клиент должен получить ответ (даже если это ошибка)
# за секунды, а не висеть до исчерпания воркерных ретраев.
_INTERACTIVE_SEND_TIMEOUT_S = 10.0
_INTERACTIVE_SEND_MAX_RETRIES = 1

# #tgsupport-web review M5: без LIMIT каждое монтирование виджета на старом
# треде отдавало бы ВЕСЬ лог переписки. См. `web_support_storage.list_messages`.
_LIST_MESSAGES_LIMIT = 200


def _require_username(request: Request) -> str:
    """Достаёт X-Authenticated-User. rbac_guard (app/main.py) уже гарантирует его
    наличие в проде для non-public путей — этот guard здесь defence-in-depth и
    делает роутер тестируемым без поднятия всего app.main (см. tests/test_support.py,
    как test_trade_in_lead.py для /lead). `.strip()` (review L4) — лишний пробел
    от прокси иначе завёл бы ВТОРОЙ тред на, по сути, того же пользователя
    (username — UNIQUE ключ треда, "alice" != "alice ")."""
    username = (request.headers.get("x-authenticated-user") or "").strip()
    if not username:
        raise HTTPException(status_code=401, detail="no authenticated user")
    return username


def _bot_configured() -> bool:
    """TELEGRAM_BOT_TOKEN и TELEGRAM_SUPPORT_CHAT_ID оба обязательны — без них
    зеркалировать в топик некуда (см. app/tgbot_main.py._should_run для токена,
    bridge.py для chat_id)."""
    return bool(settings.telegram_bot_token) and bool(settings.telegram_support_chat_id)


class SupportMessageInput(BaseModel):
    text: str = Field(min_length=1, max_length=MAX_MESSAGE_LENGTH)

    @field_validator("text")
    @classmethod
    def _not_blank(cls, value: str) -> str:
        stripped = value.strip()
        if not stripped:
            raise ValueError("text must not be blank")
        return stripped


class SupportMessageOut(BaseModel):
    id: int
    direction: Literal["in", "out"]
    text_body: str
    operator_tg_id: int | None = None
    created_at: str

    @field_validator("created_at", mode="before")
    @classmethod
    def _isoformat(cls, value: object) -> str:
        if hasattr(value, "isoformat"):
            return value.isoformat()  # type: ignore[no-any-return]
        return str(value)


class UnreadOut(BaseModel):
    unread: int


class StatusOut(BaseModel):
    status: Literal["ok"] = "ok"


def _format_mirror_text(username: str, message_text: str) -> str:
    """Помечает зеркало как пришедшее С САЙТА, от какого пользователя — оператор
    иначе не отличит веб-обращение от Telegram-клиента (#tgsupport-web AC).
    Использует ТОЛЬКО username — thread_id здесь не нужен (см. H1 в docstring
    модуля), это то, что делает возможным отложить БД-запись до после отправки."""
    return f"[С САЙТА] {username}:\n{message_text}"


@router.post("/support/messages", response_model=SupportMessageOut)
async def send_support_message(
    payload: SupportMessageInput,
    username: Annotated[str, Depends(_require_username)],
    db: Annotated[Session, Depends(get_db)],
) -> SupportMessageOut:
    """Отправляет сообщение от лица *username* в support-топик (`sendMessage` —
    не `copyMessage`: у веб-сообщения нет исходного Telegram-сообщения для копии).

    Порядок операций см. H1 в docstring модуля: rate-limit проверяется (но НЕ
    расходуется, review L3) до отправки, thread создаётся ТОЛЬКО после успешного
    `send_message` — до этого момента с БД не происходит ничего.
    """
    if not _bot_configured():
        # Предсказуемое поведение вместо 500 (#tgsupport-web AC): бот не настроен
        # (пустой TELEGRAM_BOT_TOKEN, dev/staging) или support-топик не задан —
        # мирроринг невозможен физически, ничего не пишем в БД.
        raise HTTPException(status_code=503, detail=SERVICE_UNAVAILABLE_TEXT)

    # review L3: peek без расхода бюджета — неудачная отправка НЕ должна стоить
    # пользователю попытки (расходуем `.record()` только на успех, ниже).
    retry_after = _send_limiter.retry_after(username)
    if retry_after is not None:
        raise HTTPException(
            status_code=429,
            detail="Слишком много сообщений. Попробуйте через минуту.",
            headers={"Retry-After": str(int(retry_after) + 1)},
        )

    client = TelegramClient(settings.telegram_bot_token)
    try:
        mirrored = await client.send_message(
            chat_id=settings.telegram_support_chat_id,
            text=_format_mirror_text(username, payload.text),
            message_thread_id=settings.telegram_support_topic_id or None,
            # review H1: узкий интерактивный бюджет — НЕ воркерные 5 ретраев/минуты.
            timeout=_INTERACTIVE_SEND_TIMEOUT_S,
            max_retries=_INTERACTIVE_SEND_MAX_RETRIES,
        )
    except TelegramApiError:
        # НЕ логируем payload.text (переписка — ПДн) и НЕ логируем токен (его в
        # TelegramApiError и не бывает — см. client.py docstring про redaction).
        logger.exception(
            "web support: не удалось отправить зеркало в топик (username=%s)", username
        )
        raise HTTPException(status_code=502, detail=SERVICE_UNAVAILABLE_TEXT) from None

    # Отправка удалась — теперь и только теперь расходуем rate-limit бюджет.
    _send_limiter.record(username)

    topic_message_id = mirrored.get("message_id") if isinstance(mirrored, dict) else None
    if topic_message_id is None:
        # review L1: без topic_message_id реплай оператора на это сообщение
        # НИКОГДА не смаршрутизируется обратно (find_thread_by_topic_message ищет
        # именно по этому полю) — тихая, но зафиксированная в логе деградация.
        logger.warning(
            "web support: Telegram sendMessage не вернул message_id (username=%s) — "
            "ответ оператора на это сообщение не будет смаршрутизирован",
            username,
        )

    # review H1: БД-операция ПОСЛЕ успешной отправки — см. docstring модуля.
    thread_id = storage.get_or_create_thread(db, username)
    row = storage.record_inbound(
        db,
        thread_id=thread_id,
        text_body=payload.text,
        topic_message_id=topic_message_id,
        support_chat_id=settings.telegram_support_chat_id,
    )
    db.commit()

    logger.info("web support: message sent username=%s thread_id=%d", username, thread_id)
    return SupportMessageOut(**row)


@router.get("/support/messages", response_model=list[SupportMessageOut])
def list_support_messages(
    username: Annotated[str, Depends(_require_username)],
    db: Annotated[Session, Depends(get_db)],
    since: Annotated[int, Query(ge=0)] = 0,
) -> list[SupportMessageOut]:
    """Сообщения СВОЕГО треда с id > since. Тред резолвится по username — чужой
    тред недостижим (нет параметра, которым его можно адресовать).

    Обычный (sync) `def`, не `async def` (review M3): тело — только синхронные
    psycopg-вызовы, ни одного `await`; как `async def` это исполнялось бы прямо в
    event loop (а фронт поллит эту ручку постоянно). Starlette гонит sync-handlers
    в threadpool автоматически — тот же паттерн, что `trade_in.py:get_estimate`.
    """
    thread_id = storage.find_thread_id(db, username)
    if thread_id is None:
        return []
    rows = storage.list_messages(
        db, thread_id=thread_id, since_id=since, limit=_LIST_MESSAGES_LIMIT
    )
    return [SupportMessageOut(**r) for r in rows]


@router.get("/support/unread", response_model=UnreadOut)
def get_support_unread(
    username: Annotated[str, Depends(_require_username)],
    db: Annotated[Session, Depends(get_db)],
) -> UnreadOut:
    """Sync `def` (review M3) — см. `list_support_messages`."""
    thread_id = storage.find_thread_id(db, username)
    if thread_id is None:
        return UnreadOut(unread=0)
    return UnreadOut(unread=storage.count_unread(db, thread_id=thread_id))


@router.post("/support/read", response_model=StatusOut)
def mark_support_read(
    username: Annotated[str, Depends(_require_username)],
    db: Annotated[Session, Depends(get_db)],
) -> StatusOut:
    """Sync `def` (review M3) — см. `list_support_messages`."""
    thread_id = storage.find_thread_id(db, username)
    if thread_id is not None:
        storage.mark_read(db, thread_id=thread_id)
        db.commit()
    return StatusOut()


# ---------------------------------------------------------------------------
# Анонимная ветка — поддержка без входа (см. блок в докстринге модуля)
# ---------------------------------------------------------------------------

_ANON_COOKIE_NAME = "tradein_support_anon"
# 30 дней: тред должен пережить «напишу вечером — отвечут утром», но не жить вечно.
_ANON_COOKIE_MAX_AGE_S = 30 * 24 * 3600
# Двоеточие → структурная невозможность коллизии с реальным логином (докстринг).
_ANON_THREAD_PREFIX = "anon:"
# Форма того, что МЫ выдаём (`token_urlsafe(18)` → 24 символа из [A-Za-z0-9_-]).
# Кука клиент-контролируема: без этой проверки в ключ треда (а значит в SQL-параметр
# и в лог) уехала бы произвольная строка из браузера. Не матчится — считаем куку
# отсутствующей и выдаём новую, а не пытаемся «починить» присланное.
_ANON_TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]{16,64}\Z")

# Публичная ручка записи в общий Telegram-топик — поверхность для спама, которой у
# авторизованной ветки нет. Два независимых бюджета:
#   1) per-token (`_send_limiter`, 12/мин — тот же объект, ключи не пересекаются:
#      анонимные начинаются с "anon:", что невозможно для username);
#   2) per-IP — именно он ловит обход ротацией куки (сбросил куку → новый токен →
#      бюджет (1) снова пуст). Окно широкое и щедрое для живого диалога: реальный
#      сценарий — «не могу войти, помогите», несколько сообщений подряд.
_ANON_IP_RATE_LIMIT = 10
_ANON_IP_RATE_WINDOW_S = 600.0
_anon_ip_limiter = SlidingWindowLimiter(limit=_ANON_IP_RATE_LIMIT, window_s=_ANON_IP_RATE_WINDOW_S)


def _anon_display_id(token: str) -> str:
    """Стабильный НЕсекретный ярлык треда для оператора — см. докстринг модуля.

    sha256, а не префикс токена: префикс — это часть bearer'а, а зеркало уходит
    в Telegram-топик, который читают люди и пересылают дальше.
    """
    return f"anon-{hashlib.sha256(token.encode('utf-8')).hexdigest()[:6]}"


def _read_anon_token(request: Request) -> str | None:
    """Токен из куки, если он валидной формы; иначе None (кука считается отсутствующей)."""
    raw = request.cookies.get(_ANON_COOKIE_NAME)
    if raw is None or not _ANON_TOKEN_RE.match(raw):
        return None
    return raw


def _anon_thread_key(token: str) -> str:
    return f"{_ANON_THREAD_PREFIX}{token}"


def _set_anon_cookie(response: Response, token: str) -> None:
    response.set_cookie(
        key=_ANON_COOKIE_NAME,
        value=token,
        max_age=_ANON_COOKIE_MAX_AGE_S,
        httponly=True,
        secure=True,
        samesite="lax",
        path="/",
    )


@router.post("/support/anon/messages", response_model=SupportMessageOut)
async def send_anon_support_message(
    payload: SupportMessageInput,
    request: Request,
    response: Response,
    db: Annotated[Session, Depends(get_db)],
) -> SupportMessageOut:
    """Сообщение в поддержку БЕЗ входа. Порядок операций — как в авторизованной
    ветке (H1 в докстринге модуля): БД трогаем только после успешного sendMessage.

    Кука выставляется тоже только на успехе — иначе первая же неудачная попытка
    (бот не настроен / Telegram лёг) закрепляла бы за посетителем пустой тред.
    """
    if not _bot_configured():
        raise HTTPException(status_code=503, detail=SERVICE_UNAVAILABLE_TEXT)

    token = _read_anon_token(request)
    is_new_token = token is None
    if token is None:
        token = secrets.token_urlsafe(18)
    thread_key = _anon_thread_key(token)
    ip = _client_ip(request)

    # Оба бюджета — non-destructive peek (review L3): неудачная отправка не
    # должна стоить посетителю попытки. `.record()` только на успех, ниже.
    for retry_after in (_send_limiter.retry_after(thread_key), _anon_ip_limiter.retry_after(ip)):
        if retry_after is not None:
            raise HTTPException(
                status_code=429,
                detail="Слишком много сообщений. Попробуйте позже.",
                headers={"Retry-After": str(int(retry_after) + 1)},
            )

    display_id = _anon_display_id(token)
    client = TelegramClient(settings.telegram_bot_token)
    try:
        mirrored = await client.send_message(
            chat_id=settings.telegram_support_chat_id,
            text=_format_anon_mirror_text(display_id, payload.text),
            message_thread_id=settings.telegram_support_topic_id or None,
            timeout=_INTERACTIVE_SEND_TIMEOUT_S,
            max_retries=_INTERACTIVE_SEND_MAX_RETRIES,
        )
    except TelegramApiError:
        # Ни текст сообщения (ПДн), ни токен (bearer треда) в лог не попадают.
        logger.exception(
            "web support (anon): не удалось отправить зеркало в топик (%s)", display_id
        )
        raise HTTPException(status_code=502, detail=SERVICE_UNAVAILABLE_TEXT) from None

    _send_limiter.record(thread_key)
    _anon_ip_limiter.record(ip)

    topic_message_id = mirrored.get("message_id") if isinstance(mirrored, dict) else None
    if topic_message_id is None:
        logger.warning(
            "web support (anon): Telegram sendMessage не вернул message_id (%s) — "
            "ответ оператора на это сообщение не будет смаршрутизирован",
            display_id,
        )

    thread_id = storage.get_or_create_thread(db, thread_key)
    row = storage.record_inbound(
        db,
        thread_id=thread_id,
        text_body=payload.text,
        topic_message_id=topic_message_id,
        support_chat_id=settings.telegram_support_chat_id,
    )
    db.commit()

    if is_new_token:
        _set_anon_cookie(response, token)
    logger.info("web support (anon): message sent %s thread_id=%d", display_id, thread_id)
    return SupportMessageOut(**row)


def _format_anon_mirror_text(display_id: str, message_text: str) -> str:
    """Помечает зеркало как пришедшее с сайта ОТ НЕЗАЛОГИНЕННОГО посетителя.

    Оператору это ключевой контекст: у такого обращения нет аккаунта, по которому
    можно посмотреть историю, и самая вероятная причина написать — как раз
    невозможность войти (инцидент 2026-07-31).
    """
    return f"[С САЙТА · БЕЗ ВХОДА] {display_id}:\n{message_text}"


@router.get("/support/anon/messages", response_model=list[SupportMessageOut])
def list_anon_support_messages(
    request: Request,
    db: Annotated[Session, Depends(get_db)],
    since: Annotated[int, Query(ge=0)] = 0,
) -> list[SupportMessageOut]:
    """Свой тред по куке. Нет куки / нет треда → пустой список, НЕ 401: виджет
    поллит эту ручку и до первого сообщения, 401 там был бы ложной ошибкой.

    Sync `def` (review M3) — см. `list_support_messages`.
    """
    token = _read_anon_token(request)
    if token is None:
        return []
    thread_id = storage.find_thread_id(db, _anon_thread_key(token))
    if thread_id is None:
        return []
    rows = storage.list_messages(
        db, thread_id=thread_id, since_id=since, limit=_LIST_MESSAGES_LIMIT
    )
    return [SupportMessageOut(**r) for r in rows]


@router.get("/support/anon/unread", response_model=UnreadOut)
def get_anon_support_unread(
    request: Request,
    db: Annotated[Session, Depends(get_db)],
) -> UnreadOut:
    """Sync `def` (review M3) — см. `list_support_messages`."""
    token = _read_anon_token(request)
    if token is None:
        return UnreadOut(unread=0)
    thread_id = storage.find_thread_id(db, _anon_thread_key(token))
    if thread_id is None:
        return UnreadOut(unread=0)
    return UnreadOut(unread=storage.count_unread(db, thread_id=thread_id))


@router.post("/support/anon/read", response_model=StatusOut)
def mark_anon_support_read(
    request: Request,
    db: Annotated[Session, Depends(get_db)],
) -> StatusOut:
    """Sync `def` (review M3) — см. `list_support_messages`."""
    token = _read_anon_token(request)
    if token is None:
        return StatusOut()
    thread_id = storage.find_thread_id(db, _anon_thread_key(token))
    if thread_id is not None:
        storage.mark_read(db, thread_id=thread_id)
        db.commit()
    return StatusOut()