"""Сквозной сценарий scan-gateway против compose-инфраструктуры (задача 6.1, R-7, R-10).

Настоящий lifespan приложения (`create_app` → S3, NATS, воркер, `NullNormalizer`) и эмулятор ИС Сканирование
(`src.scan_emulator.run` в потоке) работают в процессе теста; NATS 127.0.0.1:4222 и MinIO 127.0.0.1:9000 — из
`api/.env.test` (`make up`), горячая папка — tmp-каталог.

Сценарии (R-10: обе реальные фикстуры `fixtures/ocr/*.json` — ответы ИС с ошибкой `INVALID_DOCUMENT`):
1. `aep_5.pdf` (реальная фикстура) → ack `processing`, затем `INVALID_DOCUMENT` с сообщением ИС и `raw_s3_key`;
   повтор того же запроса во время ожидания ответа ИС и после результата — ещё один ack, второго файла нет;
2. `ok_doc.pdf` (синтетический успешный ответ ИС) → ack, затем `NOT_IMPLEMENTED` с `raw_s3_key`;
3. `unknown.pdf` (нет фикстуры) → ack, затем `INVALID_DOCUMENT` эмулятора.
Каждый конверт проверяется схемой `recognition-result-message`, `service.name` и `Nats-Msg-Id`; сырой ответ в MinIO
совпадает с ответом эмулятора байт в байт; входящий файл совпадает с объектом S3.

Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Чужие конверты на `recognition.result`
отсекаются по `job_id` теста; ключи MinIO уникальны (`uploads/itest-<uuid>/…`, `raw/<job_id>.json`) и удаляются.
Подписка шлюза идёт в queue group `scan-gateway`: если в этом NATS живёт другой экземпляр scan-gateway, он может
забрать запрос теста (тест тогда упадёт по таймауту ожидания ack). Перед запуском занятость группы можно проверить
мониторингом NATS: `curl -s 'http://127.0.0.1:8222/subsz?subs=1'` — подписчиков `recognition.image.request` быть
не должно. Автоматической защиты в тесте нет намеренно (она была бы гонкой). Недоступная инфраструктура — падение
теста с понятным сообщением, не skip; нет репозитория fixtures — skip сценариев на реальных данных (маркер `fixtures`).
"""

import asyncio
import json
import shutil
import socket
import threading
import uuid
from collections.abc import AsyncIterator
from contextlib import AsyncExitStack, suppress
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import httpx
import pytest
from fastapi import FastAPI
from jsonschema import Draft202012Validator, FormatChecker
from miniopy_async.error import S3Error
from nats.aio.client import Client as NATS
from nats.aio.msg import Msg

from src.app import create_app
from src.contract import IMAGE_REQUEST_SUBJECT, NATS_MSG_ID_HEADER, RESULT_SUBJECT, request_msg_id
from src.core.clients import JSON_CONTENT_TYPE, S3Storage
from src.core.config import Settings, get_settings
from src.normalizer import NOT_IMPLEMENTED_MESSAGE
from src.scan_emulator import EmulatorSettings, run

pytestmark = [
    pytest.mark.integration,
    pytest.mark.usefixtures("log_lines"),
    # miniopy-async отдаёт aiohttp тело объекта байтами; на теле > 1 МиБ (реальный скан aep_5.PDF) aiohttp пишет
    # ResourceWarning «Sending a large body directly…». Предупреждение библиотеки, по умолчанию Python его не
    # показывает; здесь отключено точечно, чтобы прогон с `-W error` проверял остальные предупреждения
    pytest.mark.filterwarnings("ignore:Sending a large body directly with raw bytes:ResourceWarning"),
]

API_DIR = Path(__file__).resolve().parents[2]
CONTRACTS_DIR = API_DIR / "contracts"
# `<корень scan-gateway>/../fixtures/ocr` (workspace и worktree `.wt/<срез>/`), как в unit-тестах
REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"

CONNECT_WAIT = 10.0  # секунд на подключение к NATS (тест и фоновая задача приложения)
SCENARIO_WAIT = 20.0  # секунд на ожидание конвертов одного сценария
QUIET_PERIOD = 1.0  # секунд тишины после повторного запроса: второго результата быть не должно
EMULATOR_DELAY = 0.5  # секунд «распознавания»: окно, в котором повторный запрос приходит во время ожидания ИС
POLL_SECONDS = 0.2  # SCAN_POLL_SECONDS шлюза
TIMEOUT_SECONDS = 15  # SCAN_TIMEOUT_SECONDS: зависший эмулятор даёт TIMEOUT раньше, чем истечёт SCENARIO_WAIT

