"""Регрессии AURORA-1442: новая попытка после сбоя обмена, без повторной загрузки."""

import asyncio
from datetime import timedelta
from http import HTTPStatus

import pytest
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession

from src.api.dependencies.audit import RequestAudit
from src.api.dependencies.directories import get_app_settings
from src.common.exceptions import ConflictError
from src.contract import CONTRACT_VERSION
from src.domain.policy import VisibilityScope
from src.models.audit_ex import AuditEvent
from src.models.consolidation import Consolidation, ConsolidationMember, ConsolidationVersion
from src.models.enums import ConsolidationStatus, FileStatus, RecognitionPipeline, RecognitionStatus
from src.models.files import RecognitionJob, StoredFile
from src.services.audit import AuditService
from src.services.file_retry import FileRetryService
from src.services.recognition_exchange import handle_message, publish_pending
from tests.conftest import AuthClientFactory
from tests.factories import make_document, make_stored_file, make_version
from tests.test_api_consolidation_members import _user, _version
from tests.test_api_files_read import KA_RF, _rostov_entity
from tests.test_recognition_exchange import FakeNats
from tests.test_recognition_inbox import envelope, fake_msg

pytestmark = pytest.mark.integration


@pytest.mark.parametrize("state,code", [(RecognitionStatus.timeout, "TIMEOUT"), (RecognitionStatus.dlq, "NO_ACK")])
async def test_retry_preserves_old_job_and_records_audit(
    seeded: AsyncSession, auth_client: AuthClientFactory, state: RecognitionStatus, code: str
) -> None:
    entity = await _rostov_entity(seeded)
    file, old = await make_stored_file(seeded, entity=entity, status=FileStatus.ERROR, job_status=state)
    old.max_attempts = 3
    old.error_code = code
    old.metrics = {"republish_count": 3}
    source = file.s3_object_id
    await seeded.flush()
    client = await auth_client(KA_RF)

    response = await client.post(f"/files/{file.id}/retry")
    assert response.status_code == HTTPStatus.ACCEPTED, response.text
    await seeded.refresh(file)
    await seeded.refresh(old)
    new = await seeded.get(RecognitionJob, file.current_recognition_job_id)
    assert new is not None and new.id != old.id
    assert (new.status, new.attempt, new.max_attempts) == (RecognitionStatus.queued, 2, 3)
    assert new.error_code is None and new.metrics is None and new.result_s3_object_id is None
    assert file.s3_object_id == source and file.status == FileStatus.UPLOADED
    assert old.status == state and old.error_code == code
    event = (
        await seeded.execute(
            select(AuditEvent).where(AuditEvent.object_id == file.id, AuditEvent.action == "FILE_REPROCESS")
        )
    ).scalar_one()
    assert event.before_value == {"job_id": str(old.id), "attempt": 1, "status": state.value, "error_code": code}
    assert event.after_value == {"job_id": str(new.id), "attempt": 2}
    again = await client.post(f"/files/{file.id}/retry")
    assert again.status_code == HTTPStatus.CONFLICT


@pytest.mark.parametrize("reason", ["limit", "ocr", "service_error", "document"])
async def test_retry_rejects_ineligible_file(seeded: AsyncSession, auth_client: AuthClientFactory, reason: str) -> None:
    entity = await _rostov_entity(seeded)
    file, job = await make_stored_file(
        seeded, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
    )
    job.error_code = "TIMEOUT"
    job.max_attempts = 3
    if reason == "limit":
        job.attempt = 3
    elif reason == "ocr":
        job.pipeline = RecognitionPipeline.image
    elif reason == "service_error":
        job.status = RecognitionStatus.error
    else:
        await make_document(seeded, file=file, job=job)
    await seeded.flush()
    client = await auth_client(KA_RF)
    response = await client.post(f"/files/{file.id}/retry")
    assert response.status_code == HTTPStatus.CONFLICT, response.text
    expected = "MAX_ATTEMPTS_REACHED" if reason == "limit" else "FILE_NOT_RETRYABLE"
    assert response.json()["detail"]["code"] == expected
    await seeded.refresh(file)
    assert file.current_recognition_job_id == job.id


