"""Outbox-публикатор и вахтёр сроков `recognition_job` (spec recognition-job-exchange: «Outbox публикует задания из
`queued`», «Запрос без ack публикуется повторно, затем — `dlq`», «Таймаут ожидания результата»; design D-6). Срез 2c:
`timeout` вахтёра и `dlq` публикатора переводят файл текущего задания в `ERROR` (spec primary-files «Жизненный цикл
статуса файла», design D-5).

Интеграционные тесты работают против реального PostgreSQL (БД сессии pytest) и NATS из `.env.test`. Публикатор и
вахтёр ведут собственные транзакции, поэтому данные фиксируются реально (`CommittedJobs`) и удаляются в teardown.
Subject'ы запросов общие (`files.*_uploaded`): тестовый подписчик фильтрует сообщения по `job_id` своих заданий.
Время в тестах повторов — инжектируемые часы `now`, без реальных ожиданий таймаутов.
"""

import asyncio
import json
import time
import uuid
from collections.abc import AsyncIterator, Callable
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any, get_args

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

from src.contract import NATS_MSG_ID_HEADER, RecognitionRequest
from src.contract.request import FileType as ContractFileType
from src.core.config import get_settings
from src.core.nats import NatsConnection, NatsPublishError
from src.models.enums import FileStatus, FileType, RecognitionPipeline, RecognitionStatus
from src.models.files import RecognitionJob
from src.services import recognition_exchange as rx
from src.services.recognition_exchange import (
    WATCHDOG_PERIOD_SECONDS,
    ExchangeTasks,
    expire_deadlines,
    publish_pending,
    run_publisher_loop,
    run_watchdog_loop,
)
from tests.factories import CommittedJobs

CONTRACTS_DIR = Path(__file__).resolve().parents[1] / "contracts"
REQUEST_SCHEMA = json.loads((CONTRACTS_DIR / "recognition-request.schema.json").read_text(encoding="utf-8"))
REQUEST_VALIDATOR = Draft202012Validator(REQUEST_SCHEMA, format_checker=FormatChecker())
REQUEST_SUBJECTS = ("files.tabular_uploaded", "files.img_uploaded")

ACK = timedelta(seconds=30)
MAX_REPUBLISH = 3
READY_TIMEOUT = 5.0
CONCURRENT_JOBS = 20
MIN_LOOP_PASSES = 3  # цикл пережил несколько ошибок прохода подряд
EXCHANGE_TASKS = 2  # публикатор и вахтёр


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


class Clock:
    """Инжектируемые часы: `now` для публикатора и вахтёра, `advance` — сдвиг без реального ожидания."""

    def __init__(self) -> None:
        self.current = datetime.now(UTC)

    def __call__(self) -> datetime:
        return self.current

    def advance(self, delta: timedelta) -> None:
        self.current += delta


class RequestSubscriber:
    """Тестовый подписчик на subject'ы запросов: «голый» клиент nats-py, сообщения фильтруются по `job_id`."""

    def __init__(self, subjects: tuple[str, ...] = REQUEST_SUBJECTS) -> None:
        self.subjects = subjects
        self.messages: list[Msg] = []
        self._nc = NATS()

    async def start(self) -> None:
        url = get_settings().OPENBAO.NATS.NATS_URL
        try:
            await self._nc.connect(url, connect_timeout=2, allow_reconnect=False)
        except Exception as exc:
            pytest.fail(f"NATS недоступен ({url}, .env.test): {type(exc).__name__}", pytrace=False)

        async def collect(msg: Msg) -> None:
            self.messages.append(msg)

        for subject in self.subjects:
            await self._nc.subscribe(subject, cb=collect)
        await self._nc.flush()

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

    def for_job(self, job_id: uuid.UUID) -> list[Msg]:
        return [m for m in self.messages if json.loads(m.data).get("job_id") == str(job_id)]

    async def settle(self) -> None:
        """Всё, что сервер принял до этого момента, доставлено подписчику (PING/PONG после сообщений) и обработано."""
        await self._nc.flush()
        await asyncio.sleep(0.1)

    async def wait_for(self, job_id: uuid.UUID, count: int, timeout: float = 10.0) -> list[Msg]:
        deadline = time.monotonic() + timeout
        while len(self.for_job(job_id)) < count:
            if time.monotonic() > deadline:
                raise AssertionError(f"за {timeout} с получено {len(self.for_job(job_id))} из {count} сообщений")
            await asyncio.sleep(0.02)
        return self.for_job(job_id)


