"""Session initialization and management for streaming transcription."""

from __future__ import annotations

import asyncio
from typing import TYPE_CHECKING

import grpc

from noteflow.config.constants import (
    DEFAULT_MEETING_TITLE,
    ERROR_MSG_MEETING_PREFIX,
    STREAM_INIT_LOCK_TIMEOUT_SECONDS,
)
from noteflow.infrastructure.diarization import SpeakerTurn
from noteflow.infrastructure.logging import get_logger

from .._types import GrpcContext
from ..converters import parse_meeting_id_or_none
from ..errors import abort_failed_precondition
from ._types import StreamSessionInit

if TYPE_CHECKING:
    from noteflow.domain.entities import Meeting
    from noteflow.domain.ports.unit_of_work import UnitOfWork

    from ..protocols import ServicerHost

logger = get_logger(__name__)


def _build_session_error(
    error_code: grpc.StatusCode,
    error_message: str,
) -> StreamSessionInit:
    """Build a failed StreamSessionInit result."""
    return StreamSessionInit(
        next_segment_id=0,
        error_code=error_code,
        error_message=error_message,
    )


async def _trigger_recording_webhook(
    host: ServicerHost,
    meeting_id: str,
    title: str,
) -> None:
    """Fire recording.started webhook (fire-and-forget).

    Silently logs and suppresses any exceptions.
    """
    if host.webhook_service is None:
        return
    try:
        await host.webhook_service.trigger_recording_started(
            meeting_id=meeting_id,
            title=title,
        )
    # INTENTIONAL BROAD HANDLER: Fire-and-forget webhook pattern
    # - Webhook failures must never block RPC operations
    # - All exceptions logged for debugging but suppressed
    except Exception:
        logger.exception("Failed to trigger recording.started webhooks")


async def _prepare_meeting_for_streaming(
    host: ServicerHost,
    repo: UnitOfWork,
    meeting: Meeting,
    meeting_id: str,
) -> StreamSessionInit | None:
    """Prepare meeting DEK and recording state, commit if needed.

    Returns StreamSessionInit on error, None on success.
    """
    _dek, _wrapped_dek, dek_updated = host.ensure_meeting_dek(meeting)
    recording_updated, error_msg = host.start_meeting_if_needed(meeting)

    if error_msg:
        return _build_session_error(grpc.StatusCode.INVALID_ARGUMENT, error_msg)

    if dek_updated or recording_updated:
        await repo.meetings.update(meeting)
        await repo.commit()

    if recording_updated:
        await _trigger_recording_webhook(
            host, meeting_id, meeting.title or DEFAULT_MEETING_TITLE
        )

    return None


def _init_audio_writer(
    host: ServicerHost,
    meeting_id: str,
    dek: bytes,
    wrapped_dek: bytes,
    asset_path: str | None,
) -> StreamSessionInit | None:
    """Initialize the meeting audio writer.

    Returns StreamSessionInit on error, None on success.
    """
    try:
        host.open_meeting_audio_writer(
            meeting_id, dek, wrapped_dek, asset_path=asset_path
        )
        return None
    except OSError as e:
        logger.error("Failed to create audio writer for %s: %s", meeting_id, e)
        return _build_session_error(
            grpc.StatusCode.INTERNAL,
            f"Failed to initialize audio storage: {e}",
        )


