"""Бизнес-логика конференций: создание, «Мои конференции», календарь, резолв, join, правки.

Тонкий API-роутер (`api/conferences.py`) делегирует сюда всю логику; проверки
статуса/пароля и генерация ответа join вынесены в `services/conference_access.py`.
"""

import logging
import uuid
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any, cast

from sqlalchemy import null
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from core.config import get_settings
from core.plugins.config import SummaryRecipientsMode
from core.security import hash_password
from models.conference import Conference
from models.guest import GuestAccess
from models.invitee import ConferenceInvitee
from models.user import User
from repositories.conferences import ConferenceInviteeRepository, ConferenceRepository
from schemas.conferences import (
    ConferenceCreateIn,
    ConferenceOut,
    ConferenceUpdateIn,
    GuestJoinIn,
    InviteeIn,
    InviteeOut,
    JoinOut,
    OccurrenceOut,
)
from services import hand_queue
from services.avatars import avatar_url as resolve_avatar_url
from services.conference_access import build_join, ensure_joinable
from services.conference_ids import generate_number, generate_slug
from services.instance_settings import InstanceSettingsService
from services.invitations_producer import enqueue_invitations
from services.recurrence import RecurrenceRule, expand_occurrences
from services.room_control import MuteSource, mute_participant_track

logger = logging.getLogger(__name__)

# Число попыток сгенерировать уникальные номер/slug при коллизии unique-constraint
# (крайне маловероятной при выбранной энтропии — см. ADR-001, п.4).
MAX_ID_GENERATION_ATTEMPTS = 5

# Горизонт поиска "следующего вхождения" закреплённой конференции с повторением
# для «Моих конференций»; с запасом покрывает самый частый шаг (weekly/monthly).
NEXT_OCCURRENCE_HORIZON = timedelta(days=400)

# Длительность вхождения календаря по умолчанию, если у разовой плановой
# конференции не указан `duration_minutes`.
DEFAULT_OCCURRENCE_DURATION_MINUTES = 60


class ConferenceNotFoundError(Exception):
    """Конференция с таким id не найдена."""


class NotConferenceOwnerError(Exception):
    """Действие разрешено только владельцу конференции или администратору."""


class ConferenceActiveError(Exception):
    """Нельзя удалить конференцию, которая сейчас активна."""


class InvalidConferenceStateError(Exception):
    """Итоговое состояние конференции после правки нарушает бизнес-инварианты."""


class InviteeUserNotFoundError(Exception):
    """Один или несколько `user_id` в составе приглашённых не существуют (ADR-003)."""

    def __init__(self, missing_user_ids: set[uuid.UUID]) -> None:
        self.missing_user_ids = missing_user_ids
        super().__init__(f"неизвестные user_id приглашённых: {missing_user_ids}")


