arm-core/api/tests/test_recognition_inbox.py

"""Inbox обмена `recognition_job`: `handle_message()` (spec recognition-job-exchange: «Ack сервиса переводит задание в
`processing`», «Inbox принимает результат идемпотентно»; design D-6; рулинги R-7, R-12, R-22).

Быстрые тесты (без маркера) — разбор заголовка `Nats-Msg-Id`, слияние метрик, отказ невалидным сообщениям без
обращения к БД, ошибка БД не выходит наружу, шаблонный `nats_callback` удалён.
Интеграционные — `handle_message` с фейковым Msg (`SimpleNamespace(data, headers, subject)`) на реальном PostgreSQL
(БД сессии pytest; данные фиксируются `CommittedJobs` и удаляются в teardown) и полный цикл через реальный NATS
(уникальный result-subject: общий `recognition.result` не трогаем).
"""

import asyncio
import importlib.util
import json
import time
import uuid
from collections.abc import AsyncIterator, Iterator
from datetime import UTC, datetime, timedelta
from functools import partial
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest
import pytest_asyncio
from jsonschema import Draft202012Validator, FormatChecker
from loguru import logger
from nats.aio.client import Client as NATS
from nats.aio.msg import Msg
from sqlalchemy import JSON, func, select
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine

from src.contract import NATS_MSG_ID_HEADER, RecognitionResultMessage, ack_msg_id, result_msg_id
from src.core.config import get_settings
from src.core.nats import NatsConnection
from src.models.enums import RecognitionStatus, S3BucketRole
from src.models.files import RecognitionJob, S3Object
from src.services.recognition_exchange import (
    FINAL_STATUSES,
    NoopMapper,
    expire_deadlines,
    handle_message,
    header_msg_id,
    merge_metrics,
    publish_pending,
)
from tests.factories import CommittedJobs

API_DIR = Path(__file__).resolve().parents[1]
MESSAGE_SCHEMA = json.loads((API_DIR / "contracts" / "recognition-result-message.schema.json").read_text("utf-8"))
MESSAGE_VALIDATOR = Draft202012Validator(MESSAGE_SCHEMA, format_checker=FormatChecker())

SERVICE = {"name": "tabular-parser", "version": "0.1.0"}
BUCKET = "arm-ka"
READY_TIMEOUT = 5.0
RACE_JOBS = 10
LOCK_HOLD_SECONDS = 0.5


# --- инфраструктура ---------------------------------------------------------------------------------------------


@pytest.fixture
def logs() -> Iterator[list[Any]]:
    """Записи loguru (уровень DEBUG и выше), выпущенные за время теста."""
    records: list[Any] = []
    handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG")
    try:
        yield records
    finally:
        logger.remove(handler_id)


def _logged(records: list[Any], fragment: str, level: str | None = None) -> list[Any]:
    return [r for r in records if fragment in r["message"] and (level is None or r["level"].name == level)]


def _result_key(file_id: uuid.UUID, job_id: uuid.UUID) -> str:
    return f"results/{file_id}/{job_id}/result.json"


def envelope(job_id: uuid.UUID, file_id: uuid.UUID, status: str, **extra: Any) -> dict[str, Any]:
    """Конверт `recognition-result-message` v1.0; для done/needs_check по умолчанию — `result_s3` и версия схемы."""
    body: dict[str, Any] = {
        "schema_version": "1.0",
        "job_id": str(job_id),
        "file_id": str(file_id),
        "status": status,
        "service": SERVICE,
    }
    if status in ("done", "needs_check"):
        body["result_s3"] = {"bucket": BUCKET, "key": _result_key(file_id, job_id)}
        body["result_schema_version"] = "1.0"
    body.update(extra)
    return body


def fake_msg(body: Any, msg_id: str | None, *, header: str = NATS_MSG_ID_HEADER, validate: bool = True) -> Any:
    """Msg-подобный объект nats-py: `.data`, `.headers`, `.subject`. Тело по умолчанию проверяется JSON Schema."""
    if validate:
        MESSAGE_VALIDATOR.validate(body)
    data = body if isinstance(body, bytes) else json.dumps(body, ensure_ascii=False).encode("utf-8")
    headers = None if msg_id is None else {header: msg_id}
    return SimpleNamespace(data=data, headers=headers, subject="recognition.result")


