"""Параллельная обработка ограничена (spec image-recognition-gateway, задача 5.4; D-3, `SCAN_MAX_INFLIGHT`).

Сценарий spec «Пять заданий подряд»: пять запросов при `SCAN_MAX_INFLIGHT = 4` → пять ack'ов сразу, четыре файла во
входящем каталоге, пятый — только после завершения одного из четырёх. Плюс строгая очерёдность при одном слоте и
остановка воркера, когда задания ждут слот (T12.9).

Детерминизм без «поспать и посмотреть»: пауза между опросами горячей папки подменена (`SteppedSleep`) — задание,
записавшее файл и опросившее исходящий каталог, «паркуется» и опрашивает снова только по шагу теста. Ожидания —
по событиям (`asyncio.Condition`) с явным тайм-аутом. Ответ ИС кладёт сам тест (`env.answer`) и ровно для того
задания, слот которого освобождается. Чтение S3 у фейка не уступает цикл событий, поэтому задание, прошедшее мимо
семафора, читает S3 в том же проходе цикла, что и его ack, — счётчик чтений S3 различает «ждёт слот» и «слот не нужен»
без гонок.
"""

import asyncio
import json
from collections import Counter
from typing import Any

import pytest

from tests.worker_helpers import SUCCESS_ANSWER, Env, EnvFactory

# Логи воркера перехватываются (`log_lines` из conftest): вывод pytest чистый
pytestmark = pytest.mark.usefixtures("log_lines")

WAIT_TIMEOUT = 3.0


class SteppedSleep:
    """Подмена `sleep` воркера: пауза между опросами длится, пока тест не сделает шаг (`step`) или не откроет её.

    Поколение растёт с каждым шагом: `wait_parked(n)` ждёт n заданий, припаркованных уже после последнего шага, —
    так ожидание не срабатывает на заданиях, которые только что отпущены, но ещё не успели проснуться.
    """

    def __init__(self) -> None:
        self._generation = 0
        self._parked: Counter[int] = Counter()
        self._gate = asyncio.Event()
        self._opened = False
        self._changed = asyncio.Condition()

    @property
    def parked(self) -> int:
        """Сколько заданий припарковано после последнего шага."""
        return self._parked[self._generation]

    async def __call__(self, seconds: float) -> None:
        if self._opened:
            await asyncio.sleep(0)
            return
        generation, gate = self._generation, self._gate
        async with self._changed:
            self._parked[generation] += 1
            self._changed.notify_all()
        try:
            await gate.wait()
        finally:
            self._parked[generation] -= 1

    def step(self) -> None:
        """Отпустить всех припаркованных: каждое задание опросит исходящий каталог ещё раз."""
        self._gate.set()
        self._gate = asyncio.Event()
        self._generation += 1

    def open(self) -> None:
        """Снять паузы совсем: дальше опросы идут без ожидания."""
        self._opened = True
        self._gate.set()

    async def wait_parked(self, count: int) -> None:
        async with self._changed:
            await asyncio.wait_for(self._changed.wait_for(lambda: self.parked >= count), WAIT_TIMEOUT)


def job_of(name: str) -> str:
    """`job_id` из имени файла входящего каталога `<job_id>_<original_name>`."""
    return name.split("_", 1)[0]


def answer(env: Env, job_id: str) -> None:
    """Ответ ИС для задания (синтетический успех R-10 → `NOT_IMPLEMENTED`)."""
    env.answer(job_id, json.dumps({**SUCCESS_ANSWER, "file_uid": job_id}).encode())


def index_of(env: Env, event: tuple[str, str]) -> int:
    assert env.events.count(event) == 1, (event, env.events)
    return env.events.index(event)


def source_key(request: dict[str, Any]) -> str:
    return str(request["s3"]["key"])


async def finish_all(env: Env, poll_sleep: SteppedSleep, jobs: list[str]) -> None:
    """Ответить на все задания, снять паузы, дождаться конца; у каждого задания — ровно один ack и один итог."""
    for job in jobs:
        if not env.local.answer_path(job).exists():
            answer(env, job)
    poll_sleep.open()
    await asyncio.wait_for(env.worker.wait_idle(), WAIT_TIMEOUT)
    for job in jobs:
        assert env.acks(job) == [f"ack:{job}:1"]
        [(msg_id, body)] = env.results(job)
        assert msg_id == f"result:{job}:1"
        assert body["status"] == "error"
        assert body["error"]["code"] == "NOT_IMPLEMENTED"
    assert env.worker.active_jobs == 0


