import asyncio
import hashlib
import io
import json
import zipfile
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from datetime import UTC, date, datetime
from pathlib import Path
from uuid import UUID

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

from backend.auth import COOKIE_NAME, hash_token
from backend.models import (
    BackupImport,
    BackupPreflight,
    Habit,
    HabitLog,
    HabitPause,
    RecurrenceException,
    Session,
    Task,
    User,
)
from tests.test_mvp_backend import boot


def _zip_entries(content: bytes) -> dict[str, bytes]:
    with zipfile.ZipFile(io.BytesIO(content)) as archive:
        return {name: archive.read(name) for name in archive.namelist()}


def _make_zip(entries: dict[str, bytes], *, backup_id: str = "11111111-1111-4111-8111-111111111111") -> bytes:
    entity_names = {
        "folders", "lists", "tasks", "recurrences", "recurrence_exceptions", "habits",
        "habit_logs", "habit_pauses", "countdowns", "memos", "attachments",
    }
    complete_entries = {f"data/{name}.json": b"[]" for name in entity_names}
    complete_entries["data/lists.json"] = json.dumps([{
        "id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
        "name": "Inbox", "is_inbox": True, "position": 0,
    }]).encode()
    complete_entries.update(entries)
    entries = complete_entries
    checksums = {name: hashlib.sha256(value).hexdigest() for name, value in entries.items()}
    manifest = {
        "format": "dodo-backup",
        "version": 2,
        "backup_id": backup_id,
        "entities": {
            name: len(json.loads(entries[f"data/{name}.json"])) for name in entity_names
        },
        "checksums": checksums,
    }
    output = io.BytesIO()
    with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("manifest.json", json.dumps(manifest))
        for name, value in entries.items():
            archive.writestr(name, value)
    return output.getvalue()


def _preflight(client, content: bytes, mode: str = "merge"):
    return client.post(
        "/api/v1/backup/preflight",
        params={"mode": mode},
        files={"file": ("backup.zip", content, "application/zip")},
    )


def _replace_entities(content: bytes, replacements: dict[str, list[dict]]) -> bytes:
    entries = _zip_entries(content)
    manifest = json.loads(entries.pop("manifest.json"))
    for entity, rows in replacements.items():
        name = f"data/{entity}.json"
        entries[name] = json.dumps(rows, allow_nan=True).encode()
        manifest["entities"][entity] = len(rows)
        manifest["checksums"][name] = hashlib.sha256(entries[name]).hexdigest()
    output = io.BytesIO()
    with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("manifest.json", json.dumps(manifest))
        for name, value in entries.items():
            archive.writestr(name, value)
    return output.getvalue()


def _archive_rows(content: bytes, entity: str) -> list[dict]:
    return json.loads(_zip_entries(content)[f"data/{entity}.json"])


