"""Удаление файла `DELETE /files/{file_id}` (spec primary-files «Удаление файла»; audit-trail `FILE_DELETE`;
design D-10).

Основные сценарии — поверх фикстуры `app` (транзакция теста). Гонки с подтверждением и циклом маппинга — на
`committed_*` (временная БД, сессия запроса как в проде).
"""

import asyncio
import datetime
import json
import uuid
from pathlib import Path
from typing import Any

import pytest
from sqlalchemy import func, select, text, update
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from src.core.config import get_settings
from src.models.access import AppUser
from src.models.audit_ex import AuditEvent, Comment
from src.models.directories import LegalEntity, RegionalOffice
from src.models.enums import AuditAction, AuditSection, DocumentStatus, FileStatus, RecognitionStatus
from src.models.files import Document, RecognitionJob, S3Object, StoredFile
from src.models.osv import OsvEntry
from src.models.statements import (
    FinancialStatementColumn,
    FinancialStatementLine,
    FinancialStatementMeta,
    FinancialStatementValue,
)
from src.services.result_mapping import process_once
from tests.conftest import AuthClientFactory, FakeStorage
from tests.factories import (
    make_company_group,
    make_document,
    make_entries_from_result,
    make_legal_entity,
    make_result_object,
    make_stored_file,
)
from tests.fixtures_helpers import GOLDEN_STATEMENTS, map_golden_snapshot

HTTP_NO_CONTENT = 204
HTTP_FORBIDDEN = 403
HTTP_NOT_FOUND = 404
HTTP_CONFLICT = 409

KA_RF = "ka_rf@armka.local"  # Ростовский филиал, DOCUMENT_VIEW + DOCUMENT_EDIT
RKA = "rka@armka.local"  # Ростовский филиал, только DOCUMENT_VIEW
CLIENT_MANAGER = "client_manager@armka.local"  # Краснодарский филиал
ROSTOV = "Ростовский региональный филиал"
MINIMAL_RESULT = Path(__file__).resolve().parents[1] / "contracts" / "examples" / "result-osv-minimal.json"
LOCK_WAIT_SECONDS = 10.0

pytestmark = pytest.mark.integration


async def _rostov_entity(session: AsyncSession, name: str = "ООО Удаление") -> LegalEntity:
    office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
    group = await make_company_group(session, name=f"ГК {name}", office=office)
    return await make_legal_entity(session, name=name, group=group)


def _minimal() -> dict[str, Any]:
    return json.loads(MINIMAL_RESULT.read_text("utf-8"))


async def _count(session: AsyncSession, model: type, *where: Any) -> int:
    return (await session.execute(select(func.count()).select_from(model).where(*where))).scalar_one()


async def _delete_events(session: AsyncSession) -> list[AuditEvent]:
    rows = await session.execute(select(AuditEvent).where(AuditEvent.action == AuditAction.FILE_DELETE))
    return list(rows.scalars())


# --- успешное удаление ----------------------------------------------------------------------------------------------


