vidconf/backend/tests/test_livekit_webhook.py

"""Интеграционные тесты `/api/v1/livekit/webhook`: полный цикл, дедуп, fallback,
невалидная подпись, пропуск невалидного identity, гостевой identity, жизненный
цикл закреплённой/незакреплённой конференции — на фикстурах payload'ов LiveKit.
"""

import asyncio
import base64
import hashlib
import json
import uuid
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from pathlib import Path
from unittest.mock import AsyncMock, Mock

import httpx
import jwt
import pytest
from google.protobuf.json_format import ParseDict
from livekit.protocol.webhook import WebhookEvent
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

import services.egress as egress_module
import services.webhook_handlers as webhook_handlers_module
from core.config import get_settings
from core.security import hash_password
from models.audio_track import SessionAudioTrack
from models.conference import Conference
from models.guest import GuestAccess
from models.instance_setting import InstanceSetting
from models.participant import ConferenceParticipant
from models.session import ConferenceSession
from models.user import User
from services.conference_ids import generate_number, generate_slug
from services.egress import EgressStartResult

FIXTURES_DIR = Path(__file__).parent / "fixtures" / "livekit"
WEBHOOK_URL = "/api/v1/livekit/webhook"


def _load_fixture(name: str, **placeholders: str) -> bytes:
    """Загрузить JSON-фикстуру webhook-события, подставив плейсхолдеры `{name}`."""
    raw = (FIXTURES_DIR / name).read_text()
    for key, value in placeholders.items():
        raw = raw.replace(f"{{{key}}}", value)
    return raw.encode()


def _sign(body: bytes) -> str:
    """Подписать тело webhook-запроса так же, как это делает LiveKit server."""
    settings = get_settings()
    sha256_b64 = base64.b64encode(hashlib.sha256(body).digest()).decode()
    payload = {
        "iss": settings.livekit_api_key,
        "sha256": sha256_b64,
        "exp": datetime.now(UTC) + timedelta(minutes=5),
    }
    return jwt.encode(payload, settings.livekit_api_secret, algorithm="HS256")


def _parse_fixture_event(name: str, **placeholders: str) -> WebhookEvent:
    """Разобрать фикстуру в `WebhookEvent` — для тестов диспатчера без HTTP-слоя."""
    return ParseDict(json.loads(_load_fixture(name, **placeholders)), WebhookEvent())


async def _set_transcriber_enabled(session: AsyncSession, *, enabled: bool) -> None:
    """Выставить `instance_settings.transcriber.enabled`.

    Пишется тем же `db_session` (savepoint), что и обработчик webhook (подмена
    `get_session` в фикстуре `app`) — видна обработчику без реального коммита
    в dev-БД. Тесты, которым важен `track_published`, обязаны выставлять флаг
    ЯВНО: значение в dev-БД непредсказуемо, а обработчик с версии 0.0.12
    выходит на выключенной транскрибации раньше всех остальных проверок.
    """
    value: dict[str, object] = {
        "enabled": enabled,
        "provider": "faster_whisper_cpu" if enabled else "null",
        "model": "small" if enabled else None,
        "language": "ru",
        "options": {},
    }
    await session.execute(
        pg_insert(InstanceSetting)
        .values(key="transcriber", value=value)
        .on_conflict_do_update(index_elements=["key"], set_={"value": value})
    )


def _use_test_session_in_background(
    monkeypatch: pytest.MonkeyPatch, db_session: AsyncSession
) -> None:
    """Заставить фоновую задачу egress работать с тестовой (savepoint) сессией.

    `run_track_egress` намеренно берёт СВОЮ сессию (`async_session_maker`):
    в бою сессия запроса к моменту фоновой задачи уже закрыта. В тестах такое
    подключение шло бы мимо откатываемой транзакции и не увидело бы ни
    конференции, ни участника — поэтому подменяем фабрику на тестовую сессию.
    """

    @asynccontextmanager
    async def _maker() -> AsyncGenerator[AsyncSession, None]:
        yield db_session

    monkeypatch.setattr(egress_module, "async_session_maker", _maker)


async def _post_webhook(client: httpx.AsyncClient, body: bytes) -> httpx.Response:
    return await client.post(
        WEBHOOK_URL,
        content=body,
        headers={"Authorization": _sign(body), "Content-Type": "application/json"},
    )