def test_zip_v2_round_trip_includes_history_exception_and_attachment_bytes(client, tmp_path):
    inbox = boot(client)
    from backend.config import get_settings

    get_settings().attachment_dir = str(tmp_path / "attachments")
    task = client.post(
        "/api/v1/tasks",
        json={"title": "完整备份任务", "list_id": inbox["id"], "due_at": "2026-09-07T09:00:00Z"},
    ).json()
    recurrence = client.post(
        "/api/v1/recurrences", json={"task_id": task["id"], "rrule": "FREQ=WEEKLY"}
    ).json()
    habit = client.post(
        "/api/v1/habits", json={"name": "备份习惯", "kind": "numeric", "target": 2, "schedule_type": "daily"}
    ).json()
    today = datetime.now(UTC).date().isoformat()
    assert client.put(f"/api/v1/habits/{habit['id']}/logs/{today}", json={"value": 2}).status_code == 200
    assert client.post(
        f"/api/v1/habits/{habit['id']}/pauses",
        json={"start_date": "2026-09-18", "end_date": "2026-09-19"},
    ).status_code == 201
    attachment_bytes = b"\x00real attachment bytes\xff"
    uploaded = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("proof.bin", attachment_bytes, "application/zip")},
    )
    assert uploaded.status_code == 201

    async def seed_exception():
        from backend.db import get_engine
        async with AsyncSession(get_engine()) as db:
            db.add(RecurrenceException(
                template_id=UUID(recurrence["id"]), occurrence_at=datetime(2026, 9, 14, 9, tzinfo=UTC),
                title="例外标题", completed=True,
            ))
            await db.commit()

    asyncio.run(seed_exception())
    exported = client.get("/api/v1/backup/export.zip")
    assert exported.status_code == 200
    assert exported.headers["content-type"] == "application/zip"
    assert "dodo-backup-v2.zip" in exported.headers["content-disposition"]
    entries = _zip_entries(exported.content)
    manifest = json.loads(entries["manifest.json"])
    assert manifest["version"] == 2
    assert set(manifest["entities"]) >= {
        "folders", "lists", "tasks", "recurrences", "recurrence_exceptions",
        "habits", "habit_logs", "habit_pauses", "countdowns", "memos", "attachments",
    }
    attachment_row = json.loads(entries["data/attachments.json"])[0]
    assert entries[attachment_row["archive_path"]] == attachment_bytes
    assert "password_hash" not in exported.content.decode("utf-8", errors="ignore")
    assert "token_hash" not in exported.content.decode("utf-8", errors="ignore")

    merge_preflight = _preflight(client, exported.content)
    merged = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": merge_preflight.json()["preflight_token"], "mode": "merge"},
    )
    assert merged.status_code == 200, merged.text

    preflight = _preflight(client, exported.content, "replace")
    assert preflight.status_code == 200
    assert preflight.json()["valid"] is True
    restored = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": preflight.json()["preflight_token"], "mode": "replace"},
    )
    assert restored.status_code == 200, restored.text
    restored_task = client.get("/api/v1/tasks", params={"q": "完整备份任务"}).json()["items"][0]
    restored_attachment = client.get(f"/api/v1/tasks/{restored_task['id']}/attachments").json()[0]
    assert client.get(f"/api/v1/attachments/{restored_attachment['id']}").content == attachment_bytes

    async def assert_history():
        from backend.db import get_engine
        async with AsyncSession(get_engine()) as db:
            assert await db.scalar(select(func.count()).select_from(HabitLog)) == 1
            assert await db.scalar(select(func.count()).select_from(HabitPause)) == 1
            assert await db.scalar(select(func.count()).select_from(RecurrenceException)) == 1

    asyncio.run(assert_history())


def test_merge_same_backup_is_idempotent_via_import_ledger(client):
    inbox = boot(client)
    client.post("/api/v1/tasks", json={"title": "只导入一次", "list_id": inbox["id"]})
    content = client.get("/api/v1/backup/export.zip").content
    first_token = _preflight(client, content).json()["preflight_token"]
    second_token = _preflight(client, content).json()["preflight_token"]
    assert client.post("/api/v1/backup/restore", json={"preflight_token": first_token, "mode": "merge"}).status_code == 200
    second = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": second_token, "mode": "merge"},
    )
    assert second.status_code == 200
    assert second.json()["already_imported"] is True

    async def counts():
        from backend.db import get_engine
        async with AsyncSession(get_engine()) as db:
            return (
                await db.scalar(select(func.count()).select_from(Task).where(Task.title == "只导入一次")),
                await db.scalar(select(func.count()).select_from(BackupImport)),
            )

    assert asyncio.run(counts()) == (1, 1)