async def test_delete_error_file(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    """«Удаление ошибочного файла»: 204, записей нет, `expires_at` у объектов, аудит с карточкой в `before_value`."""
    entity = await _rostov_entity(seeded)
    file, job = await make_stored_file(
        seeded, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.error
    )
    job.error_code = "PARSE_ERROR"
    result_object = await make_result_object(seeded, job=job, storage=FakeStorage(), result={})
    file_id, job_id, s3_id = file.id, job.id, file.s3_object_id
    client = await auth_client(KA_RF)

    response = await client.delete(f"/files/{file_id}")

    assert response.status_code == HTTP_NO_CONTENT, response.text
    assert response.content == b""
    assert await _count(seeded, StoredFile, StoredFile.id == file_id) == 0
    assert await _count(seeded, RecognitionJob, RecognitionJob.file_id == file_id) == 0
    for object_id in (s3_id, result_object.id):  # строки s3_object остаются, им ставится expires_at
        s3 = await seeded.get(S3Object, object_id, populate_existing=True)
        assert s3 is not None
        assert s3.expires_at is not None
    [event] = await _delete_events(seeded)
    assert event.section is AuditSection.PRIMARY_DOCUMENTS
    assert (event.object_type, event.object_id) == ("stored_file", file_id)
    assert event.user_id is not None
    assert event.before_value is not None
    assert event.before_value["id"] == str(file_id)
    assert event.before_value["status"] == "ERROR"
    assert event.before_value["job"]["id"] == str(job_id)
    assert event.before_value["job"]["error_code"] == "PARSE_ERROR"
    assert event.after_value is None


MEZH_BALANCE = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3" / "1_Форма1_бух_09.json"


async def _form_file_with_comment(session: AsyncSession, fixtures_dir: Path) -> uuid.UUID:
    """Файл с формой ББ (golden) и комментарием к её документу; возвращает id файла."""
    entity = await _rostov_entity(session)
    (document,) = await map_golden_snapshot(session, entity=entity, path=fixtures_dir / MEZH_BALANCE, uploaded_by=None)
    author = (await session.execute(select(AppUser.id).where(AppUser.email == KA_RF))).scalar_one()
    session.add(
        Comment(
            object_type="document",
            object_id=document.id,
            text="проверить",
            author_id=author,
            created_at=datetime.datetime.now(datetime.UTC),
        )
    )
    await session.flush()
    return document.stored_file_id


async def _assert_form_rows_gone(session: AsyncSession, file_id: uuid.UUID, document_ids: list[uuid.UUID]) -> None:
    assert await _count(session, StoredFile, StoredFile.id == file_id) == 0
    assert await _count(session, Document, Document.stored_file_id == file_id) == 0
    for model in (FinancialStatementMeta, FinancialStatementColumn, FinancialStatementLine, FinancialStatementValue):
        assert await _count(session, model) == 0, model.__tablename__
    assert await _count(session, Comment, Comment.object_id.in_(document_ids)) == 0


@pytest.mark.fixtures
@pytest.mark.parametrize("superseded", [False, True])
async def test_delete_file_with_statement_form(
    seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory, superseded: bool
) -> None:
    """C-1, T8.2: удаление файла с формой ББ/ОФР убирает шапку, колонки, строки, значения и комментарии документов
    (раньше — падение по FK `financial_statement_meta`); то же для файла `SUPERSEDED`."""
    file_id = await _form_file_with_comment(seeded, fixtures_dir)
    document_ids = list((await seeded.execute(select(Document.id).where(Document.stored_file_id == file_id))).scalars())
    assert await _count(seeded, FinancialStatementValue) > 0 and await _count(seeded, Comment) == 1
    if superseded:
        await seeded.execute(update(StoredFile).where(StoredFile.id == file_id).values(status=FileStatus.SUPERSEDED))
        await seeded.flush()
    client = await auth_client(KA_RF)

    response = await client.delete(f"/files/{file_id}")

    assert response.status_code == HTTP_NO_CONTENT, response.text
    await _assert_form_rows_gone(seeded, file_id, document_ids)


async def test_delete_expires_only_own_objects(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    """M-3: `expires_at` ставится только объектам папок файла (`uploads/{file_id}/`, `results/{file_id}/`); ссылка
    задания на объект другого файла (inbox переиспользует `s3_object` при конфликте `(bucket, key)`) его не трогает."""
    entity = await _rostov_entity(seeded)
    file, job = await make_stored_file(
        seeded, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.error
    )
    other_file, _ = await make_stored_file(seeded, entity=entity, status=FileStatus.ERROR, original_name="other.xls")
    job.result_s3_object_id = other_file.s3_object_id  # результат «указывает» на загрузку другого файла
    await seeded.flush()
    file_id, own_id, foreign_id = file.id, file.s3_object_id, other_file.s3_object_id
    client = await auth_client(KA_RF)

    response = await client.delete(f"/files/{file_id}")

    assert response.status_code == HTTP_NO_CONTENT, response.text
    own = await seeded.get(S3Object, own_id, populate_existing=True)
    foreign = await seeded.get(S3Object, foreign_id, populate_existing=True)
    assert own is not None and own.expires_at is not None
    assert foreign is not None and foreign.expires_at is None


async def test_delete_processed_file_removes_rows(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    """Файл `PROCESSED` без подтверждения: удаляются строки ОСВ, документы, задания и файл."""
    entity = await _rostov_entity(seeded)
    file, job = await make_stored_file(
        seeded, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done
    )
    documents = await make_entries_from_result(seeded, file, job, _minimal())
    assert documents
    entry_count = await _count(seeded, OsvEntry, OsvEntry.recognition_job_id == job.id)
    assert entry_count > 0
    file_id, job_id = file.id, job.id
    client = await auth_client(KA_RF)

    response = await client.delete(f"/files/{file_id}")

    assert response.status_code == HTTP_NO_CONTENT, response.text
    assert await _count(seeded, OsvEntry, OsvEntry.recognition_job_id == job_id) == 0
    assert await _count(seeded, Document, Document.stored_file_id == file_id) == 0
    assert await _count(seeded, RecognitionJob, RecognitionJob.id == job_id) == 0
    assert await _count(seeded, StoredFile, StoredFile.id == file_id) == 0
    [event] = await _delete_events(seeded)
    assert event.before_value is not None
    assert [item["id"] for item in event.before_value["documents"]] == [str(d.id) for d in documents]


@pytest.mark.parametrize("status", [FileStatus.UPLOADED, FileStatus.PROCESSING, FileStatus.SUPERSEDED])
async def test_delete_file_in_other_statuses(
    seeded: AsyncSession, auth_client: AuthClientFactory, status: FileStatus
) -> None:
    """Файл в `UPLOADED`/`PROCESSING` и `SUPERSEDED` (ни на что не влияет) удаляется."""
    entity = await _rostov_entity(seeded)
    file, _ = await make_stored_file(seeded, entity=entity, status=status)
    file_id = file.id
    client = await auth_client(KA_RF)

    response = await client.delete(f"/files/{file_id}")

    assert response.status_code == HTTP_NO_CONTENT, response.text
    assert await _count(seeded, StoredFile, StoredFile.id == file_id) == 0


async def test_delete_superseded_file_keeps_new_one(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    entity = await _rostov_entity(seeded)
    new, _ = await make_stored_file(seeded, entity=entity, status=FileStatus.PROCESSED)
    old, _ = await make_stored_file(seeded, entity=entity, status=FileStatus.SUPERSEDED)
    old.superseded_by_file_id = new.id
    await seeded.flush()
    old_id, new_id = old.id, new.id
    client = await auth_client(KA_RF)

    response = await client.delete(f"/files/{old_id}")

    assert response.status_code == HTTP_NO_CONTENT, response.text
    assert await _count(seeded, StoredFile, StoredFile.id == new_id) == 1


# --- отказы ---------------------------------------------------------------------------------------------------------


async def test_confirmed_document_blocks_delete(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    entity = await _rostov_entity(seeded)
    file, job = await make_stored_file(
        seeded, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done
    )
    await make_document(seeded, file=file, job=job, account="60", status=DocumentStatus.NEEDS_CHECK)
    await make_document(seeded, file=file, job=job, account="62", status=DocumentStatus.CONFIRMED)
    file_id, job_id = file.id, job.id
    client = await auth_client(KA_RF)

    response = await client.delete(f"/files/{file_id}")

    assert response.status_code == HTTP_CONFLICT, response.text
    assert response.json()["detail"]["code"] == "FILE_HAS_CONFIRMED_DOCUMENTS"
    assert await _count(seeded, StoredFile, StoredFile.id == file_id) == 1
    assert await _count(seeded, RecognitionJob, RecognitionJob.id == job_id) == 1
    assert await _count(seeded, Document, Document.stored_file_id == file_id) == 2  # noqa: PLR2004
    assert await _delete_events(seeded) == []
    s3 = await seeded.get(S3Object, file.s3_object_id, populate_existing=True)
    assert s3 is not None
    assert s3.expires_at is None


async def test_file_that_superseded_others_is_not_deletable(
    seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:
    entity = await _rostov_entity(seeded)
    new, _ = await make_stored_file(seeded, entity=entity, status=FileStatus.PROCESSED)
    old, _ = await make_stored_file(seeded, entity=entity, status=FileStatus.SUPERSEDED)
    old.superseded_by_file_id = new.id
    await seeded.flush()
    new_id = new.id
    client = await auth_client(KA_RF)

    response = await client.delete(f"/files/{new_id}")

    assert response.status_code == HTTP_CONFLICT, response.text
    assert response.json()["detail"]["code"] == "FILE_SUPERSEDES_OTHERS"
    assert await _count(seeded, StoredFile, StoredFile.id == new_id) == 1
    assert await _delete_events(seeded) == []


async def test_foreign_and_missing_file_are_same_404(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    entity = await _rostov_entity(seeded)
    file, _ = await make_stored_file(seeded, entity=entity, status=FileStatus.ERROR)
    file_id = file.id
    client = await auth_client(CLIENT_MANAGER)  # Краснодарский филиал: ЮЛ Ростова не видит, DOCUMENT_EDIT есть

    foreign = await client.delete(f"/files/{file_id}")
    missing = await client.delete(f"/files/{uuid.uuid4()}")

    assert foreign.status_code == missing.status_code == HTTP_NOT_FOUND
    assert foreign.json() == missing.json()
    assert await _count(seeded, StoredFile, StoredFile.id == file_id) == 1
    assert await _delete_events(seeded) == []


async def test_delete_requires_document_edit(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    entity = await _rostov_entity(seeded)
    file, _ = await make_stored_file(seeded, entity=entity, status=FileStatus.ERROR)
    file_id = file.id
    client = await auth_client(RKA)  # свой филиал, только DOCUMENT_VIEW

    assert (await client.get(f"/files/{file_id}")).status_code == 200  # noqa: PLR2004
    response = await client.delete(f"/files/{file_id}")

    assert response.status_code == HTTP_FORBIDDEN
    assert await _count(seeded, StoredFile, StoredFile.id == file_id) == 1


async def test_repeated_delete_and_read_after_delete_are_404(
    seeded: AsyncSession, auth_client: AuthClientFactory
) -> None:
    entity = await _rostov_entity(seeded)
    file, _ = await make_stored_file(seeded, entity=entity, status=FileStatus.ERROR)
    file_id = file.id
    client = await auth_client(KA_RF)

    assert (await client.delete(f"/files/{file_id}")).status_code == HTTP_NO_CONTENT
    again = await client.delete(f"/files/{file_id}")
    read = await client.get(f"/files/{file_id}")

    assert again.status_code == read.status_code == HTTP_NOT_FOUND
    assert len(await _delete_events(seeded)) == 1


# --- гонки (временная БД) -------------------------------------------------------------------------------------------


async def _committed_file(
    maker: async_sessionmaker[AsyncSession], storage: FakeStorage | None = None
) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]:
    """Файл `PROCESSING` ЮЛ Ростовского филиала с документом `NEEDS_CHECK` (и результатом в `storage`, если задан);
    `(file_id, job_id, document_id)`."""
    async with maker() as session, session.begin():
        entity = await _rostov_entity(session, "ООО Гонка удаления")
        file, job = await make_stored_file(
            session, entity=entity, status=FileStatus.PROCESSING, job_status=RecognitionStatus.done
        )
        document = await make_document(session, file=file, job=job, status=DocumentStatus.NEEDS_CHECK)
        if storage is not None:
            await make_result_object(
                session,
                job=job,
                storage=storage,
                result=_minimal() | {"job_id": str(job.id), "file_id": str(file.id)},
            )
        return file.id, job.id, document.id


async def _wait_for_lock_waiter(maker: async_sessionmaker[AsyncSession]) -> None:
    deadline = asyncio.get_running_loop().time() + LOCK_WAIT_SECONDS
    while asyncio.get_running_loop().time() < deadline:
        async with maker() as session:
            waiting = (
                await session.execute(
                    text(
                        "SELECT count(*) FROM pg_stat_activity "
                        "WHERE datname = current_database() AND wait_event_type = 'Lock'"
                    )
                )
            ).scalar_one()
        if waiting:
            return
        await asyncio.sleep(0.05)
    raise AssertionError("удаление не встало в ожидание блокировки файла")


async def test_delete_rereads_state_after_locks(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """Подтверждение держит файл под `FOR UPDATE` и фиксирует документ `CONFIRMED`; удаление, начатое в это время,
    ждёт блокировку, перечитывает документы и отвечает 409 — ничего не удалено."""
    file_id, job_id, document_id = await _committed_file(committed_sessionmaker)
    client = await committed_client(KA_RF)

    async with committed_sessionmaker() as confirmer, confirmer.begin():
        await confirmer.execute(select(StoredFile.id).where(StoredFile.id == file_id).with_for_update())
        deletion = asyncio.create_task(client.delete(f"/files/{file_id}"))
        await _wait_for_lock_waiter(committed_sessionmaker)
        assert not deletion.done()
        await confirmer.execute(
            update(Document).where(Document.id == document_id).values(status=DocumentStatus.CONFIRMED)
        )
    response = await asyncio.wait_for(deletion, LOCK_WAIT_SECONDS)

    assert response.status_code == HTTP_CONFLICT, response.text
    assert response.json()["detail"]["code"] == "FILE_HAS_CONFIRMED_DOCUMENTS"
    async with committed_sessionmaker() as session:
        assert await _count(session, StoredFile, StoredFile.id == file_id) == 1
        assert await _count(session, RecognitionJob, RecognitionJob.id == job_id) == 1
        assert await _delete_events(session) == []


async def test_delete_of_file_removed_while_waiting_is_404(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """Файл удалён параллельной транзакцией, пока удаление ждало блокировку: 404, не 500."""
    file_id, job_id, _ = await _committed_file(committed_sessionmaker)
    client = await committed_client(KA_RF)

    async with committed_sessionmaker() as other, other.begin():
        await other.execute(select(RecognitionJob.id).where(RecognitionJob.id == job_id).with_for_update())
        deletion = asyncio.create_task(client.delete(f"/files/{file_id}"))
        await _wait_for_lock_waiter(committed_sessionmaker)
        file = await other.get(StoredFile, file_id)
        assert file is not None
        file.current_recognition_job_id = None
        await other.flush()
        await other.execute(text("DELETE FROM document WHERE stored_file_id = :id"), {"id": file_id})
        await other.execute(text("DELETE FROM recognition_job WHERE file_id = :id"), {"id": file_id})
        await other.execute(text("DELETE FROM stored_file WHERE id = :id"), {"id": file_id})
    response = await asyncio.wait_for(deletion, LOCK_WAIT_SECONDS)

    assert response.status_code == HTTP_NOT_FOUND, response.text
    async with committed_sessionmaker() as session:
        assert await _delete_events(session) == []


async def test_mapping_loop_ignores_deleted_file(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """Файл `PROCESSING` с готовым результатом удалён до цикла маппинга: `process_once` не падает, кандидатов нет,
    документы и строки для файла не создаются; объект результата получил `expires_at`."""
    storage = FakeStorage()
    file_id, job_id, _ = await _committed_file(committed_sessionmaker, storage)
    client = await committed_client(KA_RF)

    response = await client.delete(f"/files/{file_id}")

    assert response.status_code == HTTP_NO_CONTENT, response.text
    assert await process_once(committed_sessionmaker, storage, get_settings()) is False
    async with committed_sessionmaker() as session:
        assert await _count(session, StoredFile, StoredFile.id == file_id) == 0
        assert await _count(session, Document, Document.stored_file_id == file_id) == 0
        assert await _count(session, OsvEntry, OsvEntry.recognition_job_id == job_id) == 0
        result_key = f"results/{file_id}/{job_id}/result.json"
        expires = (await session.execute(select(S3Object.expires_at).where(S3Object.key == result_key))).scalar_one()
        assert expires is not None