"""Protocol contracts for gRPC service mixins."""

from __future__ import annotations

import asyncio
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar, Final, Protocol

if TYPE_CHECKING:
    from collections import deque
    from collections.abc import AsyncIterator
    from datetime import datetime
    from uuid import UUID

    import numpy as np
    from numpy.typing import NDArray
    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

    from noteflow.application.services.calendar_service import CalendarService
    from noteflow.application.services.ner_service import NerService
    from noteflow.application.services.project_service import ProjectService
    from noteflow.application.services.summarization_service import SummarizationService
    from noteflow.application.services.webhook_service import WebhookService
    from noteflow.domain.entities import Integration, Meeting, Segment, Summary, SyncRun
    from noteflow.domain.ports.unit_of_work import UnitOfWork
    from noteflow.domain.value_objects import MeetingId
    from noteflow.grpc._mixins.preferences import PreferencesRepositoryProvider
    from noteflow.infrastructure.asr import FasterWhisperEngine, Segmenter, StreamingVad
    from noteflow.infrastructure.audio.partial_buffer import PartialAudioBuffer
    from noteflow.infrastructure.audio.writer import MeetingAudioWriter
    from noteflow.infrastructure.auth.oidc_registry import OidcAuthService
    from noteflow.infrastructure.diarization import (
        DiarizationEngine,
        DiarizationSession,
        SpeakerTurn,
    )
    from noteflow.infrastructure.persistence.repositories import DiarizationJob
    from noteflow.infrastructure.persistence.repositories.preferences_repo import (
        PreferenceWithMetadata,
    )
    from noteflow.infrastructure.persistence.unit_of_work import SqlAlchemyUnitOfWork
    from noteflow.infrastructure.security.crypto import AesGcmCryptoBox

    from ..meeting_store import MeetingStore
    from ..proto import noteflow_pb2
    from ..stream_state import MeetingStreamState
    from ._types import GrpcContext, GrpcStatusContext
    from .streaming._types import StreamSessionInit


