import json
import os
import shutil
import signal
import subprocess
import sys
import threading
import urllib.request
import wave
from pathlib import Path
from unittest.mock import MagicMock

import boto3
import pytest
from src.app import main, make_worker
from src.atmos.models import AtmosValidationFindingCode
from src.config import get_settings
from src.worker import Worker
from src.worker.clients.step_functions import StaleTaskToken
from src.worker.task_protection import TaskProtectionResult

# moto can't simulate ECS task protection or SFN callbacks — those are mocked.
# Everything else (SQS, S3, mediainfo on presigned URLs) runs for real.

pytestmark = pytest.mark.skipif(
    shutil.which("mediainfo") is None,
    reason="mediainfo not installed",
)

_FIXTURES_DIR = Path(__file__).parent.parent / "fixtures"
_DOLBY_RENDERER_FIXTURE = _FIXTURES_DIR / "dolby_renderer_silent_2s.wav"

_QUEUE_NAME = "spatial-audio-validation-integration"
_ATMOS_BUCKET = "test-atmos-bucket"
_ATMOS_KEY = "integration/atmos.wav"
_STEREO_BUCKET = "test-stereo-bucket"
_STEREO_KEY = "integration/stereo.flac"


@pytest.fixture(autouse=True)
def _clear_settings_cache() -> None:
    get_settings.cache_clear()


@pytest.fixture(autouse=True)
def _reset_moto() -> None:
    endpoint = os.environ["AWS_ENDPOINT_URL"]
    request = urllib.request.Request(f"{endpoint}/moto-api/reset", method="POST")
    urllib.request.urlopen(request, timeout=5)


def _write_stereo_wav(path: Path, num_samples: int = 96000) -> None:
    # 2 s of silent stereo (48 kHz, 24-bit), matching the atmos fixture duration.
    with wave.open(str(path), "wb") as writer:
        writer.setnchannels(2)
        writer.setsampwidth(3)
        writer.setframerate(48000)
        writer.writeframes(b"\x00" * num_samples * 2 * 3)


@pytest.fixture
def populated_s3(tmp_path: Path) -> None:
    s3 = boto3.client("s3")
    s3.create_bucket(Bucket=_ATMOS_BUCKET)
    s3.create_bucket(Bucket=_STEREO_BUCKET)

    s3.upload_file(str(_DOLBY_RENDERER_FIXTURE), _ATMOS_BUCKET, _ATMOS_KEY)

    stereo_path = tmp_path / "stereo.wav"
    _write_stereo_wav(stereo_path)
    s3.upload_file(str(stereo_path), _STEREO_BUCKET, _STEREO_KEY)


@pytest.fixture
def queue_url(monkeypatch: pytest.MonkeyPatch) -> str:
    response = boto3.client("sqs").create_queue(QueueName=_QUEUE_NAME)
    url = str(response["QueueUrl"])
    monkeypatch.setenv("ENVIRONMENT", "dev")
    monkeypatch.setenv("SQS_QUEUE_URL", url)
    monkeypatch.setenv("MESSAGE_PROCESSING_TIMEOUT_MINUTES", "30")
    monkeypatch.setenv("HEARTBEAT_INTERVAL_SECONDS", "60")
    return url


def _send(queue_url: str, task_token: str) -> None:
    body = {
        "task_token": task_token,
        "asset_upload_type": "atmos",
        "atmos_bucket": _ATMOS_BUCKET,
        "atmos_key": _ATMOS_KEY,
        "stereo_reference_bucket": _STEREO_BUCKET,
        "stereo_reference_key": _STEREO_KEY,
    }
    boto3.client("sqs").send_message(QueueUrl=queue_url, MessageBody=json.dumps(body))


def _queue_depth(queue_url: str) -> int:
    response = boto3.client("sqs").get_queue_attributes(
        QueueUrl=queue_url, AttributeNames=["ApproximateNumberOfMessages"]
    )
    return int(response["Attributes"]["ApproximateNumberOfMessages"])


