"""Unit-тесты воркера `src/core/worker.py` с фейками `Storage` / `Publisher` / `Parser` (design D-2, D-5, D-7).

Сценарии spec tabular-recognition-service: ack раньше результата, повторный запрос (во время обработки и после),
заглушка `NOT_IMPLEMENTED`, `FILE_READ_ERROR`, `SCHEMA_MISMATCH` без ack, исключение парсера → `UNKNOWN_ERROR`,
подмена парсера → `done` с файлом результата, повтор публикации и `NatsPublishError`. NATS и S3 не нужны.
"""

import asyncio
import copy
import json
from collections.abc import Iterator
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest
from arm_statements import RulesError
from jsonschema import Draft202012Validator, FormatChecker
from loguru import logger
from referencing import Registry, Resource

from src.contract import (
    NATS_MSG_ID_HEADER,
    RESULT_SUBJECT,
    ErrorCode,
    FileType,
    FileWarning,
    MessageError,
    Metrics,
    OsvDocument,
    RecognitionResult,
    RecognitionResultMessage,
    ResultStatus,
    StatementDocument,
)
from src.core.config import NATSSettings
from src.core.nats import NatsConnection, NatsPublishError
from src.core.worker import RecognitionWorker, result_key
from src.parser import NullParser, ParseOutcome, get_parser, require_statements_version

API_DIR = Path(__file__).resolve().parents[2]
CONTRACTS_DIR = API_DIR / "contracts"
EXAMPLES_DIR = CONTRACTS_DIR / "examples"

REQUEST = json.loads((EXAMPLES_DIR / "request-tabular.json").read_text(encoding="utf-8"))
JOB_ID = REQUEST["job_id"]
FILE_ID = REQUEST["file_id"]
BUCKET = REQUEST["s3"]["bucket"]
SOURCE_KEY = REQUEST["s3"]["key"]
RESULT_KEY = f"results/{FILE_ID}/{JOB_ID}/result.json"
SERVICE = {"name": "tabular-parser", "version": "9.9.9", "statements_version": "abc1234"}
OTHER_JOB_ID = "11111111-2222-4333-8444-555555555555"
PARSER_ROWS = 3  # metrics.rows, которые возвращает подменённый парсер


def _validators() -> dict[str, Draft202012Validator]:
    names = {
        "message": "recognition-result-message.schema.json",
        "result": "recognition-result.schema.json",
    }
    schemas = {kind: json.loads((CONTRACTS_DIR / name).read_text(encoding="utf-8")) for kind, name in names.items()}
    registry: Registry = Registry()
    for schema in schemas.values():
        registry = registry.with_resource(schema["$id"], Resource.from_contents(schema))
    return {
        kind: Draft202012Validator(schema, registry=registry, format_checker=FormatChecker())
        for kind, schema in schemas.items()
    }


VALIDATORS = _validators()


# --- фейки ------------------------------------------------------------------------------------------------------------
class FakeStorage:
    """`Storage` в памяти; пишет события в общий журнал `events`."""

    def __init__(self, events: list[tuple[str, str]]) -> None:
        self.events = events
        self.objects: dict[tuple[str, str], bytes] = {}
        self.json_objects: dict[tuple[str, str], Any] = {}
        self.get_calls: list[tuple[str, str]] = []
        self.gate: asyncio.Event | None = None  # задан — get_bytes ждёт его (обработка «в процессе»)
        self.get_started = asyncio.Event()

    async def get_bytes(self, bucket: str, key: str) -> bytes:
        self.events.append(("s3_get", key))
        self.get_calls.append((bucket, key))
        self.get_started.set()
        if self.gate is not None:
            await self.gate.wait()
        try:
            return self.objects[(bucket, key)]
        except KeyError:
            raise FileNotFoundError(f"NoSuchKey: {key}") from None

    async def put_json(self, bucket: str, key: str, obj: Any) -> None:
        self.events.append(("s3_put", key))
        json.dumps(obj)  # объект обязан быть JSON-сериализуемым
        self.json_objects[(bucket, key)] = obj


class FakePublisher:
    """`Publisher`: копит опубликованное; `fail_msg_ids` — msg_id, публикация которых падает; `gates` — msg_id,
    публикация которых ждёт события (`waiting` выставляется, когда публикация встала на ожидание)."""

    def __init__(self, events: list[tuple[str, str]]) -> None:
        self.events = events
        self.published: list[tuple[str, dict[str, Any], str]] = []
        self.fail_msg_ids: set[str] = set()
        self.gates: dict[str, asyncio.Event] = {}
        self.waiting = asyncio.Event()

    async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
        gate = self.gates.get(msg_id)
        if gate is not None:
            self.waiting.set()
            await gate.wait()
        if msg_id in self.fail_msg_ids:
            self.events.append(("publish_failed", msg_id))
            raise NatsPublishError(f"публикация {msg_id} не удалась")
        body = json.loads(payload)
        self.events.append(("publish", msg_id))
        self.published.append((subject, body, msg_id))