class RecordingMapper:
    """Подставной mapper: запоминает вызовы и видимое состояние задания; `fail=True` — исключение; `delay` — пауза
    внутри транзакции (строка задания остаётся заблокированной)."""

    def __init__(self, *, fail: bool = False, delay: float = 0) -> None:
        self.fail = fail
        self.delay = delay
        self.calls: list[tuple[uuid.UUID, RecognitionStatus, RecognitionResultMessage]] = []

    async def map_result(self, session: AsyncSession, job: RecognitionJob, message: RecognitionResultMessage) -> None:
        self.calls.append((job.id, job.status, message))
        if self.delay:
            await asyncio.sleep(self.delay)
        if self.fail:
            raise RuntimeError("подставная ошибка маппинга")


@pytest.fixture
def session_factory(db_engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(db_engine, expire_on_commit=False)


@pytest_asyncio.fixture
async def jobs(session_factory: async_sessionmaker[AsyncSession]) -> AsyncIterator[CommittedJobs]:
    committed = CommittedJobs(session_factory)
    try:
        yield committed
    finally:
        await committed.cleanup()


async def _row(session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> dict[str, Any]:
    """Снимок всех колонок строки `recognition_job`."""
    async with session_factory() as session:
        row = (await session.execute(select(RecognitionJob.__table__).where(RecognitionJob.id == job_id))).one()
        return dict(row._mapping)


async def _s3_rows(session_factory: async_sessionmaker[AsyncSession], key: str) -> list[dict[str, Any]]:
    async with session_factory() as session:
        rows = await session.execute(select(S3Object.__table__).where(S3Object.key == key))
        return [dict(r._mapping) for r in rows]


async def _s3_total(session_factory: async_sessionmaker[AsyncSession]) -> int:
    async with session_factory() as session:
        return int((await session.execute(select(func.count()).select_from(S3Object))).scalar_one())


async def _file_id(jobs: CommittedJobs, job_id: uuid.UUID) -> uuid.UUID:
    return (await jobs.get(job_id)).file_id


async def _handle(session_factory: async_sessionmaker[AsyncSession], msg: Any, **kwargs: Any) -> None:
    await handle_message(msg, session_factory=session_factory, **kwargs)


# --- быстрые тесты: заголовок, метрики, разбор ------------------------------------------------------------------


@pytest.mark.parametrize(
    ("headers", "expected"),
    [
        ({"Nats-Msg-Id": "result:x:1"}, "result:x:1"),
        ({"nats-msg-id": "result:x:1"}, "result:x:1"),
        ({"NATS-MSG-ID": "ack:x:1", "other": "y"}, "ack:x:1"),
        ({"other": "y"}, None),
        ({"Nats-Msg-Id": ""}, None),
        ({}, None),
        (None, None),
    ],
)
def test_header_msg_id_is_case_insensitive(headers: dict[str, str] | None, expected: str | None):
    assert header_msg_id(headers) == expected


@pytest.mark.parametrize(
    ("existing", "incoming", "expected"),
    [
        ({"republish_count": 2}, {"duration_ms": 5}, {"republish_count": 2, "duration_ms": 5}),
        ({"republish_count": 2, "rows": 1}, {"rows": 7}, {"republish_count": 2, "rows": 7}),
        (None, {"duration_ms": 5}, {"duration_ms": 5}),  # SQL NULL / jsonb null (старые строки, сырой SQL)
        ([None, {"x": 1}], {"duration_ms": 5}, {"duration_ms": 5}),  # не объект — как пустой
        ({"republish_count": 1}, None, {"republish_count": 1}),  # метрик сервиса нет — без изменений
        (None, None, None),
    ],
)
def test_merge_metrics_keeps_republish_count(existing: Any, incoming: Any, expected: Any):
    assert merge_metrics(existing, incoming) == expected


class _UntouchableFactory:
    """session_factory, к которому нельзя обращаться: доказывает, что сообщение отвергнуто до БД."""

    def __call__(self) -> Any:
        raise AssertionError("невалидное сообщение не должно открывать сессию БД")


_JOB = uuid.UUID("11111111-1111-4111-8111-111111111111")
_FILE = uuid.UUID("22222222-2222-4222-8222-222222222222")


@pytest.mark.parametrize(
    "data",
    [
        b"{not json",
        b"\xff\xfe",
        b"[1, 2]",
        b'"text"',
        json.dumps({"schema_version": "1.0", "status": "done"}).encode(),  # нет job_id
        json.dumps(envelope(_JOB, _FILE, "done", schema_version="2.0")).encode(),  # чужой MAJOR
        json.dumps({k: v for k, v in envelope(_JOB, _FILE, "done").items() if k != "result_s3"}).encode(),
        json.dumps(envelope(_JOB, _FILE, "error")).encode(),  # error без поля error
        json.dumps(envelope(_JOB, _FILE, "finished")).encode(),  # статус вне enum
        json.dumps(envelope(_JOB, _FILE, "done", service={"name": "other", "version": "1"})).encode(),
        json.dumps({**envelope(_JOB, _FILE, "done"), "job_id": "not-a-uuid"}).encode(),
    ],
)
async def test_invalid_message_is_logged_and_does_not_touch_db(data: bytes, logs: list[Any]):
    msg = SimpleNamespace(data=data, headers={NATS_MSG_ID_HEADER: "result:x:1"}, subject="recognition.result")
    await handle_message(msg, session_factory=_UntouchableFactory())  # type: ignore[arg-type]
    rejected = _logged(logs, "отклонено", "WARNING")
    assert len(rejected) == 1
    # тело целиком в лог не попадает
    assert "result.json" not in rejected[0]["message"]


async def test_invalid_message_log_carries_job_id_when_readable(logs: list[Any]):
    body = envelope(_JOB, _FILE, "error")  # нет обязательного error
    msg = fake_msg(body, "result:x:1", validate=False)
    await handle_message(msg, session_factory=_UntouchableFactory())  # type: ignore[arg-type]
    (record,) = _logged(logs, "отклонено", "WARNING")
    assert str(_JOB) in record["message"]


async def test_invalid_message_job_id_is_escaped_in_log(logs: list[Any]):
    """job_id невалидного сообщения с переводом строки попадает в лог через repr: подделать строку лога нельзя (M-7)."""
    forged = "not-a-uuid\n2026-10-04 | INFO | подделанная запись\r\x1b[31m"
    body = {**envelope(_JOB, _FILE, "done"), "job_id": forged}
    msg = fake_msg(body, "result:x:1", validate=False)
    await handle_message(msg, session_factory=_UntouchableFactory())  # type: ignore[arg-type]
    (record,) = _logged(logs, "отклонено", "WARNING")
    message = record["message"]
    assert "\n" not in message and "\r" not in message and "\x1b" not in message
    assert repr(forged) in message


async def test_database_error_is_logged_and_not_raised(logs: list[Any]):
    """Ошибка БД (сервер недоступен) — `logger.exception`, без повторов, наружу не выходит."""
    engine = create_async_engine("postgresql+asyncpg://nobody:nothing@127.0.0.1:1/none")
    try:
        factory = async_sessionmaker(engine, expire_on_commit=False)
        await handle_message(
            fake_msg(envelope(_JOB, _FILE, "processing"), ack_msg_id(_JOB, 1)), session_factory=factory
        )
    finally:
        await engine.dispose()
    (record,) = _logged(logs, "транзакция откачена", "ERROR")
    assert record["exception"] is not None


async def test_noop_mapper_does_nothing():
    message = RecognitionResultMessage.model_validate(envelope(_JOB, _FILE, "done"))
    assert await NoopMapper().map_result(None, None, message) is None  # type: ignore[arg-type]


def test_final_statuses():
    """R-22 (design #34): финальные для inbox — done, needs_check, error, timeout, dlq."""
    assert {s.value for s in FINAL_STATUSES} == {"done", "needs_check", "error", "timeout", "dlq"}


def test_template_nats_callback_removed():
    """R-3 (design #16): шаблонный `core/nats_callback.py` удалён и нигде в `src/` не упоминается."""
    assert importlib.util.find_spec("src.core.nats_callback") is None
    offenders = [p for p in (API_DIR / "src").rglob("*.py") if "nats_callback" in p.read_text(encoding="utf-8")]
    assert offenders == []


# --- ack: processing ---------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_ack_moves_published_job_to_processing(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """Сценарий «Сервис подтвердил»."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.published, request_msg_id=f"{uuid.uuid4()}:1")
    file_id = await _file_id(jobs, job_id)
    before = await _row(session_factory, job_id)
    moment = datetime(2026, 10, 4, 12, 0, tzinfo=UTC)

    await _handle(
        session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), now=lambda: moment
    )

    after = await _row(session_factory, job_id)
    assert after["status"] is RecognitionStatus.processing
    assert after["started_at"] == moment
    assert (after["service_name"], after["service_version"]) == ("tabular-parser", "0.1.0")
    changed = {k for k in after if after[k] != before[k]}
    assert changed == {"status", "started_at", "service_name", "service_version"}
    assert _logged(logs, "ack", "INFO")


@pytest.mark.integration
async def test_repeated_ack_changes_nothing(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.published)
    file_id = await _file_id(jobs, job_id)
    ack = fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1))
    await _handle(session_factory, ack)
    first = await _row(session_factory, job_id)

    await _handle(session_factory, ack, now=lambda: datetime.now(UTC) + timedelta(minutes=1))

    assert await _row(session_factory, job_id) == first


