"""Phase 3 mpack unpack invariants — idempotency and cold-start safety.

wire_push_unpack_mpack is synchronous and inline (no background job for the
sync path itself; commit-graph indexing is a separate background job).
These tests verify the invariants that matter for the current design:

  1. Idempotency — calling unpack-mpack twice with the same mpack produces
     no duplicate rows and leaves the DB in the correct final state.
  2. Correctness — a single unpack-mpack call writes all commits, snapshots,
     and blobs with the right storage_uri and counts.
  3. Cold-start — mpack already in storage (e.g. after a crash before DB
     writes) is correctly handled by a fresh unpack-mpack call.
"""
from __future__ import annotations

import datetime
import pathlib

import pytest
import pytest_asyncio

from httpx import AsyncClient, ASGITransport
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
from musehub.db.musehub_repo_models import MusehubObject, MusehubCommit, MusehubSnapshot
from musehub.db.database import get_db
from musehub.main import app

from muse.core.object_store import write_object
from muse.core.mpack import build_mpack, build_wire_mpack
from muse.core.paths import muse_dir
from muse.core.snapshot import compute_commit_id, compute_snapshot_id
from muse.core.commits import CommitRecord, write_commit
from muse.core.refs import write_branch_ref
from muse.core.snapshots import SnapshotRecord, write_snapshot
from muse.core.types import blob_id
from musehub.types.json_types import JSONObject


_AUTH_CTX = MSignContext(
    handle="gabriel",
    identity_id="sha256:" + "0" * 64,
    is_agent=False,
    is_admin=True,
)

_N_FILES       = 8
_N_COMMITS     = 4
_FILES_CHANGED = 2
_BLOB_SIZE     = 128


# ── fixtures ────────────────────────────────────────────────────────────────

@pytest_asyncio.fixture()
async def client(db_session: AsyncSession) -> None:
    async def _override_get_db() -> None:
        yield db_session

    app.dependency_overrides[get_db] = _override_get_db
    app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX
    app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="https://localhost:1337",
    ) as c:
        yield c

    app.dependency_overrides.clear()


@pytest_asyncio.fixture()
async def repo(client: AsyncClient) -> None:
    resp = await client.post(
        "/api/repos",
        json={"owner": "gabriel", "name": "phase3-retry-test", "visibility": "public", "initialize": False},
    )
    assert resp.status_code in (200, 201), resp.text
    data = resp.json()
    yield data["slug"]
    await client.delete(f"/api/repos/{data['repoId']}")


def _make_repo(tmp: pathlib.Path) -> tuple[pathlib.Path, str, bytes, dict]:
    """Build a local repo and return (path, head_commit_id, wire_bytes, raw_mpack)."""
    tmp.mkdir(parents=True, exist_ok=True)
    dot = muse_dir(tmp)
    dot.mkdir()
    (dot / "repo.json").write_text('{"repo_id":"phase3-test","owner":"gabriel"}')
    for d in ("commits", "snapshots", "objects"):
        (dot / d).mkdir()
    (dot / "refs" / "heads").mkdir(parents=True)
    (dot / "HEAD").write_text("ref: refs/heads/main\n")
    (dot / "config.toml").write_text("")

    blob_ids: list[str] = []
    for i in range(_N_FILES):
        data = f"base-{i:04d}".encode() + b"x" * _BLOB_SIZE
        oid = blob_id(data)
        write_object(tmp, oid, data)
        blob_ids.append(oid)

    base_manifest = {f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)}
    parent = None
    tip = ""
    ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)

    for i in range(_N_COMMITS):
        manifest = dict(base_manifest)
        for j in range(_FILES_CHANGED):
            idx = (i * _FILES_CHANGED + j) % _N_FILES
            raw = f"c{i:04d}-f{j}".encode() + b"y" * _BLOB_SIZE
            oid = blob_id(raw)
            write_object(tmp, oid, raw)
            manifest[f"src/file_{idx:04d}.py"] = oid

        sid = compute_snapshot_id(manifest)
        write_snapshot(tmp, SnapshotRecord(snapshot_id=sid, manifest=manifest))
        msg = f"commit-{i:05d}"
        cid = compute_commit_id(
            parent_ids=[parent] if parent else [],
            snapshot_id=sid,
            message=msg,
            committed_at_iso=ts.isoformat(),
            author="gabriel",
        )
        write_commit(tmp, CommitRecord(
            commit_id=cid, branch="main",
            snapshot_id=sid, message=msg, committed_at=ts,
            parent_commit_id=parent, parent2_commit_id=None,
            author="gabriel", metadata={}, structured_delta=None,
            sem_ver_bump="none", breaking_changes=[],
            agent_id="", model_id="", toolchain_id="",
            prompt_hash="", signature="", signer_key_id="",
        ))
        parent = cid
        tip = cid
        ts += datetime.timedelta(seconds=60)

    write_branch_ref(tmp, "main", tip)
    raw_mpack = build_mpack(tmp, [tip], have=[])
    wire_bytes = build_wire_mpack(raw_mpack)
    return tmp, tip, wire_bytes, raw_mpack


