"""Репозитории доступа к `conferences` (сущность) и `conference_sessions` (сеанс, ADR-001).

`ConferenceSessionRepository` используется webhook-обработчиками LiveKit
(`services/webhook_handlers.py`) и beat-задачей обслуживания
(`workers/tasks/maintenance.py`) — методы get-or-create/guard-стиля, чтобы
быть безопасными при пропущенных или дублирующихся событиях (шаги
пост-обработки идемпотентны).
"""

import uuid
from datetime import datetime
from typing import cast

from sqlalchemy import delete, func, or_, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from models.audio_track import SessionAudioTrack
from models.conference import Conference
from models.invitee import ConferenceInvitee
from models.participant import ConferenceParticipant
from models.session import ConferenceSession
from models.user import User


class ConferenceRepository:
    """Инкапсулирует SQL-запросы к конференциям (`conferences`)."""

    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def get_by_id(self, conference_id: uuid.UUID) -> Conference | None:
        """Найти конференцию по id."""
        return await self._session.get(Conference, conference_id)

    async def get_by_slug(self, slug: str) -> Conference | None:
        """Найти конференцию по постоянной ссылке (= имени LiveKit-комнаты)."""
        result = await self._session.execute(select(Conference).where(Conference.slug == slug))
        return result.scalar_one_or_none()

    async def get_by_number(self, number: str) -> Conference | None:
        """Найти конференцию по человеко-диктуемому номеру."""
        result = await self._session.execute(select(Conference).where(Conference.number == number))
        return result.scalar_one_or_none()

    async def get_status_by_id(self, conference_id: uuid.UUID) -> str | None:
        """Прочитать АКТУАЛЬНЫЙ статус конференции, минуя identity map сессии.

        В отличие от `get_by_id`/`session.get(...)`, SELECT одной колонки не
        возвращает уже загруженный в эту сессию ORM-объект `Conference` из
        кэша identity map — а при `expire_on_commit=False` (`core/db.py`)
        такой объект, однажды загруженный долгоживущей WS-сессией чата,
        никогда сам не увидит статус, изменённый вебхуком в ДРУГОЙ сессии/
        процессе (например, `room_finished` -> `ended`). Используется
        `ChatService.persist_and_publish` перед созданием новой сессии
        пайплайна.
        """
        result = await self._session.execute(
            select(Conference.status).where(Conference.id == conference_id)
        )
        return result.scalar_one_or_none()

    async def add(self, conference: Conference) -> Conference:
        """Добавить конференцию в сессию и сделать flush (нарушение unique — здесь же)."""
        self._session.add(conference)
        await self._session.flush()
        return conference

    async def delete(self, conference: Conference) -> None:
        """Удалить конференцию (сеансы/гости удаляются каскадом на уровне БД)."""
        await self._session.delete(conference)
        await self._session.flush()

    async def list_owned(
        self, user_id: uuid.UUID, *, email: str, now: datetime
    ) -> list[Conference]:
        """Конференции владельца ИЛИ приглашённого для «Моих конференций» (решение от
        2026-07-20 поверх ADR-003: приглашённый видит конференцию в своих списках) —
        закреплённые + предстоящие разовые.

        «Приглашённый» — есть строка `conference_invitees` с `user_id == user_id`
        ИЛИ с `lower(email) == lower(этого email)` (внешнее приглашение на адрес,
        под которым человек впоследствии зарегистрировался). `EXISTS`-подзапрос
        не размножает строки `Conference` — `DISTINCT` не требуется.
        """
        is_invitee = (
            select(ConferenceInvitee.id)
            .where(
                ConferenceInvitee.conference_id == Conference.id,
                or_(
                    ConferenceInvitee.user_id == user_id,
                    func.lower(ConferenceInvitee.email) == email.lower(),
                ),
            )
            .exists()
        )
        result = await self._session.execute(
            select(Conference)
            .where(
                or_(Conference.owner_id == user_id, is_invitee),
                (Conference.is_pinned.is_(True))
                | (
                    (Conference.scheduled_at.isnot(None))
                    & (Conference.scheduled_at >= now)
                    & (Conference.status == "scheduled")
                ),
            )
            .order_by(Conference.is_pinned.desc(), Conference.scheduled_at.asc().nulls_last())
        )
        return list(result.scalars().all())

    async def list_calendar_candidates(self, user_id: uuid.UUID, *, email: str) -> list[Conference]:
        """Конференции владельца ИЛИ приглашённого, потенциально дающие вхождения в календаре.

        Тот же принцип видимости приглашённого, что и `list_owned` (решение от
        2026-07-20). Развёртка диапазона — в сервисном слое
        (`services/conferences.py::list_calendar`), здесь только грубая выборка
        кандидатов (есть recurrence или указано scheduled_at).
        """
        is_invitee = (
            select(ConferenceInvitee.id)
            .where(
                ConferenceInvitee.conference_id == Conference.id,
                or_(
                    ConferenceInvitee.user_id == user_id,
                    func.lower(ConferenceInvitee.email) == email.lower(),
                ),
            )
            .exists()
        )
        result = await self._session.execute(
            select(Conference).where(
                or_(Conference.owner_id == user_id, is_invitee),
                (Conference.recurrence.isnot(None)) | (Conference.scheduled_at.isnot(None)),
            )
        )
        return list(result.scalars().all())

    async def list_expired_unpinned_scheduled(self, *, now: datetime) -> list[Conference]:
        """Незакреплённые плановые конференции без единого сеанса (кандидаты на `ended`).

        Точная проверка истечения времени (с учётом `duration_minutes` и
        запаса) — в `workers/tasks/maintenance.py`; здесь — грубая выборка по
        `NOT EXISTS` сеанса, чтобы не тянуть в память всё лишнее.
        """
        has_session = (
            select(ConferenceSession.id)
            .where(ConferenceSession.conference_id == Conference.id)
            .exists()
        )
        result = await self._session.execute(
            select(Conference).where(
                Conference.is_pinned.is_(False),
                Conference.status == "scheduled",
                Conference.scheduled_at.isnot(None),
                Conference.scheduled_at < now,
                ~has_session,
            )
        )
        return list(result.scalars().all())