MESSAGE_VALIDATOR = Draft202012Validator(
    json.loads((CONTRACTS_DIR / "recognition-result-message.schema.json").read_text(encoding="utf-8")),
    format_checker=FormatChecker(),
)
REQUEST_EXAMPLE: dict[str, Any] = json.loads(
    (CONTRACTS_DIR / "examples" / "request-image.json").read_text(encoding="utf-8")
)

# Синтетический успешный ответ ИС (R-10): копия `aep_5.json` с `error = null` и непустым `data_ocr` по форме
# docs/context/05-domain-rules.md §11 (документы с label/organization/inn/units/date/table_structure.rows)
SUCCESS_DATA_OCR: list[dict[str, Any]] = [
    {
        "pages": [1],
        "label": {"form_name": "Бухгалтерский баланс", "confidence": 98.4},
        "organization": {"full_name": "ООО «Русский бисквит»", "confidence": 97.1},
        "inn": {"inn": "7701234567", "confidence": 99.2},
        "units": {"measurement": "тыс. руб.", "confidence": 96.0},
        "date": {"date": "30.09.2025", "confidence": 95.5},
        "table_structure": {
            "rows": [
                {
                    "row_index": 0,
                    "cells": [
                        {"column_index": 0, "value": "Наименование показателя", "confidence": 99.0},
                        {"column_index": 1, "value": "Код", "confidence": 99.0},
                        {"column_index": 2, "value": "На 30 сентября 2025 г.", "confidence": 97.0},
                        {"column_index": 3, "value": "На 31 декабря 2024 г.", "confidence": 97.0},
                    ],
                },
                {
                    "row_index": 1,
                    "cells": [
                        {"column_index": 0, "value": "БАЛАНС", "confidence": 98.0},
                        {"column_index": 1, "value": "1600", "confidence": 99.0},
                        {"column_index": 2, "value": "1 234 567", "confidence": 96.5},
                        {"column_index": 3, "value": "(12 345)", "confidence": 95.0},
                    ],
                },
            ]
        },
    }
]


def _require_port(host: str, port: int, what: str) -> None:
    try:
        with socket.create_connection((host, port), timeout=2):
            return
    except OSError as exc:
        pytest.fail(f"{what} недоступен на {host}:{port} ({exc!r}): поднимите инфраструктуру `make up`")


def _real_fixture(stem: str, suffix: str) -> Path:
    """Файл реальной фикстуры без учёта регистра расширения (`aep_5.PDF`); нет репозитория fixtures — skip."""
    if not REAL_FIXTURES.is_dir():
        pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace")
    for path in sorted(REAL_FIXTURES.iterdir()):
        if path.stem == stem and path.suffix.lower() == suffix:
            return path
    pytest.fail(f"в {REAL_FIXTURES} нет {stem}{suffix} (без учёта регистра расширения)")


@dataclass
class Envelope:
    msg_id: str
    body: dict[str, Any]