class FakeNats:
    """Подставной NATS: состояние соединения и счётчик публикаций; первые `fail_first` публикаций — ошибка."""

    def __init__(self, *, connected: bool = True, fail_first: int = 0, delegate: Any = None, delay: float = 0) -> None:
        self.is_connected_value = connected
        self.fail_first = fail_first
        self.delegate = delegate
        self.delay = delay
        self.calls: list[tuple[str, str]] = []

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

    async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
        self.calls.append((subject, msg_id))
        if len(self.calls) <= self.fail_first:
            raise NatsPublishError(f"подставная ошибка публикации {msg_id}")
        if self.delay:
            await asyncio.sleep(self.delay)
        if self.delegate is not None:
            await self.delegate.publish(subject, payload, msg_id)


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


@pytest_asyncio.fixture
async def nats() -> AsyncIterator[NatsConnection]:
    conn = NatsConnection(get_settings().OPENBAO.NATS)
    await conn.start()
    try:
        if not await conn.wait_ready(timeout=READY_TIMEOUT):
            pytest.fail(f"NATS недоступен ({get_settings().OPENBAO.NATS.NATS_URL}, .env.test)", pytrace=False)
        yield conn
    finally:
        await conn.stop()


@pytest_asyncio.fixture
async def subscriber() -> AsyncIterator[RequestSubscriber]:
    sub = RequestSubscriber()
    await sub.start()
    try:
        yield sub
    finally:
        await sub.stop()


async def _set_status(
    session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, status: RecognitionStatus
) -> None:
    async with session_factory() as session, session.begin():
        await session.execute(update(RecognitionJob).where(RecognitionJob.id == job_id).values(status=status))


def _publish(
    session_factory: async_sessionmaker[AsyncSession], nats: Any, now: Callable[[], datetime] | None = None
) -> Any:
    return publish_pending(session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, now=now)


# --- юнит: соответствие enum, константы ------------------------------------------------------------------------


def test_orm_file_type_matches_contract_enum():
    """`stored_file.file_type` (DBML) и `file_type` схемы запроса — одно множество значений."""
    assert {m.value for m in FileType} == set(get_args(ContractFileType))
    assert set(REQUEST_SCHEMA["properties"]["file_type"]["enum"]) == {m.value for m in FileType}


def test_orm_pipeline_matches_request_subjects():
    assert {m.value for m in RecognitionPipeline} == {"tabular", "image"}


def test_watchdog_period_is_30_seconds():
    assert WATCHDOG_PERIOD_SECONDS == 30  # noqa: PLR2004 - значение из spec


# --- публикация -------------------------------------------------------------------------------------------------


