"""Phase 3: On-disk refs as canonical branch pointers — TDD (RED → GREEN).

Seven tiers:
  Tier 1 — write_ref writes refs/heads/<branch> via atomic rename
  Tier 2 — read_ref round-trips what write_ref wrote
  Tier 3 — read_ref returns None for missing branches
  Tier 4 — write_ref creates parent dirs on first write
  Tier 5 — write_ref is truly atomic (tmp file disappears, final file appears)
  Tier 6 — wire_push_stream writes disk ref after DB commit
  Tier 7 — GET /repos/{repo_id}/branches/{name}/repair heals DB from disk ref
"""
from __future__ import annotations

import secrets
from collections.abc import AsyncGenerator
from pathlib import Path
from unittest.mock import patch

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.auth.request_signing import MSignContext, optional_signed_request, require_signed_request
from muse.core.types import long_id, now_utc_iso
from musehub.core.genesis import compute_identity_id
from musehub.db.musehub_models import MusehubIdentity, MusehubRepo
from musehub.main import app
from musehub.types.json_types import JSONObject, StrDict


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

def _oid() -> str:
    """Return a valid sha256-prefixed object ID (128 hex chars)."""
    return long_id(secrets.token_hex(32))


def _repo_root(tmp_path: Path, owner: str = "gabriel", slug: str = "test-repo") -> Path:
    """Create a minimal server-side repo tree and return its root."""
    root = tmp_path / owner / slug
    (root / "refs" / "heads").mkdir(parents=True, exist_ok=True)
    (root / "objects").mkdir(parents=True, exist_ok=True)
    return root


# ── fixtures for Tier 6/7 integration tests ───────────────────────────────────

_OWNER = "ref-test-user"
_SLUG = "ref-test-repo"
_IDENTITY_ID = compute_identity_id(b"ref-test-user")

_TEST_CONTEXT = MSignContext(
    handle=_OWNER,
    identity_id=_IDENTITY_ID,
    is_agent=False,
    is_admin=False,
)


@pytest_asyncio.fixture
async def async_session(db_session: AsyncSession) -> AsyncSession:
    """Alias: expose the conftest db_session as async_session."""
    return db_session


@pytest_asyncio.fixture
async def async_client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
    """Async HTTP client wired to the test app + DB."""
    from tests.conftest import _Asgi24Wrapper
    transport = ASGITransport(app=_Asgi24Wrapper(app))
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest_asyncio.fixture
async def owner(db_session: AsyncSession) -> str:
    """Create a test identity and return its handle."""
    identity = MusehubIdentity(
        identity_id=_IDENTITY_ID,
        handle=_OWNER,
        display_name="Ref Test User",
        identity_type="human",
    )
    db_session.add(identity)
    await db_session.commit()
    return _OWNER


@pytest_asyncio.fixture
async def slug() -> str:
    return _SLUG


@pytest_asyncio.fixture
async def repo_id(owner: str, db_session: AsyncSession) -> str:
    """Create a minimal test repo row and return its repo_id."""
    from datetime import datetime, timezone
    from musehub.core.genesis import compute_repo_id, compute_branch_id
    from musehub.db.musehub_models import MusehubBranch

    created_at = datetime.now(tz=timezone.utc)
    rid = compute_repo_id(_IDENTITY_ID, _SLUG, "code", created_at.isoformat())
    repo = MusehubRepo(
        repo_id=rid,
        owner=owner,
        slug=_SLUG,
        name="Ref Test Repo",
        owner_user_id=_IDENTITY_ID,
        visibility="public",
        default_branch="main",
        created_at=created_at,
    )
    db_session.add(repo)
    await db_session.commit()
    branch = MusehubBranch(
        branch_id=compute_branch_id(rid, "main"),
        repo_id=rid,
        name="main",
    )
    db_session.add(branch)
    await db_session.commit()
    return rid


@pytest.fixture
def authed_headers(owner: str) -> StrDict:
    """Inject auth context and return minimal JSON headers."""
    app.dependency_overrides[require_signed_request] = lambda: _TEST_CONTEXT
    app.dependency_overrides[optional_signed_request] = lambda: _TEST_CONTEXT
    yield {"Content-Type": "application/json"}
    app.dependency_overrides.pop(require_signed_request, None)
    app.dependency_overrides.pop(optional_signed_request, None)


# ── Tier 1: write_ref writes refs/heads/<branch> ─────────────────────────────