async def _make_conference(
    session: AsyncSession, slug: str, *, is_pinned: bool = False
) -> Conference:
    conference = Conference(
        number=generate_number(), slug=slug, title=slug, is_pinned=is_pinned, status="scheduled"
    )
    session.add(conference)
    await session.flush()
    return conference


async def _make_user(session: AsyncSession, email: str) -> User:
    user = User(
        email=email,
        name_user="Participant",
        password_hash=hash_password("password123"),
        email_verified=True,
    )
    session.add(user)
    await session.flush()
    return user


async def _make_guest(session: AsyncSession, conference: Conference) -> GuestAccess:
    guest = GuestAccess(
        conference_id=conference.id, display_name="Guest Tester", email="guest-track@example.com"
    )
    session.add(guest)
    await session.flush()
    return guest


async def test_full_cycle_joined_left_finished(
    client: httpx.AsyncClient, db_session: AsyncSession
) -> None:
    conference = await _make_conference(db_session, generate_slug())
    user = await _make_user(db_session, "webhook-user-1@example.com")
    await db_session.commit()

    started = _load_fixture(
        "room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
    )
    resp = await _post_webhook(client, started)
    assert resp.status_code == 200
    assert resp.json()["status"] == "ok"

    await db_session.refresh(conference)
    assert conference.status == "active"

    session_record = await db_session.scalar(
        select(ConferenceSession).where(
            ConferenceSession.conference_id == conference.id, ConferenceSession.t_end.is_(None)
        )
    )
    assert session_record is not None
    assert session_record.t_start is not None

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    resp = await _post_webhook(client, joined)
    assert resp.status_code == 200

    participant = await db_session.scalar(
        select(ConferenceParticipant).where(
            ConferenceParticipant.session_id == session_record.id,
            ConferenceParticipant.user_id == user.id,
        )
    )
    assert participant is not None
    assert participant.joined_at is not None
    assert participant.left_at is None
    assert participant.guest_id is None

    left = _load_fixture(
        "participant_left.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    resp = await _post_webhook(client, left)
    assert resp.status_code == 200

    await db_session.refresh(participant)
    assert participant.left_at is not None

    finished = _load_fixture(
        "room_finished.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
    )
    resp = await _post_webhook(client, finished)
    assert resp.status_code == 200

    await db_session.refresh(session_record)
    assert session_record.t_end is not None
    await db_session.refresh(conference)
    assert conference.status == "ended"
    assert conference.ended_at is not None


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

    started = _load_fixture(
        "room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
    )
    assert (await _post_webhook(client, started)).status_code == 200

    finished = _load_fixture(
        "room_finished.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
    )
    resp = await _post_webhook(client, finished)
    assert resp.status_code == 200

    await db_session.refresh(conference)
    assert conference.status == "scheduled"
    assert conference.ended_at is None


async def test_guest_identity_creates_participant_with_guest_id(
    client: httpx.AsyncClient, db_session: AsyncSession
) -> None:
    conference = await _make_conference(db_session, generate_slug())
    guest = GuestAccess(
        conference_id=conference.id, display_name="Guest Tester", email="guest@example.com"
    )
    db_session.add(guest)
    await db_session.commit()

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=f"guest:{guest.id}",
    )
    resp = await _post_webhook(client, joined)
    assert resp.status_code == 200

    session_record = await db_session.scalar(
        select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
    )
    assert session_record is not None
    participant = await db_session.scalar(
        select(ConferenceParticipant).where(
            ConferenceParticipant.session_id == session_record.id,
            ConferenceParticipant.guest_id == guest.id,
        )
    )
    assert participant is not None
    assert participant.user_id is None


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

    event_id = f"evt-{uuid.uuid4()}"
    body = _load_fixture("room_started.json", event_id=event_id, room_name=conference.slug)

    first = await _post_webhook(client, body)
    assert first.status_code == 200
    assert first.json()["status"] == "ok"

    second = await _post_webhook(client, body)
    assert second.status_code == 200
    assert second.json()["status"] == "duplicate"

    sessions = (
        await db_session.scalars(
            select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
        )
    ).all()
    assert len(sessions) == 1


