import asyncio
from datetime import timedelta
from pathlib import Path
from uuid import UUID

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

from backend.models import Attachment, BackupImportEntity
from tests.test_backup_v2 import _preflight
from tests.test_mvp_backend import boot


async def _attachment_state(attachment_id: str) -> tuple[Attachment, int]:
    from backend.db import get_engine

    async with AsyncSession(get_engine()) as db:
        attachment = await db.get(Attachment, UUID(attachment_id))
        ledger_count = await db.scalar(
            select(func.count()).select_from(BackupImportEntity).where(
                BackupImportEntity.entity_type == "attachments",
                BackupImportEntity.source_id == UUID(attachment_id),
            )
        )
        return attachment, ledger_count or 0


@pytest.mark.parametrize("damage", ["missing", "corrupt"])
def test_merge_existing_attachment_without_ledger_never_certifies_bad_file(
    client, tmp_path, damage
):
    from backend.config import get_settings

    inbox = boot(client)
    root = tmp_path / "attachments"
    get_settings().attachment_dir = str(root)
    task = client.post(
        "/api/v1/tasks", json={"title": "attachment collision", "list_id": inbox["id"]}
    ).json()
    payload = b"trusted attachment bytes"
    attachment = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.txt", payload, "text/plain")},
    ).json()
    archive = client.get("/api/v1/backup/export.zip").content
    row, ledger_count = asyncio.run(_attachment_state(attachment["id"]))
    assert ledger_count == 0
    stored = root / row.storage_name
    if damage == "missing":
        stored.unlink()
    else:
        stored.write_bytes(b"x" * len(payload))

    token = _preflight(client, archive, "merge").json()["preflight_token"]
    response = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": token, "mode": "merge"},
    )

    assert response.status_code == 409
    assert response.json()["detail"]["code"] in {
        "backup_entity_missing",
        "backup_entity_conflict",
    }
    _, ledger_count = asyncio.run(_attachment_state(attachment["id"]))
    assert ledger_count == 0


def test_merge_existing_attachment_without_ledger_hashes_equal_bytes_before_ledger(
    client, tmp_path
):
    from backend.config import get_settings

    inbox = boot(client)
    root = tmp_path / "attachments"
    get_settings().attachment_dir = str(root)
    task = client.post(
        "/api/v1/tasks", json={"title": "attachment collision", "list_id": inbox["id"]}
    ).json()
    payload = b"trusted attachment bytes"
    attachment = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.txt", payload, "text/plain")},
    ).json()
    archive = client.get("/api/v1/backup/export.zip").content

    token = _preflight(client, archive, "merge").json()["preflight_token"]
    response = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": token, "mode": "merge"},
    )

    assert response.status_code == 200, response.text
    row, ledger_count = asyncio.run(_attachment_state(attachment["id"]))
    assert ledger_count == 1
    assert Path(root / row.storage_name).read_bytes() == payload


async def _business_counts() -> tuple[int, ...]:
    from backend.backup.service import ENTITY_MODELS
    from backend.db import get_engine

    async with AsyncSession(get_engine()) as db:
        counts = []
        for model in ENTITY_MODELS.values():
            counts.append((await db.scalar(select(func.count()).select_from(model))) or 0)
        return tuple(counts)


def _base_graph() -> dict[str, list[dict]]:
    list_id = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
    task_id = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"
    recurrence_id = "cccccccc-cccc-4ccc-8ccc-cccccccccccc"
    return {
        "lists": [{"id": list_id, "name": "Inbox", "is_inbox": True, "position": 0}],
        "tasks": [{
            "id": task_id,
            "list_id": list_id,
            "parent_id": None,
            "title": "valid task",
            "description": "",
            "priority": 0,
            "completed": False,
            "completed_at": None,
            "due_at": "2026-09-20T08:00:00+00:00",
            "due_has_time": True,
            "version": 1,
            "position": 0,
            "external_id": None,
        }],
        "recurrences": [{
            "id": recurrence_id,
            "task_id": task_id,
            "rrule": "FREQ=WEEKLY;BYDAY=MO",
            "starts_at": "2026-09-20T08:00:00+00:00",
            "ends_at": None,
            "trigger_mode": "scheduled",
            "after_completion_days": None,
            "last_completed_at": None,
        }],
    }


