import asyncio
import os
import shutil
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, datetime
from pathlib import Path
from uuid import UUID

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

from backend.config import get_settings
from backend.db import get_engine
from backend.models import (
    Attachment,
    PurgeOperation,
    RecurrenceException,
    RecurrenceTemplate,
    Task,
    TaskList,
    User,
)


def boot(client):
    response = client.post(
        "/api/v1/setup/initialize",
        json={"username": "owner", "password": "correct horse battery staple"},
    )
    assert response.status_code == 201


def create_list(client, name="待永久删除"):
    response = client.post("/api/v1/lists", json={"name": name})
    assert response.status_code == 201
    return response.json()


def attachment_storage_name(attachment_id: str) -> str:
    async def load():
        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
        async with session_factory() as db:
            return (await db.get(Attachment, UUID(attachment_id))).storage_name

    return asyncio.run(load())


def test_purge_requires_archive_protects_inbox_and_hides_other_users_lists(client):
    boot(client)
    inbox = client.get("/api/v1/lists").json()[0]
    active = create_list(client, "活动清单")

    assert client.delete(f"/api/v1/lists/{inbox['id']}/purge").status_code == 409
    active_response = client.delete(f"/api/v1/lists/{active['id']}/purge")
    assert active_response.status_code == 409
    assert active_response.json()["detail"] == "请先归档再永久删除"

    async def seed_foreign_archived_list():
        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
        async with session_factory() as db:
            foreign = User(username="other", password_hash="unused")
            db.add(foreign)
            await db.flush()
            foreign_list = TaskList(
                user_id=foreign.id,
                name="别人的归档清单",
                deleted_at=datetime.now(UTC),
            )
            db.add(foreign_list)
            await db.commit()
            return foreign_list.id

    foreign_list_id = asyncio.run(seed_foreign_archived_list())
    assert client.delete(f"/api/v1/lists/{foreign_list_id}/purge").status_code == 404
    assert asyncio.run(row_counts(foreign_list_id))["lists"] == 1


def test_repeated_purge_is_not_found_and_does_not_touch_other_lists(client):
    boot(client)
    target = create_list(client, "目标")
    survivor = create_list(client, "保留")
    survivor_task = client.post(
        "/api/v1/tasks", json={"title": "必须保留", "list_id": survivor["id"]}
    ).json()
    assert client.delete(f"/api/v1/lists/{target['id']}").status_code == 204
    assert client.delete(f"/api/v1/lists/{target['id']}/purge").status_code == 204

    assert client.delete(f"/api/v1/lists/{target['id']}/purge").status_code == 404
    assert client.get(f"/api/v1/tasks/{survivor_task['id']}").status_code == 200


def test_purge_uses_unique_operation_directory_and_preserves_legacy_collision(
    client, tmp_path: Path
):
    boot(client)
    attachment_root = tmp_path / "attachments"
    get_settings().attachment_dir = str(attachment_root)
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
    ).json()
    upload = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.txt", b"proof", "text/plain")},
    )
    assert upload.status_code == 201
    legacy_collision = attachment_root / ".purge-trash" / task_list["id"]
    legacy_collision.mkdir(parents=True)
    sentinel = legacy_collision / "owned-by-another-request"
    sentinel.write_text("keep", encoding="utf-8")
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204

    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")

    assert response.status_code == 204
    assert sentinel.read_text(encoding="utf-8") == "keep"
    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 0


def test_operation_directory_collision_does_not_delete_unowned_directory(
    client, tmp_path: Path, monkeypatch
):
    import backend.main as main_module

    boot(client)
    attachment_root = tmp_path / "attachments"
    get_settings().attachment_dir = str(attachment_root)
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
    ).json()
    client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.txt", b"proof", "text/plain")},
    )
    operation_id = UUID("aaaaaaaa-aaaa-4aaa-aaaa-aaaaaaaaaaaa")
    collision = attachment_root / ".purge-trash" / str(operation_id)
    collision.mkdir(parents=True)
    sentinel = collision / "owned-by-another-request"
    sentinel.write_text("keep", encoding="utf-8")
    operation_ids = iter(
        [operation_id, UUID("bbbbbbbb-bbbb-4bbb-bbbb-bbbbbbbbbbbb")]
    )
    monkeypatch.setattr(main_module, "uuid4", lambda: next(operation_ids))
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204

    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")

    assert response.status_code == 204
    assert sentinel.read_text(encoding="utf-8") == "keep"
    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 0