async def test_missing_room_started_falls_back_on_participant_joined(
    client: httpx.AsyncClient, db_session: AsyncSession
) -> None:
    conference = await _make_conference(db_session, generate_slug())
    user = await _make_user(db_session, "webhook-user-fallback@example.com")
    await db_session.commit()

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    resp = await _post_webhook(client, joined)
    assert resp.status_code == 200

    session_record = await db_session.scalar(
        select(ConferenceSession).where(
            ConferenceSession.conference_id == conference.id, ConferenceSession.t_end.is_(None)
        )
    )
    assert session_record is not None

    participant = await db_session.scalar(
        select(ConferenceParticipant).where(
            ConferenceParticipant.session_id == session_record.id,
            ConferenceParticipant.user_id == user.id,
        )
    )
    assert participant is not None


async def test_invalid_signature_returns_401(client: httpx.AsyncClient) -> None:
    body = _load_fixture(
        "room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name="does-not-matter"
    )
    resp = await client.post(
        WEBHOOK_URL,
        content=body,
        headers={"Authorization": "not-a-valid-jwt", "Content-Type": "application/json"},
    )
    assert resp.status_code == 401


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

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity="not-a-uuid",
    )
    resp = await _post_webhook(client, joined)
    assert resp.status_code == 200

    # Невалидный identity парсится раньше get-or-create сеанса, поэтому для
    # этой конференции не должно появиться ни сеанса, ни участника.
    session_record = await db_session.scalar(
        select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
    )
    assert session_record is None


async def test_track_published_by_guest_starts_egress_and_creates_track_row(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """№8 плана: гость (`identity="guest:<uuid>"`) публикует аудиотрек — AC-1."""
    started_at = datetime.now(UTC)
    mock_start = AsyncMock(
        return_value=EgressStartResult(egress_id="EG_guest_track", started_at=started_at)
    )
    monkeypatch.setattr(egress_module, "start_track_egress", mock_start)
    _use_test_session_in_background(monkeypatch, db_session)

    await _set_transcriber_enabled(db_session, enabled=True)
    conference = await _make_conference(db_session, generate_slug())
    guest = await _make_guest(db_session, conference)
    await db_session.commit()

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=f"guest:{guest.id}",
    )
    assert (await _post_webhook(client, joined)).status_code == 200

    session_record = await db_session.scalar(
        select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
    )
    assert session_record is not None
    participant = await db_session.scalar(
        select(ConferenceParticipant).where(
            ConferenceParticipant.session_id == session_record.id,
            ConferenceParticipant.guest_id == guest.id,
        )
    )
    assert participant is not None

    track_sid = f"TR_{uuid.uuid4().hex[:8]}"
    published = _load_fixture(
        "track_published.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=f"guest:{guest.id}",
        track_sid=track_sid,
    )
    resp = await _post_webhook(client, published)
    assert resp.status_code == 200

    settings = get_settings()
    expected_filepath = (
        f"{settings.recordings_dir}/{session_record.id}/{participant.id}_{track_sid}.ogg"
    )
    mock_start.assert_awaited_once_with(conference.slug, track_sid, expected_filepath)

    track_row = await db_session.scalar(
        select(SessionAudioTrack).where(
            SessionAudioTrack.session_id == session_record.id,
            SessionAudioTrack.track_sid == track_sid,
        )
    )
    assert track_row is not None
    assert track_row.participant_id == participant.id
    assert participant.guest_id == guest.id
    assert participant.user_id is None
    assert track_row.egress_id == "EG_guest_track"
    assert track_row.file_path == expected_filepath
    assert track_row.status == "recording"