@dataclass
class Roundtrip:
    """Запущенный шлюз (lifespan приложения), эмулятор и слушатель `recognition.result`."""

    settings: Settings
    app: FastAPI
    root: Path
    s3: S3Storage
    publisher: NATS
    received: list[Msg] = field(default_factory=list)
    jobs: set[str] = field(default_factory=set)
    keys: list[str] = field(default_factory=list)

    @property
    def input_dir(self) -> Path:
        return self.root / self.settings.SCAN_INPUT_DIR

    def answer_path(self, job_id: str) -> Path:
        return self.root / self.settings.SCAN_OUTPUT_DIR / job_id / f"{job_id}.json"

    # --- запрос ---------------------------------------------------------------------------------------------------
    async def request(self, original_name: str, data: bytes) -> dict[str, Any]:
        """Кладёт исходник в MinIO под уникальным ключом и строит запрос по `examples/request-image.json`."""
        job_id, file_id = str(uuid.uuid4()), str(uuid.uuid4())
        key = f"uploads/itest-{uuid.uuid4()}/{file_id}/{job_id}/{original_name}"
        self.jobs.add(job_id)
        self.keys += [key, f"raw/{job_id}.json"]
        await self.s3.put_bytes(key, data, content_type="application/pdf")
        deadline = datetime.now(UTC) + timedelta(seconds=60)
        return {
            **REQUEST_EXAMPLE,
            "job_id": job_id,
            "file_id": file_id,
            "original_name": original_name,
            "s3": {"bucket": self.s3.bucket, "key": key},
            "reply_deadline": deadline.isoformat().replace("+00:00", "Z"),
            "attempt": 1,
        }

    async def publish(self, body: dict[str, Any]) -> None:
        """Публикация как в ядре: `recognition.image.request`, `Nats-Msg-Id = <job_id>:<attempt>`."""
        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
        headers = {NATS_MSG_ID_HEADER: request_msg_id(body["job_id"], body["attempt"])}
        await self.publisher.publish(IMAGE_REQUEST_SUBJECT, payload, headers=headers)
        await self.publisher.flush()

    # --- конверты -------------------------------------------------------------------------------------------------
    def envelopes(self, job_id: str) -> list[Envelope]:
        """Конверты задания в порядке получения; каждый проверен схемой и `service.name`."""
        out = []
        for msg in self.received:
            body = json.loads(msg.data)
            if body.get("job_id") != job_id:
                continue
            MESSAGE_VALIDATOR.validate(body)
            assert body["service"]["name"] == "scan-gateway", body
            assert body["schema_version"] == "1.0", body
            assert msg.headers is not None, body
            out.append(Envelope(msg.headers[NATS_MSG_ID_HEADER], body))
        return out

    def acks(self, job_id: str) -> list[str]:
        return [e.msg_id for e in self.envelopes(job_id) if e.body["status"] == "processing"]

    def results(self, job_id: str) -> list[Envelope]:
        return [e for e in self.envelopes(job_id) if e.body["status"] != "processing"]

    async def wait_for(self, what: str, predicate: Any, timeout: float = SCENARIO_WAIT) -> None:
        loop = asyncio.get_running_loop()
        deadline = loop.time() + timeout
        while not predicate():
            if loop.time() > deadline:
                raise AssertionError(f"за {timeout} с не дождались: {what}")
            await asyncio.sleep(0.01)

    async def health(self) -> dict[str, Any]:
        transport = httpx.ASGITransport(app=self.app)
        async with httpx.AsyncClient(transport=transport, base_url="http://scan-gateway") as client:
            response = await client.get("/healthcheck")
        assert response.status_code == 200
        body: dict[str, Any] = response.json()
        return body

    # --- проверки -------------------------------------------------------------------------------------------------
    def inbox(self, job_id: str) -> list[str]:
        return sorted(p.name for p in self.input_dir.iterdir() if p.name.startswith(job_id))

    async def run(self, original_name: str, data: bytes) -> tuple[dict[str, Any], Envelope]:
        """Запрос → ack → итоговый конверт; общие проверки сквозного сценария. Возвращает запрос и итог."""
        body = await self.request(original_name, data)
        job_id = body["job_id"]
        await self.publish(body)
        await self.wait_for(f"итоговый конверт {job_id}", lambda: bool(self.results(job_id)))
        envelopes = self.envelopes(job_id)
        assert [e.msg_id for e in envelopes] == [f"ack:{job_id}:1", f"result:{job_id}:1"]  # ack раньше итога
        assert envelopes[0].body["status"] == "processing"
        assert envelopes[0].body["file_id"] == body["file_id"]
        await self.check_files(body, envelopes[1], data)
        return body, envelopes[1]

    async def check_files(self, body: dict[str, Any], result: Envelope, data: bytes) -> None:
        """Итог `error` с `raw_s3_key`; сырой ответ в MinIO = ответ эмулятора; входящий файл = объект S3."""
        job_id = body["job_id"]
        assert result.body["status"] == "error"
        assert result.body["file_id"] == body["file_id"]
        assert result.body["raw_s3_key"] == f"raw/{job_id}.json"
        raw = await self.s3.get_bytes(f"raw/{job_id}.json")
        assert raw == self.answer_path(job_id).read_bytes()  # как есть, байт в байт
        assert json.loads(raw)["file_uid"] == job_id
        stat = await self.s3.client.stat_object(self.s3.bucket, f"raw/{job_id}.json")
        assert stat.content_type == JSON_CONTENT_TYPE
        # результат-файл пишется только для done/needs_check
        with pytest.raises(S3Error):
            await self.s3.client.stat_object(self.s3.bucket, f"results/{body['file_id']}/{job_id}/result.json")
        name = f"{job_id}_{body['original_name']}.done"
        assert self.inbox(job_id) == [name]
        assert (self.input_dir / name).read_bytes() == await self.s3.get_bytes(body["s3"]["key"]) == data


