"""Тесты обслуживания конференций (`workers.tasks.maintenance.cleanup_conferences_async`).

Как и старый `test_auto_release.py`, не использует savepoint-фикстуру
`db_session`: `cleanup_conferences_async` открывает собственную сессию с
отдельным engine (`workers/db.py::open_session`), которая не видит
незакоммиченные изменения другой сессии. Тестовые данные заводятся и
коммитятся напрямую через `core.db.engine`, очистка — вручную по завершении
теста. Время передаётся параметром `now` — без freezegun.
"""

import uuid
from collections.abc import AsyncGenerator
from datetime import UTC, datetime, timedelta
from typing import cast
from unittest.mock import AsyncMock, MagicMock

import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncConnection

from core.db import engine
from core.plugins.config import ChatConfig, InstanceConfig, SummarizerConfig, TranscriberConfig
from services.conference_ids import generate_number, generate_slug
from workers.tasks import maintenance as maintenance_module

NOW = datetime.now(UTC)


class _Fixture:
    """Id тестовой конференции + (опционально) её сеанса/участника."""

    def __init__(self) -> None:
        self.conference_id = uuid.uuid4()
        self.session_id = uuid.uuid4()
        self.user_id = uuid.uuid4()
        self.number = generate_number()
        self.slug = generate_slug()


@pytest.fixture
async def fx() -> AsyncGenerator[_Fixture, None]:
    f = _Fixture()
    async with engine.connect() as conn:
        await conn.execute(
            text(
                "INSERT INTO users (id, email, name_user, password_hash) "
                "VALUES (:id, :email, 'Maintenance Tester', 'x')"
            ),
            {"id": f.user_id, "email": f"maintenance-{f.user_id}@example.com"},
        )
        await conn.commit()
    yield f
    async with engine.connect() as conn:
        await conn.execute(
            text("DELETE FROM conference_participants WHERE session_id = :id"),
            {"id": f.session_id},
        )
        await conn.execute(
            text("DELETE FROM conference_sessions WHERE conference_id = :id"),
            {"id": f.conference_id},
        )
        await conn.execute(text("DELETE FROM conferences WHERE id = :id"), {"id": f.conference_id})
        await conn.execute(text("DELETE FROM users WHERE id = :id"), {"id": f.user_id})
        await conn.commit()


async def _insert_conference(
    conn: AsyncConnection,
    fx: _Fixture,
    *,
    is_pinned: bool,
    status: str,
    scheduled_at: datetime | None = None,
    duration_minutes: int | None = None,
) -> None:
    await conn.execute(
        text(
            "INSERT INTO conferences "
            "(id, number, slug, title, status, is_pinned, scheduled_at, duration_minutes) "
            "VALUES (:id, :number, :slug, 'Maintenance test', :status, :is_pinned, "
            ":scheduled_at, :duration_minutes)"
        ),
        {
            "id": fx.conference_id,
            "number": fx.number,
            "slug": fx.slug,
            "status": status,
            "is_pinned": is_pinned,
            "scheduled_at": scheduled_at,
            "duration_minutes": duration_minutes,
        },
    )


async def _insert_session(
    conn: AsyncConnection, fx: _Fixture, *, t_start: datetime, t_end: datetime | None = None
) -> None:
    await conn.execute(
        text(
            "INSERT INTO conference_sessions (id, conference_id, title, t_start, t_end) "
            "VALUES (:id, :conference_id, 'Maintenance session', :t_start, :t_end)"
        ),
        {
            "id": fx.session_id,
            "conference_id": fx.conference_id,
            "t_start": t_start,
            "t_end": t_end,
        },
    )


async def _insert_summarizing_session(
    conn: AsyncConnection,
    fx: _Fixture,
    *,
    t_end: datetime,
    summary_data: str | None = None,
) -> None:
    """Сеанс на шаге суммаризации — для тестов `recover_stuck_summaries_async`."""
    await conn.execute(
        text(
            "INSERT INTO conference_sessions "
            "(id, conference_id, title, t_start, t_end, pipeline_status, summary_data) "
            "VALUES (:id, :conference_id, 'Recovery session', :t_start, :t_end, "
            "'summarizing', :summary_data)"
        ),
        {
            "id": fx.session_id,
            "conference_id": fx.conference_id,
            "t_start": t_end - timedelta(minutes=30),
            "t_end": t_end,
            "summary_data": summary_data,
        },
    )


async def _insert_active_participant(conn: AsyncConnection, fx: _Fixture) -> None:
    await conn.execute(
        text(
            "INSERT INTO conference_participants (id, session_id, user_id, joined_at) "
            "VALUES (:id, :session_id, :user_id, :joined_at)"
        ),
        {"id": uuid.uuid4(), "session_id": fx.session_id, "user_id": fx.user_id, "joined_at": NOW},
    )