def test_invalid_zip_variants_are_rejected_before_any_write(client):
    inbox = boot(client)
    before = len(client.get("/api/v1/tasks", params={"list_id": inbox["id"]}).json()["items"])
    valid_data = json.dumps([]).encode()
    cases = []
    cases.append(_make_zip({"../escape": b"x", "data/tasks.json": valid_data}))

    duplicate = io.BytesIO()
    with zipfile.ZipFile(duplicate, "w") as archive:
        archive.writestr("manifest.json", "{}")
        archive.writestr("data/tasks.json", "[]")
        archive.writestr("data/tasks.json", "[]")
    cases.append(duplicate.getvalue())

    bad_checksum = _make_zip({"data/tasks.json": valid_data})
    entries = _zip_entries(bad_checksum)
    manifest = json.loads(entries["manifest.json"])
    manifest["checksums"]["data/tasks.json"] = "0" * 64
    cases.append(_make_zip({"data/tasks.json": valid_data}, backup_id=manifest["backup_id"]))
    # Replace the checksum after helper generation.
    output = io.BytesIO()
    with zipfile.ZipFile(output, "w") as archive:
        archive.writestr("manifest.json", json.dumps(manifest))
        archive.writestr("data/tasks.json", valid_data)
    cases[-1] = output.getvalue()

    dangling = {"data/tasks.json": json.dumps([{
        "id": "22222222-2222-4222-8222-222222222222", "list_id": "missing", "title": "bad"
    }]).encode()}
    cases.append(_make_zip(dangling))

    missing_attachment = {"data/attachments.json": json.dumps([{
        "id": "33333333-3333-4333-8333-333333333333",
        "task_id": "22222222-2222-4222-8222-222222222222",
        "filename": "x", "mime_type": "text/plain", "size": 1,
        "archive_path": "attachments/33333333-3333-4333-8333-333333333333/content",
    }]).encode()}
    cases.append(_make_zip(missing_attachment))

    for content in cases:
        response = _preflight(client, content)
        assert response.status_code == 422
        assert response.json()["detail"]["code"].startswith("backup_")
    after = len(client.get("/api/v1/tasks", params={"list_id": inbox["id"]}).json()["items"])
    assert after == before


def test_legacy_json_and_csv_restore_remain_supported(client):
    inbox = boot(client)
    client.post("/api/v1/tasks", json={"title": "legacy", "list_id": inbox["id"]})
    exported_json = client.get("/api/v1/export")
    exported_csv = client.get("/api/v1/export.csv")
    assert client.post("/api/v1/restore?mode=merge", json=exported_json.json()).status_code == 200
    assert client.post(
        "/api/v1/restore.csv?mode=merge",
        files={"file": ("backup.csv", exported_csv.content, "text/csv")},
    ).status_code == 200


def test_backup_routes_require_auth_csrf_bind_tokens_and_consume_once(client):
    boot(client)
    exported = client.get("/api/v1/backup/export.zip")
    token = _preflight(client, exported.content).json()["preflight_token"]

    anonymous = client.__class__(client.app)
    with anonymous:
        assert anonymous.get("/api/v1/backup/export.zip").status_code == 401
        assert _preflight(anonymous, exported.content).status_code == 401
        assert anonymous.post(
            "/api/v1/backup/restore", json={"preflight_token": token, "mode": "merge"}
        ).status_code == 401

    csrf = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": token, "mode": "merge"},
        headers={"origin": "https://dodo.example", "x-csrf-token": "wrong"},
    )
    assert csrf.status_code == 403

    async def add_other_user_session():
        from backend.db import get_engine
        async with AsyncSession(get_engine()) as db:
            other = User(username="other", password_hash="unused")
            db.add(other)
            await db.flush()
            session_token = "other-user-session-token"
            db.add(Session(
                token_hash=hash_token(session_token), user_id=other.id,
                expires_at=datetime(2099, 1, 1, tzinfo=UTC),
            ))
            await db.commit()
            return session_token

    other_session = asyncio.run(add_other_user_session())
    other = client.__class__(client.app)
    with other:
        other.cookies.set(COOKIE_NAME, other_session)
        wrong_user = other.post(
            "/api/v1/backup/restore", json={"preflight_token": token, "mode": "merge"}
        )
        assert wrong_user.status_code == 409

    restored = client.post(
        "/api/v1/backup/restore", json={"preflight_token": token, "mode": "merge"}
    )
    assert restored.status_code == 200
    reused = client.post(
        "/api/v1/backup/restore", json={"preflight_token": token, "mode": "merge"}
    )
    assert reused.status_code == 409