@pytest.mark.integration
@pytest.mark.parametrize(
    ("pipeline", "file_type", "subject"),
    [
        (RecognitionPipeline.tabular, FileType.XLSX, "files.tabular_uploaded"),
        (RecognitionPipeline.image, FileType.PDF, "files.img_uploaded"),
    ],
)
async def test_publish_request_to_service(  # noqa: PLR0913,PLR0917 - фикстуры и параметры
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
    pipeline: RecognitionPipeline,
    file_type: FileType,
    subject: str,
):
    """Сценарий «Публикация запроса парсеру»: сообщение по схеме, `attempt = 1`, строка `published`."""
    job_id = await jobs.make_queued_job(pipeline=pipeline, file_type=file_type, original_name="ОСВ 60.xlsx")
    before = datetime.now(UTC)

    report = await _publish(session_factory, nats)

    assert report.published == 1
    (msg,) = await subscriber.wait_for(job_id, 1)
    body = json.loads(msg.data)
    errors = [e.message for e in REQUEST_VALIDATOR.iter_errors(body)]
    assert errors == []
    request = RecognitionRequest.model_validate_json(msg.data)
    job = await jobs.get(job_id)
    assert msg.subject == subject
    assert msg.headers == {NATS_MSG_ID_HEADER: f"{job_id}:1"}
    assert request.job_id == job_id
    assert request.file_id == job.file_id
    assert request.pipeline == pipeline.value
    assert request.file_type == file_type.value
    assert request.original_name == "ОСВ 60.xlsx"
    assert request.s3.bucket == "arm-ka"
    assert request.s3.key == await jobs.s3_key(job_id)
    assert request.attempt == 1
    assert request.schema_version == "1.0"
    # reply_deadline не был задан: вычислен при первой публикации (парсер — 5 мин, OCR — 20 мин) и сохранён в задание
    assert job.reply_deadline is not None
    assert request.reply_deadline == job.reply_deadline
    expected = timedelta(minutes=20 if pipeline is RecognitionPipeline.image else 5)
    assert expected - timedelta(minutes=1) < job.reply_deadline - before < expected + timedelta(minutes=1)

    assert job.status is RecognitionStatus.published
    assert job.request_msg_id == f"{job_id}:1"
    assert job.request_subject == subject
    assert job.published_at is not None and job.published_at >= before
    assert job.requested_at is not None
    assert job.attempt == 1


@pytest.mark.integration
async def test_publish_uses_configured_request_subject(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
) -> None:
    """Review Focus 4: subject — из настроек окружения, `request_subject` задания — фактический subject публикации."""
    run = uuid.uuid4().hex[:8]
    subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"}
    custom = RequestSubscriber(subjects=tuple(subjects.values()))
    await custom.start()
    try:
        job_id = await jobs.make_queued_job(
            pipeline=RecognitionPipeline.tabular, file_type=FileType.XLSX, original_name="ОСВ 60.xlsx"
        )
        await publish_pending(
            session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, request_subjects=subjects
        )
        (msg,) = await custom.wait_for(job_id, 1)
    finally:
        await custom.stop()
    assert msg.subject == subjects["tabular"]
    assert (await jobs.get(job_id)).request_subject == subjects["tabular"]


@pytest.mark.integration
async def test_republish_updates_request_subject(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
) -> None:
    """I-1: повторная публикация уходит на актуальный subject и обновляет `request_subject` задания."""
    run = uuid.uuid4().hex[:8]
    subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"}
    clock = Clock()
    job_id = await jobs.make_queued_job(
        pipeline=RecognitionPipeline.tabular,
        file_type=FileType.XLSX,
        original_name="ОСВ 61.xlsx",
        reply_deadline=clock() + timedelta(hours=1),
    )
    assert (await _publish(session_factory, nats, clock)).published == 1
    old_subject = (await jobs.get(job_id)).request_subject
    assert old_subject != subjects["tabular"]

    custom = RequestSubscriber(subjects=tuple(subjects.values()))
    await custom.start()
    try:
        clock.advance(ACK + timedelta(seconds=1))
        report = await publish_pending(
            session_factory,
            nats,
            ack_timeout=ACK,
            max_republish=MAX_REPUBLISH,
            now=clock,
            request_subjects=subjects,
        )
        assert report.republished == 1
        (msg,) = await custom.wait_for(job_id, 1)
    finally:
        await custom.stop()
    assert msg.subject == subjects["tabular"]
    assert (await jobs.get(job_id)).request_subject == subjects["tabular"]


@pytest.mark.integration
async def test_publish_keeps_existing_reply_deadline(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
):
    deadline = datetime.now(UTC).replace(microsecond=0) + timedelta(hours=2)
    job_id = await jobs.make_queued_job(reply_deadline=deadline)

    await _publish(session_factory, nats)

    (msg,) = await subscriber.wait_for(job_id, 1)
    assert RecognitionRequest.model_validate_json(msg.data).reply_deadline == deadline
    assert (await jobs.get(job_id)).reply_deadline == deadline


