"""Resource leak detection tests.

Detects file descriptor, memory, and coroutine leaks under load conditions.
These tests verify that resources are properly released during cleanup cycles.
"""


from __future__ import annotations

import asyncio
import gc
import os
import sys
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

# Test constants
STREAMING_CYCLES = 50
AUDIO_WRITER_CYCLES = 20
MEMORY_TEST_CYCLES = 100
INTERLEAVE_CYCLES = 20
DIARIZATION_CYCLES = 50
WRITER_THREAD_CYCLES = 10
FD_LEAK_TOLERANCE = 10
THREAD_LEAK_TOLERANCE = 2
TASK_TEST_COUNT = 5
TASK_SLEEP_SECONDS = 10
WEBHOOK_EXECUTOR_CYCLES = 5


from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from threading import Thread

    import httpx


def _get_executor_client(executor: object) -> httpx.AsyncClient | None:
    """Access WebhookExecutor._client for stress test verification.

    Stress tests need to verify HTTP client cleanup state.
    Uses object.__getattribute__ to bypass pyright's protected access check.
    """
    client = object.__getattribute__(executor, "_client")
    # Runtime type narrowing for httpx.AsyncClient | None
    if client is None:
        return None
    import httpx as httpx_mod

    if isinstance(client, httpx_mod.AsyncClient):
        return client
    return None


async def _ensure_executor_client(executor: object) -> httpx.AsyncClient:
    """Call WebhookExecutor._ensure_client for stress test setup.

    Uses object.__getattribute__ to get the method, then calls it.
    """
    import httpx as httpx_mod

    method = object.__getattribute__(executor, "_ensure_client")
    result = await method()
    if not isinstance(result, httpx_mod.AsyncClient):
        msg = "Expected httpx.AsyncClient"
        raise TypeError(msg)
    return result


def _get_session_pipeline(session: object) -> object | None:
    """Access DiarizationSession._pipeline for stress test verification.

    Stress tests need to verify pipeline cleanup state.
    """
    return object.__getattribute__(session, "_pipeline")


def _get_writer_flush_thread(writer: object) -> Thread | None:
    """Access MeetingAudioWriter._flush_thread for stress test verification.

    Stress tests need to verify background thread cleanup.
    """
    from threading import Thread

    thread = object.__getattribute__(writer, "_flush_thread")
    if thread is None:
        return None
    if isinstance(thread, Thread):
        return thread
    return None


def _get_fd_count() -> int:
    """Get current file descriptor count for this process."""
    if sys.platform == "linux":
        return len(os.listdir(f"/proc/{os.getpid()}/fd"))
    elif sys.platform == "darwin":
        # macOS: use lsof to get actual FD count
        import subprocess

        try:
            output = subprocess.check_output(
                ["lsof", "-p", str(os.getpid())],
                stderr=subprocess.DEVNULL,
            )
            # Subtract 1 for header line
            return max(0, len(output.splitlines()) - 1)
        except (subprocess.CalledProcessError, FileNotFoundError):
            return -1  # Signal that counting failed
    # Other platforms: return -1 to skip test
    return -1