@pytest.mark.integration
@pytest.mark.parametrize(
    "status",
    [
        RecognitionStatus.queued,
        RecognitionStatus.done,
        RecognitionStatus.needs_check,
        RecognitionStatus.error,
        RecognitionStatus.timeout,
        RecognitionStatus.dlq,
    ],
)
async def test_ack_for_other_status_is_logged_and_ignored(
    session_factory, jobs: CommittedJobs, logs: list[Any], status: RecognitionStatus
):
    job_id = await jobs.make_queued_job(status=status)
    file_id = await _file_id(jobs, job_id)
    before = await _row(session_factory, job_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))

    assert await _row(session_factory, job_id) == before
    (record,) = _logged(logs, "задание не изменено", "WARNING")
    assert status.value in record["message"]


# --- результат ---------------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_done_result_registers_s3_object_and_updates_job(session_factory, jobs: CommittedJobs):
    """Сценарий «Успешный результат»: s3_object (results), result_msg_id, finished_at, метрики слиты."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics={"republish_count": 1})
    file_id = await _file_id(jobs, job_id)
    moment = datetime(2026, 10, 4, 12, 30, tzinfo=UTC)
    metrics = {"duration_ms": 1234, "rows": 50, "fields_total": 300, "fields_low_confidence": 4,
               "fields_unrecognized": 2, "warnings": ["лист без шапки"]}  # fmt: skip
    body = envelope(job_id, file_id, "done", metrics=metrics, result_schema_version="1.1")
    before_total = await _s3_total(session_factory)

    await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1)), now=lambda: moment)

    job = await _row(session_factory, job_id)
    (s3,) = await _s3_rows(session_factory, _result_key(file_id, job_id))
    assert await _s3_total(session_factory) == before_total + 1
    assert job["status"] is RecognitionStatus.done
    assert job["result_msg_id"] == f"result:{job_id}:1"
    assert job["result_s3_object_id"] == s3["id"]
    assert job["finished_at"] == moment
    assert job["result_schema_version"] == "1.1"
    assert (job["service_name"], job["service_version"]) == ("tabular-parser", "0.1.0")
    assert job["metrics"] == {"republish_count": 1, **metrics}
    assert (job["fields_total"], job["fields_low_confidence"], job["fields_unrecognized"]) == (300, 4, 2)
    assert (job["error_code"], job["error_message"]) == (None, None)
    assert s3["bucket_role"] is S3BucketRole.results
    assert (s3["bucket"], s3["created_by"], s3["created_at"]) == (BUCKET, "tabular-parser", moment)


@pytest.mark.integration
async def test_needs_check_result(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "needs_check"), result_msg_id(job_id, 1)))

    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.needs_check
    assert job["result_s3_object_id"] is not None
    assert job["metrics"] is None  # метрик не было ни у задания, ни у сервиса: колонка не трогается


@pytest.mark.integration
@pytest.mark.parametrize(
    ("stored", "expected_state"),
    [(None, (True, None)), (JSON.NULL, (False, "null"))],
    ids=["sql_null", "json_null"],
)
async def test_result_metrics_merge_over_null_metrics(
    session_factory, jobs: CommittedJobs, stored: Any, expected_state: Any
):
    """Слияние метрик в inbox (Python `merge_metrics`) работает и для SQL NULL, и для jsonb null (финальное ревью I-1);
    результат без метрик колонку не трогает: SQL NULL остаётся SQL NULL."""
    with_metrics = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics=stored)
    without_metrics = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics=stored)
    assert await jobs.metrics_sql_state(with_metrics) == expected_state

    file_id = await _file_id(jobs, with_metrics)
    body = envelope(with_metrics, file_id, "done", metrics={"duration_ms": 10})
    await _handle(session_factory, fake_msg(body, result_msg_id(with_metrics, 1)))
    file_id = await _file_id(jobs, without_metrics)
    await _handle(
        session_factory, fake_msg(envelope(without_metrics, file_id, "done"), result_msg_id(without_metrics, 1))
    )

    job = await _row(session_factory, with_metrics)
    assert job["status"] is RecognitionStatus.done
    assert job["metrics"] == {"duration_ms": 10}
    assert await jobs.metrics_sql_state(with_metrics) == (False, "object")
    assert (await _row(session_factory, without_metrics))["status"] is RecognitionStatus.done
    assert await jobs.metrics_sql_state(without_metrics) == expected_state


@pytest.mark.integration
async def test_redelivery_changes_nothing(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """Сценарий «Повторная доставка»: ни recognition_job, ни s3_object не меняются; mapper не вызывается."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    msg = fake_msg(envelope(job_id, file_id, "done", metrics={"rows": 3}), result_msg_id(job_id, 1))
    mapper = RecordingMapper()
    await _handle(session_factory, msg, mapper=mapper)
    job_before = await _row(session_factory, job_id)
    s3_before = await _s3_rows(session_factory, _result_key(file_id, job_id))
    total_before = await _s3_total(session_factory)

    await _handle(session_factory, msg, mapper=mapper, now=lambda: datetime.now(UTC) + timedelta(hours=1))

    assert await _row(session_factory, job_id) == job_before
    assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == s3_before
    assert await _s3_total(session_factory) == total_before
    assert len(mapper.calls) == 1
    assert _logged(logs, "дубликат", "DEBUG")
    assert not _logged(logs, "финальном статусе")