@pytest.mark.integration
async def test_first_publication_msg_id_follows_attempt(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
):
    """`queued` с `attempt = 2` и устаревшим `request_msg_id` (будущий replay): id и тело — по текущему attempt."""
    job_id = await jobs.make_queued_job(attempt=2, request_msg_id=f"stale-{uuid.uuid4()}")

    await _publish(session_factory, nats)

    (msg,) = await subscriber.wait_for(job_id, 1)
    assert msg.headers == {NATS_MSG_ID_HEADER: f"{job_id}:2"}
    assert RecognitionRequest.model_validate_json(msg.data).attempt == 2  # noqa: PLR2004 - attempt задания
    job = await jobs.get(job_id)
    assert (job.request_msg_id, job.attempt) == (f"{job_id}:2", 2)


@pytest.mark.integration
@pytest.mark.parametrize("current", [True, False], ids=["current", "non_current"])
async def test_invalid_request_marks_job_error_and_pass_continues(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
    current: bool,
):
    """Запрос не собирается по схеме (нет `original_name`) — задание `error/INVALID_REQUEST`, проход идёт дальше; файл
    текущего задания — `ERROR` (финальная ошибка задания, spec primary-files), файл чужого задания не меняется."""
    clock = Clock()
    file_id, broken = await jobs.make_file_job(original_name=None)
    # новое задание — `processing`, чтобы публикатор его не трогал (в проходе только старое задание файла)
    new_job = None if current else await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True)
    good = await jobs.make_queued_job()

    report = await _publish(session_factory, nats, clock)

    assert (report.invalid, report.published) == (1, 1)
    await subscriber.wait_for(good, 1)
    await subscriber.settle()
    assert subscriber.for_job(broken) == []
    job = await jobs.get(broken)
    assert job.status is RecognitionStatus.error
    assert job.error_code == "INVALID_REQUEST"
    assert job.error_message and "original_name" in job.error_message
    assert job.finished_at is not None
    assert job.request_msg_id is None
    file = await jobs.file(file_id)
    if current:
        assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())
    else:
        assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)


# --- повторы без ack и dlq ---------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_republish_without_ack_then_dlq(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
):
    """Сценарий «Сервис недоступен»: 1 + 3 публикации с одним `Nats-Msg-Id`, затем `dlq / NO_ACK`."""
    clock = Clock()
    job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1), metrics={"note": "сохранить"})

    assert (await _publish(session_factory, nats, clock)).published == 1
    first = await jobs.get(job_id)
    # раньше ACK_TIMEOUT повтора нет
    clock.advance(ACK - timedelta(seconds=1))
    assert (await _publish(session_factory, nats, clock)).republished == 0

    for n in range(1, MAX_REPUBLISH + 1):
        clock.advance(timedelta(seconds=2))  # с последней публикации прошло больше ACK_TIMEOUT
        report = await _publish(session_factory, nats, clock)
        assert report.republished == 1
        job = await jobs.get(job_id)
        assert job.metrics == {"note": "сохранить", "republish_count": n}
        assert job.published_at == clock()
        assert job.status is RecognitionStatus.published
        clock.advance(ACK - timedelta(seconds=1))

    messages = await subscriber.wait_for(job_id, 1 + MAX_REPUBLISH)
    assert {m.headers[NATS_MSG_ID_HEADER] for m in messages} == {f"{job_id}:1"}
    assert len({m.data for m in messages}) == 1  # то же тело: тот же attempt и reply_deadline

    clock.advance(timedelta(seconds=2))
    report = await _publish(session_factory, nats, clock)
    assert (report.dlq, report.republished, report.published) == (1, 0, 0)
    job = await jobs.get(job_id)
    assert job.status is RecognitionStatus.dlq
    assert job.error_code == "NO_ACK"
    assert job.error_message
    assert job.finished_at == clock()
    assert job.request_msg_id == first.request_msg_id == f"{job_id}:1"
    assert job.attempt == 1
    assert job.metrics == {"note": "сохранить", "republish_count": MAX_REPUBLISH}

    clock.advance(ACK * 2)
    assert (await _publish(session_factory, nats, clock)).total == 0
    await subscriber.settle()
    assert len(subscriber.for_job(job_id)) == 1 + MAX_REPUBLISH