def test_concurrent_quarantine_allocation_uses_distinct_owned_directories(tmp_path: Path):
    from backend.main import _create_unique_quarantine

    attachment_root = tmp_path / "attachments"
    with ThreadPoolExecutor(max_workers=8) as pool:
        allocations = list(pool.map(lambda _: _create_unique_quarantine(attachment_root), range(20)))

    operation_ids = [operation_id for operation_id, _ in allocations]
    trash_dirs = [trash_dir for _, trash_dir in allocations]
    assert len(set(operation_ids)) == 20
    assert len(set(trash_dirs)) == 20
    assert all(trash_dir.is_dir() for trash_dir in trash_dirs)


def test_attachment_pre_move_failure_keeps_database_and_file(
    client, tmp_path: Path, monkeypatch
):
    boot(client)
    get_settings().attachment_dir = str(tmp_path / "attachments")
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
    ).json()
    upload = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.txt", b"proof", "text/plain")},
    )
    attachment_path = Path(get_settings().attachment_dir) / attachment_storage_name(upload.json()["id"])
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204

    def fail_replace(source, destination):
        raise OSError("disk unavailable")

    monkeypatch.setattr(os, "replace", fail_replace)
    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")

    assert response.status_code == 500
    assert response.json()["detail"] == "附件隔离失败，清单未删除"
    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
    assert asyncio.run(row_counts(task_list["id"]))["attachments"] == 1
    assert attachment_path.read_bytes() == b"proof"


def test_pre_move_restore_failure_preserves_quarantined_copy(
    client, tmp_path: Path, monkeypatch
):
    boot(client)
    attachment_root = tmp_path / "attachments"
    get_settings().attachment_dir = str(attachment_root)
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "两个附件", "list_id": task_list["id"]}
    ).json()
    first = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("one.txt", b"one", "text/plain")},
    ).json()
    second = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("two.txt", b"two", "text/plain")},
    ).json()
    first_name = attachment_storage_name(first["id"])
    second_name = attachment_storage_name(second["id"])
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
    original_replace = os.replace
    calls = 0

    def fail_second_move_and_restore(source, destination):
        nonlocal calls
        calls += 1
        if calls >= 2:
            raise OSError("disk unavailable")
        return original_replace(source, destination)

    monkeypatch.setattr(os, "replace", fail_second_move_and_restore)
    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")

    assert response.status_code == 500
    assert "部分附件恢复失败" in response.json()["detail"]
    trash_dirs = list((attachment_root / ".purge-trash").iterdir())
    assert len(trash_dirs) == 1
    assert (trash_dirs[0] / first_name).read_bytes() == b"one"
    assert (attachment_root / second_name).read_bytes() == b"two"
    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
    assert asyncio.run(row_counts(task_list["id"]))["attachments"] == 2


def test_database_commit_failure_restores_quarantined_file_and_rows(
    client, tmp_path: Path, monkeypatch
):
    boot(client)
    get_settings().attachment_dir = str(tmp_path / "attachments")
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
    ).json()
    upload = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.txt", b"proof", "text/plain")},
    )
    attachment_path = Path(get_settings().attachment_dir) / attachment_storage_name(upload.json()["id"])
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
    original_commit = AsyncSession.commit

    async def fail_commit(db):
        raise OSError("database unavailable")

    monkeypatch.setattr(AsyncSession, "commit", fail_commit)
    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
    monkeypatch.setattr(AsyncSession, "commit", original_commit)

    assert response.status_code == 500
    assert response.json()["detail"] == "数据库提交失败，清单未删除"
    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
    assert asyncio.run(row_counts(task_list["id"]))["attachments"] == 1
    assert attachment_path.read_bytes() == b"proof"
    assert not (Path(get_settings().attachment_dir) / ".purge-trash").exists()