@pytest.mark.integration
async def test_service_error_without_result_s3(session_factory, jobs: CommittedJobs):
    """Сценарий «Ошибка сервиса»: NOT_IMPLEMENTED без result_s3."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    total_before = await _s3_total(session_factory)
    body = envelope(job_id, file_id, "error", error={"code": "NOT_IMPLEMENTED", "message": "разбор ОСВ — Срез 1"})

    await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1)))

    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.error
    assert (job["error_code"], job["error_message"]) == ("NOT_IMPLEMENTED", "разбор ОСВ — Срез 1")
    assert job["result_s3_object_id"] is None
    assert job["result_msg_id"] == f"result:{job_id}:1"
    assert job["finished_at"] is not None
    assert job["result_schema_version"] is None
    assert await _s3_total(session_factory) == total_before


@pytest.mark.integration
@pytest.mark.parametrize(
    "status",
    [
        RecognitionStatus.timeout,
        RecognitionStatus.dlq,
        RecognitionStatus.error,
        RecognitionStatus.done,
        RecognitionStatus.needs_check,
    ],
)
async def test_result_for_final_status_is_logged_and_ignored(
    session_factory, jobs: CommittedJobs, logs: list[Any], status: RecognitionStatus
):
    """Сценарий «Результат пришёл после таймаута» (и прочие финальные статусы, R-22): задание не меняется."""
    other = f"result:{uuid.uuid4()}:9" if status in (RecognitionStatus.done, RecognitionStatus.needs_check) else None
    job_id = await jobs.make_queued_job(status=status, result_msg_id=other)
    file_id = await _file_id(jobs, job_id)
    before = await _row(session_factory, job_id)
    total_before = await _s3_total(session_factory)
    mapper = RecordingMapper()

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), mapper=mapper)

    assert await _row(session_factory, job_id) == before
    assert await _s3_total(session_factory) == total_before
    assert mapper.calls == []
    (record,) = _logged(logs, "финальном статусе", "WARNING")
    assert status.value in record["message"]


@pytest.mark.integration
async def test_unknown_job_is_logged_and_creates_nothing(session_factory, logs: list[Any]):
    job_id, file_id = uuid.uuid4(), uuid.uuid4()
    total_before = await _s3_total(session_factory)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))

    assert await _s3_total(session_factory) == total_before
    assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == []
    (record,) = _logged(logs, "неизвестного задания", "WARNING")
    assert str(job_id) in record["message"]


@pytest.mark.integration
async def test_file_id_mismatch_is_logged_and_ignored(session_factory, jobs: CommittedJobs, logs: list[Any]):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    before = await _row(session_factory, job_id)
    total_before = await _s3_total(session_factory)

    await _handle(session_factory, fake_msg(envelope(job_id, uuid.uuid4(), "done"), result_msg_id(job_id, 1)))

    assert await _row(session_factory, job_id) == before
    assert await _s3_total(session_factory) == total_before
    assert _logged(logs, "file_id", "WARNING")


@pytest.mark.integration
@pytest.mark.parametrize("status", [RecognitionStatus.published, RecognitionStatus.queued])
async def test_result_without_ack_is_accepted(session_factory, jobs: CommittedJobs, status: RecognitionStatus):
    """R-22: потерянный ack не блокирует результат; `started_at` остаётся пустым (начало обработки неизвестно)."""
    job_id = await jobs.make_queued_job(status=status)
    file_id = await _file_id(jobs, job_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))

    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.done
    assert job["started_at"] is None
    assert job["service_name"] == "tabular-parser"


@pytest.mark.integration
async def test_msg_id_header_case_insensitive(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)

    await _handle(
        session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1), header="nats-msg-id")
    )

    assert (await _row(session_factory, job_id))["result_msg_id"] == f"result:{job_id}:1"


@pytest.mark.integration
async def test_missing_msg_id_header_uses_derived_id(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """Нет `Nats-Msg-Id` — предупреждение и производный `result:<job_id>:<attempt>`; повтор — дубликат."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, attempt=2)
    file_id = await _file_id(jobs, job_id)
    msg = fake_msg(envelope(job_id, file_id, "done"), None)

    await _handle(session_factory, msg)
    job = await _row(session_factory, job_id)
    await _handle(session_factory, msg)

    assert job["status"] is RecognitionStatus.done
    assert job["result_msg_id"] == f"result:{job_id}:2"
    assert await _row(session_factory, job_id) == job
    assert len(_logged(logs, "без заголовка Nats-Msg-Id", "WARNING")) == 2  # noqa: PLR2004 - оба сообщения