@pytest.mark.integration
@pytest.mark.parametrize(
    ("stored", "expected_state"),
    [
        (None, (True, None)),  # Python None → SQL NULL (JSONB(none_as_null=True), финальное ревью I-1)
        (JSON.NULL, (False, "null")),  # jsonb null: строки, записанные до исправления или сырым SQL
    ],
    ids=["sql_null", "json_null"],
)
async def test_republish_counter_on_null_metrics(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, stored: Any, expected_state: Any
):
    """Обход для не-объекта в `metrics` (`jsonb_typeof` CASE) работает и для SQL NULL, и для jsonb null."""
    clock = Clock()
    job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1), metrics=stored)
    assert await jobs.metrics_sql_state(job_id) == expected_state
    fake = FakeNats()

    assert (await _publish(session_factory, fake, clock)).published == 1
    assert await jobs.metrics_sql_state(job_id) == expected_state  # первая публикация metrics не трогает
    clock.advance(ACK + timedelta(seconds=1))
    assert (await _publish(session_factory, fake, clock)).republished == 1

    assert (await jobs.get(job_id)).metrics == {"republish_count": 1}
    assert await jobs.metrics_sql_state(job_id) == (False, "object")


@pytest.mark.integration
async def test_ack_after_first_republish_stops_republishing(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
):
    """Сценарий «Сервис появился после первого повтора»: после ack (`processing`) повторов нет."""
    clock = Clock()
    job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1))
    await _publish(session_factory, nats, clock)
    clock.advance(ACK + timedelta(seconds=1))
    assert (await _publish(session_factory, nats, clock)).republished == 1
    await subscriber.wait_for(job_id, 2)

    await _set_status(session_factory, job_id, RecognitionStatus.processing)  # имитация ack
    for _ in range(MAX_REPUBLISH + 1):
        clock.advance(ACK + timedelta(seconds=1))
        assert (await _publish(session_factory, nats, clock)).total == 0

    await subscriber.settle()
    assert len(subscriber.for_job(job_id)) == 2  # noqa: PLR2004 - первая публикация и один повтор
    job = await jobs.get(job_id)
    assert job.status is RecognitionStatus.processing
    assert job.metrics == {"republish_count": 1}


@pytest.mark.integration
async def test_publisher_skips_published_job_with_passed_deadline(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
):
    """`published` с прошедшим `reply_deadline` не републикуется и не уходит в dlq — его забирает вахтёр."""
    now = datetime.now(UTC)
    job_id = await jobs.make_queued_job(
        status=RecognitionStatus.published,
        published_at=now - timedelta(hours=1),
        reply_deadline=now - timedelta(minutes=1),
        request_msg_id=f"{uuid.uuid4()}:1",
        metrics={"republish_count": MAX_REPUBLISH},
    )

    assert (await _publish(session_factory, nats)).total == 0
    await subscriber.settle()
    assert subscriber.for_job(job_id) == []
    assert (await jobs.get(job_id)).status is RecognitionStatus.published

    assert await expire_deadlines(session_factory) == 1
    job = await jobs.get(job_id)
    assert (job.status, job.error_code) == (RecognitionStatus.timeout, "TIMEOUT")


# --- NATS недоступен ---------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_pass_skipped_when_nats_disconnected(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
):
    job_id = await jobs.make_queued_job()
    fake = FakeNats(connected=False)

    report = await _publish(session_factory, fake)

    assert report.skipped
    assert fake.calls == []
    assert (await jobs.get(job_id)).status is RecognitionStatus.queued