class StreamSessionManager:
    """Manage stream session initialization and lock handling."""

    @staticmethod
    async def init_stream_for_meeting(
        host: ServicerHost,
        meeting_id: str,
        context: GrpcContext,
    ) -> StreamSessionInit | None:
        """Initialize streaming for a meeting.

        Uses unified repository provider for both DB and memory backends.
        Protected by lock to prevent race conditions on concurrent init attempts.

        Args:
            host: The servicer host.
            meeting_id: Meeting ID string.
            context: gRPC context for error handling.

        Returns:
            Initialization result, or None if error was sent.
        """
        # Atomic check-and-add protected by lock with timeout to prevent deadlock
        try:
            async with asyncio.timeout(STREAM_INIT_LOCK_TIMEOUT_SECONDS):
                async with host.stream_init_lock:
                    if meeting_id in host.active_streams:
                        await abort_failed_precondition(
                            context, f"{ERROR_MSG_MEETING_PREFIX}{meeting_id} already streaming"
                        )
                    host.active_streams.add(meeting_id)
        except TimeoutError:
            logger.error(
                "Stream initialization lock timeout for meeting %s after %.1fs",
                meeting_id,
                STREAM_INIT_LOCK_TIMEOUT_SECONDS,
            )
            await abort_failed_precondition(
                context, "Stream initialization timed out - server may be overloaded"
            )

        init_result = await StreamSessionManager._init_stream_session(host, meeting_id)

        if not init_result.success:
            host.active_streams.discard(meeting_id)
            error_code = init_result.error_code if init_result.error_code is not None else grpc.StatusCode.INTERNAL
            await context.abort(error_code, init_result.error_message or "")

        return init_result

    @staticmethod
    async def _init_stream_session(
        host: ServicerHost,
        meeting_id: str,
    ) -> StreamSessionInit:
        """Initialize stream session using unified repository provider.

        Args:
            host: The servicer host.
            meeting_id: Meeting ID string.

        Returns:
            Stream session initialization result.
        """
        parsed_meeting_id = parse_meeting_id_or_none(meeting_id)
        if parsed_meeting_id is None:
            return _build_session_error(grpc.StatusCode.INVALID_ARGUMENT, "Invalid meeting_id")

        async with host.create_repository_provider() as repo:
            meeting = await repo.meetings.get(parsed_meeting_id)
            if meeting is None:
                return _build_session_error(
                    grpc.StatusCode.NOT_FOUND,
                    f"{ERROR_MSG_MEETING_PREFIX}{meeting_id} not found",
                )

            error = await _prepare_meeting_for_streaming(host, repo, meeting, meeting_id)
            if error:
                return error

            dek, wrapped_dek, _ = host.ensure_meeting_dek(meeting)
            next_segment_id = await repo.segments.compute_next_segment_id(meeting.id)

            if error := _init_audio_writer(
                host, meeting_id, dek, wrapped_dek, meeting.asset_path
            ):
                return error

            host.init_streaming_state(meeting_id, next_segment_id)

            # Load any persisted streaming turns (crash recovery) - DB only
            if repo.supports_diarization_jobs:
                await StreamSessionManager._load_persisted_diarization_turns(host, repo, meeting_id)

            return StreamSessionInit(next_segment_id=next_segment_id)

    @staticmethod
    async def _load_persisted_diarization_turns(
        host: ServicerHost,
        repo: UnitOfWork,
        meeting_id: str,
    ) -> None:
        """Load persisted streaming diarization turns for crash recovery.

        Args:
            host: The servicer host.
            repo: Repository provider with diarization_jobs support.
            meeting_id: Meeting identifier.
        """
        if not repo.supports_diarization_jobs:
            return

        persisted_turns = await repo.diarization_jobs.get_streaming_turns(meeting_id)
        if not persisted_turns:
            return

        domain_turns = [
            SpeakerTurn(
                speaker=t.speaker,
                start=t.start_time,
                end=t.end_time,
                confidence=t.confidence,
            )
            for t in persisted_turns
        ]
        host.diarization_turns[meeting_id] = domain_turns
        # Advance stream time to avoid overlapping recovered turns
        last_end = max(t.end_time for t in persisted_turns)
        host.diarization_stream_time[meeting_id] = max(
            host.diarization_stream_time.get(meeting_id, 0.0),
            last_end,
        )
        logger.info(
            "Loaded %d streaming diarization turns for meeting %s",
            len(domain_turns),
            meeting_id,
        )
