vidconf/backend/tests/test_chat_ws.py

"""Тесты WS-чата конференции (`WS /api/v1/conferences/{id}/chat`).

Интеграционные тесты используют `ws_client` — in-process ASGI websocket
клиент (см. `tests/conftest.py::ASGIWebSocketSession`), т.к. ни
`httpx.AsyncClient`, ни `starlette.testclient.TestClient` не подходят для
websocket-тестов поверх нашей savepoint-сессии БД (см. докстринг класса).
"""

import uuid
from collections.abc import Callable
from datetime import UTC, datetime
from typing import Any

import httpx
import jwt
import pytest
from pydantic import ValidationError
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from core.redis import redis_client
from core.security import create_access_token, hash_password
from models.chat import ChatMessage
from models.conference import Conference
from models.guest import GuestAccess
from models.session import ConferenceSession
from models.user import User
from repositories.chat import ChatMessageRepository
from repositories.conferences import ConferenceSessionRepository
from schemas.chat import ChatMessageIn, ChatMessageOut
from services.chat import ChatService, chat_channel
from services.conference_ids import generate_number, generate_slug
from services.instance_settings import InstanceSettingsService, SettingsUpdateIn
from services.livekit_tokens import create_room_access_token
from tests.conftest import ASGIWebSocketSession

WSFactory = Callable[[str], ASGIWebSocketSession]


# --- Хелперы ---------------------------------------------------------------


async def _make_user(session: AsyncSession, *, name: str = "Chat Tester") -> User:
    user = User(
        email=f"{uuid.uuid4()}@example.com",
        name_user=name,
        password_hash=await hash_password("password123"),
        email_verified=True,
    )
    session.add(user)
    await session.flush()
    return user


async def _make_conference(session: AsyncSession, *, status: str = "active") -> Conference:
    conference = Conference(
        number=generate_number(), slug=generate_slug(), title="Chat Test", status=status
    )
    session.add(conference)
    await session.flush()
    return conference


async def _make_guest(session: AsyncSession, conference: Conference, *, name: str) -> GuestAccess:
    guest = GuestAccess(conference_id=conference.id, display_name=name)
    session.add(guest)
    await session.flush()
    return guest


def _chat_path(conference_id: uuid.UUID) -> str:
    return f"/api/v1/conferences/{conference_id}/chat"


def _user_token(conference: Conference, user: User) -> str:
    return create_room_access_token(
        room_name=conference.slug, identity=str(user.id), name=user.name_user
    )


def _guest_token(conference: Conference, guest: GuestAccess) -> str:
    return create_room_access_token(
        room_name=conference.slug, identity=f"guest:{guest.id}", name=guest.display_name
    )


async def _connect_and_auth(session: ASGIWebSocketSession, token: str) -> dict[str, Any]:
    """Подключиться, аутентифицироваться и вернуть первое сообщение (`history`).

    После `history` сервер сразу шлёт снапшот очереди поднятых рук
    (`{"type":"hand_queue",...}`, задача B1) — здесь он молча вычитывается
    и отбрасывается, чтобы не путать существующие тесты чата, которым он
    не интересен (см. `tests/test_hand_queue_ws.py` для тестов самой очереди).
    """
    accept = await session.connect()
    assert accept["type"] == "websocket.accept"
    await session.send_json({"type": "auth", "token": token})
    history = await session.receive_json()
    await session.receive_json()
    return history


# --- Основной сценарий: обмен сообщениями + история -------------------------


async def test_two_clients_exchange_messages_and_history_on_reconnect(
    db_session: AsyncSession, ws_client: WSFactory
) -> None:
    conference = await _make_conference(db_session)
    user1 = await _make_user(db_session, name="Alice")
    user2 = await _make_user(db_session, name="Bob")
    await db_session.commit()

    token1 = _user_token(conference, user1)
    token2 = _user_token(conference, user2)
    path = _chat_path(conference.id)

    ws1 = ws_client(path)
    history1 = await _connect_and_auth(ws1, token1)
    assert history1 == {"type": "history", "messages": []}

    ws2 = ws_client(path)
    history2 = await _connect_and_auth(ws2, token2)
    assert history2 == {"type": "history", "messages": []}

    await ws1.send_json({"type": "message", "text": "hello from alice"})

    echo = await ws1.receive_json()
    assert echo["type"] == "message"
    assert echo["message"]["text"] == "hello from alice"
    assert echo["message"]["author_name"] == "Alice"
    assert echo["message"]["is_guest"] is False
    assert echo["message"]["author_id"] == str(user1.id)
    assert echo["message"]["created_at"].endswith("Z")

    broadcast = await ws2.receive_json()
    assert broadcast == echo

    await ws1.aclose()
    await ws2.aclose()

    ws3 = ws_client(path)
    history3 = await _connect_and_auth(ws3, _user_token(conference, user1))
    assert len(history3["messages"]) == 1
    assert history3["messages"][0]["text"] == "hello from alice"