class ConferenceInviteeRepository:
    """Инкапсулирует SQL-запросы к приглашённым на конференцию (`conference_invitees`, ADR-003)."""

    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def list_with_user(
        self, conference_id: uuid.UUID
    ) -> list[tuple[ConferenceInvitee, str | None, str | None, str | None]]:
        """Приглашённые конференции + имя/email/путь аватара их пользователя (LEFT JOIN).

        Для внешних приглашённых (`user_id IS NULL`) три последних элемента
        кортежа — `None` (нет привязанного `User`).
        """
        result = await self._session.execute(
            select(ConferenceInvitee, User.name_user, User.email, User.avatar_path)
            .outerjoin(User, ConferenceInvitee.user_id == User.id)
            .where(ConferenceInvitee.conference_id == conference_id)
        )
        # `User.name_user`/`User.email` типизированы как non-optional (NOT NULL
        # в модели) — но при LEFT JOIN без совпадения (внешний приглашённый,
        # `user_id IS NULL`) значения реально приходят `NULL`; mypy не видит
        # nullability, вносимую `outerjoin`, отсюда явный `cast`.
        return cast(
            "list[tuple[ConferenceInvitee, str | None, str | None, str | None]]",
            list(result.all()),
        )

    async def existing_user_ids(self, user_ids: set[uuid.UUID]) -> set[uuid.UUID]:
        """Подмножество `user_ids`, реально существующее в `users` (проверка перед вставкой)."""
        if not user_ids:
            return set()
        result = await self._session.execute(select(User.id).where(User.id.in_(user_ids)))
        return set(result.scalars().all())

    async def exists_for_user(
        self, conference_id: uuid.UUID, *, user_id: uuid.UUID, email: str
    ) -> bool:
        """Приглашён ли `user_id` на конференцию — по `user_id` или по `lower(email)`.

        Используется проверкой доступа к детальной карточке (`GET /conferences/{id}`,
        решение от 2026-07-20 поверх ADR-003) — приглашённый должен её видеть
        наравне с владельцем/администратором.
        """
        result = await self._session.execute(
            select(ConferenceInvitee.id)
            .where(
                ConferenceInvitee.conference_id == conference_id,
                or_(
                    ConferenceInvitee.user_id == user_id,
                    func.lower(ConferenceInvitee.email) == email.lower(),
                ),
            )
            .limit(1)
        )
        return result.scalar_one_or_none() is not None

    async def replace_all(
        self, conference_id: uuid.UUID, invitees: list[ConferenceInvitee]
    ) -> None:
        """Полностью заменить состав приглашённых конференции (ADR-003, п.3 — PUT-семантика)."""
        await self._session.execute(
            delete(ConferenceInvitee).where(ConferenceInvitee.conference_id == conference_id)
        )
        for invitee in invitees:
            self._session.add(invitee)
        await self._session.flush()