class OsvParser:
    """Подмена парсера (spec «Подмена реализации в тесте»): один `osvDocument` из примера контракта."""

    def __init__(self) -> None:
        example = json.loads((EXAMPLES_DIR / "result-osv-minimal.json").read_text(encoding="utf-8"))
        self.document = OsvDocument.model_validate(example["documents"][0])
        self.calls: list[tuple[bytes, FileType]] = []

    def parse(self, data: bytes, file_type: FileType) -> ParseOutcome:
        self.calls.append((data, file_type))
        return ParseOutcome(status=ResultStatus.DONE, documents=(self.document,), metrics=Metrics(rows=PARSER_ROWS))


class ExplodingParser:
    def parse(self, data: bytes, file_type: FileType) -> ParseOutcome:
        raise ValueError("лист без заголовка")


class FakeClock:
    """Монотонные часы: каждое обращение сдвигает время на `step` секунд."""

    def __init__(self, step: float = 0.25) -> None:
        self.now = 100.0
        self.step = step

    def __call__(self) -> float:
        self.now += self.step
        return self.now


def _msg(body: Any, *, raw: bytes | None = None, msg_id: str | None = None) -> SimpleNamespace:
    data = raw if raw is not None else json.dumps(body).encode("utf-8")
    if msg_id is None and isinstance(body, dict) and "job_id" in body:
        msg_id = f"{body['job_id']}:{body.get('attempt')}"
    headers = {NATS_MSG_ID_HEADER: msg_id} if msg_id else None
    return SimpleNamespace(subject="files.tabular_uploaded", data=data, headers=headers)


def _request(**overrides: Any) -> dict[str, Any]:
    body = copy.deepcopy(REQUEST)
    body.update(overrides)
    return body


@pytest.fixture
def log_records() -> Iterator[list[Any]]:
    records: list[Any] = []
    handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG")
    yield records
    logger.remove(handler_id)


class Env:
    def __init__(
        self,
        parser: Any = None,
        *,
        concurrency: int = 1,
        done_cache_size: int = 1000,
        result_subject: str | None = None,
    ) -> None:
        self.events: list[tuple[str, str]] = []
        self.storage = FakeStorage(self.events)
        self.storage.objects[(BUCKET, SOURCE_KEY)] = b"PK\x03\x04 xlsx bytes"
        self.publisher = FakePublisher(self.events)
        self.worker = RecognitionWorker(
            storage=self.storage,
            publisher=self.publisher,
            parser=parser if parser is not None else NullParser(),
            service_name=SERVICE["name"],
            service_version=SERVICE["version"],
            statements_version=SERVICE["statements_version"],
            concurrency=concurrency,
            done_cache_size=done_cache_size,
            clock=FakeClock(),
            **({} if result_subject is None else {"result_subject": result_subject}),
        )

    def bodies(self) -> list[dict[str, Any]]:
        return [body for _subject, body, _msg_id in self.publisher.published]

    def msg_ids(self) -> list[str]:
        return [msg_id for _subject, _body, msg_id in self.publisher.published]


@pytest.fixture
async def env() -> Any:
    environment = Env()
    environment.worker.start()
    yield environment
    await environment.worker.stop(timeout=1)


async def _make_env(parser: Any = None, **kwargs: Any) -> Env:
    environment = Env(parser, **kwargs)
    environment.worker.start()
    return environment


def _assert_message_valid(body: dict[str, Any]) -> None:
    errors = [error.message for error in VALIDATORS["message"].iter_errors(body)]
    assert errors == [], errors
    RecognitionResultMessage.model_validate(body)


async def test_ack_and_result_go_to_configured_result_subject() -> None:
    env = await _make_env(result_subject="test.files.tabular_parsed")
    try:
        await env.worker.on_request(_msg(_request()))
        await env.worker.join()
    finally:
        await env.worker.stop(timeout=1)

    assert [(subject, msg_id) for subject, _body, msg_id in env.publisher.published] == [
        ("test.files.tabular_parsed", f"ack:{JOB_ID}:1"),
        ("test.files.tabular_parsed", f"result:{JOB_ID}:1"),
    ]


