"""Stress tests for NoteFlowServicer concurrent stream handling.

Detects race conditions when multiple clients stream simultaneously.
Verifies cleanup_streaming_state prevents memory leaks.
"""

from __future__ import annotations

from uuid import uuid4

import pytest

from noteflow.grpc.service import NoteFlowServicer

# Test constants
MANY_MEETINGS_COUNT = 500
"""Number of concurrent meetings for stress tests."""


class TestStreamingStateInitialization:
    """Test streaming state initialization correctness."""

    @pytest.mark.stress
    def testinit_streaming_state_creates_all_state(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Initialize streaming state creates entries in all state dictionaries."""
        meeting_id = str(uuid4())

        memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)

        # Verify all state dictionaries have the meeting_id entry
        state_dicts = {
            "partial_buffers": memory_servicer.partial_buffers,
            "vad_instances": memory_servicer.vad_instances,
            "segmenters": memory_servicer.segmenters,
            "was_speaking": memory_servicer.was_speaking,
            "segment_counters": memory_servicer.segment_counters,
            "last_partial_time": memory_servicer.last_partial_time,
            "last_partial_text": memory_servicer.last_partial_text,
            "diarization_turns": memory_servicer.diarization_turns,
            "diarization_stream_time": memory_servicer.diarization_stream_time,
        }
        for name, state_dict in state_dicts.items():
            assert meeting_id in state_dict, f"{name} missing meeting_id after init"

        memory_servicer.cleanup_streaming_state(meeting_id)

    @pytest.mark.stress
    def test_init_with_different_segment_ids(self, memory_servicer: NoteFlowServicer) -> None:
        """Initialize with different segment IDs sets counter correctly."""
        meeting_id1 = str(uuid4())
        meeting_id2 = str(uuid4())

        memory_servicer.init_streaming_state(meeting_id1, next_segment_id=0)
        memory_servicer.init_streaming_state(meeting_id2, next_segment_id=42)

        expected_counter1 = 0
        expected_counter2 = 42
        assert memory_servicer.segment_counters[meeting_id1] == expected_counter1, "meeting1 counter should be 0"
        assert memory_servicer.segment_counters[meeting_id2] == expected_counter2, "meeting2 counter should be 42"

        memory_servicer.cleanup_streaming_state(meeting_id1)
        memory_servicer.cleanup_streaming_state(meeting_id2)


class TestCleanupStreamingState:
    """Test cleanup_streaming_state removes all per-meeting resources."""

    @pytest.mark.stress
    def test_cleanup_removes_all_state_dictionaries(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify cleanup removes entries from all state dictionaries."""
        meeting_id = str(uuid4())

        memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)
        memory_servicer.active_streams.add(meeting_id)

        # Verify state was created
        assert meeting_id in memory_servicer.partial_buffers, "partial_buffers should have entry"
        assert meeting_id in memory_servicer.vad_instances, "vad_instances should have entry"
        assert meeting_id in memory_servicer.segmenters, "segmenters should have entry"

        memory_servicer.cleanup_streaming_state(meeting_id)
        memory_servicer.active_streams.discard(meeting_id)

        # Verify all state dictionaries are cleaned
        state_to_check = [
            ("partial_buffers", memory_servicer.partial_buffers),
            ("vad_instances", memory_servicer.vad_instances),
            ("segmenters", memory_servicer.segmenters),
            ("was_speaking", memory_servicer.was_speaking),
            ("segment_counters", memory_servicer.segment_counters),
            ("stream_formats", memory_servicer.stream_formats),
            ("last_partial_time", memory_servicer.last_partial_time),
            ("last_partial_text", memory_servicer.last_partial_text),
            ("diarization_turns", memory_servicer.diarization_turns),
            ("diarization_stream_time", memory_servicer.diarization_stream_time),
            ("diarization_streaming_failed", memory_servicer.diarization_streaming_failed),
            ("active_streams", memory_servicer.active_streams),
        ]
        for name, state_dict in state_to_check:
            assert meeting_id not in state_dict, f"{name} still has meeting_id after cleanup"

    @pytest.mark.stress
    def test_cleanup_idempotent(self, memory_servicer: NoteFlowServicer) -> None:
        """Cleanup is idempotent - multiple calls don't raise."""
        meeting_id = str(uuid4())

        memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)

        memory_servicer.cleanup_streaming_state(meeting_id)
        memory_servicer.cleanup_streaming_state(meeting_id)
        memory_servicer.cleanup_streaming_state(meeting_id)

    @pytest.mark.stress
    def test_cleanup_nonexistent_meeting_no_error(self, memory_servicer: NoteFlowServicer) -> None:
        """Cleanup of non-existent meeting ID doesn't raise."""
        nonexistent_id = str(uuid4())

        memory_servicer.cleanup_streaming_state(nonexistent_id)