def test_cleanup_failure_can_be_retried_by_original_list_id(
    client, tmp_path: Path, monkeypatch
):
    boot(client)
    attachment_root = tmp_path / "attachments"
    get_settings().attachment_dir = str(attachment_root)
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
    ).json()
    upload = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.txt", b"proof", "text/plain")},
    )
    storage_name = attachment_storage_name(upload.json()["id"])
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
    original_rmtree = shutil.rmtree
    cleanup_attempts = 0

    def fail_cleanup_once(path, *args, **kwargs):
        nonlocal cleanup_attempts
        if Path(path).parent.name == ".purge-trash" and cleanup_attempts == 0:
            cleanup_attempts += 1
            raise OSError("disk unavailable")
        return original_rmtree(path, *args, **kwargs)

    monkeypatch.setattr(shutil, "rmtree", fail_cleanup_once)
    first = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
    assert first.status_code == 500
    operation_id = first.json()["detail"].split("操作ID: ", 1)[1].split(",", 1)[0]
    trash_dir = attachment_root / ".purge-trash" / operation_id
    assert (trash_dir / storage_name).read_bytes() == b"proof"

    second = client.delete(f"/api/v1/lists/{task_list['id']}/purge")

    assert second.status_code == 204
    assert not trash_dir.exists()
    assert asyncio.run(purge_operation_count(task_list["id"])) == 0


async def purge_operation_count(list_id):
    session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
    async with session_factory() as db:
        return await db.scalar(
            select(func.count()).select_from(PurgeOperation).where(
                PurgeOperation.list_id == UUID(str(list_id))
            )
        )


def test_final_quarantine_cleanup_failure_is_reported_and_keeps_residue_isolated(
    client, tmp_path: Path, monkeypatch, caplog
):
    boot(client)
    attachment_root = tmp_path / "attachments"
    get_settings().attachment_dir = str(attachment_root)
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
    ).json()
    upload = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.txt", b"proof", "text/plain")},
    )
    storage_name = attachment_storage_name(upload.json()["id"])
    attachment_path = attachment_root / storage_name
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
    original_rmtree = shutil.rmtree

    def fail_trash_cleanup(path, *args, **kwargs):
        if Path(path).parent.name == ".purge-trash":
            raise OSError("disk unavailable")
        return original_rmtree(path, *args, **kwargs)

    monkeypatch.setattr(shutil, "rmtree", fail_trash_cleanup)
    with caplog.at_level("ERROR", logger="backend.main"):
        response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")

    trash_dirs = [path for path in (attachment_root / ".purge-trash").iterdir()]
    assert len(trash_dirs) == 1
    trash_dir = trash_dirs[0]
    assert response.status_code == 500
    assert "清单数据已删除，但附件清理未完成" in response.json()["detail"]
    assert f"操作ID: {trash_dir.name}" in response.json()["detail"]
    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 0
    assert not attachment_path.exists()
    assert (trash_dir / storage_name).read_bytes() == b"proof"
    assert str(trash_dir) in caplog.text


def test_malicious_attachment_path_rejects_purge_without_touching_file_or_database(
    client, tmp_path: Path
):
    boot(client)
    attachment_root = tmp_path / "attachments"
    get_settings().attachment_dir = str(attachment_root)
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "恶意附件", "list_id": task_list["id"]}
    ).json()
    outside_path = tmp_path / "outside.txt"
    outside_path.write_text("keep", encoding="utf-8")

    async def seed_malicious_attachment():
        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
        async with session_factory() as db:
            owner_id = await db.scalar(select(TaskList.user_id).where(TaskList.id == UUID(task_list["id"])))
            db.add(
                Attachment(
                    user_id=owner_id,
                    task_id=UUID(task["id"]),
                    filename="outside.txt",
                    storage_name="../outside.txt",
                    mime_type="text/plain",
                    size=4,
                )
            )
            await db.commit()

    asyncio.run(seed_malicious_attachment())
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204

    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")

    assert response.status_code == 409
    assert response.json()["detail"] == "附件存储路径无效，无法永久删除清单"
    assert outside_path.read_text(encoding="utf-8") == "keep"
    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
    assert asyncio.run(row_counts(task_list["id"]))["attachments"] == 1