def _write_fixtures(fixtures: Path) -> None:
    """Каталог фикстур эмулятора: копия реальной `aep_5.json` и синтетическая успешная `ok_doc.json` (R-10)."""
    fixtures.mkdir()
    template: dict[str, Any] = {"file_uid": "aep_5"}
    if REAL_FIXTURES.is_dir():
        shutil.copy(_real_fixture("aep_5", ".json"), fixtures / "aep_5.json")
        template = json.loads((fixtures / "aep_5.json").read_text(encoding="utf-8"))
    success = {**template, "file_uid": "ok_doc", "error": None, "data_ocr": SUCCESS_DATA_OCR}
    (fixtures / "ok_doc.json").write_text(json.dumps(success, ensure_ascii=False, indent=2), encoding="utf-8")


def _settings(monkeypatch: pytest.MonkeyPatch, root: Path) -> Settings:
    """Настройки шлюза из `.env.test` и окружения (как в проде) с tmp-корнем; инфраструктура недоступна — fail."""
    monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "local")
    monkeypatch.setenv("SCAN_HOTFOLDER_PATH", str(root))
    monkeypatch.setenv("SCAN_POLL_SECONDS", str(POLL_SECONDS))
    monkeypatch.setenv("SCAN_TIMEOUT_SECONDS", str(TIMEOUT_SECONDS))
    monkeypatch.setenv("NATS_ENABLED", "true")
    settings = get_settings()
    parsed = urlparse(settings.OPENBAO.NATS.NATS_URL)
    _require_port(parsed.hostname or "127.0.0.1", parsed.port or 4222, "NATS")
    host, _, port = settings.OPENBAO.MINIO.MINIO_URL.partition(":")
    _require_port(host, int(port or 9000), "MinIO")
    return settings


async def _start_gateway(stack: AsyncExitStack, app: FastAPI) -> None:
    """Lifespan приложения; возврат — когда шлюз подписан на `recognition.image.request`."""
    await stack.enter_async_context(app.router.lifespan_context(app))
    [connect_task] = [t for t in asyncio.all_tasks() if t.get_name() == "nats-connect"]
    # задача завершается после SUB с flush: дальше подписка шлюза действует на сервере
    await asyncio.wait_for(asyncio.shield(connect_task), CONNECT_WAIT)
    assert app.state.nats.is_connected, "шлюз не подписался на запросы (см. лог ERROR)"


def _start_emulator(stack: AsyncExitStack, settings: Settings, root: Path, fixtures: Path) -> None:
    """Эмулятор ИС (`src.scan_emulator.run`) в потоке; при закрытии стека — штатная остановка и ожидание потока."""
    stop = threading.Event()
    emulator_settings = EmulatorSettings(
        hotfolder_path=root,
        input_dir=settings.SCAN_INPUT_DIR,
        output_dir=settings.SCAN_OUTPUT_DIR,
        fixtures_dir=fixtures,
        delay=EMULATOR_DELAY,
        poll_interval=0.05,
    )
    emulator = asyncio.create_task(asyncio.to_thread(run, emulator_settings, stop), name="scan-emulator")

    async def stop_emulator() -> None:
        stop.set()
        await emulator

    stack.push_async_callback(stop_emulator)


@pytest.fixture
async def roundtrip(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, log_lines: list[str]) -> AsyncIterator[Roundtrip]:
    root = tmp_path / "hotfolder"
    root.mkdir()
    _write_fixtures(tmp_path / "fixtures")
    settings = _settings(monkeypatch, root)
    nats_url = settings.OPENBAO.NATS.NATS_URL

    async with AsyncExitStack() as stack:
        s3 = S3Storage.from_settings(settings.OPENBAO.MINIO)
        stack.push_async_callback(s3.close)
        client = NATS()
        try:
            await asyncio.wait_for(client.connect(nats_url, max_reconnect_attempts=0), CONNECT_WAIT)
        except Exception as exc:
            pytest.fail(f"нет подключения к NATS {nats_url} ({exc!r}): `make up`")
        stack.push_async_callback(client.drain)
        rt = Roundtrip(settings=settings, app=create_app(settings), root=root, s3=s3, publisher=client)

        async def remove_objects() -> None:
            for key in rt.keys:
                with suppress(S3Error):
                    await s3.client.remove_object(s3.bucket, key)

        stack.push_async_callback(remove_objects)

        async def on_result(msg: Msg) -> None:
            with suppress(ValueError):  # чужие сообщения на общем subject могут быть чем угодно
                body = json.loads(msg.data)
                if isinstance(body, dict) and body.get("job_id") in rt.jobs:
                    rt.received.append(msg)

        # core NATS без JetStream не хранит сообщения: слушатель подписан до первой публикации запроса
        await client.subscribe(RESULT_SUBJECT, cb=on_result)
        await client.flush()
        await _start_gateway(stack, rt.app)
        _start_emulator(stack, settings, root, tmp_path / "fixtures")
        yield rt

    # lifespan остановлен: заданий и фоновых задач приложения не осталось, ошибок в логе нет
    assert rt.app.state.worker.active_jobs == 0
    leaked = [t.get_name() for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()]
    assert leaked == [], f"незавершённые задачи после остановки: {leaked}"
    errors = [line for line in log_lines if line.startswith(("ERROR|", "CRITICAL|"))]
    assert errors == [], errors


