"""Phase 5: Pack file support + GC — TDD (RED → GREEN).

Seven tiers:
  Tier 1 — pack_loose_objects creates pack + index files in objects/pack/
  Tier 2 — packed objects are readable via read_packed_object
  Tier 3 — LocalBackend.get() falls through to pack when loose object removed
  Tier 4 — pack_loose_objects is idempotent: same objects → same pack hash
  Tier 5 — gc_packs deletes pack whose objects are all present in a newer pack
  Tier 6 — pack_loose_objects returns correct counts (packed, bytes_saved)
  Tier 7 — round-trip: write loose → pack → remove loose → still readable via backend
"""
from __future__ import annotations

import secrets
from pathlib import Path

import pytest
import msgpack

from muse.core.types import blob_id


# ── helpers ───────────────────────────────────────────────────────────────────

def _oid(content: bytes | None = None) -> str:
    if content is not None:
        return blob_id(content)
    return blob_id(secrets.token_bytes(16))


def _repo_root(tmp_path: Path, owner: str = "gabriel", slug: str = "pack-test") -> Path:
    root = tmp_path / owner / slug
    (root / "refs" / "heads").mkdir(parents=True, exist_ok=True)
    (root / "objects").mkdir(parents=True, exist_ok=True)
    return root


def _write_loose(repo_root: Path, object_id: str, data: bytes) -> Path:
    """Write a loose object file and return its path."""
    from muse.core.object_store import object_path
    from muse.core.paths import server_objects_dir
    p = object_path(repo_root, object_id, objects_base=server_objects_dir(repo_root))
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_bytes(data)
    return p


def _count_loose(repo_root: Path) -> int:
    """Count loose object files under objects/sha256/."""
    base = repo_root / "objects" / "sha256"
    if not base.exists():
        return 0
    return sum(1 for p in base.rglob("*") if p.is_file())


def _pack_dir(repo_root: Path) -> Path:
    return repo_root / "objects" / "pack"


# ── Tier 1: pack_loose_objects creates pack + index ───────────────────────────