# --- spec «Пять заданий подряд» -----------------------------------------------------------------------------------
async def test_five_jobs_with_four_slots(make_env: EnvFactory) -> None:
    poll_sleep = SteppedSleep()
    env = await make_env(SCAN_MAX_INFLIGHT=4, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep)
    requests = [env.request() for _ in range(5)]
    jobs = [request["job_id"] for request in requests]
    for request in requests:
        await env.deliver(request)

    # Четыре слота заняты: каждое из четырёх заданий записало файл и ждёт ответа ИС
    await poll_sleep.wait_parked(4)
    # пять ack'ов уже опубликованы — ни один слот ещё не освободился, итогов нет
    assert [env.acks(job) for job in jobs] == [[f"ack:{job}:1"] for job in jobs]
    assert all(env.results(job) == [] for job in jobs)
    # во входящем каталоге ровно четыре файла, пятое задание даже не читало S3
    assert len(env.storage.get_calls) == 4
    assert len(env.hotfolder.puts) == 4
    assert env.inbox() == sorted(env.hotfolder.puts)
    busy = [job_of(name) for name in env.hotfolder.puts]
    [waiting] = [job for job in jobs if job not in busy]
    assert poll_sleep.parked == 4
    assert env.worker.active_jobs == 5

    # Ещё один круг опроса занятых слотов: пятое задание по-прежнему ждёт
    poll_sleep.step()
    await poll_sleep.wait_parked(4)
    assert len(env.hotfolder.puts) == 4
    assert len(env.storage.get_calls) == 4
    assert waiting not in {job_of(name) for name in env.inbox()}

    # Ответ ИС для одного задания: его слот освобождается, и только тогда появляется пятый файл
    freed = busy[0]
    answer(env, freed)
    poll_sleep.step()
    await poll_sleep.wait_parked(4)  # три прежних задания + пятое, записавшее файл
    assert [job_of(name) for name in env.hotfolder.puts] == [*busy, waiting]
    assert len(env.inbox()) == 5
    assert all(env.results(job) == [] for job in jobs if job != freed)
    freed_result = index_of(env, ("publish", f"result:{freed}:1"))
    waiting_request = requests[jobs.index(waiting)]
    assert max(index_of(env, ("publish", f"ack:{job}:1")) for job in jobs) < freed_result
    assert freed_result < index_of(env, ("s3_get", source_key(waiting_request)))
    assert freed_result < index_of(env, ("put", f"{waiting}_ok_doc.pdf"))

    # Все пять заданий заканчиваются конвертами
    await finish_all(env, poll_sleep, jobs)
    assert len(env.inbox()) == 5


async def test_single_slot_serialises_jobs(make_env: EnvFactory) -> None:
    poll_sleep = SteppedSleep()
    env = await make_env(SCAN_MAX_INFLIGHT=1, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep)
    requests = [env.request() for _ in range(3)]
    jobs = [request["job_id"] for request in requests]
    for request in requests:
        await env.deliver(request)

    for done in range(3):
        await poll_sleep.wait_parked(1)
        # в работе ровно одно задание; ack'и всех трёх — сразу
        assert len(env.hotfolder.puts) == done + 1
        assert len(env.storage.get_calls) == done + 1
        assert len(env.inbox()) == done + 1
        assert all(env.acks(job) == [f"ack:{job}:1"] for job in jobs)
        current = job_of(env.hotfolder.puts[-1])
        assert env.results(current) == []
        answer(env, current)
        poll_sleep.step()
    await asyncio.wait_for(env.worker.wait_idle(), WAIT_TIMEOUT)

    # Строгая очерёдность в журнале: следующий файл записан только после итога предыдущего задания
    order = [job_of(name) for name in env.hotfolder.puts]
    assert sorted(order) == sorted(jobs)
    files_and_results = [
        (kind, ref) for kind, ref in env.events if kind == "put" or (kind == "publish" and ref.startswith("result:"))
    ]
    expected: list[tuple[str, str]] = []
    for job in order:
        expected += [("put", f"{job}_ok_doc.pdf"), ("publish", f"result:{job}:1")]
    assert files_and_results == expected
    await finish_all(env, poll_sleep, jobs)


# --- остановка при очереди на слот (T12.9) ------------------------------------------------------------------------
async def test_stop_while_jobs_wait_for_slot(make_env: EnvFactory, log_lines: list[str]) -> None:
    poll_sleep = SteppedSleep()
    env = await make_env(SCAN_MAX_INFLIGHT=2, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep)
    tasks_before = asyncio.all_tasks()
    requests = [env.request() for _ in range(5)]
    for request in requests:
        await env.deliver(request)
    await poll_sleep.wait_parked(2)
    assert env.worker.active_jobs == 5
    assert len(env.hotfolder.puts) == 2  # три задания ждут слот
    assert all(env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"] for request in requests)
    job_tasks = asyncio.all_tasks() - tasks_before
    assert len(job_tasks) == 5

    # без исключения; `asyncio.timeout`, а не `wait_for`: отдельная задача-обёртка дала бы отменённым заданиям
    # лишние проходы цикла, и проверка «stop() дождался заданий» стала бы слепой
    async with asyncio.timeout(WAIT_TIMEOUT):
        await env.worker.stop()

    # Все задачи заданий завершены отменой, новых задач не осталось, итоговых конвертов нет
    assert all(task.cancelled() for task in job_tasks)
    assert asyncio.all_tasks() - tasks_before == set()
    assert env.worker.active_jobs == 0
    assert all(env.results(request["job_id"]) == [] for request in requests)
    assert not [line for line in log_lines if line.startswith(("ERROR|", "CRITICAL|"))]
    # После остановки файлы во входящий каталог больше не пишутся и S3 не читается, даже если цикл событий крутится
    for _ in range(20):
        await asyncio.sleep(0)
    assert len(env.hotfolder.puts) == 2
    assert len(env.inbox()) == 2
    assert len(env.storage.get_calls) == 2
    assert poll_sleep.parked == 0

    # Слоты не утекли: после повторного запуска оба слота свободны
    env.worker.start()
    again = [env.request() for _ in range(3)]
    for request in again:
        await env.deliver(request)
    await poll_sleep.wait_parked(2)
    assert len(env.hotfolder.puts) == 4
    assert len(env.storage.get_calls) == 4
    await finish_all(env, poll_sleep, [request["job_id"] for request in again])