def test_preflight_binds_mode_and_persists_only_staged_metadata(client):
    boot(client)
    content = client.get("/api/v1/backup/export.zip").content
    token = _preflight(client, content, "replace").json()["preflight_token"]
    wrong_mode = client.post(
        "/api/v1/backup/restore", json={"preflight_token": token, "mode": "merge"}
    )
    assert wrong_mode.status_code == 409
    restored = client.post(
        "/api/v1/backup/restore", json={"preflight_token": token, "mode": "replace"}
    )
    assert restored.status_code == 200

    async def assert_persisted_consumption():
        from backend.db import get_engine
        from backend.models import BackupPreflight
        async with AsyncSession(get_engine()) as db:
            row = await db.scalar(select(BackupPreflight).where(BackupPreflight.token_hash.is_not(None)))
            assert row is not None
            assert row.consumed_at is not None
            assert row.archive_sha256 == hashlib.sha256(content).hexdigest()

    asyncio.run(assert_persisted_consumption())


def test_preflight_rejects_upload_over_compressed_limit_without_unbounded_read(client, monkeypatch):
    boot(client)
    import importlib
    router_module = importlib.import_module("backend.backup.router")
    monkeypatch.setattr(router_module, "MAX_ARCHIVE_BYTES", 32)
    response = _preflight(client, b"x" * 33)
    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "backup_size_invalid"


def test_quarantine_compensates_value_error_after_first_move(tmp_path, monkeypatch):
    from backend.backup import storage
    root = tmp_path / "attachments"
    root.mkdir()
    first = root / "first"
    first.write_bytes(b"first")
    original = storage.contained_file

    def fail_second(storage_root, name):
        if name == "bad":
            raise ValueError("bad path")
        return original(storage_root, name)

    monkeypatch.setattr(storage, "contained_file", fail_second)
    with pytest.raises(ValueError):
        storage.quarantine_files(root, ["first", "bad"], tmp_path / "quarantine")
    assert first.read_bytes() == b"first"


def test_preflight_rejects_unique_and_invalid_task_graph_constraints(client):
    boot(client)
    list_id = "11111111-aaaa-4111-8111-111111111111"
    parent_id = "22222222-aaaa-4222-8222-222222222222"
    child_id = "33333333-aaaa-4333-8333-333333333333"
    base_list = {"id": list_id, "name": "Inbox", "is_inbox": True, "position": 0}
    cases = [
        {"data/lists.json": json.dumps([base_list, {**base_list, "id": "44444444-aaaa-4444-8444-444444444444"}]).encode()},
        {"data/lists.json": json.dumps([base_list]).encode(), "data/tasks.json": json.dumps([
            {"id": parent_id, "list_id": list_id, "parent_id": child_id, "title": "p", "external_id": "same"},
            {"id": child_id, "list_id": list_id, "parent_id": parent_id, "title": "c", "external_id": "same"},
        ]).encode()},
    ]
    for entries in cases:
        response = _preflight(client, _make_zip(entries))
        assert response.status_code == 422
        assert response.json()["detail"]["code"] == "backup_constraint_invalid"


def test_preflight_rejects_per_user_pending_quota(client, monkeypatch):
    boot(client)
    import importlib
    router_module = importlib.import_module("backend.backup.router")
    monkeypatch.setattr(router_module, "MAX_PENDING_PREFLIGHTS_PER_USER", 1)
    content = client.get("/api/v1/backup/export.zip").content
    assert _preflight(client, content).status_code == 200
    response = _preflight(client, content)
    assert response.status_code == 429
    assert response.json()["detail"]["code"] == "backup_preflight_quota"