@pytest.mark.integration
async def test_publish_error_keeps_queued_and_stops_pass(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
):
    """Сценарий «NATS недоступен»: задание остаётся `queued`, проход прерван; следующий проход публикует."""
    first = await jobs.make_queued_job()
    second = await jobs.make_queued_job()
    fake = FakeNats(fail_first=1)

    report = await _publish(session_factory, fake)

    assert (report.failed, report.published) == (1, 0)
    assert len(fake.calls) == 1  # второе задание в этом проходе не пытались публиковать
    for job_id in (first, second):
        job = await jobs.get(job_id)
        assert job.status is RecognitionStatus.queued
        assert job.request_msg_id is None
        assert job.published_at is None
        assert job.reply_deadline is None  # откат: срок не сохранён

    report = await _publish(session_factory, nats)

    assert report.published == 2  # noqa: PLR2004 - оба задания
    for job_id in (first, second):
        await subscriber.wait_for(job_id, 1)
        assert (await jobs.get(job_id)).status is RecognitionStatus.published


@pytest.mark.integration
async def test_republish_error_keeps_published_state(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
):
    clock = Clock()
    job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1))
    await _publish(session_factory, nats, clock)
    published_at = (await jobs.get(job_id)).published_at
    clock.advance(ACK + timedelta(seconds=1))

    report = await _publish(session_factory, FakeNats(fail_first=1), clock)

    assert report.failed == 1
    job = await jobs.get(job_id)
    assert job.status is RecognitionStatus.published
    assert job.published_at == published_at
    assert job.metrics is None


# --- конкурентность -----------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_concurrent_publishers_publish_each_job_once(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
):
    """Два одновременных прохода (SKIP LOCKED): каждое задание опубликовано ровно один раз."""
    job_ids = [await jobs.make_queued_job() for _ in range(CONCURRENT_JOBS)]
    # задержка внутри публикации держит блокировку строки, пока второй проход выбирает задания
    first, second = FakeNats(delegate=nats, delay=0.02), FakeNats(delegate=nats, delay=0.02)

    reports = await asyncio.gather(_publish(session_factory, first), _publish(session_factory, second))

    assert sum(r.published for r in reports) == CONCURRENT_JOBS
    assert all(r.published > 0 for r in reports)  # оба прохода реально работали параллельно
    published_ids = [msg_id for fake in (first, second) for _, msg_id in fake.calls]
    assert sorted(published_ids) == sorted(f"{job_id}:1" for job_id in job_ids)
    await subscriber.settle()
    for job_id in job_ids:
        assert len(await subscriber.wait_for(job_id, 1)) == 1
        assert (await jobs.get(job_id)).status is RecognitionStatus.published


# --- вахтёр ----------------------------------------------------------------------------------------------------------


@pytest.mark.integration
@pytest.mark.parametrize(
    ("status", "deadline_offset", "expected"),
    [
        (RecognitionStatus.processing, timedelta(minutes=-1), RecognitionStatus.timeout),
        (RecognitionStatus.published, timedelta(minutes=-1), RecognitionStatus.timeout),
        (RecognitionStatus.processing, timedelta(minutes=5), RecognitionStatus.processing),
        (RecognitionStatus.published, timedelta(minutes=5), RecognitionStatus.published),
        (RecognitionStatus.processing, None, RecognitionStatus.processing),
        (RecognitionStatus.queued, timedelta(minutes=-1), RecognitionStatus.queued),
        (RecognitionStatus.done, timedelta(minutes=-1), RecognitionStatus.done),
        (RecognitionStatus.dlq, timedelta(minutes=-1), RecognitionStatus.dlq),
    ],
)
async def test_expire_deadlines(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    status: RecognitionStatus,
    deadline_offset: timedelta | None,
    expected: RecognitionStatus,
):
    """Сценарий «Сервис взял, но не ответил»: `processing`/`published` с прошедшим сроком → `timeout`."""
    clock = Clock()
    deadline = None if deadline_offset is None else clock() + deadline_offset
    job_id = await jobs.make_queued_job(status=status, reply_deadline=deadline)

    expired = await expire_deadlines(session_factory, now=clock)

    job = await jobs.get(job_id)
    assert job.status is expected
    if expected is RecognitionStatus.timeout:
        assert expired == 1
        assert job.error_code == "TIMEOUT"
        assert job.error_message
        assert job.finished_at == clock()
        assert job.attempt == 1
    else:
        assert expired == 0
        assert job.error_code is None
        assert job.finished_at is None