@pytest.fixture
def mock_protection(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
    mock = MagicMock()
    mock.enable.return_value = TaskProtectionResult.ENABLED
    mock.disable.return_value = True
    monkeypatch.setattr("src.worker.worker.task_protection.enable", mock.enable)
    monkeypatch.setattr("src.worker.worker.task_protection.disable", mock.disable)

    return mock


@pytest.fixture
def mock_step_functions(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
    mock = MagicMock()
    monkeypatch.setattr("src.worker.worker.step_functions.send_task_success", mock.send_task_success)
    monkeypatch.setattr("src.worker.worker.step_functions.send_task_failure", mock.send_task_failure)
    monkeypatch.setattr("src.worker.worker.step_functions.send_task_heartbeat", mock.send_task_heartbeat)

    return mock


class TestWorker:
    def test_main_processes_message_and_exits_on_sigterm(
        self,
        populated_s3: None,
        queue_url: str,
        mock_protection: MagicMock,
        mock_step_functions: MagicMock,
    ) -> None:
        # Calls main() (the production entry point) so the SIGTERM handler is
        # actually installed. SIGTERM fires after the first send_task_success;
        # the handler sets shutdown, the loop exits before processing message #2.
        _send(queue_url, task_token="main-test-1")
        _send(queue_url, task_token="main-test-2")

        def fire_sigterm(*_: object, **__: object) -> None:
            os.kill(os.getpid(), signal.SIGTERM)

        mock_step_functions.send_task_success.side_effect = fire_sigterm

        # Save and restore the SIGTERM handler so pytest's own handling isn't
        # affected after the test.
        original_handler = signal.getsignal(signal.SIGTERM)
        try:
            main()
        finally:
            signal.signal(signal.SIGTERM, original_handler)

        assert mock_step_functions.send_task_success.call_count == 1
        assert _queue_depth(queue_url) == 1

    def test_real_dolby_fixture_passes_all_validations(
        self,
        populated_s3: None,
        queue_url: str,
        mock_protection: MagicMock,
        mock_step_functions: MagicMock,
    ) -> None:
        shutdown = threading.Event()
        mock_step_functions.send_task_success.side_effect = lambda *_, **__: shutdown.set()

        _send(queue_url, task_token="integration-token-1")

        make_worker().process_messages(shutdown)

        assert _queue_depth(queue_url) == 0
        mock_step_functions.send_task_failure.assert_not_called()
        mock_step_functions.send_task_success.assert_called_once()

        call = mock_step_functions.send_task_success.call_args.kwargs
        assert call["task_token"] == "integration-token-1"
        assert call["output"]["is_valid"] is True
        assert call["output"]["errors"] == {}
        # Both fixtures are silent, so the stereo render and reference cross-correlate to r≈0 — a
        # genuine CONTENT_MISMATCH (silence can't confirm the same program). The silent 7.1.2 bed
        # also has silent height channels and no objects, so SILENT_HEIGHT fires (no height content
        # anywhere). Everything else is clean (no sync/loudness/true-peak/LFE warnings), and warnings
        # don't fail the delivery.
        assert set(call["output"]["warnings"]) == {
            AtmosValidationFindingCode.CONTENT_MISMATCH.metadata_key,
            AtmosValidationFindingCode.SILENT_HEIGHT.metadata_key,
        }
        assert call["output"]["metadata"]["codec"] == "PCM"

    def test_drains_a_batch_of_messages(
        self,
        populated_s3: None,
        queue_url: str,
        mock_protection: MagicMock,
        mock_step_functions: MagicMock,
    ) -> None:
        message_count = 3
        shutdown = threading.Event()
        processed = 0

        def track_and_shutdown(*_: object, **__: object) -> None:
            nonlocal processed
            processed += 1
            if processed >= message_count:
                shutdown.set()

        mock_step_functions.send_task_success.side_effect = track_and_shutdown

        for i in range(message_count):
            _send(queue_url, task_token=f"batch-token-{i}")

        make_worker().process_messages(shutdown)

        assert _queue_depth(queue_url) == 0
        assert mock_step_functions.send_task_success.call_count == message_count
        sent_tokens = {call.kwargs["task_token"] for call in mock_step_functions.send_task_success.call_args_list}
        assert sent_tokens == {f"batch-token-{i}" for i in range(message_count)}

    def test_missing_s3_object_sends_task_failure(
        self,
        populated_s3: None,
        queue_url: str,
        mock_protection: MagicMock,
        mock_step_functions: MagicMock,
    ) -> None:
        # A missing S3 object causes mediainfo to emit {"media": null}, which
        # run_mediainfo maps to AssetFetchError.
        shutdown = threading.Event()
        mock_step_functions.send_task_failure.side_effect = lambda *_, **__: shutdown.set()

        body = {
            "task_token": "missing-object-token",
            "asset_upload_type": "atmos",
            "atmos_bucket": _ATMOS_BUCKET,
            "atmos_key": "does-not-exist.wav",
            "stereo_reference_bucket": _STEREO_BUCKET,
            "stereo_reference_key": _STEREO_KEY,
        }
        boto3.client("sqs").send_message(QueueUrl=queue_url, MessageBody=json.dumps(body))

        make_worker().process_messages(shutdown)

        assert _queue_depth(queue_url) == 0
        mock_step_functions.send_task_success.assert_not_called()
        mock_step_functions.send_task_failure.assert_called_once()
        assert mock_step_functions.send_task_failure.call_args.kwargs["error"] == "AssetFetchError"

    def test_stale_token_at_pickup_skips_validation(
        self,
        queue_url: str,
        mock_protection: MagicMock,
        mock_step_functions: MagicMock,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        # The initial heartbeat at the start of _handle_message returns
        # StaleTaskToken — the worker should skip validation entirely (no
        # subprocess, no SFN success/failure) and move on.
        shutdown = threading.Event()
        validate_called = threading.Event()

        def track(*_: object, **__: object) -> dict[str, object]:
            validate_called.set()
            return {"is_valid": True}

        monkeypatch.setattr(Worker, "_run_subprocess", track)
        _send(queue_url, task_token="stale-pickup-token")
        mock_step_functions.send_task_heartbeat.side_effect = StaleTaskToken("retried by SFN")
        mock_protection.disable.side_effect = lambda: shutdown.set()

        make_worker().process_messages(shutdown)

        assert _queue_depth(queue_url) == 0
        assert not validate_called.is_set()
        mock_step_functions.send_task_success.assert_not_called()
        mock_step_functions.send_task_failure.assert_not_called()

    def test_stale_token_mid_validation_aborts_silently(
        self,
        queue_url: str,
        mock_protection: MagicMock,
        mock_step_functions: MagicMock,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        # Replace the validator subprocess with a long-running sleeper so the
        # heartbeat thread has time to fire before validation completes. Make
        # the second heartbeat call return StaleTaskToken to simulate SFN
        # giving up on the task. The heartbeat thread should kill the
        # subprocess and the worker should bail silently — no
        # send_task_success/failure to SFN.
        monkeypatch.setenv("HEARTBEAT_INTERVAL_SECONDS", "0.2")

        def spawn_sleeper(*_: object, **__: object) -> subprocess.Popen[str]:
            return subprocess.Popen(
                [sys.executable, "-c", "import time; time.sleep(2)"],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                start_new_session=True,
            )

        monkeypatch.setattr(Worker, "_spawn_subprocess", spawn_sleeper)

        heartbeat_calls = 0

        def heartbeat_with_stale_on_second(*_: object, **__: object) -> None:
            nonlocal heartbeat_calls
            heartbeat_calls += 1
            if heartbeat_calls >= 2:
                raise StaleTaskToken("token expired")

        mock_step_functions.send_task_heartbeat.side_effect = heartbeat_with_stale_on_second

        shutdown = threading.Event()

        def disable_and_shutdown() -> bool:
            shutdown.set()
            return True

        mock_protection.disable.side_effect = disable_and_shutdown

        _send(queue_url, task_token="stale-mid-run-token")

        make_worker().process_messages(shutdown)

        mock_step_functions.send_task_success.assert_not_called()
        mock_step_functions.send_task_failure.assert_not_called()
        assert heartbeat_calls >= 2