class TestConcurrentStreamInitialization:
    """Test concurrent stream initialization."""

    @pytest.mark.stress
    @pytest.mark.asyncio
    async def test_concurrent_init_different_meetings(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Multiple concurrent init calls for different meetings succeed."""
        meeting_ids = [str(uuid4()) for _ in range(20)]

        def init_meeting(meeting_id: str, segment_id: int) -> None:
            memory_servicer.init_streaming_state(meeting_id, segment_id)

        # Run all initializations - synchronous operations don't need async
        for idx, mid in enumerate(meeting_ids):
            init_meeting(mid, idx)

        expected_count = len(meeting_ids)
        assert len(memory_servicer.vad_instances) == expected_count, "vad_instances count mismatch"
        assert len(memory_servicer.segmenters) == expected_count, "segmenters count mismatch"
        assert len(memory_servicer.partial_buffers) == expected_count, "partial_buffers count mismatch"

        for mid in meeting_ids:
            memory_servicer.cleanup_streaming_state(mid)

    @pytest.mark.stress
    @pytest.mark.asyncio
    async def test_concurrent_cleanup_different_meetings(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Multiple concurrent cleanup calls for different meetings succeed."""
        meeting_ids = [str(uuid4()) for _ in range(20)]

        for idx, mid in enumerate(meeting_ids):
            memory_servicer.init_streaming_state(mid, idx)

        def cleanup_meeting(meeting_id: str) -> None:
            memory_servicer.cleanup_streaming_state(meeting_id)

        # Run all cleanups - synchronous operations don't need async
        for mid in meeting_ids:
            cleanup_meeting(mid)

        expected_empty = 0
        assert len(memory_servicer.vad_instances) == expected_empty, "vad_instances should be empty"
        assert len(memory_servicer.segmenters) == expected_empty, "segmenters should be empty"
        assert len(memory_servicer.partial_buffers) == expected_empty, "partial_buffers should be empty"


class TestNoMemoryLeaksUnderLoad:
    """Test no memory leaks after many stream cycles."""

    @pytest.mark.stress
    def test_stream_cycles_cleanup_completely(self, memory_servicer: NoteFlowServicer) -> None:
        """Many init/cleanup cycles leave no leaked state."""
        for _ in range(100):
            meeting_id = str(uuid4())
            memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)
            memory_servicer.active_streams.add(meeting_id)
            memory_servicer.cleanup_streaming_state(meeting_id)
            memory_servicer.active_streams.discard(meeting_id)

        # Verify all state dictionaries are empty after cleanup cycles
        state_dicts = {
            "active_streams": memory_servicer.active_streams,
            "partial_buffers": memory_servicer.partial_buffers,
            "vad_instances": memory_servicer.vad_instances,
            "segmenters": memory_servicer.segmenters,
            "was_speaking": memory_servicer.was_speaking,
            "segment_counters": memory_servicer.segment_counters,
            "last_partial_time": memory_servicer.last_partial_time,
            "last_partial_text": memory_servicer.last_partial_text,
            "diarization_turns": memory_servicer.diarization_turns,
            "diarization_stream_time": memory_servicer.diarization_stream_time,
            "diarization_streaming_failed": memory_servicer.diarization_streaming_failed,
        }
        for name, state_dict in state_dicts.items():
            assert len(state_dict) == 0, f"{name} should be empty after cleanup cycles"

    @pytest.mark.stress
    @pytest.mark.slow
    def test_many_concurrent_meetings_no_leak(self, memory_servicer: NoteFlowServicer) -> None:
        """Many meetings initialized then cleaned up leave no state."""
        meeting_ids = [str(uuid4()) for _ in range(MANY_MEETINGS_COUNT)]

        for idx, mid in enumerate(meeting_ids):
            memory_servicer.init_streaming_state(mid, idx)
            memory_servicer.active_streams.add(mid)

        assert len(memory_servicer.vad_instances) == MANY_MEETINGS_COUNT, "VAD instances mismatch"
        assert len(memory_servicer.segmenters) == MANY_MEETINGS_COUNT, "segmenters mismatch"

        for mid in meeting_ids:
            memory_servicer.cleanup_streaming_state(mid)
            memory_servicer.active_streams.discard(mid)

        assert len(memory_servicer.active_streams) == 0, "active_streams should be empty"
        assert len(memory_servicer.vad_instances) == 0, "vad_instances should be empty"
        assert len(memory_servicer.segmenters) == 0, "segmenters should be empty"

    @pytest.mark.stress
    @pytest.mark.asyncio
    async def test_interleaved_init_cleanup(self, memory_servicer: NoteFlowServicer) -> None:
        """Interleaved init and cleanup doesn't leak or corrupt."""
        for _ in range(50):
            meeting_ids = [str(uuid4()) for _ in range(10)]

            for idx, mid in enumerate(meeting_ids):
                memory_servicer.init_streaming_state(mid, idx)

            for mid in meeting_ids[:5]:
                memory_servicer.cleanup_streaming_state(mid)

            for mid in meeting_ids[5:]:
                memory_servicer.cleanup_streaming_state(mid)

        expected_empty = 0
        assert len(memory_servicer.vad_instances) == expected_empty, "vad_instances should be empty"
        assert len(memory_servicer.segmenters) == expected_empty, "segmenters should be empty"