@pytest.mark.parametrize(
    ("mutate", "label"),
    [
        (lambda graph: graph["recurrences"][0].update(trigger_mode="after_completion", rrule=None, after_completion_days=None), "after completion days required"),
        (lambda graph: graph["recurrences"][0].update(trigger_mode="after_completion", rrule="FREQ=DAILY", after_completion_days=1), "after completion excludes rrule"),
        (lambda graph: graph["recurrences"][0].update(trigger_mode="after_completion", rrule=None, after_completion_days=0), "after completion range"),
        (lambda graph: graph["recurrences"][0].update(trigger_mode="scheduled", rrule=None, after_completion_days=None), "scheduled requires rrule"),
        (lambda graph: graph["recurrences"][0].update(trigger_mode="scheduled", after_completion_days=1), "scheduled excludes days"),
        (lambda graph: graph["recurrences"][0].update(rrule="FREQ=NOPE"), "rrule parses"),
        (lambda graph: graph["recurrences"][0].update(ends_at="2026-09-19T08:00:00+00:00"), "ends after starts"),
        (lambda graph: graph["recurrences"][0].update(last_completed_at="2026-09-21T08:00:00+00:00"), "last completion before start"),
        (lambda graph: graph["tasks"][0].update(due_at=None, due_has_time=False), "recurring task has due"),
        (lambda graph: graph["tasks"][0].update(parent_id="dddddddd-dddd-4ddd-8ddd-dddddddddddd"), "recurring task top level"),
    ],
)
def test_recurrence_preflight_rejects_invalid_contract_without_business_writes(client, mutate, label):
    boot(client)
    graph = _base_graph()
    if label == "recurring task top level":
        graph["tasks"].append({
            **graph["tasks"][0],
            "id": "dddddddd-dddd-4ddd-8ddd-dddddddddddd",
            "title": "parent",
            "parent_id": None,
        })
    mutate(graph)
    before = asyncio.run(_business_counts())

    content = _make_archive_from_graph(graph)
    response = _preflight(client, content, "replace")

    assert response.status_code == 422, (label, response.text)
    assert response.json()["detail"]["code"] == "backup_recurrence_invalid"
    assert asyncio.run(_business_counts()) == before


def _make_archive_from_graph(
    graph: dict[str, list[dict]],
    files: dict[str, bytes] | None = None,
    *,
    backup_id: str = "11111111-1111-4111-8111-111111111111",
) -> bytes:
    from tests.test_backup_v2 import _make_zip

    entries = {
        f"data/{entity}.json": __import__("json").dumps(rows).encode()
        for entity, rows in graph.items()
    }
    entries.update(files or {})
    return _make_zip(entries, backup_id=backup_id)


def _task_tree_graph(parent_ids: list[str | None]) -> dict[str, list[dict]]:
    graph = _base_graph()
    graph["recurrences"] = []
    template = graph["tasks"][0]
    task_ids = [
        "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
        "cccccccc-cccc-4ccc-8ccc-cccccccccccc",
        "dddddddd-dddd-4ddd-8ddd-dddddddddddd",
    ]
    graph["tasks"] = [
        {
            **template,
            "id": task_id,
            "parent_id": parent_id,
            "title": f"tree task {index}",
            "position": index,
        }
        for index, (task_id, parent_id) in enumerate(zip(task_ids, parent_ids, strict=True))
    ]
    return graph


def test_preflight_rejects_three_level_task_tree_without_business_writes(client):
    boot(client)
    graph = _task_tree_graph([
        None,
        "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
        "cccccccc-cccc-4ccc-8ccc-cccccccccccc",
    ])
    before = asyncio.run(_business_counts())

    response = _preflight(client, _make_archive_from_graph(graph), "replace")

    assert response.status_code == 422, response.text
    assert response.json()["detail"]["code"] == "backup_constraint_invalid"
    assert asyncio.run(_business_counts()) == before