class ConferenceSessionRepository:
    """Инкапсулирует SQL-запросы к сеансам конференций и их участникам."""

    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def get_open_by_conference(self, conference_id: uuid.UUID) -> ConferenceSession | None:
        """Вернуть открытый (`t_end IS NULL`) сеанс конференции, если есть."""
        result = await self._session.execute(
            select(ConferenceSession)
            .where(
                ConferenceSession.conference_id == conference_id,
                ConferenceSession.t_end.is_(None),
            )
            .order_by(ConferenceSession.t_start.desc())
            .limit(1)
        )
        return result.scalar_one_or_none()

    async def create(
        self, *, conference_id: uuid.UUID, title: str | None, t_start: datetime
    ) -> ConferenceSession:
        """Создать новый (открытый) сеанс конференции."""
        record = ConferenceSession(conference_id=conference_id, title=title, t_start=t_start)
        self._session.add(record)
        await self._session.flush()
        return record

    async def get_or_create_open(
        self, *, conference_id: uuid.UUID, title: str | None, t_start: datetime
    ) -> ConferenceSession:
        """Get-or-create открытого сеанса конференции.

        Гвард на случай, если `room_started` было пропущено и первым пришло
        `participant_joined`.
        """
        existing = await self.get_open_by_conference(conference_id)
        if existing is not None:
            return existing
        return await self.create(conference_id=conference_id, title=title, t_start=t_start)

    async def close(self, session_record: ConferenceSession, *, t_end: datetime) -> None:
        """Закрыть сеанс, проставив `t_end`."""
        session_record.t_end = t_end

    async def list_open(self) -> list[ConferenceSession]:
        """Список всех открытых (`t_end IS NULL`) сеансов.

        Используется maintenance-задачей (`workers/tasks/maintenance.py`)
        для обхода всех "зависших" сеансов разом.
        """
        result = await self._session.execute(
            select(ConferenceSession).where(ConferenceSession.t_end.is_(None))
        )
        return list(result.scalars().all())

    async def list_stuck_summarizing(self, *, older_than: datetime) -> list[ConferenceSession]:
        """Сеансы, зависшие на шаге суммаризации: `pipeline_status='summarizing'`,
        `summary_data` ещё не заполнен, а сеанс завершился раньше `older_than`.

        Кандидаты на повторную постановку `summarize_session` —
        `workers/tasks/maintenance.py::recover_stuck_summaries` (уровень 2
        защиты от потери постановки задачи при сбое брокера в `run_pipeline`).
        """
        result = await self._session.execute(
            select(ConferenceSession).where(
                ConferenceSession.pipeline_status == "summarizing",
                ConferenceSession.summary_data.is_(None),
                ConferenceSession.t_end.isnot(None),
                ConferenceSession.t_end < older_than,
            )
        )
        return list(result.scalars().all())

    async def list_stuck_notifying(self, *, older_than: datetime) -> list[ConferenceSession]:
        """Сеансы, зависшие на шаге уведомления: саммари готово, но `notify_session`
        так и не перевела пайплайн в `notified`.

        `pipeline_status='summarizing'` + `summary_data IS NOT NULL` +
        `t_end < older_than` — кандидаты на повторную постановку
        `notify_session` (`workers/tasks/maintenance.py::recover_stuck_notifications`,
        уровень 2 защиты от потери постановки задачи при сбое брокера в
        `summarize_session`, аналог `list_stuck_summarizing`).
        """
        result = await self._session.execute(
            select(ConferenceSession).where(
                ConferenceSession.pipeline_status == "summarizing",
                ConferenceSession.summary_data.isnot(None),
                ConferenceSession.t_end.isnot(None),
                ConferenceSession.t_end < older_than,
            )
        )
        return list(result.scalars().all())

    async def count_by_pipeline_status(self) -> dict[str, int]:
        """Число сеансов в каждом статусе пайплайна (`pipeline_status`).

        Используется метриками Prometheus (`backend/api/metrics.py`,
        блок 3) для gauge `vidconf_pipeline_sessions{status=...}` — считается
        заново при каждом scrape, не кешируется. Статусы без единого сеанса в
        результат не попадают (пустая группа), это ожидаемо: вызывающая
        сторона сама проставляет 0 для отсутствующих в словаре статусов
        (полный перечень — `models.session.PipelineStatus`), чтобы метрика не
        "пропадала" из `/metrics` между сборами.
        """
        result = await self._session.execute(
            select(ConferenceSession.pipeline_status, func.count()).group_by(
                ConferenceSession.pipeline_status
            )
        )
        return {status: count for status, count in result.all()}

    async def has_active_participants(self, session_id: uuid.UUID) -> bool:
        """Есть ли у сеанса хотя бы один участник без `left_at` (кто-то ещё внутри)."""
        result = await self._session.execute(
            select(ConferenceParticipant.id)
            .where(
                ConferenceParticipant.session_id == session_id,
                ConferenceParticipant.left_at.is_(None),
            )
            .limit(1)
        )
        return result.scalar_one_or_none() is not None

    async def add_participant(
        self,
        *,
        session_id: uuid.UUID,
        user_id: uuid.UUID | None,
        guest_id: uuid.UUID | None,
        joined_at: datetime,
    ) -> ConferenceParticipant:
        """Get-or-create активной (не покинувшей) записи участия — пользователя ИЛИ гостя.

        Защищает от дублей при повторной доставке `participant_joined` для
        уже присутствующего участника. Ровно один из `user_id`/`guest_id`
        должен быть передан (см. CHECK-constraint модели).
        """
        stmt = select(ConferenceParticipant).where(
            ConferenceParticipant.session_id == session_id,
            ConferenceParticipant.left_at.is_(None),
        )
        stmt = stmt.where(
            ConferenceParticipant.user_id == user_id
            if user_id is not None
            else ConferenceParticipant.guest_id == guest_id
        )
        existing = (await self._session.execute(stmt)).scalar_one_or_none()
        if existing is not None:
            return existing

        participant = ConferenceParticipant(
            session_id=session_id, user_id=user_id, guest_id=guest_id, joined_at=joined_at
        )
        self._session.add(participant)
        await self._session.flush()
        return participant

    async def get_active_participant(
        self,
        session_id: uuid.UUID,
        *,
        user_id: uuid.UUID | None = None,
        guest_id: uuid.UUID | None = None,
    ) -> ConferenceParticipant | None:
        """Найти активную (`left_at IS NULL`) запись присутствия пользователя/гостя в сеансе.

        Используется атрибуцией аудиотрека к участнику (`track_published`,
        ADR-002): ровно один из `user_id`/`guest_id` должен быть передан.
        """
        stmt = select(ConferenceParticipant).where(
            ConferenceParticipant.session_id == session_id,
            ConferenceParticipant.left_at.is_(None),
        )
        stmt = stmt.where(
            ConferenceParticipant.user_id == user_id
            if user_id is not None
            else ConferenceParticipant.guest_id == guest_id
        )
        stmt = stmt.order_by(ConferenceParticipant.joined_at.desc()).limit(1)
        return (await self._session.execute(stmt)).scalar_one_or_none()

    async def mark_participant_left(
        self,
        *,
        session_id: uuid.UUID,
        user_id: uuid.UUID | None,
        guest_id: uuid.UUID | None,
        left_at: datetime,
    ) -> None:
        """Проставить `left_at` последней открытой записи участника (пользователя или гостя)."""
        stmt = select(ConferenceParticipant).where(
            ConferenceParticipant.session_id == session_id,
            ConferenceParticipant.left_at.is_(None),
        )
        stmt = stmt.where(
            ConferenceParticipant.user_id == user_id
            if user_id is not None
            else ConferenceParticipant.guest_id == guest_id
        )
        stmt = stmt.order_by(ConferenceParticipant.joined_at.desc()).limit(1)
        result = await self._session.execute(stmt)
        participant = result.scalar_one_or_none()
        if participant is not None:
            participant.left_at = left_at

    async def close_all_open_participants(
        self, *, session_id: uuid.UUID, left_at: datetime
    ) -> None:
        """Закрыть все записи участников сеанса с `left_at IS NULL` (`room_finished`)."""
        open_participants = await self._session.scalars(
            select(ConferenceParticipant).where(
                ConferenceParticipant.session_id == session_id,
                ConferenceParticipant.left_at.is_(None),
            )
        )
        for participant in open_participants:
            participant.left_at = left_at