# --- ack раньше результата ------------------------------------------------------------------------------------------
async def test_ack_is_published_before_s3_read_and_result(env: Env) -> None:
    await env.worker.on_request(_msg(_request()))
    await env.worker.join()

    assert env.events == [
        ("publish", f"ack:{JOB_ID}:1"),
        ("s3_get", SOURCE_KEY),
        ("publish", f"result:{JOB_ID}:1"),
    ]
    subjects = {subject for subject, _body, _msg_id in env.publisher.published}
    assert subjects == {RESULT_SUBJECT}
    ack = env.bodies()[0]
    assert ack == {
        "schema_version": "2.0",
        "job_id": JOB_ID,
        "file_id": FILE_ID,
        "status": "processing",
        "service": SERVICE,
    }
    for body in env.bodies():
        _assert_message_valid(body)


async def test_ack_is_published_by_on_request_itself_before_queueing(env: Env) -> None:
    gate = asyncio.Event()
    env.storage.gate = gate

    await env.worker.on_request(_msg(_request()))

    # on_request вернулся — ack уже опубликован, хотя обработка ещё не завершена
    assert env.msg_ids() == [f"ack:{JOB_ID}:1"]
    gate.set()
    await env.worker.join()
    assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]


# --- заглушка парсера -----------------------------------------------------------------------------------------------
async def test_null_parser_gives_not_implemented_with_duration(env: Env) -> None:
    await env.worker.on_request(_msg(_request()))
    await env.worker.join()

    result = env.bodies()[1]
    _assert_message_valid(result)
    assert result["status"] == "error"
    assert result["job_id"] == JOB_ID
    assert result["file_id"] == FILE_ID
    assert result["service"] == SERVICE
    assert result["error"] == {"code": "NOT_IMPLEMENTED", "message": "разбор XLS/XLSX/XLSB реализуется в Срезе 2"}
    assert "result_s3" not in result
    assert isinstance(result["metrics"]["duration_ms"], int)
    assert result["metrics"]["duration_ms"] >= 0
    assert env.storage.json_objects == {}  # при ошибке файл результата не пишется


# --- повторный запрос -----------------------------------------------------------------------------------------------
async def test_repeated_request_during_processing_gets_second_ack_only(env: Env) -> None:
    gate = asyncio.Event()
    env.storage.gate = gate

    await env.worker.on_request(_msg(_request()))
    await asyncio.wait_for(env.storage.get_started.wait(), 1)
    await env.worker.on_request(_msg(_request()))  # та же доставка ещё раз, пока первая в работе
    gate.set()
    await env.worker.join()

    assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]
    assert env.storage.get_calls == [(BUCKET, SOURCE_KEY)]


async def test_repeated_request_while_queued_is_not_processed_twice() -> None:
    environment = Env()  # воркер не запущен: задание стоит в очереди
    await environment.worker.on_request(_msg(_request()))
    await environment.worker.on_request(_msg(_request()))
    environment.worker.start()
    await environment.worker.join()
    await environment.worker.stop(timeout=1)

    assert environment.msg_ids() == [f"ack:{JOB_ID}:1", f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]
    assert environment.storage.get_calls == [(BUCKET, SOURCE_KEY)]


async def test_repeated_request_after_done_gets_ack_with_its_own_attempt(env: Env) -> None:
    await env.worker.on_request(_msg(_request()))
    await env.worker.join()
    await env.worker.on_request(_msg(_request(attempt=2)))
    await env.worker.join()

    assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1", f"ack:{JOB_ID}:2"]
    assert env.storage.get_calls == [(BUCKET, SOURCE_KEY)]


async def test_failed_job_counts_as_processed(env: Env) -> None:
    del env.storage.objects[(BUCKET, SOURCE_KEY)]
    await env.worker.on_request(_msg(_request()))
    await env.worker.join()
    await env.worker.on_request(_msg(_request()))
    await env.worker.join()

    assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1", f"ack:{JOB_ID}:1"]
    assert len(env.storage.get_calls) == 1


async def test_done_cache_is_bounded_lru() -> None:
    environment = await _make_env(done_cache_size=1)
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
        await environment.worker.on_request(_msg(_request(job_id=OTHER_JOB_ID)))
        await environment.worker.join()
        # первый job_id вытеснен из LRU размера 1 — его повтор обрабатывается снова
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    assert len(environment.storage.get_calls) == 3  # noqa: PLR2004 - три запроса: первый, второй, повтор первого


