"""Tests for ``muse prune`` — surgical removal of unreachable objects.

Coverage tiers:
- Unit:        _collect_all_reachable_ids, _find_prune_candidates helpers
- Integration: dry-run lists candidates without deleting, prune removes
               unreachable objects, --json schema, object count decreases,
               reachable objects never deleted, --expire filters by age
- End-to-end:  full CLI via CliRunner
- Security:    only deletes under .muse/objects/; reachable objects safe;
               no mutation in --dry-run
- Stress:      200-object store with 50% unreachable
"""

from __future__ import annotations

import datetime
import hashlib
import json
import pathlib
import time

import pytest

from tests.cli_test_helper import CliRunner
from muse.core.object_store import write_object, has_object
from muse.core.snapshot import compute_commit_id, compute_snapshot_id
from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
from muse.core._types import Manifest

runner = CliRunner()

_REPO_ID = "prune-test"
_counter = 0


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _sha(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _init_repo(path: pathlib.Path) -> pathlib.Path:
    muse = path / ".muse"
    for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
        (muse / d).mkdir(parents=True, exist_ok=True)
    (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
    (muse / "repo.json").write_text(
        json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
    )
    return path


def _env(repo: pathlib.Path) -> dict[str, str]:
    return {"MUSE_REPO_ROOT": str(repo)}


def _commit_files(
    root: pathlib.Path,
    files: dict[str, bytes],
    branch: str = "main",
) -> str:
    global _counter
    _counter += 1
    manifest: Manifest = {}
    for rel_path, content in files.items():
        obj_id = _sha(content)
        write_object(root, obj_id, content)
        manifest[rel_path] = obj_id
        abs_path = root / rel_path
        abs_path.parent.mkdir(parents=True, exist_ok=True)
        abs_path.write_bytes(content)
    snap_id = compute_snapshot_id(manifest)
    write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
    committed_at = datetime.datetime.now(datetime.timezone.utc)
    ref_path = root / ".muse" / "refs" / "heads" / branch
    parent_id = ref_path.read_text(encoding="utf-8").strip() if ref_path.exists() else None
    parents = [parent_id] if parent_id else []
    commit_id = compute_commit_id(
        parents, snap_id, f"commit {_counter}", committed_at.isoformat()
    )
    write_commit(
        root,
        CommitRecord(
            commit_id=commit_id,
            repo_id=_REPO_ID,
            branch=branch,
            snapshot_id=snap_id,
            message=f"commit {_counter}",
            committed_at=committed_at,
            parent_commit_id=parent_id,
        ),
    )
    ref_path.write_text(commit_id, encoding="utf-8")
    return commit_id


def _invoke(repo: pathlib.Path, *args: str):
    from muse.cli.app import main as cli
    return runner.invoke(cli, ["prune", *args], env=_env(repo))


def _object_count(root: pathlib.Path) -> int:
    """Count loose objects directly in the store."""
    store = root / ".muse" / "objects"
    if not store.exists():
        return 0
    count = 0
    for shard in store.iterdir():
        if shard.is_dir():
            count += sum(1 for f in shard.iterdir() if f.is_file())
    return count


# ---------------------------------------------------------------------------
# Unit — _collect_all_reachable_ids
# ---------------------------------------------------------------------------


def test_collect_all_reachable_ids_empty_repo(tmp_path: pathlib.Path) -> None:
    from muse.cli.commands.prune import _collect_all_reachable_ids
    root = _init_repo(tmp_path)
    ids = _collect_all_reachable_ids(root)
    assert isinstance(ids, set)


def test_collect_all_reachable_ids_after_commit(tmp_path: pathlib.Path) -> None:
    from muse.cli.commands.prune import _collect_all_reachable_ids
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n", "b.py": b"# b\n"})
    ids = _collect_all_reachable_ids(root)
    assert _sha(b"# a\n") in ids
    assert _sha(b"# b\n") in ids


def test_collect_all_reachable_ids_excludes_orphan(tmp_path: pathlib.Path) -> None:
    from muse.cli.commands.prune import _collect_all_reachable_ids
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    orphan = b"orphan not in any snapshot"
    write_object(root, _sha(orphan), orphan)
    ids = _collect_all_reachable_ids(root)
    assert _sha(orphan) not in ids
    assert _sha(b"# a\n") in ids


# ---------------------------------------------------------------------------
# Unit — _find_prune_candidates
# ---------------------------------------------------------------------------


def test_find_prune_candidates_returns_orphan(tmp_path: pathlib.Path) -> None:
    from muse.cli.commands.prune import _find_prune_candidates, _collect_all_reachable_ids
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    orphan = b"i am orphaned"
    orphan_id = _sha(orphan)
    write_object(root, orphan_id, orphan)
    reachable = _collect_all_reachable_ids(root)
    candidates = _find_prune_candidates(root, reachable, expire_before=None)
    candidate_ids = {c["object_id"] for c in candidates}
    assert orphan_id in candidate_ids


def test_find_prune_candidates_excludes_reachable(tmp_path: pathlib.Path) -> None:
    from muse.cli.commands.prune import _find_prune_candidates, _collect_all_reachable_ids
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    reachable = _collect_all_reachable_ids(root)
    candidates = _find_prune_candidates(root, reachable, expire_before=None)
    candidate_ids = {c["object_id"] for c in candidates}
    assert _sha(b"# a\n") not in candidate_ids


def test_find_prune_candidates_expire_before_filters_recent(tmp_path: pathlib.Path) -> None:
    from muse.cli.commands.prune import _find_prune_candidates, _collect_all_reachable_ids
    root = _init_repo(tmp_path)
    orphan = b"recent orphan"
    orphan_id = _sha(orphan)
    write_object(root, orphan_id, orphan)
    reachable = set()
    # expire_before = 1 hour AGO — a just-written file is newer, so it should be skipped
    one_hour_ago = time.time() - 3600
    candidates = _find_prune_candidates(root, reachable, expire_before=one_hour_ago)
    candidate_ids = {c["object_id"] for c in candidates}
    assert orphan_id not in candidate_ids, "Recent orphan should be kept by --expire"


def test_find_prune_candidates_expire_old_objects(tmp_path: pathlib.Path) -> None:
    from muse.cli.commands.prune import _find_prune_candidates
    root = _init_repo(tmp_path)
    orphan = b"old orphan"
    orphan_id = _sha(orphan)
    write_object(root, orphan_id, orphan)
    # Backdate the file's mtime to 2 hours ago
    obj_path = next(
        (root / ".muse" / "objects").rglob(orphan_id[-62:]), None
    )
    if obj_path:
        two_hours_ago = time.time() - 7200
        import os
        os.utime(obj_path, (two_hours_ago, two_hours_ago))
    # expire_before = 1 hour ago — the file is older, so it IS a candidate
    one_hour_ago = time.time() - 3600
    candidates = _find_prune_candidates(root, set(), expire_before=one_hour_ago)
    candidate_ids = {c["object_id"] for c in candidates}
    assert orphan_id in candidate_ids


# ---------------------------------------------------------------------------
# Integration — --dry-run
# ---------------------------------------------------------------------------


def test_prune_dry_run_does_not_delete_objects(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    orphan = b"orphan to not delete"
    orphan_id = _sha(orphan)
    write_object(root, orphan_id, orphan)
    before = _object_count(root)
    result = _invoke(root, "--dry-run")
    assert result.exit_code == 0
    after = _object_count(root)
    assert after == before, "dry-run must not delete any objects"


def test_prune_dry_run_json_lists_candidates(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    orphan = b"orphan candidate"
    orphan_id = _sha(orphan)
    write_object(root, orphan_id, orphan)
    result = _invoke(root, "--dry-run", "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert "candidates" in data
    assert data["dry_run"] is True
    candidate_ids = [c["object_id"] for c in data["candidates"]]
    assert orphan_id in candidate_ids


def test_prune_dry_run_text_mentions_candidates(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    write_object(root, _sha(b"orphan x"), b"orphan x")
    result = _invoke(root, "--dry-run")
    assert result.exit_code == 0
    assert result.stdout.strip()


# ---------------------------------------------------------------------------
# Integration — actual pruning
# ---------------------------------------------------------------------------


def test_prune_removes_unreachable_objects(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    orphan = b"i am unreachable"
    orphan_id = _sha(orphan)
    write_object(root, orphan_id, orphan)
    assert has_object(root, orphan_id)
    result = _invoke(root)
    assert result.exit_code == 0
    assert not has_object(root, orphan_id), "Orphan blob must be deleted by prune"


def test_prune_keeps_reachable_objects(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n", "b.py": b"# b\n"})
    write_object(root, _sha(b"orphan"), b"orphan")
    result = _invoke(root)
    assert result.exit_code == 0
    assert has_object(root, _sha(b"# a\n")), "Reachable blob must survive prune"
    assert has_object(root, _sha(b"# b\n")), "Reachable blob must survive prune"


def test_prune_json_schema(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    write_object(root, _sha(b"orphan"), b"orphan")
    result = _invoke(root, "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert "pruned" in data
    assert "bytes_freed" in data
    assert "dry_run" in data
    assert data["dry_run"] is False


def test_prune_json_pruned_count(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    for i in range(3):
        write_object(root, _sha(f"orphan {i}".encode()), f"orphan {i}".encode())
    result = _invoke(root, "--json")
    data = json.loads(result.stdout)
    assert data["pruned"] >= 3


def test_prune_empty_repo_exits_zero(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    result = _invoke(root, "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["pruned"] == 0


def test_prune_no_orphans_exits_zero(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    result = _invoke(root, "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["pruned"] == 0


# ---------------------------------------------------------------------------
# Security — only deletes under .muse/objects/
# ---------------------------------------------------------------------------


def test_prune_does_not_touch_commits_or_snapshots(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    _commit_files(root, {"a.py": b"# a\n"})
    write_object(root, _sha(b"orphan"), b"orphan")
    commits_before = list((root / ".muse" / "commits").glob("*.msgpack"))
    snaps_before = list((root / ".muse" / "snapshots").glob("*.msgpack"))
    _invoke(root)
    commits_after = list((root / ".muse" / "commits").glob("*.msgpack"))
    snaps_after = list((root / ".muse" / "snapshots").glob("*.msgpack"))
    assert len(commits_before) == len(commits_after), "prune must not delete commits"
    assert len(snaps_before) == len(snaps_after), "prune must not delete snapshots"


# ---------------------------------------------------------------------------
# Stress
# ---------------------------------------------------------------------------


def test_prune_50_percent_unreachable(tmp_path: pathlib.Path) -> None:
    """200 objects: 100 reachable (committed), 100 orphaned. Prune removes exactly 100."""
    root = _init_repo(tmp_path)
    # 100 reachable
    files = {f"file_{i}.py": f"# {i}\n".encode() for i in range(100)}
    _commit_files(root, files)
    # 100 orphans
    for i in range(100):
        content = f"orphan blob {i:04d}".encode()
        write_object(root, _sha(content), content)
    result = _invoke(root, "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["pruned"] == 100