def test_replace_cleanup_failure_is_retryable_and_not_reported_as_success(client, tmp_path, monkeypatch):
    boot(client)
    from backend.backup import service
    from backend.config import get_settings
    get_settings().attachment_dir = str(tmp_path / "attachments")
    content = client.get("/api/v1/backup/export.zip").content
    token = _preflight(client, content, "replace").json()["preflight_token"]
    real_remove = service.remove_quarantine
    monkeypatch.setattr(service, "remove_quarantine", lambda _: (_ for _ in ()).throw(OSError("busy")))
    response = client.post("/api/v1/backup/restore", json={"preflight_token": token, "mode": "replace"})
    assert response.status_code == 500
    assert response.json()["detail"]["code"] == "backup_cleanup_pending"
    monkeypatch.setattr(service, "remove_quarantine", real_remove)
    retried = client.post("/api/v1/backup/restore", json={"preflight_token": token, "mode": "replace"})
    assert retried.status_code == 200
    assert retried.json()["cleanup_retried"] is True


def test_preflight_rejects_malformed_rows_without_writing_or_leaking_details(client):
    boot(client)
    malformed = _make_zip({
        "data/folders.json": json.dumps([{
            "id": "44444444-4444-4444-8444-444444444444",
            "name": "bad position",
            "position": "not-an-integer",
        }]).encode(),
    })
    preflight = _preflight(client, malformed)
    assert preflight.status_code == 422
    assert preflight.json()["detail"]["code"] == "backup_entity_invalid"


def test_preflight_rejects_invalid_habit_graph_without_any_write(client):
    boot(client)
    habit = client.post(
        "/api/v1/habits",
        json={"name": "基准", "kind": "numeric", "target": 2, "max_value": 4,
              "schedule_type": "weekly", "weekdays": [1, 3], "start_date": "2026-09-01"},
    ).json()
    content = client.get("/api/v1/backup/export.zip").content
    base_habit = _archive_rows(content, "habits")[0]
    base_log = {
        "id": "91919191-9191-4191-8191-919191919191", "habit_id": habit["id"],
        "day": "2026-09-01", "value": 2, "updated_at": "2026-09-01T08:00:00+00:00",
    }
    base_pause = {
        "id": "92929292-9292-4292-8292-929292929292", "habit_id": habit["id"],
        "start_date": "2026-09-10", "end_date": "2026-09-12",
    }
    invalid_graphs = [
        ({"habits": [{**base_habit, "kind": "counter"}]}, "kind enum"),
        ({"habits": [{**base_habit, "kind": "boolean", "target": 2, "max_value": 1}]}, "boolean target"),
        ({"habits": [{**base_habit, "target": float("inf")}]}, "finite target"),
        ({"habits": [{**base_habit, "target": 5, "max_value": 4}]}, "numeric range"),
        ({"habits": [{**base_habit, "schedule_type": "sometimes"}]}, "schedule enum"),
        ({"habits": [{**base_habit, "weekdays": "99"}]}, "weekday range"),
        ({"habits": [{**base_habit, "weekdays": "1,1"}]}, "weekday uniqueness"),
        ({"habits": [{**base_habit, "weekdays": "1, 3"}]}, "weekday storage"),
        ({"habits": [{**base_habit, "schedule_type": "monthly", "weekdays": None,
                       "month_days": "0", "interval_days": None}]}, "month range"),
        ({"habits": [{**base_habit, "schedule_type": "interval", "weekdays": None,
                       "month_days": None, "interval_days": 0}]}, "interval range"),
        ({"habits": [{**base_habit, "start_date": "2026-02-30"}]}, "start date"),
        ({"habit_logs": [{**base_log, "value": float("inf")}]}, "finite log"),
        ({"habits": [{**base_habit, "kind": "boolean", "target": 1, "max_value": 1}],
          "habit_logs": [{**base_log, "value": 2}]}, "boolean log"),
        ({"habit_logs": [{**base_log, "value": 5}]}, "numeric log max"),
        ({"habit_logs": [{**base_log, "value": -1}]}, "numeric log minimum"),
        ({"habit_pauses": [{**base_pause, "start_date": "2026-09-13"}]}, "pause order"),
        ({"habit_pauses": [base_pause, {
            **base_pause, "id": "93939393-9393-4393-8393-939393939393",
            "start_date": "2026-09-12", "end_date": "2026-09-14",
        }]}, "pause overlap"),
    ]

    async def counts():
        from backend.db import get_engine
        async with AsyncSession(get_engine()) as db:
            values = []
            for model in (Habit, HabitLog, HabitPause, BackupPreflight):
                values.append(await db.scalar(select(func.count()).select_from(model)))
            return tuple(values)

    before = asyncio.run(counts())
    for replacements, label in invalid_graphs:
        response = _preflight(client, _replace_entities(content, replacements), "replace")
        assert response.status_code == 422, (label, response.text)
        assert response.json()["detail"]["code"] in {
            "backup_entity_invalid", "backup_habit_invalid",
        }, label
        assert asyncio.run(counts()) == before, label