class ConferenceService:
    """Инкапсулирует сценарии создания/просмотра/входа/правки конференций."""

    def __init__(self, session: AsyncSession, *, media_root: Path | None = None) -> None:
        self._session = session
        self._conferences = ConferenceRepository(session)
        self._invitees = ConferenceInviteeRepository(session)
        self._media_root = media_root or Path(get_settings().media_root)

    async def create(
        self, *, owner: User, data: ConferenceCreateIn
    ) -> tuple[Conference, JoinOut | None]:
        """Создать конференцию.

        Мгновенная (`status=active`, создатель входит сразу же — второй
        элемент кортежа тогда содержит готовый `JoinOut`) — только если не
        указаны ни `scheduled_at`, ни `recurrence`; закреплённая с
        повторением без явного `scheduled_at` — плановая конференция,
        ожидающая своего первого вхождения, а не мгновенный вход.
        """
        password_hash = await hash_password(data.password) if data.password else None
        is_instant = data.scheduled_at is None and data.recurrence is None
        conference_status = "active" if is_instant else "scheduled"
        recurrence_json = data.recurrence.model_dump(mode="json") if data.recurrence else None
        # Примитивы читаются из `owner` один раз, до цикла retry: после
        # `session.rollback()` (при коллизии number/slug) ORM помечает все
        # загруженные объекты, включая `owner`, протухшими, а повторное
        # обращение к их атрибутам вне greenlet-контекста упало бы с
        # `MissingGreenlet` (ленивая подгрузка синхронным геттером).
        owner_id = owner.id
        owner_name = owner.name_user
        owner_email = owner.email
        owner_avatar_path = owner.avatar_path

        # Раннее падение при неизвестном `user_id` приглашённого — ДО
        # создания конференции (см. `InviteeUserNotFoundError`), чтобы не
        # оставлять во flush-буфере сессии несохранённую строку `conferences`.
        await self._validate_participants(data.participants)

        def _factory(number: str, slug: str) -> Conference:
            conference = Conference(
                number=number,
                slug=slug,
                title=data.title,
                owner_id=owner_id,
                status=conference_status,
                is_pinned=data.is_pinned,
                is_closed=data.is_closed,
                password_hash=password_hash,
                scheduled_at=data.scheduled_at,
                duration_minutes=data.duration_minutes,
                summary_recipients=data.summary_recipients,
            )
            # Атрибут намеренно не устанавливается вовсе, если правила нет:
            # JSONB-колонка сериализует явно присвоенный Python `None` в
            # JSON-литерал `null`, а не в SQL NULL (нет `none_as_null=True` —
            # модель не трогаем, см. блок A); неустановленный атрибут при
            # INSERT просто опускается, и колонка получает настоящий NULL.
            if recurrence_json is not None:
                conference.recurrence = recurrence_json
            return conference

        conference = await self._create_with_unique_ids(_factory)
        await self._apply_participants(conference, data.participants, owner_email=owner_email)
        await self._session.commit()

        join = None
        if is_instant:
            cfg = await InstanceSettingsService(self._session).get()
            join = build_join(
                conference,
                identity=str(owner_id),
                name=owner_name,
                chat_enabled=cfg.chat.enabled,
                hand_queue_enabled=cfg.hand_queue.enabled,
                publish_quality_cap=cfg.media_limits.publish_quality_cap,
                stage_max_tiles=cfg.media_limits.stage_max_tiles,
                virtual_background_enabled=cfg.virtual_background_enabled,
                avatar_url=resolve_avatar_url(self._media_root, owner_avatar_path),
                is_organizer=True,
            )
        else:
            # Плановая (разовая) либо закреплённая с повторением/датой — есть
            # расписание, на которое имеет смысл прислать .ics-приглашение
            # («.ics»). Ставится ПОСЛЕ commit — воркер должен
            # видеть уже зафиксированную строку конференции.
            self._enqueue_invitations_safely(conference.id)
        return conference, join

    async def list_my(self, *, owner: User) -> list[ConferenceOut]:
        """Закреплённые конференции + предстоящие разовые владельца ИЛИ приглашённого.

        Решение от 2026-07-20 (поверх ADR-003): приглашённый должен видеть
        конференцию в своих списках — `owner` здесь означает «текущий
        пользователь», а не только фактического владельца конференции.
        `participants` НЕ заполняется (пусто) — список не раздувает состав
        (ADR-003, п.5); `organizer_name`/`is_owner` теперь честно вычисляются
        на конференцию (для приглашённого — имя РЕАЛЬНОГО владельца и
        `is_owner=False`), а не берутся из `owner` — небольшой N+1 на строки,
        где владелец конференции не совпадает с viewer (`_resolve_organizer_name`),
        список короткий.
        """
        now = datetime.now(UTC)
        conferences = await self._conferences.list_owned(owner.id, email=owner.email, now=now)
        result: list[ConferenceOut] = []
        for conference in conferences:
            organizer_name = await self._resolve_organizer_name(conference, viewer=owner)
            result.append(
                self.to_out(
                    conference,
                    next_occurrence=self._next_occurrence(conference, now=now),
                    viewer_id=owner.id,
                    organizer_name=organizer_name,
                )
            )
        return result

    async def list_calendar(
        self, *, owner: User, t_from: datetime, t_to: datetime
    ) -> list[OccurrenceOut]:
        """Развернуть вхождения закреплённых/разовых конференций владельца ИЛИ приглашённого
        (решение от 2026-07-20 поверх ADR-003 — тот же принцип видимости, что `list_my`)."""
        candidates = await self._conferences.list_calendar_candidates(owner.id, email=owner.email)
        occurrences: list[OccurrenceOut] = []
        for conference in candidates:
            if conference.recurrence is not None:
                rule = RecurrenceRule.model_validate(conference.recurrence)
                for starts_at in expand_occurrences(rule, t_from, t_to):
                    occurrences.append(
                        self._occurrence_out(
                            conference,
                            starts_at=starts_at,
                            ends_at=starts_at + timedelta(minutes=rule.duration_minutes),
                        )
                    )
            elif conference.scheduled_at is not None and t_from <= conference.scheduled_at <= t_to:
                duration = conference.duration_minutes or DEFAULT_OCCURRENCE_DURATION_MINUTES
                occurrences.append(
                    self._occurrence_out(
                        conference,
                        starts_at=conference.scheduled_at,
                        ends_at=conference.scheduled_at + timedelta(minutes=duration),
                    )
                )
        occurrences.sort(key=lambda occurrence: occurrence.starts_at)
        return occurrences

    async def resolve(self, q: str) -> Conference | None:
        """Найти конференцию по slug или номеру (пробелы в номере игнорируются)."""
        query = q.strip()
        if not query:
            return None
        conference = await self._conferences.get_by_slug(query)
        if conference is not None:
            return conference
        number_candidate = query.replace(" ", "")
        if number_candidate.isdigit():
            return await self._conferences.get_by_number(number_candidate)
        return None

    async def join_as_user(
        self, conference_id: uuid.UUID, *, user: User, password: str | None
    ) -> JoinOut:
        """Войти в конференцию зарегистрированным пользователем."""
        conference = await self._get_or_raise(conference_id)
        await ensure_joinable(conference, password=password)
        cfg = await InstanceSettingsService(self._session).get()
        return build_join(
            conference,
            identity=str(user.id),
            name=user.name_user,
            chat_enabled=cfg.chat.enabled,
            hand_queue_enabled=cfg.hand_queue.enabled,
            publish_quality_cap=cfg.media_limits.publish_quality_cap,
            stage_max_tiles=cfg.media_limits.stage_max_tiles,
            virtual_background_enabled=cfg.virtual_background_enabled,
            avatar_url=resolve_avatar_url(self._media_root, user.avatar_path),
            is_organizer=conference.owner_id is not None and conference.owner_id == user.id,
        )

    async def join_as_guest(self, conference_id: uuid.UUID, *, data: GuestJoinIn) -> JoinOut:
        """Войти в конференцию гостем: создать `GuestAccess` и выдать токен."""
        conference = await self._get_or_raise(conference_id)
        await ensure_joinable(conference, password=data.password)

        guest = GuestAccess(
            conference_id=conference.id, display_name=data.display_name, email=data.email
        )
        self._session.add(guest)
        await self._session.flush()
        await self._session.commit()

        cfg = await InstanceSettingsService(self._session).get()
        return build_join(
            conference,
            identity=f"guest:{guest.id}",
            name=data.display_name,
            chat_enabled=cfg.chat.enabled,
            hand_queue_enabled=cfg.hand_queue.enabled,
            publish_quality_cap=cfg.media_limits.publish_quality_cap,
            stage_max_tiles=cfg.media_limits.stage_max_tiles,
            virtual_background_enabled=cfg.virtual_background_enabled,
        )

    async def mute_participant(
        self,
        conference_id: uuid.UUID,
        *,
        actor: User,
        target_identity: str,
        source: MuteSource,
    ) -> bool:
        """Принудительно замьютить трек участника (задача B2); владелец/администратор.

        Права — ТОЛЬКО отсюда (`_ensure_owner_or_admin` по `conference.owner_id`
        в БД), не по метаданным LiveKit-токена вызывающего: те лишь подсказка
        для UI (см. `services/conference_access.py::build_join`) и потенциально
        подделываемы клиентом. `target_identity` НИКАК не валидируется против
        состава участников заранее — если его сейчас нет в комнате LiveKit,
        `mute_participant_track` бросит `ParticipantNotInRoomError` (ловит
        API-роутер).
        """
        conference = await self._get_or_raise(conference_id)
        self._ensure_owner_or_admin(conference, actor)

        muted = await mute_participant_track(
            conference.slug, identity=target_identity, source=source
        )
        if muted:
            await hand_queue.publish_forced_mute(
                conference.id, identity=target_identity, source=source
            )
        return muted

    async def update(
        self, conference_id: uuid.UUID, *, actor: User, data: ConferenceUpdateIn
    ) -> Conference:
        """Частично обновить конференцию; разрешено владельцу или администратору."""
        conference = await self._get_or_raise(conference_id)
        self._ensure_owner_or_admin(conference, actor)

        # Раннее падение при неизвестном `user_id` приглашённого — ДО любых
        # мутаций конференции (см. `InviteeUserNotFoundError`).
        await self._validate_participants(data.participants)

        # Снимок «расписательных» полей ДО правки — чтобы после всех мутаций
        # (включая recurrence, разрешаемый ниже отдельно) определить, нужно
        # ли инкрементировать `ics_sequence` и переслать .ics-приглашение.
        title_before = conference.title
        scheduled_at_before = conference.scheduled_at
        duration_before = conference.duration_minutes

        if data.title is not None:
            conference.title = data.title
        if data.scheduled_at is not None:
            conference.scheduled_at = data.scheduled_at
        if data.duration_minutes is not None:
            conference.duration_minutes = data.duration_minutes
        if data.is_closed is not None:
            conference.is_closed = data.is_closed
        if data.password is not None:
            conference.password_hash = await hash_password(data.password)
        if "summary_recipients" in data.model_fields_set:
            # Явная передача (в т.ч. `null`) — сбросить/установить
            # переопределение; отсутствие поля в запросе значение не трогает.
            conference.summary_recipients = data.summary_recipients

        is_pinned = data.is_pinned if data.is_pinned is not None else conference.is_pinned
        conference.is_pinned = is_pinned

        # Итоговое значение recurrence считаем как обычный Python dict|None —
        # чтобы бизнес-проверка ниже не путала ORM-сентинел `null()` (см. далее)
        # с «правила нет». `model_fields_set` различает «поле не передано» и
        # «передано явным `null`» — оба случая иначе выглядят одинаково
        # (`data.recurrence is None`).
        current_recurrence = conference.recurrence
        recurrence_explicitly_cleared = (
            "recurrence" in data.model_fields_set and data.recurrence is None
        )

        # Открепление обязано снимать и правило повторения: recurrence без
        # is_pinned нарушает CHECK-constraint `ck_conferences_recurrence_requires_pinned`,
        # даже если PATCH вовсе не упоминал `recurrence` (например, только
        # `{"is_pinned": false}`).
        must_clear = recurrence_explicitly_cleared or (
            not is_pinned and current_recurrence is not None
        )

        cleared_via_null_sentinel = False
        if data.recurrence is not None:
            recurrence: dict[str, Any] | None = data.recurrence.model_dump(mode="json")
            conference.recurrence = recurrence
        elif must_clear:
            recurrence = None
            # JSONB-колонка без `none_as_null=True` (модель не трогаем, см.
            # блок A) сериализует явно присвоенный Python `None` в
            # JSON-литерал `null`, а не в SQL NULL — что уронило бы
            # CHECK-constraint. `sqlalchemy.null()` форсирует настоящий SQL
            # NULL, но помечает атрибут "expired" после flush (ORM не знает
            # результат SQL-выражения без похода в БД) — обновляем объект
            # явным awaited-рефрешем ниже перед возвратом, иначе синхронное
            # чтение `conference.recurrence` вне greenlet (например, в
            # `to_out`) упадёт `MissingGreenlet`.
            conference.recurrence = cast(Any, null())
            cleared_via_null_sentinel = True
        else:
            # Ничего менять не нужно — атрибут не трогаем вовсе, чтобы не
            # провоцировать лишнюю expiry на пустом месте (см. выше).
            recurrence = current_recurrence

        if conference.is_closed and conference.password_hash is None:
            raise InvalidConferenceStateError("closed_conference_requires_password")
        if recurrence is not None and not is_pinned:
            raise InvalidConferenceStateError("recurrence_requires_pinned")

        # Расписание изменилось, если сдвинулись заголовок/дата/длительность
        # либо правило повторения (сравниваем уже разрешённое значение
        # `recurrence`, а не «сырой» атрибут — иначе сентинел `null()` ниже
        # спутал бы сравнение). Такая правка обязана переслать .ics-приглашение
        # с новым `SEQUENCE` — иначе уже принятое
        # приглашение в календаре адресата разойдётся с фактическим временем.
        schedule_changed = (
            conference.title != title_before
            or conference.scheduled_at != scheduled_at_before
            or conference.duration_minutes != duration_before
            or recurrence != current_recurrence
        )
        if schedule_changed:
            conference.ics_sequence += 1

        # Состав приглашённых — после всех проверок бизнес-инвариантов выше
        # (иначе при `InvalidConferenceStateError` DELETE/INSERT состава
        # пришлось бы объяснять исполнением, которое всё равно не закоммитится
        # — но проще просто не выполнять его до финальной валидации).
        # `ics_sequence` НЕ растёт от одной только правки состава
        # (анти-спам) — рассылка всё равно ставится в очередь.
        owner_email = await self._resolve_owner_email(conference)
        participants_changed = await self._apply_participants(
            conference, data.participants, owner_email=owner_email
        )

        await self._session.flush()
        await self._session.commit()
        if cleared_via_null_sentinel:
            await self._session.refresh(conference, attribute_names=["recurrence"])
        if schedule_changed or participants_changed:
            # Ставится ПОСЛЕ commit — воркер должен видеть уже
            # зафиксированные `ics_sequence`/новое расписание/состав.
            self._enqueue_invitations_safely(conference.id)
        return conference

    def _enqueue_invitations_safely(self, conference_id: uuid.UUID) -> None:
        """Поставить рассылку .ics-приглашений в очередь, не роняя запрос при сбое.

        Конференция к этому моменту уже закоммичена — создание/правка это
        главный результат запроса, а рассылка приглашений вторична. При
        недоступности Redis `Celery.send_task` (используется
        `enqueue_invitations`) бросает исключение синхронно — без этого перехвата
        POST/PATCH `/conferences` отдал бы 500, хотя запись уже сохранена.
        Восстановление: пропавшая постановка компенсируется ручной рассылкой
        администратором (`POST /admin/conferences/{id}/invitations`) — в
        отличие от шагов AI-пайплайна, здесь нет beat-задачи уровня 2 защиты,
        т.к. отсутствие приглашения не блокирует использование конференции.
        """
        try:
            enqueue_invitations(conference_id)
        except Exception:  # noqa: BLE001 — недоступность брокера не должна ронять запрос
            logger.warning(
                "ConferenceService: не удалось поставить в очередь рассылку "
                "приглашений для конференции %s (Redis недоступен?) — конференция "
                "сохранена, разослать приглашения можно вручную из админки",
                conference_id,
                exc_info=True,
            )

    async def delete(self, conference_id: uuid.UUID, *, actor: User) -> None:
        """Удалить конференцию; запрещено для активной (409 на уровне роутера)."""
        conference = await self._get_or_raise(conference_id)
        self._ensure_owner_or_admin(conference, actor)
        if conference.status == "active":
            raise ConferenceActiveError
        await self._conferences.delete(conference)
        await self._session.commit()

    async def get_detail(self, conference_id: uuid.UUID, *, actor: User) -> Conference:
        """Получить конференцию для детального просмотра.

        Разрешено владельцу, администратору ИЛИ приглашённому (решение от
        2026-07-20 поверх ADR-003 — иначе ховер-карточка/детальная страница
        приглашённого получали бы 403). PATCH/DELETE эту проверку НЕ
        переиспользуют — там по-прежнему только владелец/администратор
        (`_ensure_owner_or_admin`).
        """
        conference = await self._get_or_raise(conference_id)
        await self._ensure_viewable(conference, actor)
        return conference

    async def to_detail_out(
        self, conference: Conference, *, viewer: User, join: JoinOut | None = None
    ) -> ConferenceOut:
        """Собрать `ConferenceOut` с полным составом участников (детальный ответ).

        Используется создание/правка/`GET /conferences/{id}` — списочные
        эндпоинты (`list_my`) состав не раздувают (ADR-003, п.5) и строят
        ответ через `to_out` напрямую.
        """
        organizer_name = await self._resolve_organizer_name(conference, viewer=viewer)
        participants = await self._build_participants_out(conference)
        return self.to_out(
            conference,
            join=join,
            organizer_name=organizer_name,
            participants=participants,
            viewer_id=viewer.id,
        )

    def to_out(
        self,
        conference: Conference,
        *,
        join: JoinOut | None = None,
        next_occurrence: datetime | None = None,
        viewer_id: uuid.UUID | None = None,
        organizer_name: str | None = None,
        participants: list[InviteeOut] | None = None,
    ) -> ConferenceOut:
        """Собрать `ConferenceOut` из ORM-модели."""
        recurrence = (
            RecurrenceRule.model_validate(conference.recurrence) if conference.recurrence else None
        )
        return ConferenceOut(
            id=conference.id,
            number=conference.number,
            slug=conference.slug,
            title=conference.title,
            status=conference.status,
            is_pinned=conference.is_pinned,
            is_closed=conference.is_closed,
            scheduled_at=conference.scheduled_at,
            duration_minutes=conference.duration_minutes,
            recurrence=recurrence,
            next_occurrence=next_occurrence,
            created_at=conference.created_at,
            join=join,
            summary_recipients=cast("SummaryRecipientsMode | None", conference.summary_recipients),
            owner_id=conference.owner_id,
            is_owner=viewer_id is not None and conference.owner_id == viewer_id,
            organizer_name=organizer_name,
            participants=participants or [],
        )

    async def _resolve_organizer_name(self, conference: Conference, *, viewer: User) -> str | None:
        """Имя организатора; переиспользует уже загруженного `viewer`, если он и есть владелец."""
        if conference.owner_id is None:
            return None
        if conference.owner_id == viewer.id:
            return viewer.name_user
        return await self._resolve_owner_name(conference)

    async def _resolve_owner_name(self, conference: Conference) -> str | None:
        if conference.owner_id is None:
            return None
        owner = await self._session.get(User, conference.owner_id)
        return owner.name_user if owner is not None else None

    async def _resolve_owner_email(self, conference: Conference) -> str | None:
        if conference.owner_id is None:
            return None
        owner = await self._session.get(User, conference.owner_id)
        return owner.email if owner is not None else None

    async def _build_participants_out(self, conference: Conference) -> list[InviteeOut]:
        """Состав приглашённых: организатор всегда первым (ADR-003, п.2), затем остальные."""
        items: list[InviteeOut] = []
        if conference.owner_id is not None:
            owner = await self._session.get(User, conference.owner_id)
            if owner is not None:
                items.append(
                    InviteeOut(
                        user_id=owner.id,
                        email=owner.email,
                        name=owner.name_user,
                        avatar_url=resolve_avatar_url(self._media_root, owner.avatar_path),
                        is_organizer=True,
                    )
                )
        rows = await self._invitees.list_with_user(conference.id)
        for invitee, user_name, user_email, user_avatar_path in rows:
            items.append(
                InviteeOut(
                    user_id=invitee.user_id,
                    email=invitee.email if invitee.email is not None else user_email,
                    name=user_name,
                    avatar_url=(
                        resolve_avatar_url(self._media_root, user_avatar_path)
                        if invitee.user_id is not None
                        else None
                    ),
                    is_organizer=False,
                )
            )
        return items

    async def _validate_participants(self, participants: list[InviteeIn] | None) -> None:
        """Проверить существование всех `user_id` состава — ДО любых мутаций (ADR-003)."""
        if participants is None:
            return
        user_ids = {item.user_id for item in participants if item.user_id is not None}
        if not user_ids:
            return
        missing = user_ids - await self._invitees.existing_user_ids(user_ids)
        if missing:
            raise InviteeUserNotFoundError(missing)

    async def _apply_participants(
        self,
        conference: Conference,
        participants: list[InviteeIn] | None,
        *,
        owner_email: str | None,
    ) -> bool:
        """Заменить состав приглашённых; вернуть `True`, если фактический состав изменился.

        `participants=None` — не менять (ADR-003, п.3), no-op. Организатор
        (по `user_id` владельца или его email) молча дедуплицируется — он и
        так всегда в составе (ADR-003, п.2), а неудаляем конструктивно.
        `_validate_participants` должен быть вызван заранее.
        """
        if participants is None:
            return False

        owner_id = conference.owner_id
        desired: dict[tuple[str, str], ConferenceInvitee] = {}
        for item in participants:
            if item.user_id is not None:
                if item.user_id == owner_id:
                    continue
                key = ("user", str(item.user_id))
                desired.setdefault(
                    key, ConferenceInvitee(conference_id=conference.id, user_id=item.user_id)
                )
            else:
                assert item.email is not None  # гарантировано `InviteeIn._validate`
                # email пользователя в БД может быть в смешанном регистре,
                # InviteeIn.email всегда нормализован в lower — сравниваем без регистра.
                if owner_email is not None and item.email == owner_email.lower():
                    continue
                key = ("email", item.email)
                desired.setdefault(
                    key, ConferenceInvitee(conference_id=conference.id, email=item.email)
                )

        existing_rows = await self._invitees.list_with_user(conference.id)
        existing_keys = {
            ("user", str(invitee.user_id))
            if invitee.user_id is not None
            else ("email", invitee.email)
            for invitee, _, _, _ in existing_rows
        }
        changed = existing_keys != set(desired.keys())

        await self._invitees.replace_all(conference.id, list(desired.values()))
        return changed

    async def _get_or_raise(self, conference_id: uuid.UUID) -> Conference:
        conference = await self._conferences.get_by_id(conference_id)
        if conference is None:
            raise ConferenceNotFoundError
        return conference

    def _ensure_owner_or_admin(self, conference: Conference, actor: User) -> None:
        if conference.owner_id != actor.id and actor.role != "admin":
            raise NotConferenceOwnerError

    async def _ensure_viewable(self, conference: Conference, actor: User) -> None:
        """Владелец/администратор/приглашённый — иначе `NotConferenceOwnerError` (403).

        Только для чтения (`get_detail`) — решение от 2026-07-20 поверх
        ADR-003; PATCH/DELETE используют `_ensure_owner_or_admin` (синхронный,
        без приглашённых).
        """
        if conference.owner_id == actor.id or actor.role == "admin":
            return
        if await self._invitees.exists_for_user(conference.id, user_id=actor.id, email=actor.email):
            return
        raise NotConferenceOwnerError

    def _next_occurrence(self, conference: Conference, *, now: datetime) -> datetime | None:
        if conference.recurrence is not None:
            rule = RecurrenceRule.model_validate(conference.recurrence)
            horizon = NEXT_OCCURRENCE_HORIZON
            if rule.type == "every_n_days" and rule.interval_days is not None:
                horizon = max(horizon, timedelta(days=rule.interval_days + 2))
            occurrences = expand_occurrences(rule, now, now + horizon)
            return occurrences[0] if occurrences else None
        if conference.scheduled_at is not None and conference.scheduled_at >= now:
            return conference.scheduled_at
        return None

    def _occurrence_out(
        self, conference: Conference, *, starts_at: datetime, ends_at: datetime
    ) -> OccurrenceOut:
        return OccurrenceOut(
            conference_id=conference.id,
            title=conference.title,
            starts_at=starts_at,
            ends_at=ends_at,
            number=conference.number,
            slug=conference.slug,
            is_pinned=conference.is_pinned,
            is_closed=conference.is_closed,
        )

    async def _create_with_unique_ids(
        self, factory: Callable[[str, str], Conference]
    ) -> Conference:
        """Создать конференцию, повторяя генерацию номера/slug при коллизии unique.

        До `MAX_ID_GENERATION_ATTEMPTS` попыток; откат до последнего
        savepoint обязателен после `IntegrityError` — иначе сессия
        SQLAlchemy становится непригодной для дальнейших операций (в т.ч. в
        тестах со savepoint).
        """
        last_error: IntegrityError | None = None
        for _ in range(MAX_ID_GENERATION_ATTEMPTS):
            conference = factory(generate_number(), generate_slug())
            try:
                return await self._conferences.add(conference)
            except IntegrityError as exc:
                await self._session.rollback()
                last_error = exc
        assert last_error is not None
        raise last_error