class ServicerHost(Protocol):
    """Protocol defining shared state and methods for service mixins.

    All mixins should type-hint `self` as `ServicerHost` to access these
    attributes and methods from the host NoteFlowServicer class.
    """

    # Configuration
    session_factory: async_sessionmaker[AsyncSession] | None
    memory_store: MeetingStore | None
    meetings_dir: Path
    crypto: AesGcmCryptoBox

    # Engines and services
    asr_engine: FasterWhisperEngine | None
    diarization_engine: DiarizationEngine | None
    summarization_service: SummarizationService | None
    ner_service: NerService | None
    calendar_service: CalendarService | None
    webhook_service: WebhookService | None
    project_service: ProjectService | None
    diarization_refinement_enabled: bool

    # Audio writers
    audio_writers: dict[str, MeetingAudioWriter]
    audio_write_failed: set[str]

    # VAD and segmentation state per meeting
    vad_instances: dict[str, StreamingVad]
    segmenters: dict[str, Segmenter]
    was_speaking: dict[str, bool]
    segment_counters: dict[str, int]
    stream_formats: dict[str, tuple[int, int]]
    active_streams: set[str]
    stop_requested: set[str]  # Meeting IDs with pending stop requests

    # Chunk sequence tracking for acknowledgments
    chunk_sequences: dict[str, int]  # Highest received sequence per meeting
    chunk_counts: dict[str, int]  # Chunks since last ack (emit ack every 5)
    chunk_receipt_times: dict[str, deque[float]]  # Receipt timestamps per meeting
    pending_chunks: dict[str, int]  # Pending chunks counter per meeting

    # Partial transcription state per meeting
    partial_buffers: dict[str, PartialAudioBuffer]
    last_partial_time: dict[str, float]
    last_partial_text: dict[str, str]

    # Streaming diarization state per meeting
    diarization_turns: dict[str, list[SpeakerTurn]]
    diarization_stream_time: dict[str, float]
    diarization_streaming_failed: set[str]
    diarization_sessions: dict[str, DiarizationSession]

    # Consolidated per-meeting streaming state (single lookup replaces 13+ dict accesses)
    stream_states: dict[str, MeetingStreamState]

    # Background diarization task references (for cancellation)
    diarization_jobs: dict[str, DiarizationJob]
    diarization_tasks: dict[str, asyncio.Task[None]]
    diarization_lock: asyncio.Lock
    stream_init_lock: asyncio.Lock  # Guards concurrent stream initialization

    # Integration sync runs cache
    sync_runs: dict[UUID, SyncRun]
    # Track when each sync run was cached (Sprint GAP-002: State Synchronization)
    sync_run_cache_times: dict[UUID, datetime]

    # Constants
    DEFAULT_SAMPLE_RATE: Final[int]
    SUPPORTED_SAMPLE_RATES: ClassVar[list[int]]  # Converted to frozenset when passed to validate_stream_format
    PARTIAL_CADENCE_SECONDS: Final[float]
    MIN_PARTIAL_AUDIO_SECONDS: Final[float]

    @property
    def diarization_job_ttl_seconds(self) -> float:
        """Return diarization job TTL from settings."""
        ...

    def use_database(self) -> bool:
        """Check if database persistence is configured."""
        ...

    def get_memory_store(self) -> MeetingStore:
        """Get the in-memory store, raising if not configured."""
        ...

    def create_uow(self) -> SqlAlchemyUnitOfWork:
        """Create a new Unit of Work (database-backed)."""
        ...

    def create_repository_provider(self) -> UnitOfWork:
        """Create a repository provider (database or memory backed).

        Returns a UnitOfWork implementation appropriate for the current
        configuration. Use this for operations that can work with either
        backend, eliminating the need for if/else branching.

        Returns:
            SqlAlchemyUnitOfWork if database configured, MemoryUnitOfWork otherwise.
        """
        ...

    def next_segment_id(self, meeting_id: str, fallback: int = 0) -> int:
        """Get and increment the next segment id for a meeting."""
        ...

    def init_streaming_state(self, meeting_id: str, next_segment_id: int) -> None:
        """Initialize VAD, Segmenter, speaking state, and partial buffers."""
        ...

    def cleanup_streaming_state(self, meeting_id: str) -> None:
        """Clean up streaming state for a meeting."""
        ...

    def get_stream_state(self, meeting_id: str) -> MeetingStreamState | None:
        """Get consolidated streaming state for a meeting.

        Returns None if meeting has no active stream state.
        Single lookup replaces 13+ dict accesses in hot paths.
        """
        ...

    def ensure_meeting_dek(self, meeting: Meeting) -> tuple[bytes, bytes, bool]:
        """Ensure meeting has a DEK, generating one if needed."""
        ...

    def start_meeting_if_needed(self, meeting: Meeting) -> tuple[bool, str | None]:
        """Start recording on meeting if not already recording."""
        ...

    def open_meeting_audio_writer(
        self,
        meeting_id: str,
        dek: bytes,
        wrapped_dek: bytes,
        asset_path: str | None = None,
    ) -> None:
        """Open audio writer for a meeting."""
        ...

    def close_audio_writer(self, meeting_id: str) -> None:
        """Close and remove the audio writer for a meeting."""
        ...

    # Diarization mixin methods (for internal cross-references)
    async def prune_diarization_jobs(self) -> None:
        """Prune expired diarization jobs from in-memory cache."""
        ...

    async def run_diarization_job(self, job_id: str, num_speakers: int | None) -> None:
        """Run background diarization job."""
        ...

    async def collect_speaker_ids(self, meeting_id: str) -> list[str]:
        """Collect unique speaker IDs for a meeting."""
        ...

    def run_diarization_inference(
        self,
        meeting_id: str,
        num_speakers: int | None,
    ) -> list[SpeakerTurn]:
        """Run diarization inference synchronously."""
        ...

    async def apply_diarization_turns(
        self,
        meeting_id: str,
        turns: list[SpeakerTurn],
    ) -> int:
        """Apply diarization turns to meeting segments."""
        ...

    async def refine_speaker_diarization(
        self,
        meeting_id: str,
        num_speakers: int | None = None,
    ) -> int:
        """Run post-meeting speaker diarization refinement."""
        ...

    # Diarization job management methods
    async def update_job_completed(
        self,
        job_id: str,
        job: DiarizationJob | None,
        updated_count: int,
        speaker_ids: list[str],
    ) -> None:
        """Update job status to COMPLETED."""
        ...

    async def handle_job_timeout(
        self,
        job_id: str,
        job: DiarizationJob | None,
        meeting_id: str | None,
    ) -> None:
        """Handle job timeout."""
        ...

    async def handle_job_cancelled(
        self,
        job_id: str,
        job: DiarizationJob | None,
        meeting_id: str | None,
    ) -> None:
        """Handle job cancellation."""
        ...

    async def handle_job_failed(
        self,
        job_id: str,
        job: DiarizationJob | None,
        meeting_id: str | None,
        exc: Exception,
    ) -> None:
        """Handle job failure."""
        ...

    async def start_diarization_job(
        self,
        request: noteflow_pb2.RefineSpeakerDiarizationRequest,
        context: GrpcStatusContext,
    ) -> noteflow_pb2.RefineSpeakerDiarizationResponse:
        """Start a new diarization refinement job."""
        ...

    async def persist_streaming_turns(
        self,
        meeting_id: str,
        new_turns: list[SpeakerTurn],
    ) -> None:
        """Persist streaming turns to database (fire-and-forget)."""
        ...

    async def process_streaming_diarization(
        self,
        meeting_id: str,
        audio: NDArray[np.float32],
    ) -> None:
        """Process audio chunk for streaming diarization (best-effort)."""
        ...

    # Webhook methods
    async def fire_stop_webhooks(self, meeting: Meeting) -> None:
        """Trigger webhooks for meeting stop (fire-and-forget)."""
        ...

    # OIDC service
    oidc_service: OidcAuthService | None

    def get_oidc_service(self) -> OidcAuthService:
        """Get or create the OIDC auth service."""
        ...

    # Preferences methods
    async def decode_and_validate_prefs(
        self,
        request: noteflow_pb2.SetPreferencesRequest,
        context: GrpcContext,
    ) -> dict[str, object]:
        """Decode and validate JSON preferences from request."""
        ...

    async def apply_preferences(
        self,
        repo: PreferencesRepositoryProvider,
        request: noteflow_pb2.SetPreferencesRequest,
        current_prefs: list[PreferenceWithMetadata],
        decoded_prefs: dict[str, object],
    ) -> None:
        """Apply preferences based on merge mode."""
        ...

    # Streaming methods
    async def init_stream_for_meeting(
        self,
        meeting_id: str,
        context: GrpcContext,
    ) -> StreamSessionInit | None:
        """Initialize streaming for a meeting."""
        ...

    def process_stream_chunk(
        self,
        meeting_id: str,
        chunk: noteflow_pb2.AudioChunk,
        context: GrpcContext,
    ) -> AsyncIterator[noteflow_pb2.TranscriptUpdate]:
        """Process a single audio chunk from the stream."""
        ...

    def flush_segmenter(
        self,
        meeting_id: str,
    ) -> AsyncIterator[noteflow_pb2.TranscriptUpdate]:
        """Flush remaining audio from segmenter at stream end."""
        ...

    # Summarization methods
    async def summarize_or_placeholder(
        self,
        meeting_id: MeetingId,
        segments: list[Segment],
        style_prompt: str | None = None,
    ) -> Summary:
        """Try to summarize via service, fallback to placeholder on failure."""
        ...

    def generate_placeholder_summary(
        self,
        meeting_id: MeetingId,
        segments: list[Segment],
    ) -> Summary:
        """Generate a lightweight placeholder summary when summarization fails."""
        ...

    # Sync mixin methods
    def ensure_sync_runs_cache(self) -> dict[UUID, SyncRun]:
        """Ensure the sync runs cache exists."""
        ...

    def cache_sync_run(self, sync_run: SyncRun) -> None:
        """Cache a sync run with timestamp tracking (Sprint GAP-002)."""
        ...

    def get_sync_run_expires_at(self, sync_run_id: UUID) -> str | None:
        """Get expiry timestamp for a cached sync run (Sprint GAP-002)."""
        ...

    async def resolve_integration(
        self,
        uow: UnitOfWork,
        integration_id: UUID,
        context: GrpcContext,
        request: noteflow_pb2.StartIntegrationSyncRequest,
    ) -> tuple[Integration | None, UUID]:
        """Resolve integration by ID with provider fallback."""
        ...

    async def perform_sync(
        self,
        integration_id: UUID,
        sync_run_id: UUID,
        provider: str,
    ) -> None:
        """Perform the actual sync operation (background task)."""
        ...

    async def execute_sync_fetch(self, provider: str) -> int:
        """Execute the calendar fetch and return items count."""
        ...

    async def complete_sync_run(
        self,
        integration_id: UUID,
        sync_run_id: UUID,
        items_synced: int,
    ) -> SyncRun | None:
        """Mark sync run as complete and update integration last_sync."""
        ...

    async def fail_sync_run(
        self,
        sync_run_id: UUID,
        error_message: str,
    ) -> SyncRun | None:
        """Mark sync run as failed with error message."""
        ...