# --- FILE_READ_ERROR ------------------------------------------------------------------------------------------------
async def test_missing_s3_object_gives_file_read_error_and_worker_continues(env: Env) -> None:
    del env.storage.objects[(BUCKET, SOURCE_KEY)]
    other_key = f"uploads/{FILE_ID}/{OTHER_JOB_ID}/osv.xlsx"
    env.storage.objects[(BUCKET, other_key)] = b"bytes"

    await env.worker.on_request(_msg(_request()))
    await env.worker.join()
    await env.worker.on_request(_msg(_request(job_id=OTHER_JOB_ID, s3={"bucket": BUCKET, "key": other_key})))
    await env.worker.join()

    first, second = env.bodies()[1], env.bodies()[3]
    _assert_message_valid(first)
    assert first["status"] == "error"
    assert first["error"]["code"] == "FILE_READ_ERROR"
    assert SOURCE_KEY in first["error"]["message"]
    assert "result_s3" not in first
    assert first["metrics"]["duration_ms"] >= 0
    # следующий запрос обработан как обычно
    assert second["job_id"] == OTHER_JOB_ID
    assert second["error"]["code"] == "NOT_IMPLEMENTED"


# --- SCHEMA_MISMATCH ------------------------------------------------------------------------------------------------
async def test_unsupported_major_version_gives_schema_mismatch_without_ack(env: Env) -> None:
    await env.worker.on_request(_msg(_request(schema_version="2.0")))
    await env.worker.join()

    assert env.msg_ids() == [f"result:{JOB_ID}:1"]
    result = env.bodies()[0]
    _assert_message_valid(result)
    assert result["status"] == "error"
    assert result["schema_version"] == "2.0"
    assert result["job_id"] == JOB_ID
    assert result["file_id"] == FILE_ID
    assert result["error"]["code"] == "SCHEMA_MISMATCH"
    assert "2.0" in result["error"]["message"]
    assert result["metrics"]["duration_ms"] >= 0
    assert env.storage.get_calls == []


async def test_minor_version_is_accepted(env: Env) -> None:
    await env.worker.on_request(_msg(_request(schema_version="1.7", extra_field="ignored")))
    await env.worker.join()

    assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]


@pytest.mark.parametrize(
    "mutate",
    [
        pytest.param(lambda body: body["s3"].pop("key"), id="no-s3-key"),
        pytest.param(lambda body: body.update(file_type="DOCX"), id="unknown-file-type"),
        pytest.param(lambda body: body.pop("reply_deadline"), id="no-reply-deadline"),
    ],
)
async def test_request_violating_schema_gives_schema_mismatch_without_ack(env: Env, mutate: Any) -> None:
    body = _request()
    mutate(body)

    await env.worker.on_request(_msg(body))
    await env.worker.join()

    assert env.msg_ids() == [f"result:{JOB_ID}:1"]
    result = env.bodies()[0]
    _assert_message_valid(result)
    assert result["error"]["code"] == "SCHEMA_MISMATCH"
    assert env.storage.get_calls == []


async def test_schema_mismatch_takes_attempt_from_header_when_body_attempt_is_invalid(env: Env) -> None:
    await env.worker.on_request(_msg(_request(attempt=0), msg_id=f"{JOB_ID}:3"))
    await env.worker.join()

    assert env.msg_ids() == [f"result:{JOB_ID}:3"]


@pytest.mark.parametrize(
    "message",
    [
        pytest.param(_msg(None, raw=b"\xff not json"), id="not-json"),
        pytest.param(_msg(None, raw=b"[1, 2]"), id="json-array"),
        pytest.param(_msg(_request(job_id="not-a-uuid")), id="job-id-unreadable"),
        pytest.param(_msg({k: v for k, v in REQUEST.items() if k != "file_id"}), id="no-file-id"),
        pytest.param(_msg(_request(attempt="x"), msg_id="garbage"), id="attempt-unreadable"),
    ],
)
async def test_unreadable_request_is_logged_and_dropped(env: Env, log_records: list[Any], message: Any) -> None:
    await env.worker.on_request(message)
    await env.worker.join()

    assert env.publisher.published == []
    assert env.storage.get_calls == []
    errors = [record for record in log_records if record["level"].name == "ERROR"]
    assert errors, "отброшенный запрос должен попасть в лог уровня error"
    assert "error_category=schema_mismatch" in errors[0]["message"]


# --- исключение парсера ---------------------------------------------------------------------------------------------
async def test_parser_exception_gives_unknown_error_without_traceback() -> None:
    environment = await _make_env(ExplodingParser())
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    result = environment.bodies()[1]
    _assert_message_valid(result)
    assert result["error"]["code"] == "UNKNOWN_ERROR"
    assert "лист без заголовка" in result["error"]["message"]
    assert "Traceback" not in result["error"]["message"]
    assert "\n" not in result["error"]["message"]
    assert result["metrics"]["duration_ms"] >= 0