class TestActiveStreamsTracking:
    """Test active_streams set behavior."""

    @pytest.mark.stress
    def testactive_streams_tracks_active_meetings(self, memory_servicer: NoteFlowServicer) -> None:
        """active_streams correctly tracks active meeting IDs."""
        meeting_ids = [str(uuid4()) for _ in range(5)]

        for mid in meeting_ids:
            memory_servicer.active_streams.add(mid)

        assert len(memory_servicer.active_streams) == 5, "should have 5 active streams"
        for mid in meeting_ids:
            assert mid in memory_servicer.active_streams, f"{mid} should be in active_streams"

        for mid in meeting_ids:
            memory_servicer.active_streams.discard(mid)

        assert len(memory_servicer.active_streams) == 0, "active_streams should be empty"

    @pytest.mark.stress
    def test_discard_nonexistent_no_error(self, memory_servicer: NoteFlowServicer) -> None:
        """Discarding non-existent meeting ID doesn't raise."""
        nonexistent = str(uuid4())
        memory_servicer.active_streams.discard(nonexistent)


class TestDiarizationStateCleanup:
    """Test diarization-related state cleanup."""

    @pytest.mark.stress
    def test_diarization_failed_set_cleaned(self, memory_servicer: NoteFlowServicer) -> None:
        """diarization_streaming_failed set is cleaned on cleanup."""
        meeting_id = str(uuid4())

        memory_servicer.init_streaming_state(meeting_id, 0)
        memory_servicer.diarization_streaming_failed.add(meeting_id)

        assert meeting_id in memory_servicer.diarization_streaming_failed, "should be in failed set"

        memory_servicer.cleanup_streaming_state(meeting_id)

        assert meeting_id not in memory_servicer.diarization_streaming_failed, "should be cleaned"

    @pytest.mark.stress
    def testdiarization_turns_cleaned(self, memory_servicer: NoteFlowServicer) -> None:
        """diarization_turns dict is cleaned on cleanup."""
        meeting_id = str(uuid4())

        memory_servicer.init_streaming_state(meeting_id, 0)

        assert meeting_id in memory_servicer.diarization_turns, "should have turns entry"
        assert memory_servicer.diarization_turns[meeting_id] == [], "turns should be empty list"

        memory_servicer.cleanup_streaming_state(meeting_id)

        assert meeting_id not in memory_servicer.diarization_turns, "turns should be cleaned"


class TestServicerInstantiation:
    """Test NoteFlowServicer instantiation patterns."""

    @pytest.mark.stress
    def test_servicer_starts_with_empty_state(self) -> None:
        """New servicer has empty state dictionaries."""
        servicer = NoteFlowServicer()

        # Verify all state dictionaries start empty
        state_dicts = {
            "active_streams": servicer.active_streams,
            "partial_buffers": servicer.partial_buffers,
            "vad_instances": servicer.vad_instances,
            "segmenters": servicer.segmenters,
            "was_speaking": servicer.was_speaking,
            "segment_counters": servicer.segment_counters,
            "audio_writers": servicer.audio_writers,
        }
        for name, state_dict in state_dicts.items():
            assert len(state_dict) == 0, f"{name} should start empty"

    @pytest.mark.stress
    def test_multiple_servicers_independent(self) -> None:
        """Multiple servicer instances have independent state."""
        servicer1 = NoteFlowServicer()
        servicer2 = NoteFlowServicer()

        meeting_id = str(uuid4())
        servicer1.init_streaming_state(meeting_id, 0)

        assert meeting_id in servicer1.vad_instances, "servicer1 should have meeting"
        assert meeting_id not in servicer2.vad_instances, "servicer2 should not have meeting"

        servicer1.cleanup_streaming_state(meeting_id)


class TestMemoryStoreAccess:
    """Test memory store access patterns."""

    @pytest.mark.stress
    def testget_memory_store_returns_store(self, memory_servicer: NoteFlowServicer) -> None:
        """get_memory_store returns MeetingStore when configured."""
        store = memory_servicer.get_memory_store()
        assert store is not None, "memory store should be configured"

    @pytest.mark.stress
    def testmemory_store_create_meeting(self, memory_servicer: NoteFlowServicer) -> None:
        """Memory store can create and retrieve meetings."""
        store = memory_servicer.get_memory_store()

        meeting = store.create(title="Test Meeting")
        assert meeting is not None, "meeting should be created"
        assert meeting.title == "Test Meeting", "meeting should have correct title"

        retrieved = store.get(str(meeting.id))
        assert retrieved is not None, "meeting should be retrievable"
        assert retrieved.title == "Test Meeting", "retrieved meeting should have correct title"