@pytest.mark.integration
async def test_existing_s3_object_is_reused(session_factory, jobs: CommittedJobs):
    """Ключ результата уже зарегистрирован — строка переиспользуется, дубль (bucket, key) не вставляется."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    async with session_factory() as session, session.begin():
        existing = S3Object(
            bucket_role=S3BucketRole.results,
            bucket=BUCKET,
            key=_result_key(file_id, job_id),
            created_by="tabular-parser",
            created_at=datetime.now(UTC),
        )
        session.add(existing)
        await session.flush()
        existing_id = existing.id
    jobs.track_s3_object(existing_id)

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))

    assert (await _row(session_factory, job_id))["result_s3_object_id"] == existing_id
    assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1


# --- mapper ------------------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_mapper_sees_job_and_message(session_factory, jobs: CommittedJobs):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    mapper = RecordingMapper()

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)), mapper=mapper)

    ((seen_job_id, seen_status, message),) = mapper.calls
    assert seen_job_id == job_id
    assert seen_status is RecognitionStatus.done  # mapper видит уже обновлённое задание (может его уточнить)
    assert message.job_id == job_id
    assert message.result_s3 is not None
    assert message.result_s3.key == _result_key(file_id, job_id)


@pytest.mark.integration
async def test_mapper_error_rolls_back_whole_transaction(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """Исключение mapper'а — откат всего: задание не изменено, s3_object не создан (дойдёт до timeout, D-6)."""
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    before = await _row(session_factory, job_id)
    total_before = await _s3_total(session_factory)

    msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
    await _handle(session_factory, msg, mapper=RecordingMapper(fail=True))

    assert await _row(session_factory, job_id) == before
    assert await _s3_total(session_factory) == total_before
    (record,) = _logged(logs, "Ошибка маппинга", "ERROR")
    assert record["exception"] is not None