@pytest.mark.integration
@pytest.mark.parametrize("file_status", [FileStatus.PROCESSING, FileStatus.UPLOADED])
@pytest.mark.parametrize("job_status", [RecognitionStatus.processing, RecognitionStatus.published])
async def test_expire_deadlines_moves_current_file_to_error(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    file_status: FileStatus,
    job_status: RecognitionStatus,
):
    """Сценарий «Сервис не ответил» (primary-files): задание — `timeout`, файл — `ERROR` в той же транзакции."""
    clock = Clock()
    file_id, job_id = await jobs.make_file_job(
        file_status=file_status, job_status=job_status, reply_deadline=clock() - timedelta(minutes=1)
    )

    assert await expire_deadlines(session_factory, now=clock) == 1

    job, file = await jobs.get(job_id), await jobs.file(file_id)
    assert (job.status, job.error_code) == (RecognitionStatus.timeout, "TIMEOUT")
    assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())


@pytest.mark.integration
@pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED])
async def test_expire_deadlines_keeps_processed_and_superseded_file(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, file_status: FileStatus
):
    clock = Clock()
    file_id, job_id = await jobs.make_file_job(
        file_status=file_status, job_status=RecognitionStatus.processing, reply_deadline=clock() - timedelta(minutes=1)
    )

    assert await expire_deadlines(session_factory, now=clock) == 1

    assert (await jobs.get(job_id)).status is RecognitionStatus.timeout
    file = await jobs.file(file_id)
    assert (file.status, file.updated_at) == (file_status, None)


@pytest.mark.integration
async def test_expire_deadlines_of_non_current_job_keeps_file(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
):
    """Review Focus 1: таймаут старого задания файла, у которого уже новое текущее задание, файл не трогает."""
    clock = Clock()
    file_id, old_job = await jobs.make_file_job(
        file_status=FileStatus.UPLOADED,
        job_status=RecognitionStatus.processing,
        reply_deadline=clock() - timedelta(minutes=1),
    )
    new_job = await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True)

    assert await expire_deadlines(session_factory, now=clock) == 1

    assert (await jobs.get(old_job)).status is RecognitionStatus.timeout
    file = await jobs.file(file_id)
    assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)


@pytest.mark.integration
@pytest.mark.parametrize("current", [True, False], ids=["current", "non_current"])
async def test_dlq_moves_current_file_to_error(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, current: bool
):
    """Публикатор переводит задание в `dlq` — файл текущего задания `ERROR`; файл чужого задания не меняется."""
    clock = Clock()
    file_id, job_id = await jobs.make_file_job(
        job_status=RecognitionStatus.published,
        published_at=clock() - ACK * 2,
        request_msg_id="dlq-test:1",
        reply_deadline=clock() + timedelta(hours=1),
        metrics={"republish_count": MAX_REPUBLISH},
    )
    # новое задание — `processing`, чтобы публикатор его не трогал (в проходе только задание старого)
    new_job = None if current else await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True)

    report = await _publish(session_factory, FakeNats(), clock)

    assert report.dlq == 1
    assert (await jobs.get(job_id)).status is RecognitionStatus.dlq
    file = await jobs.file(file_id)
    if current:
        assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())
    else:
        assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)


# --- фоновые циклы -----------------------------------------------------------------------------------------------


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


async def _cancel(task: asyncio.Task[Any]) -> None:
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task


@pytest.mark.integration
async def test_publisher_loop_publishes_and_stops_on_cancel(
    session_factory: async_sessionmaker[AsyncSession],
    jobs: CommittedJobs,
    nats: NatsConnection,
    subscriber: RequestSubscriber,
):
    job_id = await jobs.make_queued_job()
    task = asyncio.create_task(
        run_publisher_loop(session_factory, nats, poll_seconds=0.1, ack_timeout=ACK, max_republish=MAX_REPUBLISH)
    )
    try:
        await subscriber.wait_for(job_id, 1, timeout=2.0)
    finally:
        await _cancel(task)
    assert (await jobs.get(job_id)).status is RecognitionStatus.published