# --- Гонка history vs pubsub subscribe --------------------------------------


async def test_subscribe_happens_before_history_query(
    db_session: AsyncSession, ws_client: WSFactory, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Regression: подписка на pub/sub должна происходить ДО SELECT истории.

    Иначе сообщение, опубликованное другим клиентом в этом окне, теряется
    для подключающегося клиента.
    """
    conference = await _make_conference(db_session)
    user = await _make_user(db_session)
    await db_session.commit()

    call_order: list[str] = []

    original_history = ChatService.history

    async def _tracked_history(self: ChatService, conference: Conference) -> list[ChatMessageOut]:
        call_order.append("history")
        return await original_history(self, conference)

    monkeypatch.setattr(ChatService, "history", _tracked_history)

    import redis.asyncio.client as redis_client_module

    original_subscribe = redis_client_module.PubSub.subscribe

    async def _tracked_subscribe(self: Any, *args: Any, **kwargs: Any) -> Any:
        call_order.append("subscribe")
        return await original_subscribe(self, *args, **kwargs)

    monkeypatch.setattr(redis_client_module.PubSub, "subscribe", _tracked_subscribe)

    token = _user_token(conference, user)
    ws = ws_client(_chat_path(conference.id))
    await _connect_and_auth(ws, token)

    assert call_order == ["subscribe", "history"]


async def test_no_duplicate_when_message_already_in_history(
    db_session: AsyncSession, ws_client: WSFactory
) -> None:
    """Сообщение, уже попавшее в history, не должно продублироваться через pub/sub.

    Имитирует стык гонки: то же сообщение (тот же `id`) публикуется в канал
    уже ПОСЛЕ того, как клиент получил его в `history` — дедуп по `id`
    (`_pump_pubsub_to_websocket`) должен его отфильтровать.
    """
    conference = await _make_conference(db_session)
    user = await _make_user(db_session, name="Eve")
    await db_session.commit()

    session_record = await ConferenceSessionRepository(db_session).create(
        conference_id=conference.id, title=conference.title, t_start=datetime.now(UTC)
    )
    existing = await ChatMessageRepository(db_session).add(
        session_id=session_record.id,
        user_id=user.id,
        guest_access_id=None,
        author_name="Eve",
        text="already there",
    )
    await db_session.commit()

    token = _user_token(conference, user)
    ws = ws_client(_chat_path(conference.id))
    history = await _connect_and_auth(ws, token)
    assert len(history["messages"]) == 1
    assert history["messages"][0]["id"] == existing.id

    duplicate_payload = ChatMessageOut(
        id=existing.id,
        author_id=str(user.id),
        author_name="Eve",
        is_guest=False,
        text="already there",
        created_at=existing.created_at,
    )
    await redis_client.publish(chat_channel(conference.id), duplicate_payload.model_dump_json())

    # Настоящее новое сообщение отправляем следом — оно должно дойти БЕЗ
    # опережающего дубликата (если бы дедуп не работал, первым пришёл бы
    # повтор "already there").
    await ws.send_json({"type": "message", "text": "genuinely new"})
    received = await ws.receive_json()
    assert received["type"] == "message"
    assert received["message"]["text"] == "genuinely new"


# --- Auth: коды закрытия ----------------------------------------------------


async def test_no_auth_message_closes_4401(db_session: AsyncSession, ws_client: WSFactory) -> None:
    conference = await _make_conference(db_session)
    await db_session.commit()

    ws = ws_client(_chat_path(conference.id))
    accept = await ws.connect()
    assert accept["type"] == "websocket.accept"
    # Первым сообщением шлём НЕ auth (нарушение протокола) — трактуется как отсутствие auth.
    await ws.send_json({"type": "message", "text": "too early"})
    code = await ws.receive_close()
    assert code == 4401


async def test_garbage_token_closes_4401(db_session: AsyncSession, ws_client: WSFactory) -> None:
    conference = await _make_conference(db_session)
    await db_session.commit()

    ws = ws_client(_chat_path(conference.id))
    await ws.connect()
    await ws.send_json({"type": "auth", "token": "not-a-jwt-at-all"})
    code = await ws.receive_close()
    assert code == 4401


async def test_non_livekit_jwt_closes_4401(db_session: AsyncSession, ws_client: WSFactory) -> None:
    conference = await _make_conference(db_session)
    await db_session.commit()

    # Синтаксически валидный JWT, но подписан не тем секретом (не LiveKit).
    foreign_token = jwt.encode(
        {"sub": "someone"}, "wrong-secret-not-livekit-at-all-32-bytes", algorithm="HS256"
    )

    ws = ws_client(_chat_path(conference.id))
    await ws.connect()
    await ws.send_json({"type": "auth", "token": foreign_token})
    code = await ws.receive_close()
    assert code == 4401


async def test_wrong_room_token_closes_4403(db_session: AsyncSession, ws_client: WSFactory) -> None:
    conference_a = await _make_conference(db_session)
    conference_b = await _make_conference(db_session)
    user = await _make_user(db_session)
    await db_session.commit()

    # Токен выдан для конференции A, подключаемся к B.
    token_for_a = _user_token(conference_a, user)

    ws = ws_client(_chat_path(conference_b.id))
    await ws.connect()
    await ws.send_json({"type": "auth", "token": token_for_a})
    code = await ws.receive_close()
    assert code == 4403


async def test_chat_disabled_closes_4404_without_restart(
    db_session: AsyncSession, ws_client: WSFactory
) -> None:
    conference = await _make_conference(db_session)
    user = await _make_user(db_session)
    await db_session.commit()

    token = _user_token(conference, user)
    path = _chat_path(conference.id)

    # До выключения тоггла подключение работает штатно.
    ws1 = ws_client(path)
    history = await _connect_and_auth(ws1, token)
    assert history["type"] == "history"
    await ws1.aclose()

    await InstanceSettingsService(db_session).update(SettingsUpdateIn(chat_enabled=False))

    # Тоггл действует немедленно, без рестарта backend — следующее подключение отклоняется.
    ws2 = ws_client(path)
    await ws2.connect()
    await ws2.send_json({"type": "auth", "token": token})
    code = await ws2.receive_close()
    assert code == 4404


async def test_conference_not_found_closes_4404(
    db_session: AsyncSession, ws_client: WSFactory
) -> None:
    conference = await _make_conference(db_session)
    user = await _make_user(db_session)
    await db_session.commit()

    # Токен валиден (комната существует), но подключаемся по чужому/несуществующему id.
    token = _user_token(conference, user)
    ws = ws_client(_chat_path(uuid.uuid4()))
    await ws.connect()
    await ws.send_json({"type": "auth", "token": token})
    code = await ws.receive_close()
    assert code == 4404


# --- Сообщение после room_finished ------------------------------------------


async def test_message_after_conference_ended_rejected_without_phantom_session(
    db_session: AsyncSession, ws_client: WSFactory
) -> None:
    """LiveKit-токен (TTL 6 часов) может пережить конференцию — сообщение отклоняется.

    Сценарий: клиент подключился, пока конференция была активна; сессия
    пайплайна закрывается (`t_end` проставлен, как это делает webhook
    `room_finished`), а статус конференции переводится в `ended` СЫРЫМ SQL
    (в обход ORM identity map — имитация того, что вебхук работает в ДРУГОЙ
    сессии/процессе и `expire_on_commit=False` не даёт уже загруженному
    объекту `conference` увидеть новый статус). Отправленное после этого
    сообщение должно быть отклонено (close 4404), новая "фантомная" открытая
    сессия — НЕ создана, сообщение — НЕ сохранено.
    """
    conference = await _make_conference(db_session, status="active")
    user = await _make_user(db_session)
    await db_session.commit()

    token = _user_token(conference, user)
    ws = ws_client(_chat_path(conference.id))
    await _connect_and_auth(ws, token)

    sessions_repo = ConferenceSessionRepository(db_session)
    session_record = await sessions_repo.get_open_by_conference(conference.id)
    if session_record is None:
        session_record = await sessions_repo.create(
            conference_id=conference.id, title=conference.title, t_start=datetime.now(UTC)
        )
    await sessions_repo.close(session_record, t_end=datetime.now(UTC))

    # Сырой UPDATE — намеренно в обход ORM, чтобы не обновить закэшированный
    # в текущей сессии Python-объект `conference` (имитация другой сессии/процесса).
    await db_session.execute(
        text("UPDATE conferences SET status = 'ended' WHERE id = :id"), {"id": conference.id}
    )
    await db_session.commit()

    await ws.send_json({"type": "message", "text": "too late"})
    code = await ws.receive_close()
    assert code == 4404

    sessions = (
        (
            await db_session.execute(
                select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
            )
        )
        .scalars()
        .all()
    )
    assert len(sessions) == 1
    assert sessions[0].id == session_record.id
    assert sessions[0].t_end is not None

    stray = (
        await db_session.execute(select(ChatMessage).where(ChatMessage.text == "too late"))
    ).scalar_one_or_none()
    assert stray is None


async def test_message_before_room_started_creates_session_for_active_conference(
    db_session: AsyncSession, ws_client: WSFactory
) -> None:
    """Легитимный случай: первое сообщение до webhook `room_started`.

    Открытой сессии ещё нет, но конференция активна (не `ended`) — создание
    новой сессии допустимо (в отличие от случая, когда конференция уже
    завершена).
    """
    conference = await _make_conference(db_session, status="active")
    user = await _make_user(db_session)
    await db_session.commit()

    token = _user_token(conference, user)
    ws = ws_client(_chat_path(conference.id))
    await _connect_and_auth(ws, token)

    await ws.send_json({"type": "message", "text": "before room_started"})
    echo = await ws.receive_json()
    assert echo["message"]["text"] == "before room_started"

    sessions = (
        (
            await db_session.execute(
                select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
            )
        )
        .scalars()
        .all()
    )
    assert len(sessions) == 1
    assert sessions[0].t_end is None


# --- Гость -------------------------------------------------------------------


async def test_guest_message_persists_guest_access_id_and_author_name(
    db_session: AsyncSession, ws_client: WSFactory
) -> None:
    conference = await _make_conference(db_session)
    guest = await _make_guest(db_session, conference, name="Guest Carol")
    await db_session.commit()

    token = _guest_token(conference, guest)
    ws = ws_client(_chat_path(conference.id))
    await _connect_and_auth(ws, token)

    await ws.send_json({"type": "message", "text": "hi from guest"})
    echo = await ws.receive_json()
    assert echo["message"]["author_name"] == "Guest Carol"
    assert echo["message"]["is_guest"] is True
    assert echo["message"]["author_id"] == str(guest.id)

    await ws.aclose()

    row = (
        await db_session.execute(select(ChatMessage).where(ChatMessage.text == "hi from guest"))
    ).scalar_one()
    assert row.guest_access_id == guest.id
    assert row.user_id is None
    assert row.author_name == "Guest Carol"


# --- REST: JoinOut.chat_enabled ----------------------------------------------


async def test_join_out_reflects_chat_enabled_toggle(
    client: httpx.AsyncClient, db_session: AsyncSession
) -> None:
    user = await _make_user(db_session)
    await db_session.commit()
    headers = {"Authorization": f"Bearer {create_access_token(user.id, user.role)}"}

    response = await client.post("/api/v1/conferences", json={"title": "Standup"}, headers=headers)
    assert response.status_code == 201, response.text
    assert response.json()["join"]["chat_enabled"] is True

    await InstanceSettingsService(db_session).update(SettingsUpdateIn(chat_enabled=False))

    response = await client.post(
        "/api/v1/conferences", json={"title": "Standup 2"}, headers=headers
    )
    assert response.status_code == 201, response.text
    assert response.json()["join"]["chat_enabled"] is False


async def test_guest_join_out_reflects_chat_enabled(
    client: httpx.AsyncClient, db_session: AsyncSession
) -> None:
    conference = await _make_conference(db_session)
    await db_session.commit()

    await InstanceSettingsService(db_session).update(SettingsUpdateIn(chat_enabled=False))

    response = await client.post(
        f"/api/v1/conferences/{conference.id}/guest-join", json={"display_name": "Dave"}
    )
    assert response.status_code == 200, response.text
    assert response.json()["chat_enabled"] is False


# --- Юнит: схема протокола ----------------------------------------------------


def test_chat_message_in_rejects_empty_text_after_strip() -> None:
    with pytest.raises(ValidationError):
        ChatMessageIn(type="message", text="   ")


def test_chat_message_in_rejects_too_long_text() -> None:
    with pytest.raises(ValidationError):
        ChatMessageIn(type="message", text="a" * 2001)


def test_chat_message_in_strips_surrounding_whitespace() -> None:
    parsed = ChatMessageIn(type="message", text="  hello  ")
    assert parsed.text == "hello"


def test_chat_message_out_serializes_created_at_as_utc_z() -> None:
    out = ChatMessageOut(
        id=1,
        author_id="abc",
        author_name="Alice",
        is_guest=False,
        text="hi",
        created_at=datetime(2026, 7, 18, 12, 0, 0, tzinfo=UTC),
    )
    dumped = out.model_dump(mode="json")
    assert dumped["created_at"] == "2026-07-18T12:00:00Z"