async def test_parser_is_called_in_a_thread(monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[Any] = []
    original = asyncio.to_thread

    async def spy(func: Any, /, *args: Any, **kwargs: Any) -> Any:
        calls.append(func)
        return await original(func, *args, **kwargs)

    monkeypatch.setattr(asyncio, "to_thread", spy)
    parser = OsvParser()
    environment = await _make_env(parser)
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    assert calls == [parser.parse]
    assert parser.calls == [(b"PK\x03\x04 xlsx bytes", "XLSX")]


# --- подмена парсера → done -----------------------------------------------------------------------------------------
async def test_substituted_parser_gives_done_with_result_file() -> None:
    parser = OsvParser()
    environment = await _make_env(parser)
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    assert environment.events == [
        ("publish", f"ack:{JOB_ID}:1"),
        ("s3_get", SOURCE_KEY),
        ("s3_put", RESULT_KEY),
        ("publish", f"result:{JOB_ID}:1"),
    ]
    result_file = environment.storage.json_objects[(BUCKET, RESULT_KEY)]
    errors = [error.message for error in VALIDATORS["result"].iter_errors(result_file)]
    assert errors == [], errors
    parsed = RecognitionResult.model_validate(result_file)
    assert parsed.status is ResultStatus.DONE
    assert parsed.documents == [parser.document]
    assert str(parsed.job_id) == JOB_ID
    assert result_file["service"] == SERVICE
    assert result_file["metrics"]["rows"] == PARSER_ROWS

    envelope = environment.bodies()[1]
    _assert_message_valid(envelope)
    assert envelope["status"] == "done"
    assert envelope["result_s3"] == {"bucket": BUCKET, "key": RESULT_KEY}
    assert envelope["result_schema_version"] == "2.0"
    assert "error" not in envelope
    assert envelope["metrics"]["rows"] == PARSER_ROWS
    assert envelope["metrics"]["duration_ms"] >= 0


async def test_needs_check_outcome_is_propagated() -> None:
    parser = OsvParser()

    class NeedsCheckParser:
        def parse(self, data: bytes, file_type: FileType) -> ParseOutcome:
            return ParseOutcome(status=ResultStatus.NEEDS_CHECK, documents=(parser.document,))

    environment = await _make_env(NeedsCheckParser())
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    envelope = environment.bodies()[1]
    _assert_message_valid(envelope)
    assert envelope["status"] == "needs_check"
    assert environment.storage.json_objects[(BUCKET, RESULT_KEY)]["status"] == "needs_check"


async def test_error_outcome_metrics_are_kept_and_duration_added() -> None:
    class ErrorWithMetricsParser:
        def parse(self, data: bytes, file_type: FileType) -> ParseOutcome:
            return ParseOutcome(
                status=ResultStatus.ERROR,
                error=MessageError(code=ErrorCode.INVALID_FORMAT, message="не ОСВ"),
                metrics=Metrics(
                    warnings=[FileWarning(code="SHEET_OTHER", severity="info", message="лист пуст", locator="sheet:Л")]
                ),
            )

    environment = await _make_env(ErrorWithMetricsParser())
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    envelope = environment.bodies()[1]
    _assert_message_valid(envelope)
    assert envelope["error"] == {"code": "INVALID_FORMAT", "message": "не ОСВ"}
    assert envelope["metrics"]["warnings"] == [
        {"code": "SHEET_OTHER", "severity": "info", "message": "лист пуст", "locator": "sheet:Л"}
    ]
    assert envelope["metrics"]["duration_ms"] >= 0


async def test_result_file_write_failure_gives_error_without_result_s3() -> None:
    environment = await _make_env(OsvParser())

    async def broken_put(bucket: str, key: str, obj: Any) -> None:
        raise ConnectionError("S3 недоступен")

    environment.storage.put_json = broken_put  # type: ignore[method-assign]
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    envelope = environment.bodies()[1]
    _assert_message_valid(envelope)
    assert envelope["status"] == "error"
    assert envelope["error"]["code"] == "UNKNOWN_ERROR"
    assert "result_s3" not in envelope


def test_result_key_template() -> None:
    assert result_key(FILE_ID, JOB_ID) == RESULT_KEY


# --- публикация: повтор и исчерпание попыток ------------------------------------------------------------------------
async def test_result_publish_failure_is_logged_and_worker_continues(env: Env, log_records: list[Any]) -> None:
    env.publisher.fail_msg_ids.add(f"result:{JOB_ID}:1")
    env.storage.objects[(BUCKET, "uploads/other.xlsx")] = b"bytes"

    await env.worker.on_request(_msg(_request()))
    await env.worker.join()
    await env.worker.on_request(_msg(_request(job_id=OTHER_JOB_ID, s3={"bucket": BUCKET, "key": "uploads/other.xlsx"})))
    await env.worker.join()

    assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"ack:{OTHER_JOB_ID}:1", f"result:{OTHER_JOB_ID}:1"]
    errors = [record["message"] for record in log_records if record["level"].name == "ERROR"]
    assert any(f"result:{JOB_ID}:1" in message for message in errors)