class AudioTrackRepository:
    """Инкапсулирует SQL-запросы к записанным аудиотрекам сеансов (`session_audio_tracks`)."""

    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def get_by_session_and_track(
        self, session_id: uuid.UUID, track_sid: str
    ) -> SessionAudioTrack | None:
        """Найти строку трека по (`session_id`, `track_sid`) — ключ идемпотентности `create`."""
        result = await self._session.execute(
            select(SessionAudioTrack).where(
                SessionAudioTrack.session_id == session_id,
                SessionAudioTrack.track_sid == track_sid,
            )
        )
        return result.scalar_one_or_none()

    async def create(
        self,
        *,
        session_id: uuid.UUID,
        participant_id: uuid.UUID,
        track_sid: str,
        egress_id: str | None,
        file_path: str | None,
        started_at: datetime,
    ) -> SessionAudioTrack:
        """Идемпотентно создать строку трека по (`session_id`, `track_sid`).

        Повторная доставка `track_published` (например, гонка между двумя
        одновременными вебхуками) — `INSERT ... ON CONFLICT DO NOTHING` по
        уникальному индексу `uq_session_track`, затем возврат уже
        существующей строки. `status` по умолчанию `recording`.
        """
        insert_stmt = (
            pg_insert(SessionAudioTrack)
            .values(
                session_id=session_id,
                participant_id=participant_id,
                track_sid=track_sid,
                egress_id=egress_id,
                file_path=file_path,
                started_at=started_at,
            )
            .on_conflict_do_nothing(constraint="uq_session_track")
            .returning(SessionAudioTrack.id)
        )
        inserted_id = (await self._session.execute(insert_stmt)).scalar_one_or_none()
        if inserted_id is None:
            existing = await self.get_by_session_and_track(session_id, track_sid)
            assert existing is not None  # конфликт гарантирует существование строки
            return existing

        await self._session.flush()
        record = await self._session.get(SessionAudioTrack, inserted_id)
        assert record is not None
        return record

    async def finalize(
        self,
        *,
        egress_id: str,
        status: str,
        ended_at: datetime,
        file_path: str | None = None,
    ) -> SessionAudioTrack | None:
        """Финализировать строку трека по `egress_id` (обработка webhook `egress_ended`).

        `status` — `'recorded'` при успехе, `'failed'` при ошибке egress.
        Отсутствие строки (например, `track_published` был потерян) — не
        ошибка, лог оставляет вызывающая сторона.
        """
        result = await self._session.execute(
            select(SessionAudioTrack).where(SessionAudioTrack.egress_id == egress_id)
        )
        record = result.scalar_one_or_none()
        if record is None:
            return None

        record.status = status
        record.ended_at = ended_at
        if file_path is not None:
            record.file_path = file_path
        return record

    async def list_by_session(self, session_id: uuid.UUID) -> list[SessionAudioTrack]:
        """Список всех треков сеанса (используется оркестрацией пайплайна)."""
        result = await self._session.execute(
            select(SessionAudioTrack).where(SessionAudioTrack.session_id == session_id)
        )
        return list(result.scalars().all())