@pytest.mark.parametrize(
    "storage_name",
    [
        "/tmp/dodo-outside.txt",
        "nested/../proof.txt",
        "nested/./proof.txt",
        "nested//proof.txt",
        "C:\\temp\\proof.txt",
    ],
)
def test_absolute_navigation_or_empty_attachment_path_segment_is_rejected(
    client, tmp_path: Path, storage_name: str
):
    boot(client)
    get_settings().attachment_dir = str(tmp_path / "attachments")
    task_list = create_list(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "恶意附件", "list_id": task_list["id"]}
    ).json()

    async def seed_malicious_attachment():
        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
        async with session_factory() as db:
            owner_id = await db.scalar(select(TaskList.user_id).where(TaskList.id == UUID(task_list["id"])))
            db.add(
                Attachment(
                    user_id=owner_id,
                    task_id=UUID(task["id"]),
                    filename="proof.txt",
                    storage_name=storage_name,
                    mime_type="text/plain",
                    size=4,
                )
            )
            await db.commit()

    asyncio.run(seed_malicious_attachment())
    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
    assert client.delete(f"/api/v1/lists/{task_list['id']}/purge").status_code == 409
    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1


async def row_counts(list_id):
    list_id = UUID(str(list_id))
    session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
    async with session_factory() as db:
        task_ids = list((await db.scalars(select(Task.id).where(Task.list_id == list_id))).all())
        template_ids = list(
            (await db.scalars(select(RecurrenceTemplate.id).where(RecurrenceTemplate.task_id.in_(task_ids)))).all()
        ) if task_ids else []
        return {
            "lists": await db.scalar(select(func.count()).select_from(TaskList).where(TaskList.id == list_id)),
            "tasks": len(task_ids),
            "templates": len(template_ids),
            "exceptions": await db.scalar(
                select(func.count()).select_from(RecurrenceException).where(
                    RecurrenceException.template_id.in_(template_ids)
                )
            ) if template_ids else 0,
            "attachments": await db.scalar(
                select(func.count()).select_from(Attachment).where(Attachment.task_id.in_(task_ids))
            ) if task_ids else 0,
        }


def test_purge_archived_list_removes_all_task_data_and_attachment_files(client, tmp_path: Path):
    boot(client)
    get_settings().attachment_dir = str(tmp_path / "attachments")
    task_list = create_list(client)
    parent = client.post(
        "/api/v1/tasks",
        json={
            "title": "重复父任务",
            "list_id": task_list["id"],
            "due_at": "2030-01-01T09:00:00Z",
            "rrule": "FREQ=DAILY",
        },
    ).json()
    child = client.post(
        "/api/v1/tasks",
        json={"title": "子任务", "list_id": task_list["id"], "parent_id": parent["id"]},
    ).json()
    completed = client.post(
        "/api/v1/tasks", json={"title": "已完成", "list_id": task_list["id"]}
    ).json()
    completed = client.patch(
        f"/api/v1/tasks/{completed['id']}", json={"completed": True, "version": completed["version"]}
    ).json()
    soft_deleted = client.post(
        "/api/v1/tasks", json={"title": "软删除", "list_id": task_list["id"]}
    ).json()
    assert client.delete(f"/api/v1/tasks/{soft_deleted['id']}").status_code == 204

    upload = client.post(
        f"/api/v1/tasks/{child['id']}/attachments",
        files={"file": ("proof.txt", b"proof", "text/plain")},
    )
    assert upload.status_code == 201
    attachment_id = upload.json()["id"]

    async def seed_exception_and_get_storage_name():
        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
        async with session_factory() as db:
            template = await db.scalar(
                select(RecurrenceTemplate).where(RecurrenceTemplate.task_id == UUID(parent["id"]))
            )
            db.add(
                RecurrenceException(
                    template_id=template.id,
                    occurrence_at=datetime(2030, 1, 2, 9, tzinfo=UTC),
                    completed=True,
                )
            )
            attachment = await db.get(Attachment, UUID(attachment_id))
            storage_name = attachment.storage_name
            await db.commit()
            return storage_name

    storage_name = asyncio.run(seed_exception_and_get_storage_name())
    attachment_path = Path(get_settings().attachment_dir) / storage_name
    assert attachment_path.is_file()
    assert asyncio.run(row_counts(task_list["id"])) == {
        "lists": 1,
        "tasks": 4,
        "templates": 1,
        "exceptions": 1,
        "attachments": 1,
    }

    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")

    assert response.status_code == 204
    assert asyncio.run(row_counts(task_list["id"])) == {
        "lists": 0,
        "tasks": 0,
        "templates": 0,
        "exceptions": 0,
        "attachments": 0,
    }
    assert not attachment_path.exists()