async def test_track_published_survives_egress_unavailable(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Недоступность egress не должна ронять webhook (блок D): 200 + warning, без строки трека."""
    mock_start = AsyncMock(side_effect=RuntimeError("egress service unavailable"))
    monkeypatch.setattr(egress_module, "start_track_egress", mock_start)
    _use_test_session_in_background(monkeypatch, db_session)

    await _set_transcriber_enabled(db_session, enabled=True)
    conference = await _make_conference(db_session, generate_slug())
    user = await _make_user(db_session, "webhook-track-egress-down@example.com")
    await db_session.commit()

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    assert (await _post_webhook(client, joined)).status_code == 200

    track_sid = "TR_egress_down"
    published = _load_fixture(
        "track_published.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
        track_sid=track_sid,
    )
    resp = await _post_webhook(client, published)
    assert resp.status_code == 200
    assert resp.json()["status"] == "ok"

    mock_start.assert_awaited_once()
    tracks = (
        await db_session.scalars(
            select(SessionAudioTrack).where(SessionAudioTrack.track_sid == track_sid)
        )
    ).all()
    assert tracks == []


async def test_track_published_skips_egress_when_transcription_disabled(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Транскрибация выключена → egress не дёргается вовсе (релиз 0.0.12).

    Симметрично guard'у в `_on_room_finished`. Без него на инстансе без профиля
    `transcribe` (egress-контейнера в деплое нет) каждый микрофонный трек
    превращался в заведомо безнадёжный вызов длиной в таймаут psrpc LiveKit —
    20–25 секунд внутри открытой транзакции вебхука. На нагрузочном тесте
    28.07.2026 это выгребало пул соединений и роняло вход в конференцию в 500.
    """
    mock_start = AsyncMock()
    monkeypatch.setattr(egress_module, "start_track_egress", mock_start)
    _use_test_session_in_background(monkeypatch, db_session)

    await _set_transcriber_enabled(db_session, enabled=False)
    conference = await _make_conference(db_session, generate_slug())
    user = await _make_user(db_session, "webhook-track-transcriber-off@example.com")
    await db_session.commit()

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    assert (await _post_webhook(client, joined)).status_code == 200

    track_sid = "TR_transcriber_off"
    published = _load_fixture(
        "track_published.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
        track_sid=track_sid,
    )
    assert (await _post_webhook(client, published)).status_code == 200

    mock_start.assert_not_awaited()
    tracks = (
        await db_session.scalars(
            select(SessionAudioTrack).where(SessionAudioTrack.track_sid == track_sid)
        )
    ).all()
    assert tracks == []


async def test_track_published_does_not_call_egress_inside_transaction(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Запуск egress уходит в фон, а не выполняется внутри обработчика (релиз 0.0.12).

    Проверяется не время ответа (в тестах ASGI-транспорт дожидается фоновых
    задач), а сама суть: пока открыта транзакция вебхука, сетевого вызова не
    происходит — обработчик только планирует задачу. Именно это разгружает пул
    соединений: до правки вызов жил внутри транзакции и держал соединение
    20–25 секунд, когда egress-сервиса в деплое нет.
    """
    mock_start = AsyncMock(
        return_value=EgressStartResult(egress_id="EG_bg", started_at=datetime.now(UTC))
    )
    monkeypatch.setattr(egress_module, "start_track_egress", mock_start)

    scheduled: list[tuple[object, dict[str, object]]] = []

    def _schedule(func: object, **kwargs: object) -> None:
        scheduled.append((func, kwargs))

    await _set_transcriber_enabled(db_session, enabled=True)
    conference = await _make_conference(db_session, generate_slug())
    user = await _make_user(db_session, "webhook-track-background@example.com")
    await db_session.commit()

    dispatcher = webhook_handlers_module.WebhookDispatcher(db_session, schedule=_schedule)
    joined_event = _parse_fixture_event(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    await dispatcher.dispatch(joined_event)

    track_sid = "TR_background"
    published_event = _parse_fixture_event(
        "track_published.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
        track_sid=track_sid,
    )
    await dispatcher.dispatch(published_event)

    # Сеть не тронута: обработчик только запланировал задачу.
    mock_start.assert_not_awaited()
    assert len(scheduled) == 1
    func, kwargs = scheduled[0]
    assert func is egress_module.run_track_egress
    assert kwargs["room_name"] == conference.slug
    assert kwargs["track_sid"] == track_sid

    session_record = await db_session.scalar(
        select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
    )
    assert session_record is not None
    assert kwargs["session_id"] == session_record.id

    # А вот запущенная задача действительно ходит в egress и пишет строку.
    _use_test_session_in_background(monkeypatch, db_session)
    await egress_module.run_track_egress(**kwargs)  # type: ignore[arg-type]
    mock_start.assert_awaited_once()

    track_row = await db_session.scalar(
        select(SessionAudioTrack).where(SessionAudioTrack.track_sid == track_sid)
    )
    assert track_row is not None
    assert track_row.egress_id == "EG_bg"


async def test_start_track_egress_gives_up_on_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
    """Запуск egress не ждёт дольше `egress_start_timeout_s` (релиз 0.0.12).

    Когда egress-воркера нет, LiveKit держит вызов до собственного таймаута
    psrpc — на тесте 28.07.2026 это было 20–25 секунд на каждый микрофонный
    трек. Живой egress отвечает за доли секунды, ждать столько незачем.
    """
    settings = get_settings()
    monkeypatch.setattr(settings, "egress_start_timeout_s", 0.05, raising=False)

    closed = False

    class _HangingEgress:
        async def start_track_egress(self, _request: object) -> object:
            await asyncio.sleep(5)
            raise AssertionError("вызов должен был прерваться по таймауту")

    class _HangingApi:
        def __init__(self, *_args: object, **_kwargs: object) -> None:
            self.egress = _HangingEgress()

        async def aclose(self) -> None:
            nonlocal closed
            closed = True

    # Строковая форма: `api` в `services.egress` — реэкспорт из livekit SDK,
    # обращение к нему атрибутом mypy считает неявным экспортом.
    monkeypatch.setattr("services.egress.api.LiveKitAPI", _HangingApi)

    with pytest.raises(TimeoutError):
        await egress_module.start_track_egress("room", "TR_hang", "/recordings/x.ogg")

    # Клиент закрывается и на неуспешном пути — иначе утекали бы соединения.
    assert closed is True


async def test_track_published_video_is_noop(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """№9 плана (часть 1): video-трек — no-op, egress не запускается."""
    mock_start = AsyncMock()
    monkeypatch.setattr(egress_module, "start_track_egress", mock_start)
    _use_test_session_in_background(monkeypatch, db_session)

    await _set_transcriber_enabled(db_session, enabled=True)
    conference = await _make_conference(db_session, generate_slug())
    user = await _make_user(db_session, "webhook-track-video@example.com")
    await db_session.commit()

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    assert (await _post_webhook(client, joined)).status_code == 200

    published = _load_fixture(
        "track_published_video.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
        track_sid="TR_video",
    )
    resp = await _post_webhook(client, published)
    assert resp.status_code == 200

    mock_start.assert_not_awaited()
    tracks = (
        await db_session.scalars(
            select(SessionAudioTrack).where(SessionAudioTrack.track_sid == "TR_video")
        )
    ).all()
    assert tracks == []


async def test_track_published_repeated_webhook_creates_single_row(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """№9 плана (часть 2): повторный `track_published` того же трека → одна строка."""
    mock_start = AsyncMock(
        return_value=EgressStartResult(egress_id="EG_repeat", started_at=datetime.now(UTC))
    )
    monkeypatch.setattr(egress_module, "start_track_egress", mock_start)
    _use_test_session_in_background(monkeypatch, db_session)

    await _set_transcriber_enabled(db_session, enabled=True)
    conference = await _make_conference(db_session, generate_slug())
    user = await _make_user(db_session, "webhook-track-repeat@example.com")
    await db_session.commit()

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    assert (await _post_webhook(client, joined)).status_code == 200

    track_sid = "TR_repeat"
    for _ in range(2):
        published = _load_fixture(
            "track_published.json",
            event_id=f"evt-{uuid.uuid4()}",
            room_name=conference.slug,
            identity=str(user.id),
            track_sid=track_sid,
        )
        resp = await _post_webhook(client, published)
        assert resp.status_code == 200

    session_record = await db_session.scalar(
        select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
    )
    assert session_record is not None
    tracks = (
        await db_session.scalars(
            select(SessionAudioTrack).where(
                SessionAudioTrack.session_id == session_record.id,
                SessionAudioTrack.track_sid == track_sid,
            )
        )
    ).all()
    assert len(tracks) == 1
    mock_start.assert_awaited_once()


async def test_egress_ended_finalizes_track_success_and_failure(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """№10 плана (часть 1): `egress_ended` — 'recorded' на успехе, 'failed' на ошибке."""
    _use_test_session_in_background(monkeypatch, db_session)
    await _set_transcriber_enabled(db_session, enabled=True)
    conference = await _make_conference(db_session, generate_slug())
    user = await _make_user(db_session, "webhook-egress-ended@example.com")
    await db_session.commit()

    joined = _load_fixture(
        "participant_joined.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
    )
    assert (await _post_webhook(client, joined)).status_code == 200
    session_record = await db_session.scalar(
        select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
    )
    assert session_record is not None

    # Успешная запись.
    ok_started = datetime.now(UTC)
    monkeypatch.setattr(
        egress_module,
        "start_track_egress",
        AsyncMock(return_value=EgressStartResult(egress_id="EG_ok", started_at=ok_started)),
    )
    published_ok = _load_fixture(
        "track_published.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
        track_sid="TR_ok",
    )
    assert (await _post_webhook(client, published_ok)).status_code == 200

    ended_ok = _load_fixture(
        "egress_ended.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        egress_id="EG_ok",
        file_path=f"/recordings/{session_record.id}/track-ok.ogg",
    )
    assert (await _post_webhook(client, ended_ok)).status_code == 200

    ok_track = await db_session.scalar(
        select(SessionAudioTrack).where(SessionAudioTrack.egress_id == "EG_ok")
    )
    assert ok_track is not None
    assert ok_track.status == "recorded"
    assert ok_track.ended_at is not None
    assert ok_track.file_path == f"/recordings/{session_record.id}/track-ok.ogg"

    # Ошибка записи.
    monkeypatch.setattr(
        egress_module,
        "start_track_egress",
        AsyncMock(
            return_value=EgressStartResult(egress_id="EG_fail", started_at=datetime.now(UTC))
        ),
    )
    published_fail = _load_fixture(
        "track_published.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        identity=str(user.id),
        track_sid="TR_fail",
    )
    assert (await _post_webhook(client, published_fail)).status_code == 200

    ended_failed = _load_fixture(
        "egress_ended_failed.json",
        event_id=f"evt-{uuid.uuid4()}",
        room_name=conference.slug,
        egress_id="EG_fail",
    )
    assert (await _post_webhook(client, ended_failed)).status_code == 200

    failed_track = await db_session.scalar(
        select(SessionAudioTrack).where(SessionAudioTrack.egress_id == "EG_fail")
    )
    assert failed_track is not None
    assert failed_track.status == "failed"
    assert failed_track.ended_at is not None


async def test_room_finished_enqueues_pipeline(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """№10 плана (часть 2): `room_finished` ставит `run_pipeline` в очередь (мок)."""
    mock_enqueue = Mock()
    monkeypatch.setattr(webhook_handlers_module, "enqueue_pipeline", mock_enqueue)

    # Флаг выставляется явно: `_on_room_finished` ставит задачу в очередь только
    # при включённой транскрибации, а состояние `instance_settings` в dev-БД
    # непредсказуемо (тест падал, если в базе оставалось `enabled: false`).
    await _set_transcriber_enabled(db_session, enabled=True)
    conference = await _make_conference(db_session, generate_slug())
    await db_session.commit()

    started = _load_fixture(
        "room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
    )
    assert (await _post_webhook(client, started)).status_code == 200

    session_record = await db_session.scalar(
        select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
    )
    assert session_record is not None

    finished = _load_fixture(
        "room_finished.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
    )
    assert (await _post_webhook(client, finished)).status_code == 200

    mock_enqueue.assert_called_once_with(session_record.id)


async def test_room_finished_skips_enqueue_when_transcription_disabled(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Guard от зависания сеанса: транскрибация
    выключена в настройках инстанса (пресеты 1/2 инсталлятора — без AI) →
    `room_finished` НЕ ставит `run_pipeline` в очередь `transcription` (некому
    обслужить), сеанс сразу переводится в терминальный `pipeline_status='notified'`
    вместо зависания в `recording` навсегда."""
    mock_enqueue = Mock()
    monkeypatch.setattr(webhook_handlers_module, "enqueue_pipeline", mock_enqueue)

    # Настройка `instance_settings.transcriber` пишется тем же `db_session`
    # (savepoint), что и обработчик webhook (подмена `get_session` в фикстуре
    # `app`) — видна обработчику без реального коммита в dev-БД.
    disabled_transcriber = {
        "enabled": False,
        "provider": "null",
        "model": None,
        "language": "ru",
        "options": {},
    }
    stmt = (
        pg_insert(InstanceSetting)
        .values(key="transcriber", value=disabled_transcriber)
        .on_conflict_do_update(index_elements=["key"], set_={"value": disabled_transcriber})
    )
    await db_session.execute(stmt)

    conference = await _make_conference(db_session, generate_slug())
    await db_session.commit()

    started = _load_fixture(
        "room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
    )
    assert (await _post_webhook(client, started)).status_code == 200

    session_record = await db_session.scalar(
        select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
    )
    assert session_record is not None

    finished = _load_fixture(
        "room_finished.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
    )
    assert (await _post_webhook(client, finished)).status_code == 200

    mock_enqueue.assert_not_called()
    await db_session.refresh(session_record)
    assert session_record.pipeline_status == "notified"