async def _push_and_unpack(client: AsyncClient, repo_slug: str, wire_bytes: bytes, head: str) -> JSONObject:
    """Store mpack in MemoryBackend and call unpack-mpack. Returns the response dict.

    Uses backend.put_mpack() directly instead of the presign+httpx PUT pattern —
    the MemoryBackend patched by conftest has no real presigned URL infrastructure.
    wire_bytes must be MUSE binary format (starts with b"MUSE").
    mpack_key uses blob_id (sha256("blob <size>\\0<data>")) — same formula the
    server's integrity check uses at step 2.
    """
    import musehub.storage.backends as _backends_mod
    import msgpack as _msgpack
    from muse.core.types import blob_id as _blob_id

    mpack_key = _blob_id(wire_bytes)
    backend = _backends_mod.get_backend()
    await backend.put_mpack(mpack_key, wire_bytes)

    ur = await client.post(
        f"/gabriel/{repo_slug}/push/unpack-mpack",
        content=_msgpack.packb(
            {"mpack_key": mpack_key, "branch": "main", "head": head},
            use_bin_type=True,
        ),
        headers={"Content-Type": "application/x-msgpack"},
    )
    assert ur.status_code == 200, ur.text
    return ur.json()


# ── tests ────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_unpack_mpack_idempotent(
    client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
) -> None:
    """Calling unpack-mpack twice with the same mpack produces no duplicate rows.

    wire_push_unpack_mpack uses ON CONFLICT DO NOTHING / ON CONFLICT DO UPDATE
    throughout, so a second identical push must leave the DB in the same state
    as a single push — no extra rows, no constraint violations.
    """
    _, head, wire_bytes, raw_mpack = _make_repo(tmp_path / "repo")

    result1 = await _push_and_unpack(client, repo, wire_bytes, head)
    assert result1.get("commits_written", 0) == _N_COMMITS, result1

    # Second push — must not raise and must not duplicate rows.
    result2 = await _push_and_unpack(client, repo, wire_bytes, head)
    assert result2.get("commits_written", 0) == 0, (
        "second push of same mpack must write 0 new commits (all already exist)"
    )

    all_oids = [obj["object_id"] for obj in (raw_mpack.get("blobs") or [])]
    rows = (await db_session.execute(
        select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
    )).scalars().all()
    assert len(rows) == len(all_oids), (
        f"expected exactly {len(all_oids)} object rows after double push, got {len(rows)}"
    )


@pytest.mark.asyncio
async def test_unpack_mpack_writes_all_entities(
    client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
) -> None:
    """A single unpack-mpack call correctly writes all commits, snapshots, and blobs.

    Verifies counts in the response and presence in the DB with correct storage_uri.
    """
    _, head, wire_bytes, raw_mpack = _make_repo(tmp_path / "repo")
    result = await _push_and_unpack(client, repo, wire_bytes, head)

    assert result.get("commits_written") == _N_COMMITS, (
        f"expected {_N_COMMITS} commits_written, got {result}"
    )
    assert result.get("snapshots_written") == _N_COMMITS, (
        f"expected {_N_COMMITS} snapshots_written (one per commit), got {result}"
    )
    n_blobs = len(raw_mpack.get("blobs") or [])
    assert result.get("blobs_written") == n_blobs, (
        f"expected {n_blobs} blobs_written, got {result}"
    )

    all_oids = [obj["object_id"] for obj in (raw_mpack.get("blobs") or [])]
    rows = (await db_session.execute(
        select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
    )).scalars().all()
    assert len(rows) == n_blobs, (
        f"expected {n_blobs} musehub_objects rows, got {len(rows)}"
    )
    # All objects must have a real storage_uri — nothing pending.
    bad = [r for r in rows if r.storage_uri in (None, "pending")]
    assert not bad, f"{len(bad)} objects still have storage_uri='pending'"


@pytest.mark.asyncio
async def test_unpack_mpack_cold_start(
    client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
) -> None:
    """unpack-mpack succeeds when mpack is already in storage with no prior DB record.

    Simulates the crash-then-retry path: mpack bytes landed in storage but the
    DB transaction was rolled back before unpack-mpack ran. A cold call to
    unpack-mpack must write all entities correctly.
    """
    import musehub.storage.backends as _backends_mod
    import msgpack as _msgpack

    _, head, wire_bytes, raw_mpack = _make_repo(tmp_path / "repo")

    # Store mpack directly — simulates completed PUT with no subsequent DB writes.
    from muse.core.types import blob_id as _blob_id
    mpack_key = _blob_id(wire_bytes)
    backend = _backends_mod.get_backend()
    await backend.put_mpack(mpack_key, wire_bytes)

    # Cold unpack-mpack — no prior presign row, no prior unpack attempt in DB.
    ur = await client.post(
        f"/gabriel/{repo}/push/unpack-mpack",
        content=_msgpack.packb(
            {"mpack_key": mpack_key, "branch": "main", "head": head},
            use_bin_type=True,
        ),
        headers={"Content-Type": "application/x-msgpack"},
    )
    assert ur.status_code == 200, ur.text
    result = ur.json()

    assert result.get("commits_written") == _N_COMMITS, (
        f"expected {_N_COMMITS} commits after cold start, got {result}"
    )

    n_blobs = len(raw_mpack.get("blobs") or [])
    all_oids = [obj["object_id"] for obj in (raw_mpack.get("blobs") or [])]
    rows = (await db_session.execute(
        select(MusehubObject).where(MusehubObject.object_id.in_(all_oids))
    )).scalars().all()
    assert len(rows) == n_blobs, (
        f"expected {n_blobs} objects after cold start, got {len(rows)}"
    )