async def _fetch_conference(fx: _Fixture) -> dict[str, object]:
    async with engine.connect() as conn:
        result = await conn.execute(
            text("SELECT status, ended_at FROM conferences WHERE id = :id"),
            {"id": fx.conference_id},
        )
        row = result.mappings().one()
        return dict(row)


async def _fetch_session_t_end(fx: _Fixture) -> datetime | None:
    async with engine.connect() as conn:
        result = await conn.execute(
            text("SELECT t_end FROM conference_sessions WHERE id = :id"), {"id": fx.session_id}
        )
        return cast("datetime | None", result.scalar_one())


def _sent_tasks_for_session(mock_send_task: MagicMock, fx: _Fixture) -> list[str]:
    """Имена задач, отправленных `send_task` именно для тестового сеанса `fx`.

    Recovery-задачи сканируют ВСЮ таблицу `conference_sessions`, а общая
    dev-БД может легитимно содержать чужие зависшие сеансы (реальные данные
    разработчика) — глобальные `assert_called_once`/`assert_not_called`
    от них флэкают. Проверяем только вызовы с id нашего сеанса.
    """
    session_id = str(fx.session_id)
    return [
        c.args[0] for c in mock_send_task.call_args_list if c.kwargs.get("args") == [session_id]
    ]


def _summarizer_cfg(*, enabled: bool = True) -> InstanceConfig:
    """Конфиг для тестов `recover_stuck_summaries_async` — интересует
    только `summarizer.enabled`."""
    return InstanceConfig(
        transcriber=TranscriberConfig(),
        summarizer=SummarizerConfig(enabled=enabled, provider="fake"),
        chat=ChatConfig(),
    )


async def test_idle_session_of_unpinned_conference_closes_and_ends_conference(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="active")
        await _insert_session(conn, fx, t_start=NOW - timedelta(minutes=15))
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)

    assert await _fetch_session_t_end(fx) is not None
    conference = await _fetch_conference(fx)
    assert conference["status"] == "ended"
    assert conference["ended_at"] is not None
    mock_delete.assert_awaited_once_with(fx.slug)


async def test_idle_session_of_pinned_conference_returns_to_scheduled(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=True, status="active")
        await _insert_session(conn, fx, t_start=NOW - timedelta(minutes=15))
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)

    assert await _fetch_session_t_end(fx) is not None
    conference = await _fetch_conference(fx)
    assert conference["status"] == "scheduled"
    assert conference["ended_at"] is None


async def test_session_with_active_participant_is_not_touched(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="active")
        await _insert_session(conn, fx, t_start=NOW - timedelta(minutes=15))
        await _insert_active_participant(conn, fx)
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)

    assert await _fetch_session_t_end(fx) is None
    mock_delete.assert_not_awaited()


async def test_fresh_session_is_not_touched(fx: _Fixture, monkeypatch: pytest.MonkeyPatch) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="active")
        await _insert_session(conn, fx, t_start=NOW)
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)

    assert await _fetch_session_t_end(fx) is None
    mock_delete.assert_not_awaited()


async def test_repeated_run_is_a_noop(fx: _Fixture, monkeypatch: pytest.MonkeyPatch) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="active")
        await _insert_session(conn, fx, t_start=NOW - timedelta(minutes=15))
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)
    first_t_end = await _fetch_session_t_end(fx)
    assert first_t_end is not None

    await maintenance_module.cleanup_conferences_async(NOW + timedelta(minutes=5))
    second_t_end = await _fetch_session_t_end(fx)
    assert second_t_end == first_t_end
    mock_delete.assert_awaited_once()


async def test_expired_unpinned_scheduled_without_session_becomes_ended(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(
            conn,
            fx,
            is_pinned=False,
            status="scheduled",
            scheduled_at=NOW - timedelta(hours=2),
            duration_minutes=30,
        )
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)

    conference = await _fetch_conference(fx)
    assert conference["status"] == "ended"
    assert conference["ended_at"] is not None


async def test_scheduled_conference_within_grace_is_not_touched(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(
            conn,
            fx,
            is_pinned=False,
            status="scheduled",
            scheduled_at=NOW - timedelta(minutes=10),
            duration_minutes=None,
        )
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)

    conference = await _fetch_conference(fx)
    assert conference["status"] == "scheduled"


async def test_pinned_expired_scheduled_conference_is_not_touched(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(
            conn,
            fx,
            is_pinned=True,
            status="scheduled",
            scheduled_at=NOW - timedelta(hours=5),
            duration_minutes=30,
        )
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)

    conference = await _fetch_conference(fx)
    assert conference["status"] == "scheduled"