def test_parent_with_multiple_children_round_trips_in_arbitrary_zip_order(client):
    boot(client)
    parent_id = "cccccccc-cccc-4ccc-8ccc-cccccccccccc"
    graph = _task_tree_graph([parent_id, None, parent_id])
    graph["tasks"] = [graph["tasks"][2], graph["tasks"][1], graph["tasks"][0]]

    preflight = _preflight(client, _make_archive_from_graph(graph), "replace")
    assert preflight.status_code == 200, preflight.text
    restored = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": preflight.json()["preflight_token"], "mode": "replace"},
    )
    assert restored.status_code == 200, restored.text

    from tests.test_backup_v2 import _archive_rows

    rows = _archive_rows(client.get("/api/v1/backup/export.zip").content, "tasks")
    by_id = {row["id"]: row for row in rows}
    assert by_id[parent_id]["parent_id"] is None
    assert {
        row["id"] for row in rows if row["parent_id"] == parent_id
    } == {
        "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
        "dddddddd-dddd-4ddd-8ddd-dddddddddddd",
    }
    items = client.get("/api/v1/tasks", params={"limit": 100}).json()["items"]
    restored_parent = next(item for item in items if item["id"] == parent_id)
    assert {child["id"] for child in restored_parent["subtasks"]} == {
        "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
        "dddddddd-dddd-4ddd-8ddd-dddddddddddd",
    }


@pytest.mark.parametrize(
    "recurrence",
    [
        {
            "rrule": "FREQ=MONTHLY;BYMONTHDAY=1,15;COUNT=8",
            "trigger_mode": "scheduled",
            "after_completion_days": None,
        },
        {
            "rrule": None,
            "trigger_mode": "after_completion",
            "after_completion_days": 30,
        },
    ],
)
def test_both_recurrence_modes_round_trip_through_replace(client, recurrence):
    boot(client)
    graph = _base_graph()
    graph["recurrences"][0].update(recurrence)
    content = _make_archive_from_graph(graph)
    preflight = _preflight(client, content, "replace")
    assert preflight.status_code == 200, preflight.text
    restored = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": preflight.json()["preflight_token"], "mode": "replace"},
    )
    assert restored.status_code == 200, restored.text
    exported = client.get("/api/v1/backup/export.zip").content
    from tests.test_backup_v2 import _archive_rows

    row = _archive_rows(exported, "recurrences")[0]
    assert row["rrule"] == recurrence["rrule"]
    assert row["trigger_mode"] == recurrence["trigger_mode"]
    assert row["after_completion_days"] == recurrence["after_completion_days"]


def _entity_case(entity: str, row: dict, files: dict[str, bytes] | None = None) -> bytes:
    graph = _base_graph()
    graph[entity] = [row]
    return _make_archive_from_graph(graph, files)


