first commit

backend/repositories/conferences.py

@@ -0,0 +1,543 @@
+"""Репозитории доступа к `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())