def test_four_habit_schedules_and_history_round_trip(client):
    boot(client)
    definitions = [
        {"name": "每天", "kind": "boolean", "schedule_type": "daily"},
        {"name": "每周", "kind": "numeric", "target": 2, "max_value": 4,
         "schedule_type": "weekly", "weekdays": [1, 3]},
        {"name": "每月", "kind": "numeric", "target": 3, "max_value": 5,
         "schedule_type": "monthly", "month_days": [1, 15, 31]},
        {"name": "间隔", "kind": "numeric", "target": 1.5, "max_value": 2.5,
         "schedule_type": "interval", "interval_days": 3},
    ]
    habits = [client.post("/api/v1/habits", json={**item, "start_date": "2026-09-01"}).json()
              for item in definitions]

    async def add_history():
        from backend.db import get_engine
        async with AsyncSession(get_engine()) as db:
            for index, habit in enumerate(habits):
                db.add(HabitLog(
                    habit_id=UUID(habit["id"]), day=date(2026, 9, index + 1),
                    value=1 if habit["kind"] == "boolean" else habit["target"],
                ))
            db.add_all([
                HabitPause(habit_id=UUID(habits[1]["id"]), start_date=date(2026, 9, 20),
                           end_date=date(2026, 9, 21)),
                HabitPause(habit_id=UUID(habits[1]["id"]), start_date=date(2026, 9, 23),
                           end_date=date(2026, 9, 24)),
            ])
            row = await db.get(Habit, UUID(habits[3]["id"]))
            row.archived_at = datetime(2026, 9, 30, tzinfo=UTC)
            await db.commit()

    asyncio.run(add_history())
    content = client.get("/api/v1/backup/export.zip").content
    expected_habits = _archive_rows(content, "habits")
    expected_logs = _archive_rows(content, "habit_logs")
    expected_pauses = _archive_rows(content, "habit_pauses")
    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
    after = client.get("/api/v1/backup/export.zip").content

    def stable(rows):
        return sorted(
            [{key: value for key, value in row.items() if key not in {"created_at", "updated_at"}}
             for row in rows], key=lambda row: row["id"]
        )

    assert stable(_archive_rows(after, "habits")) == stable(expected_habits)
    assert stable(_archive_rows(after, "habit_logs")) == stable(expected_logs)
    assert stable(_archive_rows(after, "habit_pauses")) == stable(expected_pauses)


def test_merge_rejects_backup_id_reuse_with_different_archive(client):
    boot(client)
    backup_id = "55555555-5555-4555-8555-555555555555"
    first = _make_zip({"data/folders.json": b"[]"}, backup_id=backup_id)
    first_token = _preflight(client, first).json()["preflight_token"]
    assert client.post(
        "/api/v1/backup/restore", json={"preflight_token": first_token, "mode": "merge"}
    ).status_code == 200

    changed = _make_zip({"data/folders.json": json.dumps([{
        "id": "66666666-6666-4666-8666-666666666666", "name": "different", "position": 0,
    }]).encode()}, backup_id=backup_id)
    second_token = _preflight(client, changed).json()["preflight_token"]
    second = client.post(
        "/api/v1/backup/restore", json={"preflight_token": second_token, "mode": "merge"}
    )
    assert second.status_code == 409
    assert second.json()["detail"]["code"] == "backup_id_conflict"