async def test_batch_filters_perimeter_and_skips_ineligible(
    seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:

    version, inside, outside, _ = await _version(seeded, members=1, spare=1, locked_by=None)
    good, job = await make_stored_file(
        seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.dlq
    )
    job.error_code, job.max_attempts = "NO_ACK", 3
    good.consolidation_id = version.consolidation_id
    bad, _ = await make_stored_file(seeded, entity=inside[0])
    bad.consolidation_id = version.consolidation_id
    excluded, _ = await make_stored_file(seeded, entity=outside[0])
    excluded.consolidation_id = version.consolidation_id
    desk, _ = await make_stored_file(seeded, entity=inside[0])
    await seeded.flush()
    client = await auth_client(KA_RF)
    response = await client.post(f"/consolidations/versions/{version.id}/files/retry")
    assert response.status_code == HTTPStatus.ACCEPTED, response.text
    body = response.json()
    assert [row["file_id"] for row in body["accepted"]] == [str(good.id)]
    assert body["skipped"] == []
    event = (
        await seeded.execute(
            select(AuditEvent).where(AuditEvent.object_id == good.id, AuditEvent.action == "FILE_REPROCESS")
        )
    ).scalar_one()
    assert event.after_value["consolidation_id"] == str(version.consolidation_id)
    assert str(desk.id) not in str(body)


async def test_batch_rejects_verified_and_returns_empty_for_empty_perimeter(
    seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:

    version, _, _, _ = await _version(seeded, members=1, locked_by=None)
    client = await auth_client(KA_RF)
    url = f"/consolidations/versions/{version.id}/files/retry"
    response = await client.post(url)
    assert response.status_code == HTTPStatus.ACCEPTED
    assert response.json() == {"accepted": [], "skipped": []}
    version.status = ConsolidationStatus.VERIFIED
    await seeded.flush()
    response = await client.post(url)
    assert response.status_code == HTTPStatus.CONFLICT
    assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE"


async def test_concurrent_retries_create_only_one_attempt(committed_sessionmaker) -> None:

    async with committed_sessionmaker() as session, session.begin():
        entity = await _rostov_entity(session)
        file, job = await make_stored_file(
            session, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
        )
        job.error_code, job.max_attempts = "TIMEOUT", 3
        user = await _user(session, KA_RF)
        file_id, user_id = file.id, user.id

    async def retry():
        async with committed_sessionmaker() as session:
            try:
                async with session.begin():
                    audit = RequestAudit(AuditService(session), user_id, None, None)
                    return await FileRetryService(session, audit, VisibilityScope(None)).retry(file_id)
            except ConflictError as error:
                return error.code

    results = await asyncio.wait_for(asyncio.gather(retry(), retry()), timeout=10)
    assert sum(isinstance(result, str) for result in results) == 1
    assert "FILE_NOT_RETRYABLE" in results
    async with committed_sessionmaker() as session:
        jobs = list((await session.scalars(select(RecognitionJob).where(RecognitionJob.file_id == file_id))).all())
        assert sorted(job.attempt for job in jobs) == [1, 2]


async def test_retry_outbox_and_late_old_result(committed_sessionmaker) -> None:

    async with committed_sessionmaker() as session, session.begin():
        entity = await _rostov_entity(session)
        file, old = await make_stored_file(
            session, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
        )
        old.error_code, old.max_attempts = "TIMEOUT", 3
        user = await _user(session, KA_RF)
        file_id, old_id = file.id, old.id
        audit = RequestAudit(AuditService(session), user.id, None, None)
        card = await FileRetryService(session, audit, VisibilityScope(None)).retry(file.id)
        assert card.job is not None
        new_id = card.job.id

    for state in ("processing", "done"):
        await handle_message(fake_msg(envelope(old_id, file_id, state), None), session_factory=committed_sessionmaker)
    offline = FakeNats(connected=False)
    await publish_pending(committed_sessionmaker, offline, ack_timeout=timedelta(seconds=30), max_republish=3)
    async with committed_sessionmaker() as session:
        new = await session.get(RecognitionJob, new_id)
        assert new is not None and new.status is RecognitionStatus.queued
        assert new.reply_deadline is None and new.result_s3_object_id is None
    online = FakeNats()
    await publish_pending(committed_sessionmaker, online, ack_timeout=timedelta(seconds=30), max_republish=3)
    async with committed_sessionmaker() as session:
        new = await session.get(RecognitionJob, new_id)
        assert new is not None and new.status is RecognitionStatus.published
        assert new.reply_deadline is not None and new.request_msg_id == f"{new_id}:2"
        old = await session.get(RecognitionJob, old_id)
        assert old is not None and old.status is RecognitionStatus.timeout


async def test_retry_rolls_back_job_pointer_and_audit(committed_sessionmaker) -> None:
    async with committed_sessionmaker() as session, session.begin():
        entity = await _rostov_entity(session)
        file, job = await make_stored_file(
            session, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
        )
        job.error_code, job.max_attempts = "TIMEOUT", 3
        user = await _user(session, KA_RF)
        file_id, job_id, user_id = file.id, job.id, user.id
    with pytest.raises(RuntimeError, match="rollback"):
        async with committed_sessionmaker() as session, session.begin():
            audit = RequestAudit(AuditService(session), user_id, None, None)
            await FileRetryService(session, audit, VisibilityScope(None)).retry(file_id)
            raise RuntimeError("rollback")
    async with committed_sessionmaker() as session:
        jobs = list((await session.scalars(select(RecognitionJob).where(RecognitionJob.file_id == file_id))).all())
        assert [job.id for job in jobs] == [job_id]
        events = list((await session.scalars(select(AuditEvent).where(AuditEvent.object_id == file_id))).all())
        assert events == []


async def test_concurrent_batch_retry(committed_sessionmaker) -> None:
    async with committed_sessionmaker() as session, session.begin():
        version, members, _, _ = await _version(session, members=1, locked_by=None)
        file, job = await make_stored_file(
            session, entity=members[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
        )
        file.consolidation_id = version.consolidation_id
        job.error_code, job.max_attempts = "TIMEOUT", 3
        user = await _user(session, KA_RF)
        version_id, user_id = version.id, user.id

    async def retry():
        async with committed_sessionmaker() as session, session.begin():
            audit = RequestAudit(AuditService(session), user_id, None, None)
            return await FileRetryService(session, audit, VisibilityScope(None)).retry_version(version_id)

    results = await asyncio.wait_for(asyncio.gather(retry(), retry()), timeout=10)
    assert sorted(len(result.accepted) for result in results) == [0, 1]
    assert all(result.skipped == [] for result in results)


@pytest.mark.parametrize("change", ["status", "perimeter"])
async def test_batch_rechecks_version_after_waiting_for_job(
    committed_sessionmaker, monkeypatch: pytest.MonkeyPatch, change: str
) -> None:

    async with committed_sessionmaker() as session, session.begin():
        version, members, _, _ = await _version(session, members=1, locked_by=None)
        file, job = await make_stored_file(
            session, entity=members[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
        )
        file.consolidation_id = version.consolidation_id
        job.error_code, job.max_attempts = "TIMEOUT", 3
        user = await _user(session, KA_RF)
        version_id, job_id, user_id = version.id, job.id, user.id
    reached = asyncio.Event()
    original = FileRetryService._lock_files

    async def signal(self, files):
        reached.set()
        return await original(self, files)

    monkeypatch.setattr(FileRetryService, "_lock_files", signal)

    async def retry():
        async with committed_sessionmaker() as session:
            try:
                async with session.begin():
                    audit = RequestAudit(AuditService(session), user_id, None, None)
                    return await FileRetryService(session, audit, VisibilityScope(None)).retry_version(version_id)
            except ConflictError as error:
                return error.code

    async with committed_sessionmaker() as holder, holder.begin():
        await holder.execute(select(RecognitionJob).where(RecognitionJob.id == job_id).with_for_update())
        pending = asyncio.create_task(retry())
        await asyncio.wait_for(reached.wait(), timeout=10)
        version = await holder.get(ConsolidationVersion, version_id)
        assert version is not None
        if change == "status":
            version.status = ConsolidationStatus.VERIFIED
        else:
            await holder.execute(
                delete(ConsolidationMember).where(ConsolidationMember.consolidation_version_id == version_id)
            )
    result = await asyncio.wait_for(pending, timeout=10)
    if change == "status":
        assert result == "VERSION_NOT_EDITABLE"
    else:
        assert result.accepted == [] and result.skipped == []


async def test_batch_mixed_package_skips_ocr_and_exhausted_files(
    seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:
    """Смешанный пакет: подходящий файл принят, OCR в ERROR и исчерпанный лимит попыток — в skipped с кодами."""
    version, inside, _, _ = await _version(seeded, members=1, locked_by=None)
    good, good_job = await make_stored_file(
        seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
    )
    good_job.error_code, good_job.max_attempts = "TIMEOUT", 3
    ocr, ocr_job = await make_stored_file(
        seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
    )
    ocr_job.pipeline, ocr_job.error_code, ocr_job.max_attempts = RecognitionPipeline.image, "TIMEOUT", 3
    spent, spent_job = await make_stored_file(
        seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
    )
    spent_job.error_code, spent_job.max_attempts, spent_job.attempt = "TIMEOUT", 3, 3
    for item in (good, ocr, spent):
        item.consolidation_id = version.consolidation_id
    await seeded.flush()
    client = await auth_client(KA_RF)
    response = await client.post(f"/consolidations/versions/{version.id}/files/retry")
    assert response.status_code == HTTPStatus.ACCEPTED, response.text
    body = response.json()
    assert [row["file_id"] for row in body["accepted"]] == [str(good.id)]
    skipped = {row["file_id"]: row["code"] for row in body["skipped"]}
    assert skipped == {str(ocr.id): "FILE_NOT_RETRYABLE", str(spent.id): "MAX_ATTEMPTS_REACHED"}
    for skipped_file, skipped_job in ((ocr, ocr_job), (spent, spent_job)):
        await seeded.refresh(skipped_file)
        assert skipped_file.current_recognition_job_id == skipped_job.id and skipped_file.status == FileStatus.ERROR
        events = await seeded.scalars(
            select(AuditEvent).where(AuditEvent.object_id == skipped_file.id, AuditEvent.action == "FILE_REPROCESS")
        )
        assert list(events.all()) == []


async def test_batch_no_ack_at_attempt_limit_is_max_attempts_reached(
    seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:
    """NO_ACK (dlq) на пределе попыток в пакете даёт MAX_ATTEMPTS_REACHED, новое задание не создаётся."""
    version, inside, _, _ = await _version(seeded, members=1, locked_by=None)
    file, job = await make_stored_file(
        seeded, entity=inside[0], status=FileStatus.ERROR, job_status=RecognitionStatus.dlq
    )
    job.error_code, job.max_attempts, job.attempt = "NO_ACK", 3, 3
    file.consolidation_id = version.consolidation_id
    await seeded.flush()
    client = await auth_client(KA_RF)
    response = await client.post(f"/consolidations/versions/{version.id}/files/retry")
    assert response.status_code == HTTPStatus.ACCEPTED, response.text
    assert response.json() == {"accepted": [], "skipped": [{"file_id": str(file.id), "code": "MAX_ATTEMPTS_REACHED"}]}
    jobs = await seeded.scalars(select(RecognitionJob).where(RecognitionJob.file_id == file.id))
    assert [item.id for item in jobs.all()] == [job.id]


async def test_retry_uses_current_contract_and_subject(
    seeded: AsyncSession, auth_client: AuthClientFactory, monkeypatch: pytest.MonkeyPatch
) -> None:
    entity = await _rostov_entity(seeded)
    file, old = await make_stored_file(
        seeded, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
    )
    old.error_code, old.max_attempts = "TIMEOUT", 3
    old.schema_version, old.request_subject = "0.9", "old.tabular.subject"
    monkeypatch.setattr(get_app_settings().OPENBAO.NATS, "NATS_TABULAR_REQUEST_SUBJECT", "current.tabular.subject")
    await seeded.flush()
    client = await auth_client(KA_RF)
    response = await client.post(f"/files/{file.id}/retry")
    assert response.status_code == HTTPStatus.ACCEPTED, response.text
    await seeded.refresh(file)
    new = await seeded.get(RecognitionJob, file.current_recognition_job_id)
    assert new is not None
    assert new.schema_version == CONTRACT_VERSION and new.request_subject == "current.tabular.subject"
    await seeded.refresh(old)
    assert old.schema_version == "0.9" and old.request_subject == "old.tabular.subject"


@pytest.mark.parametrize(
    "normal_status", [FileStatus.UPLOADED, FileStatus.PROCESSING, FileStatus.PROCESSED, FileStatus.SUPERSEDED]
)
async def test_batch_does_not_lock_normal_files(committed_sessionmaker, normal_status: FileStatus) -> None:
    async with committed_sessionmaker() as session, session.begin():
        version, members, _, _ = await _version(session, members=1, locked_by=None)
        normal, job = await make_stored_file(
            session, entity=members[0], status=normal_status, job_status=RecognitionStatus.processing
        )
        normal.consolidation_id = version.consolidation_id
        user = await _user(session, KA_RF)
        version_id, file_id, job_id, user_id = version.id, normal.id, job.id, user.id

    async def retry():
        async with committed_sessionmaker() as session, session.begin():
            audit = RequestAudit(AuditService(session), user_id, None, None)
            return await FileRetryService(session, audit, VisibilityScope(None)).retry_version(version_id)

    async with committed_sessionmaker() as holder, holder.begin():
        await holder.execute(select(RecognitionJob).where(RecognitionJob.id == job_id).with_for_update())
        await holder.execute(select(StoredFile).where(StoredFile.id == file_id).with_for_update())
        result = await asyncio.wait_for(retry(), timeout=5)
        assert result.accepted == [] and result.skipped == []


@pytest.mark.parametrize("closed_status", [ConsolidationStatus.VERIFIED, ConsolidationStatus.CANCELLED])
async def test_single_retry_does_not_lock_closed_versions(committed_sessionmaker, closed_status) -> None:
    async with committed_sessionmaker() as session, session.begin():
        working, members, _, _ = await _version(session, members=1, locked_by=None)
        consolidation = await session.get(Consolidation, working.consolidation_id)
        user = await _user(session, KA_RF)
        closed = await make_version(
            session, consolidation=consolidation, version_no=2, members=members, initiator=user, status=closed_status
        )
        file, job = await make_stored_file(
            session, entity=members[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
        )
        file.consolidation_id = working.consolidation_id
        job.error_code, job.max_attempts = "TIMEOUT", 3
        file_id, user_id, closed_id = file.id, user.id, closed.id

    async def retry():
        async with committed_sessionmaker() as session, session.begin():
            audit = RequestAudit(AuditService(session), user_id, None, None)
            return await FileRetryService(session, audit, VisibilityScope(None)).retry(file_id)

    async with committed_sessionmaker() as holder, holder.begin():
        await holder.execute(select(ConsolidationVersion).where(ConsolidationVersion.id == closed_id).with_for_update())
        card = await asyncio.wait_for(retry(), timeout=5)
        assert card.status == FileStatus.UPLOADED