class TestFileDescriptorLeaks:
    """Detect file descriptor leaks."""

    @pytest.mark.slow
    @pytest.mark.asyncio
    async def test_streaming_fd_cleanup(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify file descriptors released after streaming cycles."""
        # Establish clean baseline
        gc.collect()
        initial_fds = _get_fd_count()
        if initial_fds < 0:
            pytest.skip("FD counting not supported on this platform")

        # Run streaming init/cleanup cycles
        for i in range(STREAMING_CYCLES):
            meeting_id = f"fd-test-{i:03d}"
            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.close_audio_writer(meeting_id)
            memory_servicer.active_streams.discard(meeting_id)

        # Force garbage collection
        gc.collect()
        await asyncio.sleep(0.1)  # Allow async cleanup

        final_fds = _get_fd_count()

        # Allow small variance (logging, etc.) but detect major leaks
        assert final_fds <= initial_fds + FD_LEAK_TOLERANCE, (
            f"File descriptor leak: started with {initial_fds}, ended with {final_fds}"
        )

    @pytest.mark.slow
    @pytest.mark.asyncio
    async def test_audio_writer_fd_cleanup(self, tmp_path: Path) -> None:
        """Verify audio writer closes file handles."""
        from noteflow.infrastructure.audio.writer import MeetingAudioWriter
        from noteflow.infrastructure.security.crypto import AesGcmCryptoBox
        from noteflow.infrastructure.security.keystore import InMemoryKeyStore

        # Establish clean baseline
        gc.collect()
        initial_fds = _get_fd_count()
        if initial_fds < 0:
            pytest.skip("FD counting not supported on this platform")

        crypto = AesGcmCryptoBox(InMemoryKeyStore())

        # Create and close writers
        for i in range(AUDIO_WRITER_CYCLES):
            writer = MeetingAudioWriter(crypto, tmp_path, buffer_size=1024)
            dek = crypto.generate_dek()
            wrapped_dek = crypto.wrap_dek(dek)
            writer.open(f"meeting-{i}", dek, wrapped_dek, sample_rate=DEFAULT_SAMPLE_RATE)
            writer.close()

        gc.collect()
        await asyncio.sleep(0.1)

        final_fds = _get_fd_count()
        assert final_fds <= initial_fds + FD_LEAK_TOLERANCE // 2, (
            f"Audio writer FD leak: {initial_fds} -> {final_fds}"
        )


class TestMemoryLeaks:
    """Detect memory leaks under load."""

    @pytest.mark.slow
    @pytest.mark.asyncio
    async def test_streaming_state_memory_stability(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify state dicts don't grow unbounded during streaming cycles."""
        for i in range(MEMORY_TEST_CYCLES):
            meeting_id = f"mem-test-{i:03d}"
            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)

        gc.collect()

        # Verify no lingering state
        assert len(memory_servicer.vad_instances) == 0, "VAD instances leaked"
        assert len(memory_servicer.segmenters) == 0, "Segmenters leaked"
        assert len(memory_servicer.partial_buffers) == 0, "Partial buffers leaked"
        assert len(memory_servicer.diarization_sessions) == 0, "Sessions leaked"
        assert len(memory_servicer.active_streams) == 0, "Active streams leaked"

    @pytest.mark.slow
    @pytest.mark.asyncio
    async def test_resource_interleaved_init_cleanup(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify memory stability with interleaved init/cleanup patterns."""
        active_meetings: list[str] = []

        # Interleaved pattern: init 3, cleanup 1, repeat
        for cycle in range(INTERLEAVE_CYCLES):
            # Init 3 meetings
            for j in range(3):
                meeting_id = f"interleave-{cycle}-{j}"
                memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)
                memory_servicer.active_streams.add(meeting_id)
                active_meetings.append(meeting_id)

            # Cleanup 1 oldest
            if active_meetings:
                old_meeting = active_meetings.pop(0)
                memory_servicer.cleanup_streaming_state(old_meeting)
                memory_servicer.active_streams.discard(old_meeting)

        # Cleanup remaining
        for meeting_id in active_meetings:
            memory_servicer.cleanup_streaming_state(meeting_id)
            memory_servicer.active_streams.discard(meeting_id)

        gc.collect()

        # Verify clean state
        assert len(memory_servicer.vad_instances) == 0, "VAD leaked after interleave"
        assert len(memory_servicer.active_streams) == 0, "Streams leaked after interleave"

    @pytest.mark.slow
    @pytest.mark.asyncio
    async def test_diarization_session_memory(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify diarization sessions don't leak memory."""
        # Create and close many sessions
        for i in range(DIARIZATION_CYCLES):
            meeting_id = f"diarize-mem-{i:03d}"
            mock_session = MagicMock()
            mock_session.close = MagicMock()

            memory_servicer.init_streaming_state(meeting_id, next_segment_id=0)
            memory_servicer.diarization_sessions[meeting_id] = mock_session
            memory_servicer.cleanup_streaming_state(meeting_id)

        gc.collect()

        assert len(memory_servicer.diarization_sessions) == 0, "Diarization sessions leaked"


class TestCoroutineLeaks:
    """Detect leaked coroutines/tasks."""

    @pytest.mark.asyncio
    async def test_no_orphaned_tasks_after_shutdown(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify no tasks remain after servicer shutdown."""
        # Create some diarization tasks with explicit type annotation
        tasks_created: list[asyncio.Task[None]] = []
        for i in range(TASK_TEST_COUNT):
            task: asyncio.Task[None] = asyncio.create_task(
                asyncio.sleep(TASK_SLEEP_SECONDS)
            )
            memory_servicer.diarization_tasks[f"job-{i}"] = task
            tasks_created.append(task)

        # Shutdown
        await memory_servicer.shutdown()

        # Verify all tasks cancelled and cleared
        assert len(memory_servicer.diarization_tasks) == 0

        # Check tasks are cancelled
        for task in tasks_created:
            assert task.done()
            assert task.cancelled()

    @pytest.mark.asyncio
    async def test_task_cleanup_on_exception(
        self, memory_servicer: NoteFlowServicer
    ) -> None:
        """Verify tasks cleaned up even if task raises exception."""

        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()

        # Cleanup
        await memory_servicer.shutdown()
        assert len(memory_servicer.diarization_tasks) == 0


class TestWebhookClientLeaks:
    """Detect webhook HTTP client leaks."""

    @pytest.mark.asyncio
    async def test_webhook_executor_cleanup(self) -> None:
        """Verify webhook executor closes HTTP client."""
        from noteflow.infrastructure.webhooks.executor import WebhookExecutor

        executor = WebhookExecutor()

        # Force client creation using type-safe helper
        await _ensure_executor_client(executor)
        assert _get_executor_client(executor) is not None

        # Close
        await executor.close()
        assert _get_executor_client(executor) is None

    @pytest.mark.asyncio
    async def test_webhook_executor_close_idempotent(self) -> None:
        """Verify webhook executor close is idempotent."""
        from noteflow.infrastructure.webhooks.executor import WebhookExecutor

        executor = WebhookExecutor()

        # Close without opening - should not raise
        await executor.close()
        await executor.close()
        await executor.close()

    @pytest.mark.asyncio
    async def test_webhook_executor_multiple_cycles(self) -> None:
        """Verify webhook executor can be opened and closed multiple times."""
        from noteflow.infrastructure.webhooks.executor import WebhookExecutor

        async def run_executor_cycle() -> tuple[bool, bool]:
            """Run one executor open/close cycle, return (client_created, client_cleared)."""
            executor = WebhookExecutor()
            await _ensure_executor_client(executor)
            client_created = _get_executor_client(executor) is not None
            await executor.close()
            client_cleared = _get_executor_client(executor) is None
            return client_created, client_cleared

        # Run cycles and collect results
        results = [await run_executor_cycle() for _ in range(WEBHOOK_EXECUTOR_CYCLES)]

        # Verify all cycles succeeded
        assert all(created for created, _ in results), "Client creation failed in some cycles"
        assert all(cleared for _, cleared in results), "Client cleanup failed in some cycles"


class TestDiarizationSessionLeaks:
    """Detect diarization session resource leaks."""

    @pytest.mark.asyncio
    async def test_session_close_releases_pipeline(self) -> None:
        """Verify diarization session close releases pipeline reference."""
        from noteflow.infrastructure.diarization.session import DiarizationSession

        # Create mock pipeline
        mock_pipeline = MagicMock()

        session = DiarizationSession(
            meeting_id="test-session",
            _pipeline=mock_pipeline,
        )

        assert _get_session_pipeline(session) is not None

        session.close()

        # Pipeline should be None after close
        assert _get_session_pipeline(session) is None
        assert session.is_closed

    @pytest.mark.asyncio
    async def test_session_close_idempotent(self) -> None:
        """Verify session close is idempotent."""
        from noteflow.infrastructure.diarization.session import DiarizationSession

        mock_pipeline = MagicMock()

        session = DiarizationSession(
            meeting_id="test-session",
            _pipeline=mock_pipeline,
        )

        # Multiple closes should not raise
        session.close()
        session.close()
        session.close()

        assert session.is_closed


class TestAudioWriterThreadLeaks:
    """Detect audio writer background thread leaks."""

    @pytest.mark.slow
    @pytest.mark.asyncio
    async def test_flush_thread_stopped_on_close(self, tmp_path: Path) -> None:
        """Verify background flush thread stops on close."""

        from noteflow.infrastructure.audio.writer import MeetingAudioWriter
        from noteflow.infrastructure.security.crypto import AesGcmCryptoBox
        from noteflow.infrastructure.security.keystore import InMemoryKeyStore

        crypto = AesGcmCryptoBox(InMemoryKeyStore())
        writer = MeetingAudioWriter(crypto, tmp_path, buffer_size=1024)
        dek = crypto.generate_dek()
        wrapped_dek = crypto.wrap_dek(dek)

        # Open writer - starts flush thread
        writer.open("test-thread", dek, wrapped_dek, sample_rate=DEFAULT_SAMPLE_RATE)

        # Verify thread is running using type-safe helper
        flush_thread = _get_writer_flush_thread(writer)
        assert flush_thread is not None
        assert flush_thread.is_alive()

        # Close writer
        writer.close()

        # Give thread time to stop
        await asyncio.sleep(0.1)

        # Thread should be stopped and cleared
        assert _get_writer_flush_thread(writer) is None

    @pytest.mark.slow
    @pytest.mark.asyncio
    async def test_multiple_writer_cycles(self, tmp_path: Path) -> None:
        """Verify no thread leaks across multiple writer cycles."""
        import threading

        from noteflow.infrastructure.audio.writer import MeetingAudioWriter
        from noteflow.infrastructure.security.crypto import AesGcmCryptoBox
        from noteflow.infrastructure.security.keystore import InMemoryKeyStore

        crypto = AesGcmCryptoBox(InMemoryKeyStore())
        initial_threads = threading.active_count()

        # Create and close writers
        for i in range(WRITER_THREAD_CYCLES):
            writer = MeetingAudioWriter(crypto, tmp_path, buffer_size=1024)
            dek = crypto.generate_dek()
            wrapped_dek = crypto.wrap_dek(dek)
            writer.open(f"thread-test-{i}", dek, wrapped_dek, sample_rate=DEFAULT_SAMPLE_RATE)
            writer.close()

        await asyncio.sleep(0.5)  # Allow threads to fully stop
        gc.collect()

        final_threads = threading.active_count()

        # Should not have leaked threads
        assert final_threads <= initial_threads + THREAD_LEAK_TOLERANCE, (
            f"Thread leak: {initial_threads} -> {final_threads}"
        )