# --- гонки --------------------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_concurrent_same_message_applied_once(session_factory, jobs: CommittedJobs, logs: list[Any]):
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
    file_id = await _file_id(jobs, job_id)
    msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
    mapper = RecordingMapper(delay=LOCK_HOLD_SECONDS)  # первый держит строку, второй ждёт его commit

    await asyncio.gather(*[_handle(session_factory, msg, mapper=mapper) for _ in range(2)])

    assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1
    assert (await _row(session_factory, job_id))["result_msg_id"] == f"result:{job_id}:1"
    assert len(mapper.calls) == 1
    assert len(_logged(logs, "дубликат", "DEBUG")) == 1
    assert not _logged(logs, "неизвестного задания")


class _SlowNats:
    """Подставной NATS: публикация «висит» `delay` секунд, пока публикатор держит блокировку строки задания."""

    def __init__(self, delay: float) -> None:
        self.delay = delay
        self.entered = asyncio.Event()

    @property
    def is_connected(self) -> bool:
        return True

    async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
        self.entered.set()
        await asyncio.sleep(self.delay)


@pytest.mark.integration
async def test_ack_waits_for_publisher_lock(session_factory, jobs: CommittedJobs):
    """Ack пришёл, пока публикатор держит строку (до его commit): обработчик ждёт блокировку и применяет ack после."""
    job_id = await jobs.make_queued_job()
    file_id = await _file_id(jobs, job_id)
    nats = _SlowNats(LOCK_HOLD_SECONDS)
    publisher = asyncio.create_task(
        publish_pending(session_factory, nats, ack_timeout=timedelta(seconds=30), max_republish=3, batch_size=1)
    )
    await asyncio.wait_for(nats.entered.wait(), 5)
    started = time.monotonic()

    await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))

    waited = time.monotonic() - started
    report = await publisher
    job = await _row(session_factory, job_id)
    assert report.published == 1
    assert waited >= LOCK_HOLD_SECONDS / 2  # ждал блокировку, а не увидел queued
    assert job["status"] is RecognitionStatus.processing
    assert job["published_at"] is not None
    assert job["started_at"] is not None