@pytest.mark.integration
async def test_watchdog_loop_expires_and_stops_on_cancel(
    session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
):
    job_id = await jobs.make_queued_job(
        status=RecognitionStatus.processing, reply_deadline=datetime.now(UTC) - timedelta(seconds=1)
    )
    task = asyncio.create_task(run_watchdog_loop(session_factory, period_seconds=0.1))
    try:
        deadline = time.monotonic() + 2.0
        while (await jobs.get(job_id)).status is not RecognitionStatus.timeout:
            assert time.monotonic() < deadline, "вахтёр не перевёл задание в timeout за 2 с"
            await asyncio.sleep(0.05)
    finally:
        await _cancel(task)


class RecordingSleep:
    """Подмена `sleep` циклов: запоминает паузы и только уступает управление."""

    def __init__(self) -> None:
        self.delays: list[float] = []

    async def __call__(self, delay: float) -> None:
        self.delays.append(delay)
        await asyncio.sleep(0)


async def test_publisher_loop_survives_pass_errors(monkeypatch: pytest.MonkeyPatch):
    calls: list[int] = []

    async def failing_pass(*args: Any, **kwargs: Any) -> Any:
        calls.append(1)
        raise RuntimeError("БД недоступна")

    monkeypatch.setattr(rx, "publish_pending", failing_pass)
    sleep = RecordingSleep()
    task = asyncio.create_task(
        run_publisher_loop(
            object(),  # type: ignore[arg-type]  # проход подменён: фабрика сессий не используется
            FakeNats(),
            poll_seconds=5,
            ack_timeout=ACK,
            max_republish=MAX_REPUBLISH,
            sleep=sleep,
        )
    )
    try:
        await _wait_until(lambda: len(calls) >= MIN_LOOP_PASSES)
    finally:
        await _cancel(task)
    assert set(sleep.delays) == {5}


async def test_watchdog_loop_survives_pass_errors(monkeypatch: pytest.MonkeyPatch):
    calls: list[int] = []

    async def failing_pass(*args: Any, **kwargs: Any) -> int:
        calls.append(1)
        raise RuntimeError("БД недоступна")

    monkeypatch.setattr(rx, "expire_deadlines", failing_pass)
    sleep = RecordingSleep()
    task = asyncio.create_task(run_watchdog_loop(object(), sleep=sleep))  # type: ignore[arg-type]
    try:
        await _wait_until(lambda: len(calls) >= MIN_LOOP_PASSES)
    finally:
        await _cancel(task)
    assert set(sleep.delays) == {WATCHDOG_PERIOD_SECONDS}


async def test_exchange_tasks_start_two_loops_and_stop(monkeypatch: pytest.MonkeyPatch):
    started: dict[str, dict[str, Any]] = {}

    async def fake_publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None:
        started["publisher"] = kwargs
        await asyncio.Event().wait()

    async def fake_watchdog(session_factory: Any, **kwargs: Any) -> None:
        started["watchdog"] = kwargs
        await asyncio.Event().wait()

    monkeypatch.setattr(rx, "run_publisher_loop", fake_publisher)
    monkeypatch.setattr(rx, "run_watchdog_loop", fake_watchdog)
    settings = get_settings().model_copy(
        update={"OUTBOX_POLL_SECONDS": 7, "ACK_TIMEOUT_SECONDS": 11, "OUTBOX_MAX_REPUBLISH": 2}
    )

    exchange = ExchangeTasks.start(object(), FakeNats(), settings)  # type: ignore[arg-type]
    await _wait_until(lambda: len(started) == EXCHANGE_TASKS)

    assert len(exchange.tasks) == EXCHANGE_TASKS
    assert started["publisher"] == {
        "poll_seconds": 7,
        "ack_timeout": timedelta(seconds=11),
        "max_republish": 2,
        "request_subjects": {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"},
    }
    assert started["watchdog"] == {"period_seconds": WATCHDOG_PERIOD_SECONDS}
    await exchange.stop()
    assert all(task.done() for task in exchange.tasks)
    await exchange.stop()  # повторная остановка безопасна