"""Signal handling and graceful shutdown tests.

Tests verify the server shuts down gracefully on SIGTERM/SIGINT,
properly cleaning up resources and active streams.
"""


from __future__ import annotations

import asyncio
from pathlib import Path
from unittest.mock import MagicMock

import pytest

from noteflow.config.constants import DEFAULT_SAMPLE_RATE
from noteflow.grpc.service import NoteFlowServicer


def _filter_not_done(tasks: list[asyncio.Task[None]]) -> list[asyncio.Task[None]]:
    """Return tasks that are not yet done."""
    return [t for t in tasks if not t.done()]


def _filter_done_not_cancelled(tasks: list[asyncio.Task[None]]) -> list[asyncio.Task[None]]:
    """Return tasks that are done but not cancelled."""
    return [t for t in tasks if t.done() and not t.cancelled()]


class TestServicerShutdown:
    """Test servicer shutdown behavior."""

    @pytest.mark.asyncio
    async def test_shutdown_with_noactive_streams(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify shutdown works when no streams are active."""
        # Empty state
        assert len(memory_servicer.active_streams) == 0, (
            f"expected no active streams, got {len(memory_servicer.active_streams)}"
        )
        assert len(memory_servicer.diarization_tasks) == 0, (
            f"expected no diarization tasks, got {len(memory_servicer.diarization_tasks)}"
        )

        # Shutdown should complete without error
        await memory_servicer.shutdown()

        # State should still be empty
        assert len(memory_servicer.active_streams) == 0, (
            f"expected no active streams after shutdown, got {len(memory_servicer.active_streams)}"
        )

    @pytest.mark.asyncio
    async def test_shutdown_cleansactive_streams(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify shutdown cleans up all active streaming state."""
        # Create multiple active streams
        for i in range(5):
            meeting_id = f"shutdown-stream-{i:03d}"
            memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)
            memory_servicer.active_streams.add(meeting_id)

            # Add mock diarization session
            mock_session = MagicMock()
            mock_session.close = MagicMock()
            memory_servicer.diarization_sessions[meeting_id] = mock_session

        assert len(memory_servicer.active_streams) == 5, (
            f"expected 5 active streams, got {len(memory_servicer.active_streams)}"
        )
        assert len(memory_servicer.diarization_sessions) == 5, (
            f"expected 5 diarization sessions, got {len(memory_servicer.diarization_sessions)}"
        )

        # Shutdown
        await memory_servicer.shutdown()

        # Verify all sessions closed
        assert len(memory_servicer.diarization_sessions) == 0, (
            f"expected all diarization sessions closed, got {len(memory_servicer.diarization_sessions)}"
        )

    @pytest.mark.asyncio
    async def test_shutdown_cancelsdiarization_tasks(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify shutdown cancels all pending diarization tasks."""
        # Create some diarization tasks
        tasks_created: list[asyncio.Task[None]] = []
        for i in range(3):
            task = asyncio.create_task(asyncio.sleep(100))
            memory_servicer.diarization_tasks[f"job-{i}"] = task
            tasks_created.append(task)

        assert len(memory_servicer.diarization_tasks) == 3, (
            f"expected 3 diarization tasks, got {len(memory_servicer.diarization_tasks)}"
        )

        # Shutdown should cancel all
        await memory_servicer.shutdown()

        assert len(memory_servicer.diarization_tasks) == 0, (
            f"expected no diarization tasks after shutdown, got {len(memory_servicer.diarization_tasks)}"
        )

        # Verify tasks are cancelled - collect non-done/non-cancelled tasks
        not_done = _filter_not_done(tasks_created)
        not_cancelled = _filter_done_not_cancelled(tasks_created)
        assert not not_done, f"all tasks should be done after shutdown, {len(not_done)} still running"
        assert not not_cancelled, f"all tasks should be cancelled, {len(not_cancelled)} completed instead"

    @pytest.mark.asyncio
    async def test_shutdown_marks_cancelled_jobs_failed(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify cancelled diarization jobs are marked as FAILED."""
        from noteflow.grpc.proto import noteflow_pb2
        from noteflow.infrastructure.persistence.repositories import DiarizationJob

        # Create task and job
        task = asyncio.create_task(asyncio.sleep(100))
        job_id = "job-cancel-test"
        memory_servicer.diarization_tasks[job_id] = task

        job = DiarizationJob(
            job_id=job_id,
            meeting_id="meeting-001",
            status=noteflow_pb2.JOB_STATUS_RUNNING,
        )
        memory_servicer.diarization_jobs[job_id] = job

        # Shutdown
        await memory_servicer.shutdown()

        # Verify job marked as failed
        assert job.status == noteflow_pb2.JOB_STATUS_FAILED, (
            f"expected job status FAILED, got {job.status}"
        )
        assert job.error_message == "ERR_TASK_CANCELLED", (
            f"expected error message 'ERR_TASK_CANCELLED', got '{job.error_message}'"
        )

    @pytest.mark.asyncio
    async def test_shutdown_idempotent(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify shutdown can be called multiple times safely."""
        # First shutdown
        await memory_servicer.shutdown()

        # Second shutdown should not raise
        await memory_servicer.shutdown()

        # Third shutdown should not raise
        await memory_servicer.shutdown()

    @pytest.mark.asyncio
    async def test_shutdown_closesaudio_writers(
        self, memory_servicer: NoteFlowServicer, tmp_path: Path
    ) -> None:
        """Verify shutdown closes all audio writers."""
        from noteflow.infrastructure.audio.writer import MeetingAudioWriter
        from noteflow.infrastructure.security.crypto import AesGcmCryptoBox
        from noteflow.infrastructure.security.keystore import InMemoryKeyStore

        # Create and open an audio writer
        crypto = AesGcmCryptoBox(InMemoryKeyStore())
        writer = MeetingAudioWriter(crypto, tmp_path, buffer_size=1024)
        dek = crypto.generate_dek()
        wrapped_dek = crypto.wrap_dek(dek)

        meeting_id = "audio-shutdown-test"
        writer.open(meeting_id, dek, wrapped_dek, sample_rate=DEFAULT_SAMPLE_RATE)

        memory_servicer.audio_writers[meeting_id] = writer
        assert writer.is_recording, "expected audio writer to be recording after open"

        # Shutdown
        await memory_servicer.shutdown()

        # Verify writer was closed and removed
        assert meeting_id not in memory_servicer.audio_writers, (
            f"expected meeting '{meeting_id}' removed from audio writers after shutdown"
        )
        assert not writer.is_recording, "expected audio writer to stop recording after shutdown"


class TestStreamingStateCleanup:
    """Test streaming state cleanup during shutdown-like scenarios."""

    @pytest.mark.asyncio
    async def test_cleanup_allactive_streams(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify all streaming state is cleaned up."""
        # Create multiple streams with various states
        meeting_ids: list[str] = []
        for i in range(10):
            meeting_id = f"cleanup-test-{i:03d}"
            meeting_ids.append(meeting_id)
            memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)
            memory_servicer.active_streams.add(meeting_id)

        # Verify state exists
        assert len(memory_servicer.active_streams) == 10, (
            f"expected 10 active streams, got {len(memory_servicer.active_streams)}"
        )
        assert len(memory_servicer.vad_instances) == 10, (
            f"expected 10 VAD instances, got {len(memory_servicer.vad_instances)}"
        )
        assert len(memory_servicer.segmenters) == 10, (
            f"expected 10 segmenters, got {len(memory_servicer.segmenters)}"
        )

        # Clean up all streams
        for meeting_id in meeting_ids:
            memory_servicer.cleanup_streaming_state(meeting_id)
            memory_servicer.active_streams.discard(meeting_id)

        # Verify all state cleaned
        assert len(memory_servicer.active_streams) == 0, (
            f"expected no active streams after cleanup, got {len(memory_servicer.active_streams)}"
        )
        assert len(memory_servicer.vad_instances) == 0, (
            f"expected no VAD instances after cleanup, got {len(memory_servicer.vad_instances)}"
        )
        assert len(memory_servicer.segmenters) == 0, (
            f"expected no segmenters after cleanup, got {len(memory_servicer.segmenters)}"
        )
        assert len(memory_servicer.partial_buffers) == 0, (
            f"expected no partial buffers after cleanup, got {len(memory_servicer.partial_buffers)}"
        )

    @pytest.mark.asyncio
    async def test_cleanup_withdiarization_sessions(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify diarization sessions are closed during cleanup."""
        meeting_id = "diarize-cleanup-test"

        # Initialize state
        memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)

        # Add mock diarization session
        mock_session = MagicMock()
        mock_session.close = MagicMock()
        memory_servicer.diarization_sessions[meeting_id] = mock_session

        # Cleanup
        memory_servicer.cleanup_streaming_state(meeting_id)

        # Verify session was closed
        mock_session.close.assert_called_once()
        assert meeting_id not in memory_servicer.diarization_sessions, (
            f"expected meeting '{meeting_id}' removed from diarization sessions after cleanup"
        )


class TestTaskCancellation:
    """Test asyncio task cancellation behavior."""

    @pytest.mark.asyncio
    async def test_long_running_task_cancellation(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify long-running tasks are properly cancelled."""
        # Create a task that would run for a long time
        task = asyncio.create_task(asyncio.sleep(3600))  # 1 hour
        memory_servicer.diarization_tasks["long-task"] = task

        # Cancel via shutdown
        await memory_servicer.shutdown()

        # Task should be cancelled and done
        assert task.done(), "expected long-running task to be done after shutdown"
        assert task.cancelled(), "expected long-running task to be cancelled after shutdown"

    @pytest.mark.asyncio
    async def test_task_with_exception_handling(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify tasks that raise exceptions are handled gracefully."""

        async def failing_task() -> None:
            await asyncio.sleep(0.1)
            raise ValueError("Task failed")

        task = asyncio.create_task(failing_task())
        memory_servicer.diarization_tasks["failing-job"] = task

        # Wait for task to fail
        with pytest.raises(ValueError, match="Task failed"):
            await task

        # Verify task is done (not stuck)
        assert task.done(), "expected failing task to be done after exception"

        # Cleanup should still work
        await memory_servicer.shutdown()
        assert len(memory_servicer.diarization_tasks) == 0, (
            f"expected no diarization tasks after shutdown, got {len(memory_servicer.diarization_tasks)}"
        )

    @pytest.mark.asyncio
    async def test_mixed_task_states_on_shutdown(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify shutdown handles mix of running, done, and cancelled tasks."""
        # Running task
        running = asyncio.create_task(asyncio.sleep(100))
        memory_servicer.diarization_tasks["running"] = running

        # Already completed task
        completed = asyncio.create_task(asyncio.sleep(0))
        await completed
        memory_servicer.diarization_tasks["completed"] = completed

        # Pre-cancelled task
        cancelled = asyncio.create_task(asyncio.sleep(100))
        cancelled.cancel()
        memory_servicer.diarization_tasks["cancelled"] = cancelled

        # Shutdown should handle all states
        await memory_servicer.shutdown()

        assert len(memory_servicer.diarization_tasks) == 0, (
            f"expected no diarization tasks after shutdown, got {len(memory_servicer.diarization_tasks)}"
        )


class TestResourceCleanupOrder:
    """Test that resources are cleaned up in correct order."""

    @pytest.mark.asyncio
    async def test_diarization_before_audio(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify diarization cleaned before audio writers."""
        cleanup_order: list[str] = []

        # Mock diarization session
        mock_session = MagicMock()

        def record_session_close() -> None:
            cleanup_order.append("diarization")

        mock_session.close = record_session_close

        # Mock audio writer
        mock_writer = MagicMock()
        mock_writer.is_recording = True

        def record_writer_close() -> None:
            cleanup_order.append("audio")
            mock_writer.is_recording = False

        mock_writer.close = record_writer_close

        meeting_id = "order-test"
        memory_servicer.diarization_sessions[meeting_id] = mock_session
        memory_servicer.audio_writers[meeting_id] = mock_writer

        # Shutdown
        await memory_servicer.shutdown()

        # Diarization should be closed before audio (based on shutdown() order)
        assert "diarization" in cleanup_order, (
            f"expected 'diarization' in cleanup order, got {cleanup_order}"
        )
        assert "audio" in cleanup_order, (
            f"expected 'audio' in cleanup order, got {cleanup_order}"
        )

    @pytest.mark.asyncio
    async def test_tasks_cancelled_before_sessions_closed(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify tasks are cancelled before sessions are closed."""
        # Create task
        task = asyncio.create_task(asyncio.sleep(100))
        memory_servicer.diarization_tasks["test-task"] = task

        # Create session
        mock_session = MagicMock()
        mock_session.close = MagicMock()
        memory_servicer.diarization_sessions["test-meeting"] = mock_session

        # Shutdown
        await memory_servicer.shutdown()

        # Both should be cleaned up
        assert task.done(), "expected task to be done after shutdown"
        mock_session.close.assert_called_once()


class TestConcurrentShutdown:
    """Test concurrent shutdown scenarios."""

    @pytest.mark.asyncio
    async def test_concurrent_shutdown_calls_safe(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify concurrent shutdown calls don't cause issues."""
        # Create some state
        for i in range(3):
            task = asyncio.create_task(asyncio.sleep(100))
            memory_servicer.diarization_tasks[f"job-{i}"] = task

        # Call shutdown concurrently
        await asyncio.gather(
            memory_servicer.shutdown(),
            memory_servicer.shutdown(),
            memory_servicer.shutdown(),
        )

        # Should be clean
        assert len(memory_servicer.diarization_tasks) == 0, (
            f"expected no diarization tasks after concurrent shutdowns, got {len(memory_servicer.diarization_tasks)}"
        )

    @pytest.mark.asyncio
    async def test_new_operations_during_shutdown(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify new operations don't interfere with shutdown."""
        # Start shutdown in background
        shutdown_task = asyncio.create_task(memory_servicer.shutdown())

        # Add new state (simulating race condition)
        memory_servicer.active_streams.add("late-stream")

        # Complete shutdown
        await shutdown_task

        # Late stream might still be there, but shutdown shouldn't fail
        # (Real implementation would need a shutdown flag to prevent new streams)