async def test_ack_publish_failure_is_logged_and_job_still_processed(env: Env, log_records: list[Any]) -> None:
    env.publisher.fail_msg_ids.add(f"ack:{JOB_ID}:1")

    await env.worker.on_request(_msg(_request()))
    await env.worker.join()

    assert env.msg_ids() == [f"result:{JOB_ID}:1"]
    assert env.storage.get_calls == [(BUCKET, SOURCE_KEY)]
    errors = [record["message"] for record in log_records if record["level"].name == "ERROR"]
    assert any(f"ack:{JOB_ID}:1" in message for message in errors)


class _FlakyNatsClient:
    """Клиент nats-py для `NatsConnection`: первая публикация результата падает, остальные проходят."""

    def __init__(self) -> None:
        self.is_connected = False
        self.is_closed = False
        self.attempts: list[tuple[str, dict[str, str]]] = []
        self.delivered: list[tuple[str, dict[str, str]]] = []
        self._failed_once = False

    async def connect(self, *_args: Any, **_kwargs: Any) -> None:
        self.is_connected = True

    async def publish(self, subject: str, payload: bytes, headers: dict[str, str]) -> None:
        msg_id = headers[NATS_MSG_ID_HEADER]
        self.attempts.append((msg_id, headers))
        if msg_id.startswith("result:") and not self._failed_once:
            self._failed_once = True
            raise ConnectionError("соединение сброшено")
        self.delivered.append((msg_id, json.loads(payload)))

    async def flush(self, timeout: float) -> None:
        return None

    async def drain(self) -> None:
        self.is_connected = False
        self.is_closed = True

    async def close(self) -> None:
        self.is_connected = False
        self.is_closed = True


async def test_result_publish_is_retried_with_same_msg_id_through_nats_connection() -> None:
    client = _FlakyNatsClient()
    pauses: list[float] = []

    async def fake_sleep(seconds: float) -> None:
        pauses.append(seconds)

    connection = NatsConnection(
        NATSSettings(NATS_URL="nats://fake:4222"),
        client_factory=lambda: client,
        sleep=fake_sleep,
        publish_retry_pause=2.0,
    )
    await connection.start()
    assert await connection.wait_ready(1)
    events: list[tuple[str, str]] = []
    storage = FakeStorage(events)
    storage.objects[(BUCKET, SOURCE_KEY)] = b"bytes"
    worker = RecognitionWorker(
        storage=storage,
        publisher=connection,
        parser=NullParser(),
        service_name=SERVICE["name"],
        service_version=SERVICE["version"],
        statements_version=SERVICE["statements_version"],
    )
    worker.start()
    try:
        await worker.on_request(_msg(_request()))
        await worker.join()
    finally:
        await worker.stop(timeout=1)
        await connection.stop()

    assert [msg_id for msg_id, _headers in client.attempts] == [
        f"ack:{JOB_ID}:1",
        f"result:{JOB_ID}:1",
        f"result:{JOB_ID}:1",
    ]
    assert [msg_id for msg_id, _body in client.delivered] == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]
    assert pauses == [2.0]
    assert storage.get_calls == [(BUCKET, SOURCE_KEY)]  # обработка не повторялась


# --- конкуренция и остановка ----------------------------------------------------------------------------------------
async def test_concurrency_processes_jobs_in_parallel() -> None:
    JOBS = 2  # noqa: N806 - число одновременных заданий
    environment = await _make_env(concurrency=2)
    gate = asyncio.Event()
    environment.storage.gate = gate
    started: list[str] = []
    all_started = asyncio.Event()
    original_get = environment.storage.get_bytes

    async def tracking_get(bucket: str, key: str) -> bytes:
        started.append(key)
        if len(started) == JOBS:
            all_started.set()
        return await original_get(bucket, key)

    environment.storage.get_bytes = tracking_get  # type: ignore[method-assign]
    other_key = "uploads/other.xlsx"
    environment.storage.objects[(BUCKET, other_key)] = b"bytes"
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.on_request(
            _msg(_request(job_id=OTHER_JOB_ID, s3={"bucket": BUCKET, "key": other_key}))
        )
        # оба задания одновременно ждут S3 (первое не отпущено): при concurrency=1 второе не стартовало бы
        await asyncio.wait_for(all_started.wait(), 1)
        assert len(started) == JOBS
        gate.set()
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)