@pytest.mark.parametrize(
    ("entity", "row", "files"),
    [
        ("folders", {"id": "10101010-1010-4010-8010-101010101010", "name": "   ", "position": 0}, None),
        ("folders", {"id": "10101010-1010-4010-8010-101010101010", "name": "x" * 121, "position": 0}, None),
        ("folders", {"id": "10101010-1010-4010-8010-101010101010", "name": "x", "position": -1}, None),
        ("lists", {"id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", "name": "x" * 121, "is_inbox": True, "position": 0}, None),
        ("lists", {"id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", "name": "Inbox", "is_inbox": True, "position": -1}, None),
        ("tasks", {**_base_graph()["tasks"][0], "title": "   "}, None),
        ("tasks", {**_base_graph()["tasks"][0], "priority": 4}, None),
        ("tasks", {**_base_graph()["tasks"][0], "version": 0}, None),
        ("tasks", {**_base_graph()["tasks"][0], "position": -1}, None),
        ("tasks", {**_base_graph()["tasks"][0], "due_at": None, "due_has_time": True}, None),
        ("tasks", {**_base_graph()["tasks"][0], "completed": False, "completed_at": "2026-09-20T08:00:00+00:00"}, None),
        ("tasks", {**_base_graph()["tasks"][0], "created_at": "2026-09-21T08:00:00+00:00", "updated_at": "2026-09-20T08:00:00+00:00"}, None),
        ("countdowns", {"id": "20202020-2020-4020-8020-202020202020", "title": "x", "event_date": "2026-09-20", "calendar_mode": "bad", "lunar_month": None, "lunar_day": None, "ignore_year": False, "kind": "countdown", "repeat_rule": "none", "icon": "x", "pinned": False}, None),
        ("countdowns", {"id": "20202020-2020-4020-8020-202020202020", "title": "x", "event_date": "2026-09-20", "calendar_mode": "solar", "lunar_month": 1, "lunar_day": 1, "ignore_year": False, "kind": "countdown", "repeat_rule": "none", "icon": "x", "pinned": False}, None),
        ("memos", {"id": "30303030-3030-4030-8030-303030303030", "title": "   ", "content": "", "version": 1}, None),
        ("memos", {"id": "30303030-3030-4030-8030-303030303030", "title": "x", "content": "", "version": 0}, None),
        ("memos", {"id": "30303030-3030-4030-8030-303030303030", "title": "x", "content": "", "version": 1, "created_at": "2026-09-21T08:00:00+00:00", "updated_at": "2026-09-20T08:00:00+00:00"}, None),
        ("attachments", {"id": "40404040-4040-4040-8040-404040404040", "task_id": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", "filename": "x" * 256, "mime_type": "text/plain", "size": 1, "archive_path": "attachments/40404040-4040-4040-8040-404040404040/content"}, {"attachments/40404040-4040-4040-8040-404040404040/content": b"x"}),
        ("attachments", {"id": "40404040-4040-4040-8040-404040404040", "task_id": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", "filename": "x", "mime_type": "application/x-danger", "size": 1, "archive_path": "attachments/40404040-4040-4040-8040-404040404040/content"}, {"attachments/40404040-4040-4040-8040-404040404040/content": b"x"}),
        ("attachments", {"id": "40404040-4040-4040-8040-404040404040", "task_id": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", "filename": "x", "mime_type": "text/plain", "size": -1, "archive_path": "attachments/40404040-4040-4040-8040-404040404040/content"}, {"attachments/40404040-4040-4040-8040-404040404040/content": b"x"}),
    ],
)
def test_all_entity_contracts_fail_preflight_without_business_writes(client, entity, row, files):
    boot(client)
    before = asyncio.run(_business_counts())

    response = _preflight(client, _entity_case(entity, row, files), "replace")

    assert response.status_code == 422, (entity, response.text)
    assert asyncio.run(_business_counts()) == before


def test_legacy_replace_is_rejected_without_mutating_data(client):
    inbox = boot(client)
    task = client.post(
        "/api/v1/tasks", json={"title": "must survive", "list_id": inbox["id"]}
    ).json()
    payload = client.get("/api/v1/export").json()

    response = client.post("/api/v1/restore?mode=replace", json=payload)

    assert response.status_code == 422
    assert response.json()["detail"] == {
        "code": "legacy_replace_unsupported",
        "message": "旧版备份仅支持合并恢复",
    }
    assert client.get(f"/api/v1/tasks/{task['id']}").status_code == 200


def test_legacy_csv_upload_has_streaming_hard_limit(client, monkeypatch):
    boot(client)
    from backend import mvp

    monkeypatch.setattr(mvp, "LEGACY_BACKUP_MAX_BYTES", 32)
    response = client.post(
        "/api/v1/restore.csv?mode=merge",
        files={"file": ("backup.csv", b"entity,data\n" + b"x" * 33, "text/csv")},
    )

    assert response.status_code == 422
    assert response.json()["detail"] == {
        "code": "legacy_backup_too_large",
        "message": "旧版备份文件过大",
    }


def test_legacy_json_rejects_excessive_records(client, monkeypatch):
    boot(client)
    from backend import mvp

    monkeypatch.setattr(mvp, "LEGACY_BACKUP_MAX_RECORDS", 1)
    payload = {"version": 1, "folders": [], "lists": [], "tasks": [
        {"id": "11111111-1111-4111-8111-111111111111"},
        {"id": "22222222-2222-4222-8222-222222222222"},
    ]}

    response = client.post("/api/v1/restore?mode=merge", json=payload)

    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "legacy_backup_too_many_records"


def test_v2_preflight_rejects_multiple_active_pinned_countdowns(client):
    boot(client)
    graph = _base_graph()
    graph["recurrences"] = []
    graph["countdowns"] = [
        {
            "id": f"{index:08d}-2020-4020-8020-202020202020",
            "title": f"pinned {index}", "event_date": "2026-09-20",
            "calendar_mode": "solar", "lunar_month": None, "lunar_day": None,
            "ignore_year": False, "kind": "countdown", "repeat_rule": "none",
            "icon": "x", "pinned": True, "archived_at": None,
        }
        for index in (1, 2)
    ]

    response = _preflight(client, _make_archive_from_graph(graph), "merge")

    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "backup_constraint_invalid"


def test_reaper_removes_failed_and_stranded_staging_and_repairs_cleanup(client, tmp_path):
    boot(client)
    from backend.backup.router import _prune
    from backend.config import get_settings
    from backend.db import get_engine
    from backend.models import BackupPreflight, utcnow

    staging_root = tmp_path / "staging"
    staging_root.mkdir()
    get_settings().backup_staging_dir = str(staging_root)
    failed_file = staging_root / "failed.zip"
    consuming_file = staging_root / "consuming.zip"
    failed_file.write_bytes(b"failed")
    consuming_file.write_bytes(b"consuming")
    cleanup = tmp_path / "cleanup"
    cleanup.mkdir()
    (cleanup / "old").write_bytes(b"old")

    async def exercise():
        async with AsyncSession(get_engine()) as db:
            user_id = await db.scalar(select(__import__("backend.models", fromlist=["User"]).User.id))
            now = utcnow() - timedelta(hours=1)
            rows = [
                BackupPreflight(
                    token_hash=str(index) * 64, user_id=user_id,
                    backup_id=UUID(f"00000000-0000-4000-8000-00000000000{index}"),
                    archive_sha256="0" * 64, archive_size=10,
                    staging_path=str(path), mode="merge", status=status,
                    expires_at=now, consumed_at=now,
                    cleanup_path=str(cleanup) if status == "cleanup_pending" else None,
                )
                for index, (status, path) in enumerate(
                    [("failed", failed_file), ("consuming", consuming_file),
                     ("cleanup_pending", staging_root / "cleanup.zip")], start=1
                )
            ]
            db.add_all(rows)
            await db.flush()
            row_ids = [row.id for row in rows]
            await db.commit()
            await _prune(db)
            statuses = {
                str(row_id): await db.scalar(
                    select(BackupPreflight.status).where(BackupPreflight.id == row_id)
                )
                for row_id in row_ids
            }
            return row_ids, statuses

    row_ids, statuses = asyncio.run(exercise())
    assert not failed_file.exists()
    assert not consuming_file.exists()
    assert not cleanup.exists()
    assert statuses[str(row_ids[2])] == "consumed"


def test_expired_repair_pending_is_repaired_not_deleted_with_quarantine(client, tmp_path):
    boot(client)
    from backend.backup.router import _prune
    from backend.config import get_settings
    from backend.db import get_engine
    from backend.models import BackupPreflight, User, utcnow

    staging_root = tmp_path / "staging"
    attachment_root = tmp_path / "attachments"
    quarantine = tmp_path / "quarantine"
    staging_root.mkdir()
    attachment_root.mkdir()
    quarantine.mkdir()
    staging = staging_root / "repair.zip"
    staging.write_bytes(b"staged")
    (quarantine / "restored.bin").write_bytes(b"original")
    get_settings().backup_staging_dir = str(staging_root)
    get_settings().attachment_dir = str(attachment_root)

    async def exercise():
        async with AsyncSession(get_engine()) as db:
            user_id = await db.scalar(select(User.id))
            row = BackupPreflight(
                token_hash="9" * 64,
                user_id=user_id,
                backup_id=UUID("99999999-9999-4999-8999-999999999999"),
                archive_sha256="0" * 64,
                archive_size=7,
                staging_path=str(staging),
                mode="replace",
                status="repair_pending",
                expires_at=utcnow() - timedelta(hours=1),
                cleanup_path=str(quarantine),
            )
            db.add(row)
            await db.flush()
            row_id = row.id
            await db.commit()
            await _prune(db)
            repaired = await db.get(BackupPreflight, row_id)
            return repaired.status, repaired.cleanup_path

    status, cleanup_path = asyncio.run(exercise())
    assert (attachment_root / "restored.bin").read_bytes() == b"original"
    assert not quarantine.exists()
    assert not staging.exists()
    assert status == "failed"
    assert cleanup_path is None


def test_repair_pending_counts_against_preflight_quota(client, tmp_path, monkeypatch):
    boot(client)
    import importlib

    from backend.config import get_settings
    from backend.db import get_engine
    from backend.models import BackupPreflight, User, utcnow

    router_module = importlib.import_module("backend.backup.router")
    monkeypatch.setattr(router_module, "MAX_PENDING_PREFLIGHTS_PER_USER", 1)
    staging_root = tmp_path / "staging"
    staging_root.mkdir()
    get_settings().backup_staging_dir = str(staging_root)
    staged = staging_root / "pending.zip"
    staged.write_bytes(b"pending")

    async def seed():
        async with AsyncSession(get_engine()) as db:
            user_id = await db.scalar(select(User.id))
            db.add(BackupPreflight(
                token_hash="8" * 64,
                user_id=user_id,
                backup_id=UUID("88888888-8888-4888-8888-888888888888"),
                archive_sha256="0" * 64,
                archive_size=7,
                staging_path=str(staged),
                mode="replace",
                status="repair_pending",
                expires_at=utcnow() + timedelta(hours=1),
                cleanup_path=str(tmp_path / "quarantine"),
            ))
            await db.commit()

    asyncio.run(seed())
    content = client.get("/api/v1/backup/export.zip").content
    response = _preflight(client, content, "merge")
    assert response.status_code == 429
    assert response.json()["detail"]["code"] == "backup_preflight_quota"


def test_repair_pending_exception_survives_router_and_same_token_only_repairs(
    client, tmp_path, monkeypatch
):
    inbox = boot(client)
    import importlib

    backup_router = importlib.import_module("backend.backup.router")
    from backend.backup import service
    from backend.config import get_settings
    from backend.db import get_engine
    from backend.models import Attachment, BackupPreflight

    root = tmp_path / "attachments"
    get_settings().attachment_dir = str(root)
    task = client.post("/api/v1/tasks", json={"title": "old", "list_id": inbox["id"]}).json()
    uploaded = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("old.txt", b"old", "text/plain")},
    ).json()
    content = client.get("/api/v1/backup/export.zip").content
    token = _preflight(client, content, "replace").json()["preflight_token"]

    real_restore = service.restore_quarantine
    monkeypatch.setattr(service, "contained_file", lambda *_: (_ for _ in ()).throw(OSError("write")))
    monkeypatch.setattr(service, "restore_quarantine", lambda *_: (_ for _ in ()).throw(OSError("repair")))
    response = client.post(
        "/api/v1/backup/restore", json={"preflight_token": token, "mode": "replace"}
    )
    assert response.status_code == 500
    assert response.json()["detail"]["code"] == "backup_repair_pending"

    async def state():
        async with AsyncSession(get_engine()) as db:
            row = await db.scalar(select(BackupPreflight).where(BackupPreflight.status == "repair_pending"))
            attachment = await db.get(Attachment, UUID(uploaded["id"]))
            return row.status, row.cleanup_path, root / attachment.storage_name

    status, cleanup_path, old_path = asyncio.run(state())
    assert status == "repair_pending"
    assert cleanup_path
    assert not old_path.exists()

    monkeypatch.setattr(service, "restore_quarantine", real_restore)
    monkeypatch.setattr(
        backup_router,
        "restore_v2",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("DB restore repeated")),
    )
    retried = client.post(
        "/api/v1/backup/restore", json={"preflight_token": token, "mode": "replace"}
    )
    assert retried.status_code == 409
    assert retried.json()["detail"]["code"] == "backup_restore_retry"
    assert old_path.read_bytes() == b"old"

    async def repaired_state():
        async with AsyncSession(get_engine()) as db:
            row = await db.scalar(select(BackupPreflight).where(BackupPreflight.token_hash.is_not(None)))
            return row.status, row.cleanup_path

    assert asyncio.run(repaired_state()) == ("failed", None)


def test_merge_preflight_rejects_archive_pin_when_user_has_different_active_pin(client):
    boot(client)
    existing = client.post(
        "/api/v1/countdowns",
        json={"title": "existing pin", "event_date": "2026-09-20", "pinned": True},
    ).json()
    graph = _base_graph()
    graph["recurrences"] = []
    graph["countdowns"] = [{
        "id": "77777777-7777-4777-8777-777777777777",
        "title": "incoming pin",
        "event_date": "2026-09-21",
        "calendar_mode": "solar",
        "lunar_month": None,
        "lunar_day": None,
        "ignore_year": False,
        "kind": "countdown",
        "repeat_rule": "none",
        "icon": "x",
        "pinned": True,
        "archived_at": None,
    }]

    response = _preflight(client, _make_archive_from_graph(graph), "merge")
    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "backup_constraint_invalid"
    active = client.get("/api/v1/countdowns").json()
    assert [row["id"] for row in active if row["pinned"]] == [existing["id"]]


def test_restore_target_ids_are_unique_across_entity_tables(client):
    boot(client)
    shared = "66666666-6666-4666-8666-666666666666"

    async def seed_folder():
        from backend.db import get_engine
        from backend.models import Folder, User
        async with AsyncSession(get_engine()) as db:
            user_id = await db.scalar(select(User.id))
            db.add(Folder(
                id=UUID(shared), user_id=user_id, name="existing entity id", position=0
            ))
            await db.commit()

    asyncio.run(seed_folder())
    current = client.get("/api/v1/backup/export.zip").content
    from tests.test_backup_v2 import _archive_rows, _replace_entities

    lists = _archive_rows(current, "lists")
    lists.append({
        "id": shared,
        "name": "cross-table collision",
        "is_inbox": False,
        "position": 1,
    })
    second_content = _replace_entities(current, {"lists": lists})
    entries = __import__("tests.test_backup_v2", fromlist=["_zip_entries"])._zip_entries(second_content)
    entities = {
        name.removeprefix("data/").removesuffix(".json"): __import__("json").loads(value)
        for name, value in entries.items()
        if name.startswith("data/") and name.endswith(".json")
    }
    second = _preflight(
        client,
        _make_archive_from_graph(
            entities,
            {
                name: value
                for name, value in entries.items()
                if name.startswith("attachments/")
            },
            backup_id="55555555-5555-4555-8555-555555555555",
        ),
        "merge",
    )
    assert second.status_code == 200, second.text
    restored = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": second.json()["preflight_token"], "mode": "merge"},
    )
    assert restored.status_code == 200, restored.text

    async def ids():
        from backend.db import get_engine
        from backend.models import Folder, TaskList
        async with AsyncSession(get_engine()) as db:
            folder_id = await db.scalar(select(Folder.id).where(Folder.name == "existing entity id"))
            list_id = await db.scalar(select(TaskList.id).where(TaskList.name == "cross-table collision"))
            return folder_id, list_id

    folder_id, list_id = asyncio.run(ids())
    assert folder_id == UUID(shared)
    assert list_id != folder_id