@pytest.mark.integration
async def test_result_vs_watchdog_is_consistent(session_factory, jobs: CommittedJobs):
    """Результат и вахтёр на заданиях с прошедшим сроком одновременно: каждое — либо done (с s3_object), либо
    timeout (без результата), но не смесь. Порядок решает блокировка строки: кто первый, тот и финальный."""
    past = datetime.now(UTC) - timedelta(minutes=1)
    job_ids = [
        await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past) for _ in range(RACE_JOBS)
    ]
    messages = [
        fake_msg(envelope(job_id, await _file_id(jobs, job_id), "done"), result_msg_id(job_id, 1)) for job_id in job_ids
    ]

    await asyncio.gather(expire_deadlines(session_factory), *[_handle(session_factory, m) for m in messages])

    for job_id in job_ids:
        job = await _row(session_factory, job_id)
        s3 = await _s3_rows(session_factory, _result_key(job["file_id"], job_id))
        if job["status"] is RecognitionStatus.done:
            assert job["error_code"] is None
            assert job["result_msg_id"] == f"result:{job_id}:1"
            assert [r["id"] for r in s3] == [job["result_s3_object_id"]]
        else:
            assert job["status"] is RecognitionStatus.timeout
            assert job["error_code"] == "TIMEOUT"
            assert job["result_msg_id"] is None
            assert job["result_s3_object_id"] is None
            assert s3 == []


@pytest.mark.integration
async def test_watchdog_waits_for_result_transaction(session_factory, jobs: CommittedJobs):
    """Вахтёр запущен, пока обработчик результата держит строку: его UPDATE ждёт commit, перепроверяет статус на новой
    версии строки (`done`) и задание не трогает."""
    past = datetime.now(UTC) - timedelta(minutes=1)
    job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)
    mapper = RecordingMapper(delay=LOCK_HOLD_SECONDS)
    msg = fake_msg(envelope(job_id, await _file_id(jobs, job_id), "done"), result_msg_id(job_id, 1))
    handler = asyncio.create_task(_handle(session_factory, msg, mapper=mapper))

    async def locked() -> None:  # обработчик внутри транзакции, строка заблокирована
        while not mapper.calls:
            await asyncio.sleep(0.01)

    await asyncio.wait_for(locked(), 5)

    expired = await expire_deadlines(session_factory)
    await handler

    assert expired == 0
    job = await _row(session_factory, job_id)
    assert job["status"] is RecognitionStatus.done
    assert job["error_code"] is None


@pytest.mark.integration
async def test_watchdog_after_result_and_result_after_watchdog(session_factory, jobs: CommittedJobs):
    past = datetime.now(UTC) - timedelta(minutes=1)
    first = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)
    second = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)

    await _handle(
        session_factory, fake_msg(envelope(first, await _file_id(jobs, first), "done"), result_msg_id(first, 1))
    )
    await expire_deadlines(session_factory)
    await _handle(
        session_factory, fake_msg(envelope(second, await _file_id(jobs, second), "done"), result_msg_id(second, 1))
    )

    assert (await _row(session_factory, first))["status"] is RecognitionStatus.done
    assert (await _row(session_factory, second))["status"] is RecognitionStatus.timeout