async def test_expired_scheduled_conference_with_session_is_not_touched(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    mock_delete = AsyncMock()
    monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)

    async with engine.connect() as conn:
        await _insert_conference(
            conn,
            fx,
            is_pinned=False,
            status="scheduled",
            scheduled_at=NOW - timedelta(hours=5),
            duration_minutes=30,
        )
        # Сеанс уже был (и закрыт) — конференция не должна считаться "без сеансов".
        await _insert_session(
            conn, fx, t_start=NOW - timedelta(hours=5), t_end=NOW - timedelta(hours=4)
        )
        await conn.commit()

    await maintenance_module.cleanup_conferences_async(NOW)

    conference = await _fetch_conference(fx)
    assert conference["status"] == "scheduled"


async def test_recover_stuck_summaries_resends_task_for_hung_session(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Сеанс завис в 'summarizing' без summary_data дольше порога — задача переставлена."""
    mock_send_task = MagicMock()
    monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="ended")
        await _insert_summarizing_session(
            conn,
            fx,
            t_end=NOW - maintenance_module.STUCK_SUMMARIZING_THRESHOLD - timedelta(minutes=1),
        )
        await conn.commit()

    await maintenance_module.recover_stuck_summaries_async(NOW, plugins_config=_summarizer_cfg())

    assert _sent_tasks_for_session(mock_send_task, fx) == [
        "workers.tasks.summarize.summarize_session"
    ]


async def test_recover_stuck_summaries_ignores_fresh_session(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Сеанс моложе порога (недавно ушёл в summarizing) — трогать рано, no-op."""
    mock_send_task = MagicMock()
    monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="ended")
        await _insert_summarizing_session(conn, fx, t_end=NOW - timedelta(minutes=5))
        await conn.commit()

    await maintenance_module.recover_stuck_summaries_async(NOW, plugins_config=_summarizer_cfg())

    assert _sent_tasks_for_session(mock_send_task, fx) == []


async def test_recover_stuck_summaries_ignores_session_with_summary_already_present(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Сеанс уже готов к уведомлению (summary_data заполнен) — recovery его не трогает."""
    mock_send_task = MagicMock()
    monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="ended")
        await _insert_summarizing_session(
            conn,
            fx,
            t_end=NOW - maintenance_module.STUCK_SUMMARIZING_THRESHOLD - timedelta(minutes=1),
            summary_data="Уже готовое саммари",
        )
        await conn.commit()

    await maintenance_module.recover_stuck_summaries_async(NOW, plugins_config=_summarizer_cfg())

    assert _sent_tasks_for_session(mock_send_task, fx) == []


async def test_recover_stuck_summaries_noop_when_summarizer_disabled(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """`summarizer.enabled=false` — recovery выходит сразу, не читая БД и не слишком часто
    впустую рассылая задачу (иначе слала бы её каждые 5 минут без толку)."""
    mock_send_task = MagicMock()
    monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="ended")
        await _insert_summarizing_session(
            conn,
            fx,
            t_end=NOW - maintenance_module.STUCK_SUMMARIZING_THRESHOLD - timedelta(minutes=1),
        )
        await conn.commit()

    await maintenance_module.recover_stuck_summaries_async(
        NOW, plugins_config=_summarizer_cfg(enabled=False)
    )

    mock_send_task.assert_not_called()


async def test_recover_stuck_notifications_resends_task_for_hung_session(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Саммари готово, но сеанс завис в 'summarizing' дольше порога — notify_session
    переставлена."""
    mock_send_task = MagicMock()
    monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="ended")
        await _insert_summarizing_session(
            conn,
            fx,
            t_end=NOW - maintenance_module.STUCK_NOTIFYING_THRESHOLD - timedelta(minutes=1),
            summary_data="Готовое саммари",
        )
        await conn.commit()

    await maintenance_module.recover_stuck_notifications_async(NOW)

    assert _sent_tasks_for_session(mock_send_task, fx) == ["workers.tasks.notify.notify_session"]


async def test_recover_stuck_notifications_ignores_fresh_session(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Сеанс моложе порога (саммари только что готово) — трогать рано, no-op."""
    mock_send_task = MagicMock()
    monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="ended")
        await _insert_summarizing_session(
            conn, fx, t_end=NOW - timedelta(minutes=5), summary_data="Готовое саммари"
        )
        await conn.commit()

    await maintenance_module.recover_stuck_notifications_async(NOW)

    assert _sent_tasks_for_session(mock_send_task, fx) == []


async def test_recover_stuck_notifications_ignores_session_without_summary(
    fx: _Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Саммари ещё не готово (summary_data NULL) — это забота recover_stuck_summaries, не notify."""
    mock_send_task = MagicMock()
    monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)

    async with engine.connect() as conn:
        await _insert_conference(conn, fx, is_pinned=False, status="ended")
        await _insert_summarizing_session(
            conn,
            fx,
            t_end=NOW - maintenance_module.STUCK_NOTIFYING_THRESHOLD - timedelta(minutes=1),
        )
        await conn.commit()

    await maintenance_module.recover_stuck_notifications_async(NOW)

    assert _sent_tasks_for_session(mock_send_task, fx) == []