class TestWriteRefCreatesFile:
    def test_write_ref_creates_ref_file(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref
        from muse.core.paths import server_ref_path

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()

        write_ref(repo_root, "main", commit_id)

        ref_file = server_ref_path(repo_root, "main")
        assert ref_file.exists(), "ref file must exist after write_ref"

    def test_write_ref_content_is_commit_id_with_newline(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref
        from muse.core.paths import server_ref_path

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()

        write_ref(repo_root, "main", commit_id)

        ref_file = server_ref_path(repo_root, "main")
        assert ref_file.read_text() == f"{commit_id}\n"

    def test_write_ref_overwrites_existing(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref
        from muse.core.paths import server_ref_path

        repo_root = _repo_root(tmp_path)
        old_id = _oid()
        new_id = _oid()

        write_ref(repo_root, "main", old_id)
        write_ref(repo_root, "main", new_id)

        ref_file = server_ref_path(repo_root, "main")
        assert ref_file.read_text() == f"{new_id}\n"

    def test_write_ref_supports_feature_branches(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref
        from muse.core.paths import server_ref_path

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()

        write_ref(repo_root, "feat/new-melody", commit_id)

        ref_file = server_ref_path(repo_root, "feat/new-melody")
        assert ref_file.exists()
        assert ref_file.read_text() == f"{commit_id}\n"


# ── Tier 2: read_ref round-trips ─────────────────────────────────────────────

class TestReadRefRoundTrips:
    def test_read_ref_returns_commit_id(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref, read_ref

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()

        write_ref(repo_root, "main", commit_id)
        result = read_ref(repo_root, "main")

        assert result == commit_id

    def test_read_ref_strips_trailing_newline(self, tmp_path: Path) -> None:
        from musehub.storage.refs import read_ref
        from muse.core.paths import server_ref_path

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()
        ref_file = server_ref_path(repo_root, "main")
        ref_file.write_text(f"{commit_id}\n")

        result = read_ref(repo_root, "main")
        assert result == commit_id

    def test_read_ref_round_trips_feature_branch(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref, read_ref

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()

        write_ref(repo_root, "feat/jazz-changes", commit_id)
        result = read_ref(repo_root, "feat/jazz-changes")

        assert result == commit_id

    def test_read_ref_multiple_branches_independent(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref, read_ref

        repo_root = _repo_root(tmp_path)
        main_id = _oid()
        dev_id = _oid()

        write_ref(repo_root, "main", main_id)
        write_ref(repo_root, "dev", dev_id)

        assert read_ref(repo_root, "main") == main_id
        assert read_ref(repo_root, "dev") == dev_id


# ── Tier 3: read_ref returns None for missing ─────────────────────────────────

class TestReadRefMissing:
    def test_read_ref_returns_none_for_unknown_branch(self, tmp_path: Path) -> None:
        from musehub.storage.refs import read_ref

        repo_root = _repo_root(tmp_path)

        assert read_ref(repo_root, "nonexistent") is None

    def test_read_ref_returns_none_empty_repo(self, tmp_path: Path) -> None:
        from musehub.storage.refs import read_ref

        repo_root = _repo_root(tmp_path)

        assert read_ref(repo_root, "main") is None

    def test_read_ref_returns_none_after_branch_deleted_from_disk(
        self, tmp_path: Path
    ) -> None:
        from musehub.storage.refs import write_ref, read_ref
        from muse.core.paths import server_ref_path

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()
        write_ref(repo_root, "temp-branch", commit_id)

        server_ref_path(repo_root, "temp-branch").unlink()

        assert read_ref(repo_root, "temp-branch") is None


# ── Tier 4: write_ref creates parent dirs ────────────────────────────────────

class TestWriteRefCreatesParentDirs:
    def test_write_ref_creates_refs_heads_dir(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref

        # Start with only the objects dir — no refs/heads/ yet
        repo_root = tmp_path / "owner" / "repo"
        (repo_root / "objects").mkdir(parents=True)

        commit_id = _oid()
        write_ref(repo_root, "main", commit_id)

        assert (repo_root / "refs" / "heads" / "main").exists()

    def test_write_ref_creates_nested_branch_dirs(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()

        write_ref(repo_root, "feat/new/nested", commit_id)

        assert (repo_root / "refs" / "heads" / "feat" / "new" / "nested").exists()


# ── Tier 5: atomic write (tmp → rename) ──────────────────────────────────────

class TestWriteRefAtomic:
    def test_no_tmp_file_after_write(self, tmp_path: Path) -> None:
        from musehub.storage.refs import write_ref
        from muse.core.paths import server_ref_path

        repo_root = _repo_root(tmp_path)
        commit_id = _oid()

        write_ref(repo_root, "main", commit_id)

        ref_file = server_ref_path(repo_root, "main")
        tmp_file = ref_file.with_suffix(".tmp")
        assert not tmp_file.exists(), ".tmp sentinel must be gone after rename"

    def test_partial_write_does_not_corrupt_existing_ref(
        self, tmp_path: Path
    ) -> None:
        """A crash mid-write (simulated by leaving a .tmp) must not corrupt the ref."""
        from musehub.storage.refs import read_ref
        from muse.core.paths import server_ref_path

        repo_root = _repo_root(tmp_path)
        good_id = _oid()

        # Write a good ref first
        ref_file = server_ref_path(repo_root, "main")
        ref_file.write_text(f"{good_id}\n")

        # Simulate a crash: write a .tmp but never rename
        tmp_file = ref_file.with_suffix(".tmp")
        tmp_file.write_text(long_id("ff" * 32) + "\n")

        # The ref still reads the good value
        assert read_ref(repo_root, "main") == good_id


# ── Tier 6: wire_push_stream writes disk ref ─────────────────────────────────

import msgpack as _msgpack
from muse.core.types import blob_id
from muse.core.mpack import MuseWireFrameWriter as _MuseWireFrameWriter
from musehub.models.wire import (
    SFRAME_COMMIT_PACK,
    SFRAME_END,
    SFRAME_HEADER,
    SFRAME_RESULT,
)

_fw = _MuseWireFrameWriter()


def _mwp_wrap(ft: str, data: JSONObject) -> bytes:
    payload = _msgpack.packb(data, use_bin_type=True)
    return _fw.wrap(frame_type=ft, payload=payload)


def _mwp_header(branch: str = "main", n_commits: int = 1) -> bytes:
    snap_id = blob_id(b"default-snap")
    return _mwp_wrap(SFRAME_HEADER, {
        "t": SFRAME_HEADER, "branch": branch, "force": False,
        "have": [], "head": snap_id, "n_objects": 0, "n_commits": n_commits,
    })


def _mwp_commit_pack(commits: list[dict], snapshots: list[dict]) -> bytes:
    return _mwp_wrap(SFRAME_COMMIT_PACK, {
        "t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots,
    })


def _mwp_end(n_commits: int = 1) -> bytes:
    return _mwp_wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": 0, "n_commits": n_commits})


def _make_commit_wire(commit_id: str, snap_id: str, author: str = _OWNER) -> JSONObject:
    from datetime import datetime, timezone
    return {
        "commit_id": commit_id,
        "parent_ids": [],
        "parent_commit_id": None,
        "parent2_commit_id": None,
        "snapshot_id": snap_id,
        "branch": "main",
        "message": "disk ref test commit",
        "author": author,
        "committed_at": now_utc_iso(),
        "signature": "",
        "signer_key_id": "",
        "agent_id": "",
        "model_id": "",
        "metadata": {},
    }


def _make_snap_wire(snap_id: str) -> JSONObject:
    from datetime import datetime, timezone
    return {
        "snapshot_id": snap_id,
        "manifest": {},
        "committed_at": now_utc_iso(),
    }


async def _run_push(db_session: AsyncSession, repo_id: str, owner: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> str:
    """Run a minimal push via wire_push_stream and return the pushed commit_id."""
    from unittest.mock import AsyncMock
    from musehub.services.musehub_wire import wire_push_stream
    from musehub.config import settings

    # Stub the R2 backend
    store: dict[str, bytes] = {}
    backend = AsyncMock()
    backend.exists = AsyncMock(side_effect=lambda oid, **kw: oid in store)
    backend.put = AsyncMock(side_effect=lambda oid, data, **kw: store.update({oid: data}) or f"local://{oid}")
    backend.get = AsyncMock(side_effect=lambda oid, **kw: store.get(oid))
    monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend)

    # Redirect repos_dir to tmp_path so disk refs land there
    monkeypatch.setattr(settings, "musehub_repos_dir", str(tmp_path))

    commit_id = blob_id(f"ref-test-commit-{secrets.token_hex(8)}".encode())
    snap_id = blob_id(b"ref-test-snap")
    commit = _make_commit_wire(commit_id, snap_id, owner)
    snap = _make_snap_wire(snap_id)

    body_frames = _mwp_header() + _mwp_commit_pack([commit], [snap]) + _mwp_end()

    async def body_iter():
        yield body_frames

    results = []
    unpacker = _msgpack.Unpacker(raw=False)
    async for chunk in wire_push_stream(db_session, repo_id, body_iter(), owner):
        unpacker.feed(chunk)
    for frame in unpacker:
        results.append(frame)

    result_frames = [f for f in results if f.get("t") == SFRAME_RESULT]
    assert result_frames and result_frames[0]["ok"] is True, f"push failed: {results}"
    return commit_id


class TestWirePushWritesDiskRef:
    """Integration: after a push the ref file must exist on disk."""

    @pytest.mark.asyncio
    async def test_push_writes_disk_ref(
        self,
        tmp_path: Path,
        async_session: AsyncSession,
        repo_id: str,
        owner: str,
        slug: str,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """After a successful push, refs/heads/main must exist on disk."""
        from musehub.storage.refs import read_ref

        commit_id = await _run_push(async_session, repo_id, owner, monkeypatch, tmp_path)

        repo_root = tmp_path / owner / slug
        result = read_ref(repo_root, "main")
        assert result == commit_id

    @pytest.mark.asyncio
    async def test_push_disk_ref_matches_db_head(
        self,
        tmp_path: Path,
        async_session: AsyncSession,
        repo_id: str,
        owner: str,
        slug: str,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """The disk ref commit_id must equal the DB branch head commit_id."""
        from musehub.storage.refs import read_ref
        from musehub.services.musehub_repository import get_branch_head_commit_id

        commit_id = await _run_push(async_session, repo_id, owner, monkeypatch, tmp_path)

        repo_root = tmp_path / owner / slug
        disk_head = read_ref(repo_root, "main")
        db_head = await get_branch_head_commit_id(async_session, repo_id, "main")
        assert disk_head == commit_id
        assert db_head == commit_id


# ── Tier 7: repair endpoint ───────────────────────────────────────────────────

class TestRepairEndpoint:
    """GET /repos/{repo_id}/branches/{name}/repair must heal DB from disk ref."""

    @pytest.mark.asyncio
    async def test_repair_heals_db_from_disk(
        self,
        tmp_path: Path,
        async_client: AsyncClient,
        async_session: AsyncSession,
        authed_headers: StrDict,
        repo_id: str,
        owner: str,
        slug: str,
    ) -> None:
        """When disk ref != DB head, repair updates DB to match disk."""
        from musehub.storage.refs import write_ref
        from musehub.services.musehub_repository import get_branch_head_commit_id
        import musehub.db.musehub_models as models
        from sqlalchemy import select, update

        canonical_id = _oid()
        stale_id = _oid()

        with patch(
            "musehub.storage.backends.settings.musehub_repos_dir",
            str(tmp_path),
        ):
            repo_root = tmp_path / owner / slug
            # Write the canonical commit to disk
            write_ref(repo_root, "main", canonical_id)

            # Manually set a stale value in DB
            await async_session.execute(
                update(models.MusehubBranch)
                .where(
                    models.MusehubBranch.repo_id == repo_id,
                    models.MusehubBranch.name == "main",
                )
                .values(head_commit_id=stale_id)
            )
            await async_session.commit()

            response = await async_client.post(
                f"/api/repos/{repo_id}/branches/main/repair",
                headers=authed_headers,
            )
            assert response.status_code == 200
            body = response.json()
            assert body["healed"] is True
            assert body["commit_id"] == canonical_id

            # DB must now match disk
            db_head = await get_branch_head_commit_id(async_session, repo_id, "main")
            assert db_head == canonical_id

    @pytest.mark.asyncio
    async def test_repair_noop_when_already_consistent(
        self,
        tmp_path: Path,
        async_client: AsyncClient,
        async_session: AsyncSession,
        authed_headers: StrDict,
        repo_id: str,
        owner: str,
        slug: str,
    ) -> None:
        """When disk and DB agree, repair returns healed=False."""
        from musehub.storage.refs import write_ref
        import musehub.db.musehub_models as models
        from sqlalchemy import update

        commit_id = _oid()

        with patch(
            "musehub.storage.backends.settings.musehub_repos_dir",
            str(tmp_path),
        ):
            repo_root = tmp_path / owner / slug
            write_ref(repo_root, "main", commit_id)

            await async_session.execute(
                update(models.MusehubBranch)
                .where(
                    models.MusehubBranch.repo_id == repo_id,
                    models.MusehubBranch.name == "main",
                )
                .values(head_commit_id=commit_id)
            )
            await async_session.commit()

            response = await async_client.post(
                f"/api/repos/{repo_id}/branches/main/repair",
                headers=authed_headers,
            )
            assert response.status_code == 200
            body = response.json()
            assert body["healed"] is False

    @pytest.mark.asyncio
    async def test_repair_404_when_no_disk_ref(
        self,
        tmp_path: Path,
        async_client: AsyncClient,
        authed_headers: StrDict,
        repo_id: str,
    ) -> None:
        """Repair returns 404 when no disk ref exists to repair from."""
        with patch(
            "musehub.storage.backends.settings.musehub_repos_dir",
            str(tmp_path),
        ):
            response = await async_client.post(
                f"/api/repos/{repo_id}/branches/ghost/repair",
                headers=authed_headers,
            )
            assert response.status_code == 404

    @pytest.mark.asyncio
    async def test_repair_requires_auth(
        self,
        tmp_path: Path,
        async_client: AsyncClient,
        repo_id: str,
    ) -> None:
        """Repair endpoint must reject unauthenticated requests."""
        with patch(
            "musehub.storage.backends.settings.musehub_repos_dir",
            str(tmp_path),
        ):
            response = await async_client.post(
                f"/api/repos/{repo_id}/branches/main/repair",
            )
            assert response.status_code in (401, 403)