async def test_stop_waits_for_in_flight_job_then_rejects_new_requests() -> None:
    environment = await _make_env()
    gate = asyncio.Event()
    environment.storage.gate = gate
    await environment.worker.on_request(_msg(_request()))
    await asyncio.wait_for(environment.storage.get_started.wait(), 1)

    stop_task = asyncio.create_task(environment.worker.stop(timeout=5))
    await asyncio.sleep(0)
    # после начала остановки новые запросы не принимаются (без ack — ядро переотправит другой инстанции)
    await environment.worker.on_request(_msg(_request(job_id=OTHER_JOB_ID)))
    gate.set()
    await asyncio.wait_for(stop_task, 2)

    assert environment.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]


async def test_stop_cancels_jobs_that_do_not_finish_in_time(log_records: list[Any]) -> None:
    environment = await _make_env()
    environment.storage.gate = asyncio.Event()  # никогда не откроется
    await environment.worker.on_request(_msg(_request()))
    await environment.worker.on_request(_msg(_request(job_id=OTHER_JOB_ID)))
    await asyncio.wait_for(environment.storage.get_started.wait(), 1)

    await asyncio.wait_for(environment.worker.stop(timeout=0.05), 2)

    assert environment.msg_ids() == [f"ack:{JOB_ID}:1", f"ack:{OTHER_JOB_ID}:1"]
    warnings = [record["message"] for record in log_records if record["level"].name == "WARNING"]
    assert any("pending=1" in message for message in warnings)


async def test_stop_is_idempotent_and_safe_without_start() -> None:
    environment = Env()
    await environment.worker.stop(timeout=0.1)
    await environment.worker.stop(timeout=0.1)


async def test_stop_during_ack_publish_drops_job_with_warning(log_records: list[Any]) -> None:
    environment = await _make_env()
    gate = asyncio.Event()
    environment.publisher.gates[f"ack:{JOB_ID}:1"] = gate
    request_task = asyncio.create_task(environment.worker.on_request(_msg(_request())))
    await asyncio.wait_for(environment.publisher.waiting.wait(), 1)

    # stop() начался, пока on_request публикует ack: очередь пуста, корутины отменены
    await asyncio.wait_for(environment.worker.stop(timeout=1), 2)
    gate.set()
    await asyncio.wait_for(request_task, 1)

    assert environment.msg_ids() == [f"ack:{JOB_ID}:1"]
    assert environment.storage.get_calls == []
    assert environment.worker._in_progress == set()
    assert environment.worker._queue.qsize() == 0
    warnings = [record["message"] for record in log_records if record["level"].name == "WARNING"]
    assert any("error_category=shutdown" in message and JOB_ID in message for message in warnings), warnings


# --- нормализация сообщения ошибки парсера ------------------------------------------------------------------------
async def test_parser_error_message_is_normalised_to_one_bounded_line() -> None:
    long_message = "строка 1\nстрока 2\t" + "x" * 3000

    class VerboseErrorParser:
        def parse(self, data: bytes, file_type: FileType) -> ParseOutcome:
            return ParseOutcome(
                status=ResultStatus.ERROR,
                error=MessageError(code=ErrorCode.INVALID_FORMAT, message=long_message),
            )

    environment = await _make_env(VerboseErrorParser())
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    envelope = environment.bodies()[1]
    _assert_message_valid(envelope)
    message = envelope["error"]["message"]
    assert envelope["error"]["code"] == "INVALID_FORMAT"
    assert message.startswith("строка 1 строка 2 xxx")
    assert "\n" not in message
    assert "\t" not in message
    assert len(message) <= 1000  # noqa: PLR2004 - ERROR_MESSAGE_LIMIT


# --- невалидное тело с job_id, который уже в работе или обработан ---------------------------------------------------
async def test_malformed_request_for_job_in_progress_is_dropped(env: Env, log_records: list[Any]) -> None:
    gate = asyncio.Event()
    env.storage.gate = gate
    await env.worker.on_request(_msg(_request()))
    await asyncio.wait_for(env.storage.get_started.wait(), 1)

    malformed = _request()
    malformed["s3"].pop("key")
    await env.worker.on_request(_msg(malformed))
    gate.set()
    await env.worker.join()

    # SCHEMA_MISMATCH под тем же result:<job_id>:<attempt> не публикуется — только настоящий результат
    assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]
    assert env.bodies()[1]["error"]["code"] == "NOT_IMPLEMENTED"
    warnings = [record["message"] for record in log_records if record["level"].name == "WARNING"]
    assert any("error_category=schema_mismatch" in message and JOB_ID in message for message in warnings)