def test_replace_rebuilds_backup_identity_ledger_after_prior_merge(client):
    boot(client)
    backup_id = "77777777-7777-4777-8777-777777777777"
    original = _make_zip({"data/folders.json": b"[]"}, backup_id=backup_id)

    merge_token = _preflight(client, original, "merge").json()["preflight_token"]
    assert client.post(
        "/api/v1/backup/restore", json={"preflight_token": merge_token, "mode": "merge"}
    ).status_code == 200
    replace_token = _preflight(client, original, "replace").json()["preflight_token"]
    assert client.post(
        "/api/v1/backup/restore", json={"preflight_token": replace_token, "mode": "replace"}
    ).status_code == 200

    changed = _make_zip({"data/folders.json": json.dumps([{
        "id": "88888888-8888-4888-8888-888888888888", "name": "new entity", "position": 0,
    }]).encode()}, backup_id=backup_id)
    changed_token = _preflight(client, changed, "merge").json()["preflight_token"]
    conflict = client.post(
        "/api/v1/backup/restore", json={"preflight_token": changed_token, "mode": "merge"}
    )

    assert conflict.status_code == 409
    assert conflict.json()["detail"]["code"] == "backup_id_conflict"


def test_restore_cleans_parsed_staging_dir_when_preflight_identity_changed(client, monkeypatch):
    boot(client)
    content = client.get("/api/v1/backup/export.zip").content
    token = _preflight(client, content, "merge").json()["preflight_token"]
    import importlib

    router_module = importlib.import_module("backend.backup.router")
    real_parse = router_module.parse_archive_path
    parsed_dirs: list[Path] = []

    def parse_with_changed_identity(path, **kwargs):
        archive = real_parse(path, **kwargs)
        parsed_dirs.append(archive.staging_dir)
        return replace(archive, archive_sha256="0" * 64)

    monkeypatch.setattr(router_module, "parse_archive_path", parse_with_changed_identity)
    response = client.post(
        "/api/v1/backup/restore", json={"preflight_token": token, "mode": "merge"}
    )

    assert response.status_code == 409
    assert response.json()["detail"]["code"] == "backup_preflight_invalid"
    assert parsed_dirs
    assert not parsed_dirs[0].exists()


def test_restore_handles_child_before_parent_task_order(client):
    inbox = boot(client)
    parent = client.post(
        "/api/v1/tasks", json={"title": "parent", "list_id": inbox["id"]}
    ).json()
    client.post(
        "/api/v1/tasks",
        json={"title": "child", "list_id": inbox["id"], "parent_id": parent["id"]},
    )
    entries = _zip_entries(client.get("/api/v1/backup/export.zip").content)
    manifest = json.loads(entries.pop("manifest.json"))
    tasks = json.loads(entries["data/tasks.json"])
    entries["data/tasks.json"] = json.dumps(list(reversed(tasks))).encode()
    manifest["checksums"]["data/tasks.json"] = hashlib.sha256(entries["data/tasks.json"]).hexdigest()
    rebuilt = io.BytesIO()
    with zipfile.ZipFile(rebuilt, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("manifest.json", json.dumps(manifest))
        for name, value in entries.items():
            archive.writestr(name, value)

    token = _preflight(client, rebuilt.getvalue(), "replace").json()["preflight_token"]
    restored = client.post(
        "/api/v1/backup/restore", json={"preflight_token": token, "mode": "replace"}
    )
    assert restored.status_code == 200, restored.text
    items = client.get("/api/v1/tasks", params={"q": "parent"}).json()["items"]
    assert items[0]["subtasks"][0]["title"] == "child"


def test_replace_restores_quarantined_files_when_database_write_fails(client, tmp_path, monkeypatch):
    inbox = boot(client)
    from backend.backup import service
    from backend.config import get_settings

    root = tmp_path / "attachments"
    get_settings().attachment_dir = str(root)
    old_bytes = b"keep me"
    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", old_bytes, "text/plain")},
    ).json()

    async def old_storage_path():
        from backend.db import get_engine
        from backend.models import Attachment
        async with AsyncSession(get_engine()) as db:
            row = await db.get(Attachment, UUID(uploaded["id"]))
            return root / row.storage_name

    old_path = asyncio.run(old_storage_path())
    content = client.get("/api/v1/backup/export.zip").content
    token = _preflight(client, content, "replace").json()["preflight_token"]

    def fail_new_attachment_write(storage_root: Path, storage_name: str):
        raise OSError("simulated write failure")

    monkeypatch.setattr(service, "contained_file", fail_new_attachment_write)
    with pytest.raises(OSError, match="simulated write failure"):
        client.post(
            "/api/v1/backup/restore", json={"preflight_token": token, "mode": "replace"}
        )

    assert old_path.read_bytes() == old_bytes
    assert client.get("/api/v1/tasks", params={"q": "old"}).json()["items"]