class TestPackCreatesFiles:
    def test_creates_pack_dir(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        data = b"object content"
        oid = _oid(data)
        _write_loose(repo_root, oid, data)

        pack_loose_objects(repo_root)

        assert _pack_dir(repo_root).exists()

    def test_creates_pack_and_index_files(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        data = b"hello pack"
        oid = _oid(data)
        _write_loose(repo_root, oid, data)

        pack_loose_objects(repo_root)

        pack_files = list(_pack_dir(repo_root).glob("*.pack"))
        idx_files = list(_pack_dir(repo_root).glob("*.idx"))
        assert len(pack_files) == 1, "must create exactly one .pack file"
        assert len(idx_files) == 1, "must create exactly one .idx file"
        assert pack_files[0].stem == idx_files[0].stem, ".pack and .idx must share a basename"

    def test_noop_when_no_loose_objects(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        result = pack_loose_objects(repo_root)

        assert result.packed == 0
        assert not list(_pack_dir(repo_root).glob("*.pack")) if _pack_dir(repo_root).exists() else True

    def test_packs_all_loose_objects(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        objects = {f"obj-{i}".encode(): None for i in range(10)}
        oids = []
        for content in [f"data-{i}".encode() for i in range(10)]:
            oid = _oid(content)
            _write_loose(repo_root, oid, content)
            oids.append(oid)

        result = pack_loose_objects(repo_root)
        assert result.packed == 10


# ── Tier 2: read_packed_object reads from pack ────────────────────────────────

class TestReadPackedObject:
    def test_reads_object_from_pack(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects, read_packed_object

        repo_root = _repo_root(tmp_path)
        data = b"packed content here"
        oid = _oid(data)
        _write_loose(repo_root, oid, data)
        pack_loose_objects(repo_root)

        result = read_packed_object(repo_root, oid)
        assert result == data

    def test_returns_none_for_unknown_object(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects, read_packed_object

        repo_root = _repo_root(tmp_path)
        data = b"something"
        oid = _oid(data)
        _write_loose(repo_root, oid, data)
        pack_loose_objects(repo_root)

        assert read_packed_object(repo_root, _oid()) is None

    def test_reads_all_objects_from_pack(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects, read_packed_object

        repo_root = _repo_root(tmp_path)
        items = [(f"data-{i}".encode(),) for i in range(5)]
        oids = []
        for (content,) in items:
            oid = _oid(content)
            _write_loose(repo_root, oid, content)
            oids.append((oid, content))

        pack_loose_objects(repo_root)

        for oid, content in oids:
            assert read_packed_object(repo_root, oid) == content

    def test_reads_from_correct_pack_when_multiple_exist(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects, read_packed_object

        repo_root = _repo_root(tmp_path)

        # First pack: 3 objects
        first_oids = []
        for i in range(3):
            content = f"first-batch-{i}".encode()
            oid = _oid(content)
            _write_loose(repo_root, oid, content)
            first_oids.append((oid, content))
        pack_loose_objects(repo_root)

        # Second pack: 3 more objects
        second_oids = []
        for i in range(3):
            content = f"second-batch-{i}".encode()
            oid = _oid(content)
            _write_loose(repo_root, oid, content)
            second_oids.append((oid, content))
        pack_loose_objects(repo_root)

        for oid, content in first_oids + second_oids:
            assert read_packed_object(repo_root, oid) == content


# ── Tier 3: LocalBackend.get() falls through to pack ─────────────────────────

class TestLocalBackendPackFallthrough:
    @pytest.mark.asyncio
    async def test_get_finds_object_in_pack_after_loose_removed(
        self, tmp_path: Path
    ) -> None:
        from musehub.storage.backends import LocalBackend
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        data = b"i will be packed"
        oid = _oid(data)
        loose_path = _write_loose(repo_root, oid, data)

        pack_loose_objects(repo_root)
        loose_path.unlink()  # simulate GC of loose objects after packing

        backend = LocalBackend(repo_root=repo_root)
        result = await backend.get(oid, repo_root=repo_root)
        assert result == data

    @pytest.mark.asyncio
    async def test_exists_returns_true_for_packed_object(
        self, tmp_path: Path
    ) -> None:
        from musehub.storage.backends import LocalBackend
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        data = b"packed existence check"
        oid = _oid(data)
        loose_path = _write_loose(repo_root, oid, data)

        pack_loose_objects(repo_root)
        loose_path.unlink()

        backend = LocalBackend(repo_root=repo_root)
        assert await backend.exists(oid, repo_root=repo_root) is True

    @pytest.mark.asyncio
    async def test_get_returns_none_when_not_loose_or_packed(
        self, tmp_path: Path
    ) -> None:
        from musehub.storage.backends import LocalBackend

        repo_root = _repo_root(tmp_path)
        backend = LocalBackend(repo_root=repo_root)
        assert await backend.get(_oid(), repo_root=repo_root) is None

    @pytest.mark.asyncio
    async def test_loose_object_takes_precedence_over_pack(
        self, tmp_path: Path
    ) -> None:
        """Loose object must be returned before hitting pack (hot path)."""
        from musehub.storage.backends import LocalBackend
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        original_data = b"original loose content"
        oid = _oid(original_data)
        _write_loose(repo_root, oid, original_data)

        pack_loose_objects(repo_root)
        # Overwrite loose with different bytes (unusual but must take precedence)
        from muse.core.object_store import object_path
        from muse.core.paths import server_objects_dir
        p = object_path(repo_root, oid, objects_base=server_objects_dir(repo_root))
        p.write_bytes(original_data)  # same content in this case

        backend = LocalBackend(repo_root=repo_root)
        result = await backend.get(oid, repo_root=repo_root)
        assert result == original_data


# ── Tier 4: pack_loose_objects is idempotent ──────────────────────────────────

class TestPackIdempotent:
    def test_same_objects_produce_same_pack_name(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects

        repo_root_a = _repo_root(tmp_path, slug="repo-a")
        repo_root_b = _repo_root(tmp_path, slug="repo-b")

        contents = [f"shared-{i}".encode() for i in range(4)]
        for content in contents:
            oid = _oid(content)
            _write_loose(repo_root_a, oid, content)
            _write_loose(repo_root_b, oid, content)

        result_a = pack_loose_objects(repo_root_a)
        result_b = pack_loose_objects(repo_root_b)

        packs_a = list(_pack_dir(repo_root_a).glob("*.pack"))
        packs_b = list(_pack_dir(repo_root_b).glob("*.pack"))
        assert packs_a[0].name == packs_b[0].name

    def test_packing_already_packed_objects_skips_them(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        data = b"pack me once"
        oid = _oid(data)
        _write_loose(repo_root, oid, data)

        result1 = pack_loose_objects(repo_root)
        # Second call: no new loose objects → nothing to pack
        result2 = pack_loose_objects(repo_root)

        assert result1.packed == 1
        assert result2.packed == 0
        assert len(list(_pack_dir(repo_root).glob("*.pack"))) == 1


# ── Tier 5: gc_packs removes superseded packs ────────────────────────────────

class TestGCPacks:
    def test_gc_removes_pack_superseded_by_newer_pack(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects, gc_packs

        repo_root = _repo_root(tmp_path)

        # Pack 1: objects A, B
        for content in [b"obj-a", b"obj-b"]:
            _write_loose(repo_root, _oid(content), content)
        pack_loose_objects(repo_root)

        # Pack 2: objects A, B, C (superset)
        for content in [b"obj-a", b"obj-b", b"obj-c"]:
            _write_loose(repo_root, _oid(content), content)
        pack_loose_objects(repo_root)

        packs_before = len(list(_pack_dir(repo_root).glob("*.pack")))
        gc_result = gc_packs(repo_root)
        packs_after = len(list(_pack_dir(repo_root).glob("*.pack")))

        assert gc_result.packs_deleted >= 1
        assert packs_after < packs_before

    def test_gc_keeps_pack_with_unique_objects(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects, gc_packs

        repo_root = _repo_root(tmp_path)

        # Pack 1: object A only — prune loose A so it doesn't bleed into pack 2
        content_a = b"unique-obj-a"
        oid_a = _oid(content_a)
        loose_a = _write_loose(repo_root, oid_a, content_a)
        pack_loose_objects(repo_root)
        loose_a.unlink()  # prune: A now lives only in pack 1

        # Pack 2: object B only (no overlap with pack 1)
        content_b = b"unique-obj-b"
        _write_loose(repo_root, _oid(content_b), content_b)
        pack_loose_objects(repo_root)

        packs_before = len(list(_pack_dir(repo_root).glob("*.pack")))
        gc_result = gc_packs(repo_root)

        assert gc_result.packs_deleted == 0, "must not delete packs with unique objects"

    def test_gc_noop_with_single_pack(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects, gc_packs

        repo_root = _repo_root(tmp_path)
        data = b"solo pack"
        _write_loose(repo_root, _oid(data), data)
        pack_loose_objects(repo_root)

        result = gc_packs(repo_root)
        assert result.packs_deleted == 0


# ── Tier 6: PackResult counts are accurate ────────────────────────────────────

class TestPackResult:
    def test_packed_count_matches_loose_objects(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        n = 7
        for i in range(n):
            content = f"count-me-{i}".encode()
            _write_loose(repo_root, _oid(content), content)

        result = pack_loose_objects(repo_root)
        assert result.packed == n

    def test_bytes_packed_sums_object_sizes(self, tmp_path: Path) -> None:
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        contents = [b"x" * 100, b"y" * 200, b"z" * 300]
        total = sum(len(c) for c in contents)
        for content in contents:
            _write_loose(repo_root, _oid(content), content)

        result = pack_loose_objects(repo_root)
        assert result.bytes_packed == total


# ── Tier 7: full round-trip ───────────────────────────────────────────────────

class TestPackRoundTrip:
    @pytest.mark.asyncio
    async def test_write_pack_remove_loose_still_readable(
        self, tmp_path: Path
    ) -> None:
        from musehub.storage.backends import LocalBackend
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        backend = LocalBackend(repo_root=repo_root)

        # 1. Write 20 loose objects via backend
        objects = {}
        for i in range(20):
            content = f"round-trip-{i}-{'x' * 50}".encode()
            oid = _oid(content)
            await backend.put(oid, content, repo_root=repo_root)
            objects[oid] = content

        # 2. Pack them
        result = pack_loose_objects(repo_root)
        assert result.packed == 20

        # 3. Remove loose files (simulate post-pack GC)
        from muse.core.paths import server_objects_dir
        algo_dir = server_objects_dir(repo_root) / "sha256"
        for shard_dir in algo_dir.iterdir():
            for obj_file in shard_dir.iterdir():
                obj_file.unlink()

        # 4. All 20 objects must still be readable via backend
        for oid, content in objects.items():
            result_data = await backend.get(oid, repo_root=repo_root)
            assert result_data == content, f"object {oid[:20]}... not found after packing"

    @pytest.mark.asyncio
    async def test_new_loose_objects_after_pack_are_readable(
        self, tmp_path: Path
    ) -> None:
        """Objects written after packing must be readable as loose files."""
        from musehub.storage.backends import LocalBackend
        from musehub.storage.pack import pack_loose_objects

        repo_root = _repo_root(tmp_path)
        backend = LocalBackend(repo_root=repo_root)

        # Pack first batch
        old_content = b"old-object"
        old_oid = _oid(old_content)
        await backend.put(old_oid, old_content, repo_root=repo_root)
        pack_loose_objects(repo_root)

        # Write new loose object AFTER packing
        new_content = b"new-object-after-pack"
        new_oid = _oid(new_content)
        await backend.put(new_oid, new_content, repo_root=repo_root)

        # Both must be readable
        assert await backend.get(new_oid, repo_root=repo_root) == new_content
        assert await backend.get(old_oid, repo_root=repo_root) == old_content