# --- e2e: реальный NATS -------------------------------------------------------------------------------------------


async def _wait_until(predicate: Any, timeout: float = 10.0) -> None:
    deadline = time.monotonic() + timeout
    while not await predicate():
        if time.monotonic() > deadline:
            raise AssertionError("условие не выполнено за отведённое время")
        await asyncio.sleep(0.05)


async def _wait_status(
    session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, status: RecognitionStatus
) -> dict[str, Any]:
    async def reached() -> bool:
        return (await _row(session_factory, job_id))["status"] is status

    await _wait_until(reached)
    return await _row(session_factory, job_id)


class _FakeService:
    """«Сервис распознавания» поверх голого nats-py: принимает запросы своего задания, публикует конверты."""

    def __init__(self, job_id: uuid.UUID, result_subject: str) -> None:
        self.job_id = job_id
        self.result_subject = result_subject
        self.requests: list[Msg] = []
        self._nc = NATS()

    async def start(self, url: str) -> None:
        await self._nc.connect(url, connect_timeout=2, allow_reconnect=False)

        async def collect(msg: Msg) -> None:
            if json.loads(msg.data).get("job_id") == str(self.job_id):
                self.requests.append(msg)

        await self._nc.subscribe("recognition.tabular.request", cb=collect)
        await self._nc.flush()

    async def send(self, body: dict[str, Any], msg_id: str) -> None:
        MESSAGE_VALIDATOR.validate(body)
        await self._nc.publish(self.result_subject, json.dumps(body).encode(), headers={NATS_MSG_ID_HEADER: msg_id})
        await self._nc.flush()

    async def stop(self) -> None:
        await self._nc.close()


@pytest.mark.integration
async def test_full_cycle_over_real_nats(session_factory, jobs: CommittedJobs, logs: list[Any]):
    """queued → published (outbox) → processing (ack) → done + s3_object (результат) → повтор без изменений."""
    settings = get_settings().OPENBAO.NATS
    result_subject = f"test.result.{uuid.uuid4().hex}"
    handler = partial(handle_message, session_factory=session_factory, mapper=NoopMapper())

    async def subscribe(conn: NatsConnection) -> None:
        await conn.subscribe_results(handler)

    core = NatsConnection(settings, on_connected=subscribe, result_subject=result_subject)
    job_id = await jobs.make_queued_job()
    file_id = await _file_id(jobs, job_id)
    service = _FakeService(job_id, result_subject)
    await core.start()
    try:
        if not await core.wait_ready(timeout=READY_TIMEOUT):
            pytest.fail(f"NATS недоступен ({settings.NATS_URL}, .env.test)", pytrace=False)
        await service.start(settings.NATS_URL)

        async def published() -> bool:
            await publish_pending(session_factory, core, ack_timeout=timedelta(seconds=30), max_republish=3)
            return bool(service.requests)

        await _wait_until(published)
        await _wait_status(session_factory, job_id, RecognitionStatus.published)
        assert service.requests[0].headers == {NATS_MSG_ID_HEADER: f"{job_id}:1"}

        await service.send(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1))
        job = await _wait_status(session_factory, job_id, RecognitionStatus.processing)
        assert job["started_at"] is not None
        assert (job["service_name"], job["service_version"]) == ("tabular-parser", "0.1.0")

        result = envelope(job_id, file_id, "done", metrics={"duration_ms": 10})
        await service.send(result, result_msg_id(job_id, 1))
        job = await _wait_status(session_factory, job_id, RecognitionStatus.done)
        (s3,) = await _s3_rows(session_factory, _result_key(file_id, job_id))
        assert (job["result_s3_object_id"], job["result_msg_id"]) == (s3["id"], f"result:{job_id}:1")
        assert job["finished_at"] is not None
        assert s3["bucket_role"] is S3BucketRole.results

        await service.send(result, result_msg_id(job_id, 1))

        async def duplicate_seen() -> bool:
            return bool(_logged(logs, "дубликат", "DEBUG"))

        await _wait_until(duplicate_seen)
        assert await _row(session_factory, job_id) == job
        assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == [s3]
    finally:
        await service.stop()
        await core.stop()