def test_archive_blob_is_never_read_whole(client, tmp_path, monkeypatch):
    inbox = boot(client)
    from backend.config import get_settings

    get_settings().attachment_dir = str(tmp_path / "attachments")
    task = client.post("/api/v1/tasks", json={"title": "stream", "list_id": inbox["id"]}).json()
    blob = bytes(range(256)) * 64
    assert client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("blob.bin", blob, "application/zip")},
    ).status_code == 201
    content = client.get("/api/v1/backup/export.zip").content
    real_read = zipfile.ZipFile.read

    def reject_blob_read(self, name, *args, **kwargs):
        filename = name.filename if isinstance(name, zipfile.ZipInfo) else name
        if str(filename).startswith("attachments/"):
            raise AssertionError("blob entry was loaded with ZipFile.read")
        return real_read(self, name, *args, **kwargs)

    monkeypatch.setattr(zipfile.ZipFile, "read", reject_blob_read)
    preflight = _preflight(client, content)
    assert preflight.status_code == 200, preflight.text
    restored = client.post(
        "/api/v1/backup/restore",
        json={"preflight_token": preflight.json()["preflight_token"], "mode": "merge"},
    )
    assert restored.status_code == 200, restored.text


def test_pending_quota_reservation_is_atomic_across_workers(client, monkeypatch):
    boot(client)
    import importlib

    router_module = importlib.import_module("backend.backup.router")
    monkeypatch.setattr(router_module, "MAX_PENDING_PREFLIGHTS_PER_USER", 1)
    content = client.get("/api/v1/backup/export.zip").content

    def upload(_):
        with client.__class__(client.app) as worker:
            worker.cookies.update(client.cookies)
            return _preflight(worker, content).status_code

    with ThreadPoolExecutor(max_workers=2) as executor:
        statuses = list(executor.map(upload, range(2)))
    assert sorted(statuses) == [200, 429]


def test_entity_ledger_survives_partial_retry_and_normalizes_relationships(client):
    inbox = boot(client)
    task = client.post("/api/v1/tasks", json={"title": "ledger-parent", "list_id": inbox["id"]}).json()
    child = client.post(
        "/api/v1/tasks",
        json={"title": "ledger-child", "list_id": inbox["id"], "parent_id": task["id"]},
    ).json()
    content = client.get("/api/v1/backup/export.zip").content
    token = _preflight(client, content).json()["preflight_token"]
    restored = client.post("/api/v1/backup/restore", json={"preflight_token": token, "mode": "merge"})
    assert restored.status_code == 200, restored.text

    async def verify_ledger():
        from backend.db import get_engine
        from backend.models import BackupImportEntity
        async with AsyncSession(get_engine()) as db:
            rows = list((await db.scalars(select(BackupImportEntity).where(
                BackupImportEntity.entity_type == "tasks"
            ))).all())
            by_source = {str(row.source_id): row for row in rows}
            assert by_source[task["id"]].target_id
            assert by_source[child["id"]].target_id
            assert len(by_source[child["id"]].content_digest) == 64

    asyncio.run(verify_ledger())