@pytest.mark.fixtures
async def test_real_fixture_aep_5_passes_is_error_through_and_repeat_gets_ack(roundtrip: Roundtrip) -> None:
    data = _real_fixture("aep_5", ".pdf").read_bytes()
    expected_error = json.loads(_real_fixture("aep_5", ".json").read_text(encoding="utf-8"))["error"]
    health = await roundtrip.health()
    assert (health["nats"], health["hotfolder"]) == ("ok", "ok"), health

    body = await roundtrip.request("aep_5.pdf", data)
    job_id = body["job_id"]
    await roundtrip.publish(body)
    inbox_file = roundtrip.input_dir / f"{job_id}_aep_5.pdf"
    await roundtrip.wait_for(f"файл {inbox_file.name} во входящем каталоге", inbox_file.exists)
    # повтор того же запроса во время ожидания ответа ИС (эмулятор «распознаёт» EMULATOR_DELAY с)
    await roundtrip.publish(body)
    await roundtrip.wait_for(f"итоговый конверт {job_id}", lambda: bool(roundtrip.results(job_id)))
    await asyncio.sleep(QUIET_PERIOD)  # второй (лишний) результат или ack успел бы прийти

    envelopes = roundtrip.envelopes(job_id)
    assert [e.msg_id for e in envelopes] == [f"ack:{job_id}:1", f"ack:{job_id}:1", f"result:{job_id}:1"]
    [result] = roundtrip.results(job_id)
    assert result.body["error"] == {"code": "INVALID_DOCUMENT", "message": expected_error["message"]}
    assert result.body["error"]["message"] == "Не удалось определить тип документа."
    await roundtrip.check_files(body, result, data)
    raw = json.loads(await roundtrip.s3.get_bytes(f"raw/{job_id}.json"))
    assert raw["error"] == expected_error

    # повтор ядра после результата (attempt 2): ещё один ack, без второго файла и второго результата
    await roundtrip.publish({**body, "attempt": 2})
    await roundtrip.wait_for(f"ack:{job_id}:2", lambda: f"ack:{job_id}:2" in roundtrip.acks(job_id))
    await asyncio.sleep(QUIET_PERIOD)
    assert roundtrip.acks(job_id) == [f"ack:{job_id}:1", f"ack:{job_id}:1", f"ack:{job_id}:2"]
    assert [e.msg_id for e in roundtrip.results(job_id)] == [f"result:{job_id}:1"]
    assert roundtrip.inbox(job_id) == [f"{job_id}_aep_5.pdf.done"]


@pytest.mark.fixtures
async def test_success_answer_gives_not_implemented(roundtrip: Roundtrip) -> None:
    data = _real_fixture("aep_5", ".pdf").read_bytes()  # байты реального скана под именем ok_doc.pdf
    body, result = await roundtrip.run("ok_doc.pdf", data)
    job_id = body["job_id"]
    assert result.body["error"] == {"code": "NOT_IMPLEMENTED", "message": NOT_IMPLEMENTED_MESSAGE}
    raw = json.loads(await roundtrip.s3.get_bytes(f"raw/{job_id}.json"))
    assert raw["error"] is None
    assert raw["data_ocr"] == SUCCESS_DATA_OCR


async def test_unknown_name_gets_emulator_invalid_document(roundtrip: Roundtrip) -> None:
    data = b"%PDF-1.4\n% scan-gateway itest unknown.pdf\n\x00\x01\x02"
    body, result = await roundtrip.run("unknown.pdf", data)
    job_id = body["job_id"]
    error = {"code": "INVALID_DOCUMENT", "message": "unknown.pdf: нет эталонного ответа в фикстурах"}
    assert result.body["error"] == error
    assert json.loads(await roundtrip.s3.get_bytes(f"raw/{job_id}.json")) == {
        "file_uid": job_id,
        "error": error,
        "data_ocr": [],
    }