async def test_malformed_request_for_done_job_is_dropped(env: Env) -> None:
    await env.worker.on_request(_msg(_request()))
    await env.worker.join()

    await env.worker.on_request(_msg(_request(job_id=JOB_ID.upper(), schema_version="2.0")))
    await env.worker.join()

    assert env.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]


# --- реальный парсер в воркере ------------------------------------------------------------------------------------
REAL_OSV_66 = Path("osv") / "rb" / "2024-FY" / "ОСВ_66 за 2024.xls"


@pytest.mark.fixtures
async def test_worker_with_real_osv_parser_gives_done_and_valid_result_file(fixtures_dir: Path) -> None:
    """Spec «Реальная ОСВ через воркер»: настоящий `OsvParser` (`get_parser`) + байты реальной ОСВ 66 без NATS/S3."""
    environment = Env(get_parser())
    environment.storage.objects[(BUCKET, SOURCE_KEY)] = (fixtures_dir / REAL_OSV_66).read_bytes()
    environment.worker.start()
    try:
        await environment.worker.on_request(_msg(_request(file_type="XLS")))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    assert environment.msg_ids() == [f"ack:{JOB_ID}:1", f"result:{JOB_ID}:1"]
    envelope = environment.bodies()[1]
    _assert_message_valid(envelope)
    assert envelope["status"] == "done"
    assert envelope["result_schema_version"] == "2.0"
    assert envelope["result_s3"] == {"bucket": BUCKET, "key": RESULT_KEY}
    stored = environment.storage.json_objects[(BUCKET, RESULT_KEY)]
    assert [error.message for error in VALIDATORS["result"].iter_errors(stored)] == []
    assert RecognitionResult.model_validate(stored).to_wire() == stored
    assert [document["document_type"] for document in stored["documents"]] == ["OSV"]
    assert stored["documents"][0]["header"]["account_code_raw"] == "66"


# --- версия общего кода в результате (контракт 2.0) -----------------------------------------------------------------
async def test_result_carries_statements_version(env: Env) -> None:
    """Spec «Версия общего кода в результате»: `service.statements_version` — в ack, конверте и файле результата;
    запрос 1.0 принимается (запрос остаётся v1), запрос 2.0 — `SCHEMA_MISMATCH` без ack."""
    await env.worker.on_request(_msg(_request()))
    await env.worker.join()

    assert [body["schema_version"] for body in env.bodies()] == ["2.0", "2.0"]
    assert all(body["service"]["statements_version"] == "abc1234" for body in env.bodies())


async def test_worker_with_form_result_has_statements_version_in_result_file() -> None:
    document = StatementDocument.model_validate(
        {
            "document_type": "DISCLOSURE",
            "form_variant": None,
            "locator": "sheet:Пояснения",
            "status": "done",
            "warnings": [],
            "header": {"form_name": "Пояснения"},
            "columns": [],
            "lines": [],
        }
    )
    parser = SimpleNamespace(parse=lambda data, file_type: ParseOutcome(ResultStatus.DONE, documents=(document,)))
    environment = Env(parser)
    environment.worker.start()
    try:
        await environment.worker.on_request(_msg(_request()))
        await environment.worker.join()
    finally:
        await environment.worker.stop(timeout=1)

    stored = environment.storage.json_objects[(BUCKET, RESULT_KEY)]
    assert stored["service"]["statements_version"] == "abc1234"
    assert [error.message for error in VALIDATORS["result"].iter_errors(stored)] == []


def test_require_statements_version_fails_without_file(tmp_path: Path) -> None:
    with pytest.raises(RuntimeError, match="VERSION"):
        require_statements_version(tmp_path / "VERSION")


def test_require_statements_version_reads_the_marker(tmp_path: Path) -> None:
    marker = tmp_path / "VERSION"
    marker.write_text("abc1234\n", encoding="utf-8")

    assert require_statements_version(marker) == "abc1234"
    assert require_statements_version()  # версия установленной библиотеки


def test_get_parser_wraps_rules_error_into_runtime_error(monkeypatch: pytest.MonkeyPatch) -> None:
    """I-2(в): ошибка словарей пакета при старте — `RuntimeError`, сервис не стартует."""

    def broken() -> object:
        raise RulesError("битый словарь форм")

    monkeypatch.setattr("src.parser.default_rules", broken)

    with pytest.raises(RuntimeError, match="битый словарь форм"):
        get_parser()