"""Wire protocol endpoints — Muse CLI push/fetch transport.

URL pattern mirrors Git's Smart HTTP protocol:

    muse remote add origin https://musehub.ai/gabriel/muse

Active endpoints:

    GET  /{owner}/{slug}/refs                — branch heads + domain metadata (pre-flight)
    POST /{owner}/{slug}/push/mpack-presign  — get presigned PUT URL for whole mpack
    POST /{owner}/{slug}/push/unpack-mpack   — server indexes mpack from R2
    POST /{owner}/{slug}/fetch/mpack         — server builds and returns fetch mpack
    POST /{owner}/{slug}/fetch/presign       — presigned GET URLs for large fetches

These routes MUST be registered before the wildcard UI router in main.py
(/{owner}/{repo_slug}/...) so FastAPI matches the concrete third-segment
paths first.
"""

import json
import logging

import msgpack
import pydantic
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status
from fastapi.responses import Response
from sqlalchemy.ext.asyncio import AsyncSession

from sqlalchemy import select
from musehub.auth.dependencies import optional_token, require_valid_token, TokenClaims
from musehub.config import get_settings
from musehub.db.database import get_db as get_session
from musehub.db.musehub_abuse_models import MusehubDailyPushBytes
from musehub.db.musehub_collaborator_models import MusehubCollaborator
from musehub.db.musehub_repo_models import MusehubBranch, MusehubObject, MusehubRepo
from musehub.models.wire import WireFetchRequest
from musehub.types.json_types import JSONObject

from musehub.api.validation import BranchQueryParam, SlugParam
from musehub.rate_limits import limiter, WIRE_PUSH_LIMIT, WIRE_FETCH_LIMIT, OBJECT_LIMIT, REPAIR_LIMIT
from musehub.services.musehub_repository import get_repo_row_by_owner_slug
from musehub.services.musehub_wire import (
    FetchCommitNotFound,
    FetchNotIndexedError,
    MPackNotReadyError,
    FetchNotReady,
    MPackValidationError,
    ObjectHashMismatch,
    record_mpack_bytes_uploaded,
    wire_fetch,
    wire_fetch_mpack,
    wire_fetch_objects,
    wire_push_mpack_presign,
    wire_push_unpack_mpack,
    wire_refs,
    wire_repair_commit,
    wire_repair_object,
    wire_repair_snapshot,
)
from musehub.storage import get_backend

logger = logging.getLogger(__name__)

router = APIRouter(tags=["Wire Protocol"])

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

async def _resolve_repo(
    session: AsyncSession,
    owner: SlugParam,
    slug: SlugParam,
) -> MusehubRepo:
    """Resolve owner/slug → repo row or raise 404."""
    repo = await get_repo_row_by_owner_slug(session, owner, slug)
    if repo is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"repo '{owner}/{slug}' not found",
        )
    return repo

async def _resolve_repo_id(
    session: AsyncSession,
    owner: SlugParam,
    slug: SlugParam,
) -> str:
    """Resolve owner/slug → repo_id or raise 404.

    Used only by endpoints that perform their own write-authorization check
    against ``repo_id`` afterward (repair-object/snapshot/commit) or that are
    read-only. Write endpoints that need ``_assert_writable`` should call
    ``_resolve_repo`` instead, since that check needs the full repo row.
    """
    return (await _resolve_repo(session, owner, slug)).repo_id

async def _assert_readable(
    repo: MusehubRepo,
    claims: TokenClaims | None,
    session: AsyncSession,
) -> None:
    """Raise 404 if *repo* is private and the caller is not the owner or a collaborator.

    Returns 404 (not 403) to avoid leaking that the repo exists.
    """
    if repo.visibility == "public":
        return
    caller_handle: str | None = claims.handle if claims else None
    if caller_handle == repo.owner:
        return
    # Check collaborators with at least read permission
    if caller_handle:
        collab_row = (await session.execute(
            select(MusehubCollaborator).where(
                MusehubCollaborator.repo_id == repo.repo_id,
                MusehubCollaborator.identity_handle == caller_handle,
                MusehubCollaborator.accepted_at.isnot(None),
            )
        )).scalar_one_or_none()
        if collab_row is not None:
            return
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="repo not found",
    )

async def _assert_writable(
    repo: MusehubRepo,
    claims: TokenClaims,
    session: AsyncSession,
) -> None:
    """Raise 403 unless the caller is the repo owner or an accepted write/admin collaborator.

    Unlike ``_assert_readable``, visibility never widens write access — a
    public repo still requires owner or write/admin collaborator status to
    push to it directly. See musehub issue #131.
    """
    caller_handle: str = claims.handle
    if caller_handle == repo.owner:
        return
    collab_row = (await session.execute(
        select(MusehubCollaborator).where(
            MusehubCollaborator.repo_id == repo.repo_id,
            MusehubCollaborator.identity_handle == caller_handle,
            MusehubCollaborator.accepted_at.isnot(None),
            MusehubCollaborator.permission.in_(["write", "admin"]),
        )
    )).scalar_one_or_none()
    if collab_row is None:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="only the repo owner or a write/admin collaborator may write to this repo",
        )

# ── Wire helpers ─────────────────────────────────────────────────────────────

def _mpack_response(data: JSONObject, request: Request) -> Response:
    """Encode *data* as msgpack based on the client's Accept header.

    Clients send ``Accept: application/x-msgpack`` and always receive
    binary msgpack.  The dict may contain ``bytes`` values (e.g. object
    content) which msgpack handles natively.
    """
    accept = request.headers.get("accept", "")
    if "application/x-msgpack" in accept:
        return Response(
            content=msgpack.packb(data, use_bin_type=True),
            media_type="application/x-msgpack",
        )
    return Response(content=json.dumps(data), media_type="application/json")

def _decode_request_body(raw: bytes, content_type: str) -> JSONObject:
    """Decode an HTTP request body from msgpack or JSON.

    Clients send ``Content-Type: application/x-msgpack``; JSON is also
    accepted as a fallback for compatibility.
    """
    if "application/x-msgpack" in content_type or "application/x-muse-mpack" in content_type:
        decoded = msgpack.unpackb(raw, raw=False)
        if not isinstance(decoded, dict):
            raise ValueError("msgpack body must be a mapping")
        return dict(decoded)
    parsed = json.loads(raw)
    if not isinstance(parsed, dict):
        raise ValueError("JSON body must be a mapping")
    return dict(parsed)

# ── wire endpoints ─────────────────────────────────────────────────────────────

@router.get(
    "/{owner}/{slug}/refs",
    summary="Get branch heads (muse pull / muse push pre-flight)",
    response_description="Repo metadata and current branch heads",
)
@limiter.limit(WIRE_FETCH_LIMIT)
async def get_refs(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    _claims: TokenClaims | None = Depends(optional_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Return branch heads and domain metadata for a repo.

    Called by ``muse push`` and ``muse pull`` as a pre-flight to determine
    what the remote already has.  Equivalent to Git's:
    ``GET /owner/repo/info/refs?service=git-upload-pack``

    Private repos are only visible to their owner — unauthenticated callers
    receive a 404 (same response as a non-existent repo, to avoid leaking
    the existence of private repos).

    Response:
    ```json
    {
      "repo_id": "...",
      "domain": "code",
      "default_branch": "main",
      "branch_heads": {"main": "sha...", "dev": "sha..."}
    }
    ```
    """
    repo = await _resolve_repo(session, owner, slug)
    await _assert_readable(repo, _claims, session)
    result = await wire_refs(session, repo.repo_id)
    if result is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repo not found")
    return _mpack_response(result.model_dump(), request)

@router.post(
    "/{owner}/{slug}/push/mpack-presign",
    summary="Return one presigned PUT URL for a whole mpack",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(OBJECT_LIMIT)
async def push_mpack_presign(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Return one presigned PUT URL so the client can upload the entire mpack.

    Request body (msgpack):
        mpack_key  str  — sha256:<hex> of the mpack bytes
        size_bytes  int  — advisory byte count

    Response (msgpack):
        upload_url  str  — presigned PUT URL valid for 1 hour
        mpack_key  str  — echoed back for the client to pass to unpack-mpack
    """
    repo = await _resolve_repo(session, owner, slug)
    await _assert_writable(repo, claims, session)
    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)
    mpack_key = str(data.get("mpack_key", "") or "")
    size_bytes = int(data.get("size_bytes", 0))
    logger.warning("[mpack-presign] received mpack_key=%s  size_bytes=%d  content_type=%r", mpack_key, size_bytes, ct)
    if not mpack_key:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="mpack-presign requires mpack_key",
        )
    # mpack_key must be "sha256:<64-hex>" — anything else will never match
    # what the client PUT to MinIO and will fail integrity check in unpack-mpack.
    if not mpack_key.startswith("sha256:") or len(mpack_key) != 7 + 64:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="mpack_key must be 'sha256:<64-hex>'",
        )
    _settings = get_settings()
    if size_bytes > _settings.mpack_max_bytes:
        raise HTTPException(
            status_code=status.HTTP_413_CONTENT_TOO_LARGE,
            detail=(
                f"mpack size {size_bytes:,} bytes exceeds limit "
                f"{_settings.mpack_max_bytes:,} bytes"
            ),
        )
    # 4a — per-user daily byte limit
    if _settings.mpack_daily_upload_limit_bytes > 0:
        from sqlalchemy import select as _select, func as _func
        import datetime as _dt
        today = _dt.date.today()
        _result = await session.execute(
            _select(_func.coalesce(
                _func.sum(MusehubDailyPushBytes.bytes_uploaded), 0
            )).where(
                MusehubDailyPushBytes.identity_id == claims.identity_id,
                MusehubDailyPushBytes.date == today,
            )
        )
        daily_total = int(_result.scalar() or 0)
        if daily_total >= _settings.mpack_daily_upload_limit_bytes:
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail=(
                    f"daily upload limit of {_settings.mpack_daily_upload_limit_bytes:,} bytes reached; "
                    "try again tomorrow"
                ),
            )
        await record_mpack_bytes_uploaded(session, claims.identity_id, size_bytes)
        await session.commit()
    result = await wire_push_mpack_presign(mpack_key, size_bytes)
    return _mpack_response(result, request)


@router.post(
    "/{owner}/{slug}/push/unpack-mpack",
    summary="Read an mpack from storage, index all contents into PG",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_PUSH_LIMIT)
async def push_unpack_mpack(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Server reads a previously uploaded mpack from MinIO and indexes it.

    Request body (msgpack):
        mpack_key  str  — sha256:<hex> used when the client called mpack-presign

    Response (msgpack):
        commits_written    int
        snapshots_written  int
        blobs_written      int
    """
    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)
    mpack_key = data.get("mpack_key", "")
    logger.warning("[unpack-mpack] received mpack_key=%s  raw_body_len=%d  content_type=%r", mpack_key, len(raw), ct)
    if not mpack_key:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="unpack-pack requires mpack_key",
        )
    branch = str(data.get("branch") or "main")
    head_commit_id = str(data.get("head") or "")
    commits_count = int(data.get("commits_count") or 0)
    blobs_count = int(data.get("blobs_count") or 0)
    force = bool(data.get("force") or False)
    _settings = get_settings()
    if commits_count > _settings.mpack_max_commits:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=f"commits_count {commits_count:,} exceeds limit {_settings.mpack_max_commits:,}",
        )
    if blobs_count > _settings.mpack_max_objects:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=f"blobs_count {blobs_count:,} exceeds limit {_settings.mpack_max_objects:,}",
        )
    repo = await _resolve_repo(session, owner, slug)
    await _assert_writable(repo, claims, session)
    repo_id = repo.repo_id
    from musehub.services.musehub_wire import NonFastForwardError as _NonFastForwardError
    try:
        result = await wire_push_unpack_mpack(
            session, repo_id, mpack_key, claims.handle,
            branch=branch, head_commit_id=head_commit_id,
            commits_count=commits_count, blobs_count=blobs_count,
            force=force,
        )
    except _NonFastForwardError as exc:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(exc),
        )
    except ValueError as exc:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        )

    # Enqueue intel + file-last-commits + gc + profile.snapshot after every push.
    try:
        from musehub.services.musehub_jobs import enqueue_push_intel as _enqueue_push_intel
        repo_row = await session.get(MusehubRepo, repo_id)
        domain_id = repo_row.domain_id if repo_row else None
        head = result.get("head", head_commit_id)
        await _enqueue_push_intel(
            session,
            repo_id,
            head,
            domain_id=domain_id,
            branch=branch,
            owner=owner,
            mpack_key=mpack_key,
        )
        await session.commit()
    except Exception:
        logger.exception(
            "enqueue_push_intel failed for repo=%s — push succeeded, intel skipped",
            repo_id[:16],
        )

    return _mpack_response(result, request)


@router.post(
    "/{owner}/{slug}/repair-object",
    summary="Replace a corrupt stored object with verified correct bytes (owner/write only)",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(REPAIR_LIMIT)
async def repair_object(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Replace a stored object's bytes with correct content, verified by SHA-256.

    Intended for operators to repair objects that were stored with wrong bytes —
    e.g. objects produced by a failed delta reconstruction where the base was
    zlib-compressed and the delta result is garbage.

    Request body (msgpack):
        object_id  str   — full "sha256:<64-hex>" content ID
        content    bytes — the correct raw bytes for this object

    The endpoint verifies SHA-256(content) == object_id before writing.
    Only the repo owner or a write/admin collaborator may call this.
    """
    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)
    object_id: str = data.get("object_id", "")
    content: bytes = data.get("content", b"")
    if not object_id or not isinstance(content, bytes):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="repair-object requires object_id (str) and content (bytes)",
        )
    repo_id = await _resolve_repo_id(session, owner, slug)
    caller_id: str | None = claims.handle
    try:
        result = await wire_repair_object(session, repo_id, object_id, content, caller_id)
    except PermissionError as exc:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
    except ObjectHashMismatch as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc))
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    return _mpack_response(result, request)

@router.post(
    "/{owner}/{slug}/repair-snapshot",
    summary="Replace a corrupt snapshot manifest with verified correct content (owner/write only)",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(REPAIR_LIMIT)
async def repair_snapshot(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Replace a stored snapshot's manifest with correct content, verified by snapshot_id.

    Intended for operators to repair snapshots that were stored with an empty
    or corrupted manifest_blob — e.g. snapshots affected by the R2 empty-object
    bug.  Uses force-overwrite (unlike the push path which uses ON CONFLICT DO
    NOTHING), so an existing row with wrong content is corrected in place.

    Request body (msgpack):
        snapshot_id  str        — full "sha256:<64-hex>" snapshot ID
        manifest     dict[str,str] — {path: object_id} mapping
        directories  list[str]  — directory paths (may be empty)

    The endpoint recomputes compute_snapshot_id(manifest, directories) and
    verifies it matches snapshot_id before writing.
    Only the repo owner or a write/admin collaborator may call this.
    """
    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)
    snapshot_id: str = data.get("snapshot_id", "")
    manifest: JSONObject = data.get("manifest", {})
    directories: list[str] = data.get("directories", [])
    if not snapshot_id or not isinstance(manifest, dict):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="repair-snapshot requires snapshot_id (str) and manifest (dict)",
        )
    repo_id = await _resolve_repo_id(session, owner, slug)
    caller_id: str | None = claims.handle
    try:
        result = await wire_repair_snapshot(
            session, repo_id, snapshot_id, manifest, directories, caller_id
        )
    except PermissionError as exc:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
    except ObjectHashMismatch as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc))
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    return _mpack_response(result, request)

@router.post(
    "/{owner}/{slug}/repair-commit",
    summary="Replace a corrupt commit record with verified-correct content (owner/write only)",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(REPAIR_LIMIT)
async def repair_commit(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Replace a stored commit's identity fields with correct content, verified by commit_id.

    Intended for operators to repair commits whose stored row no longer reproduces its
    commit_id — e.g. commits the rc10 object-store migration stamped with a
    signer_public_key without recomputing the id, so the serve path's hash check fails
    on clone and the commit (and all descendants) is dropped.

    Request body (msgpack):
        commit  dict — a wire commit record (WireCommit shape). Its identity fields
                       (parent ids, snapshot_id, message, committed_at, author,
                       signer_public_key) must reproduce commit_id.

    The endpoint recomputes the commit identity (round-tripping committed_at exactly as
    the serve path does) and verifies it matches commit_id before writing.
    Only the repo owner or a write/admin collaborator may call this.
    """
    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)
    commit: JSONObject = data.get("commit", {})
    if not isinstance(commit, dict) or not commit.get("commit_id"):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="repair-commit requires commit (dict) with a commit_id",
        )
    repo_id = await _resolve_repo_id(session, owner, slug)
    caller_id: str | None = claims.handle
    try:
        result = await wire_repair_commit(session, repo_id, commit, caller_id)
    except PermissionError as exc:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
    except ObjectHashMismatch as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc))
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    return _mpack_response(result, request)

@router.post(
    "/{owner}/{slug}/fetch/objects",
    summary="Fetch raw content for a list of object IDs",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(OBJECT_LIMIT)
async def fetch_objects(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    _claims: TokenClaims | None = Depends(optional_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Return raw bytes for each requested object ID as concatenated msgpack frames.

    Request body (msgpack):
        object_ids  list[str] — canonical sha256:<hex> IDs to fetch

    Response body: concatenated msgpack dicts, one per found object:
        {"object_id": str, "content": bytes}

    Objects not found are silently omitted.
    """
    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)
    object_ids = data.get("object_ids", [])
    if not isinstance(object_ids, list):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="fetch/objects requires object_ids (list)",
        )
    repo = await _resolve_repo(session, owner, slug)
    await _assert_readable(repo, _claims, session)
    objects = await wire_fetch_objects(session, repo.repo_id, [str(o) for o in object_ids])
    import msgpack as _mp
    body = b"".join(_mp.packb(obj, use_bin_type=True) for obj in objects)
    return Response(content=body, media_type="application/x-msgpack")

@router.post(
    "/{owner}/{slug}/fetch/presign",
    summary="Wire — presigned fetch for large repos",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_FETCH_LIMIT)
async def fetch_presign(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    _claims: TokenClaims | None = Depends(optional_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Return per-object presigned GET URLs for a large fetch delta.

    For large fetches (≥ 500 objects or ≥ 50 MB) Cloudflare's origin timeout
    kills the response before the client receives everything.  This
    endpoint generates one presigned R2 GET URL per needed object — zero object
    bytes are read server-side.  The client downloads all objects in parallel
    directly from R2, bypassing Cloudflare entirely.

    **Request body** (``Content-Type: application/x-msgpack``):
        msgpack ``{"want": [str], "have": [str], "depth": int|null, "ttl_seconds": int}``

    **Response body** (``Content-Type: application/x-msgpack``):
        msgpack ``{"presign": bool, "blob_urls": {oid: url}, "commits": [...],
                   "snapshots": [...], "branch_heads": {...}, "repo_id": str,
                   "domain": str, "default_branch": str,
                   "expires_at": str|null, "commit_count": int, "blob_count": int}``

    When ``presign=False`` the delta is below the threshold or the backend does
    not support presigned URLs — the client should fall back to ``fetch/mpack``.
    """
    from musehub.services.musehub_wire import wire_fetch_presign
    raw = await request.body()
    data = _decode_request_body(raw, "application/x-msgpack")
    ttl_seconds = int(data.pop("ttl_seconds", 3600))
    body = WireFetchRequest.model_validate(data)
    repo = await _resolve_repo(session, owner, slug)
    await _assert_readable(repo, _claims, session)
    result = await wire_fetch_presign(session, repo.repo_id, body, ttl_seconds)
    return _mpack_response(result, request)


@router.post(
    "/{owner}/{slug}/fetch",
    summary="Wire — fetch protocol step 1 (issue #68)",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_FETCH_LIMIT)
async def fetch(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    _claims: TokenClaims | None = Depends(optional_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Compute the fetch delta and return a presigned GET URL for the mpack.

    **Request body** (``Content-Type: application/x-msgpack``):
        ``{"want": [sha256:...], "have": [sha256:...]}``

    **Response body** (``Content-Type: application/x-msgpack``):
        ``{"mpack_id": "sha256:...", "mpack_url": "...", "commit_count": N, "object_count": N}``

    Returns 422 if want is empty or entries are malformed.
    Returns 404 if any want commit_id is unknown.
    Returns 503 with Retry-After if needed objects are not yet indexed.
    """
    raw = await request.body()
    data = _decode_request_body(raw, "application/x-msgpack")
    want: list[str] = data.get("want") or []
    have: list[str] = data.get("have") or []

    if not want:
        raise HTTPException(status_code=422, detail="want must be non-empty")

    repo = await _resolve_repo(session, owner, slug)
    await _assert_readable(repo, _claims, session)

    try:
        result = await wire_fetch(session, repo.repo_id, want, have)
    except MPackValidationError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    except FetchCommitNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except FetchNotReady as exc:
        from fastapi.responses import Response as _Resp
        body = msgpack.packb({"error": str(exc), "retry_after": 60}, use_bin_type=True)
        return _Resp(
            content=body,
            status_code=503,
            headers={"Retry-After": "60", "Content-Type": "application/x-msgpack"},
        )

    return _mpack_response(result, request)


@router.post(
    "/{owner}/{slug}/fetch/mpack",
    summary="Wire — single-mpack fetch (issue #47)",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_FETCH_LIMIT)
async def fetch_mpack(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    _claims: TokenClaims | None = Depends(optional_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Return the fetch delta as a single content-addressed mpack.

    Phase 2 protocol: server assembles the fetch mpack, stores it ephemerally
    in MinIO, and returns a presigned GET URL.  The client GETs the mpack
    directly from MinIO, verifies sha256, then calls apply_mpack().

    **Request body** (``Content-Type: application/x-msgpack``):
        msgpack ``{"want": [str], "have": [str], "ttl_seconds": int}``

    **Response body** (``Content-Type: application/x-msgpack``):
        msgpack ``{"mpack_url": str|null, "mpack_id": str|null,
                   "commit_count": int, "blob_count": int}``

    Returns 503 with ``Retry-After: 60`` if needed objects are not yet indexed
    (mpack.index background job still running).
    """
    raw = await request.body()
    data = _decode_request_body(raw, "application/x-msgpack")
    ttl_seconds = int(data.pop("ttl_seconds", 3600))
    want: list[str] = data.get("want") or []
    have: list[str] = data.get("have") or []
    repo = await _resolve_repo(session, owner, slug)
    await _assert_readable(repo, _claims, session)
    try:
        result = await wire_fetch_mpack(session, repo.repo_id, want, have, ttl_seconds)
    except MPackNotReadyError:
        return Response(
            content="mpack not ready — prebuild in progress. Retry shortly.",
            status_code=503,
            headers={"Retry-After": "30"},
        )
    except FetchNotIndexedError as exc:
        return Response(
            content=f"Objects not yet indexed: {exc.missing_count} missing. Retry shortly.",
            status_code=503,
            headers={"Retry-After": "60"},
        )
    return _mpack_response(result, request)


# ── release wire endpoints ─────────────────────────────────────────────────────

@router.post(
    "/{owner}/{slug}/releases",
    summary="Push a release from muse CLI",
    status_code=status.HTTP_201_CREATED,
)
@limiter.limit(WIRE_PUSH_LIMIT)
async def wire_create_release(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    background_tasks: BackgroundTasks,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Accept a ``ReleaseDict`` payload from ``muse release push``.

    The body must be a JSON object matching the CLI ``ReleaseRecord.to_dict()``
    shape (application/json).  Returns immediately with the server-assigned
    ``release_id``; semantic analysis runs as a background task so the push
    is never blocked by analysis time.

    Only the repo owner or a write/admin collaborator may push releases —
    the endpoint mirrors the auth model of ``POST /{owner}/{slug}/push``.
    """
    from musehub.services import musehub_releases as rel_svc
    from musehub.services.release_analysis import analyse_release_background

    repo = await _resolve_repo(session, owner, slug)
    await _assert_writable(repo, claims, session)

    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    data = _decode_request_body(raw, ct)

    repo_id = repo.repo_id

    try:
        response = await rel_svc.create_release_from_dict(session, repo_id, data)
    except ValueError as exc:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(exc),
        )

    await session.commit()
    logger.info("✅ wire: release %s pushed for %s/%s", response.tag, owner, slug)

    # Fire semantic analysis after the response is sent.  Opens its own
    # session so it is independent of the request lifecycle.
    background_tasks.add_task(
        analyse_release_background, repo_id, response.release_id
    )

    return _mpack_response({"release_id": response.release_id}, request)

@router.delete(
    "/{owner}/{slug}/branches/{branch_name:path}",
    summary="Delete a branch from MuseHub",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_PUSH_LIMIT)
async def wire_delete_branch(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    branch_name: str,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Delete a branch pushed by ``muse push``.

    Idiomatic equivalent of ``git push origin --delete <branch>``.  The branch
    ref is removed from MuseHub; commits and objects are unaffected.

    Only the repo owner may delete branches.  Attempting to delete the
    repository's default branch is rejected with 409.
    """
    from sqlalchemy import delete as sql_delete

    repo = await _resolve_repo(session, owner, slug)

    caller_id: str | None = claims.handle
    if caller_id != repo.owner:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="only the repo owner may delete branches",
        )

    default_branch: str = getattr(repo, "default_branch", None) or "main"
    if branch_name == default_branch:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"cannot delete the default branch '{default_branch}'",
        )

    result = await session.execute(
        sql_delete(MusehubBranch).where(
            MusehubBranch.repo_id == repo.repo_id,
            MusehubBranch.name == branch_name,
        ).returning(MusehubBranch.name)
    )
    deleted_name: str | None = result.scalar_one_or_none()
    if deleted_name is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"branch '{branch_name}' not found",
        )

    await session.commit()
    logger.info("✅ wire: branch %s deleted from %s/%s", branch_name, owner, slug)
    return _mpack_response({"deleted": branch_name}, request)

@router.delete(
    "/{owner}/{slug}/releases/{tag:path}",
    summary="Retract a release from MuseHub",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_PUSH_LIMIT)
async def wire_delete_release(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    tag: str,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Retract a release pushed by ``muse release push``.

    Removes the named release label from MuseHub.  The underlying commits and
    snapshots are not affected — they remain in the content-addressed object
    store and are still reachable by their SHA-256.

    Only the repo owner may retract releases.
    """
    from musehub.services import musehub_releases as rel_svc

    repo = await _resolve_repo(session, owner, slug)

    caller_id: str | None = claims.handle
    if caller_id != repo.owner:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="only the repo owner may retract releases",
        )

    deleted = await rel_svc.delete_release_by_tag(session, repo.repo_id, tag)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"release '{tag}' not found",
        )

    await session.commit()
    logger.info("✅ wire: release %s retracted from %s/%s", tag, owner, slug)
    return _mpack_response({"retracted": tag}, request)

# ── wire-tag endpoints ─────────────────────────────────────────────────────────

@router.post(
    "/{owner}/{slug}/tags",
    summary="Push lightweight wire tags from muse CLI",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_PUSH_LIMIT)
async def wire_push_tags(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Upsert a batch of lightweight semantic tags pushed from the Muse CLI.

    The body must be a msgpack object with a ``tags`` key holding a list of
    ``WireTag`` dicts.  The server upserts them — pushing the same tag label
    twice for the same repo is a no-op (``commit_id`` is refreshed).

    Returns ``{"stored": <count>}`` as msgpack or JSON.
    """
    from musehub.models.musehub import WireTagInput
    from musehub.services import musehub_wire_tags as tag_svc

    repo = await _resolve_repo(session, owner, slug)
    await _assert_writable(repo, claims, session)

    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    try:
        data = _decode_request_body(raw, ct)
    except (ValueError, Exception):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Request body must be a valid msgpack or JSON object",
        )

    repo_id = repo.repo_id

    tags_raw = data.get("tags", [])
    if not isinstance(tags_raw, list):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="'tags' must be a list",
        )

    tags: list[WireTagInput] = []
    for item in tags_raw:
        if not isinstance(item, dict):
            continue
        tag_id_raw = item.get("tag_id", "")
        commit_id_raw = item.get("commit_id", "")
        tag_label_raw = item.get("tag", "")
        created_at_raw = item.get("created_at", "")
        tags.append(
            WireTagInput(
                tag_id=str(tag_id_raw) if isinstance(tag_id_raw, str) else "",
                commit_id=str(commit_id_raw) if isinstance(commit_id_raw, str) else "",
                tag=str(tag_label_raw) if isinstance(tag_label_raw, str) else "",
                created_at=str(created_at_raw) if isinstance(created_at_raw, str) else "",
            )
        )

    stored = await tag_svc.store_wire_tags(session, repo_id, tags)
    await session.commit()
    logger.info("✅ wire: %d tag(s) pushed for %s/%s", stored, owner, slug)
    return _mpack_response({"stored": stored}, request)

# ── version-tag endpoints ──────────────────────────────────────────────────────

@router.post(
    "/{owner}/{slug}/version-tags",
    summary="Push semantic-version tags from muse CLI",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_PUSH_LIMIT)
async def wire_push_version_tags(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    force: bool = False,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Upsert a batch of semantic-version tags pushed from the Muse CLI.

    The body must be a msgpack object with a ``tags`` key holding a list of
    version-tag dicts.  Re-pushing the same tag label without ``force=true``
    is a no-op if the commit_id is unchanged (skipped).  With ``force=true``
    the commit_id is overwritten.

    Returns ``{"stored": <count>, "skipped": <count>}`` as msgpack or JSON.
    Returns 409 if a tag already exists with a different commit_id and
    ``force`` is not set.
    """
    from musehub.services import musehub_version_tags as vtag_svc

    repo = await _resolve_repo(session, owner, slug)
    await _assert_writable(repo, claims, session)

    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    try:
        data = _decode_request_body(raw, ct)
    except (ValueError, Exception):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Request body must be a valid msgpack or JSON object",
        )

    repo_id = repo.repo_id

    tags_raw = data.get("tags", [])
    if not isinstance(tags_raw, list):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="'tags' must be a list",
        )

    tags: list[dict] = []
    for item in tags_raw:
        if not isinstance(item, dict):
            continue
        tags.append(item)

    # Without force, detect conflicts (same tag label, different commit_id) first.
    if not force:
        from sqlalchemy import select as sa_select
        from musehub.db.musehub_repo_models import MusehubVersionTag as _VTag
        for item in tags:
            tag_label = str(item.get("tag", "")).strip()
            commit_id = str(item.get("commit_id", "")).strip()
            if not tag_label or not commit_id:
                continue
            existing = await session.scalar(
                sa_select(_VTag).where(
                    _VTag.repo_id == repo_id,
                    _VTag.tag == tag_label,
                ).limit(1)
            )
            if existing is not None and existing.commit_id != commit_id:
                raise HTTPException(
                    status_code=status.HTTP_409_CONFLICT,
                    detail=f"Tag '{tag_label}' already points to a different commit. Use force=true to override.",
                )

    stored, skipped = await vtag_svc.store_version_tags(session, repo_id, tags, force=force)
    await session.commit()
    logger.info(
        "✅ wire: %d version tag(s) stored, %d skipped for %s/%s",
        stored, skipped, owner, slug,
    )
    return _mpack_response({"stored": stored, "skipped": skipped}, request)


@router.get(
    "/{owner}/{slug}/version-tags",
    summary="Fetch all version tags for a repo",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(OBJECT_LIMIT)
async def wire_get_version_tags(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    _claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Return all semantic-version tags stored for a repo.

    Returns ``{"tags": [...]}`` as msgpack or JSON, with tags ordered by
    semver descending (newest first).
    """
    from musehub.services import musehub_version_tags as vtag_svc

    repo_id = await _resolve_repo_id(session, owner, slug)
    tags = await vtag_svc.list_version_tags(session, repo_id)
    logger.info("✅ wire: returning %d version tag(s) for %s/%s", len(tags), owner, slug)
    return _mpack_response({"tags": tags}, request)

# ── reflog endpoints (muse#54 item 4) ───────────────────────────────────────────

@router.post(
    "/{owner}/{slug}/reflog",
    summary="Push branch reflog entries from muse CLI",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_PUSH_LIMIT)
async def wire_push_reflog(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    branch: BranchQueryParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Append a batch of reflog entries pushed from the Muse CLI for one branch.

    The body must be a msgpack object with an ``entries`` key holding a list
    of entry dicts (``old_id``, ``new_id``, ``author``, ``timestamp``,
    ``operation``). Entries are immutable history, so re-pushing an entry
    already stored is a true no-op (matched by content hash) — there is no
    ``force`` flag because there is nothing to conflict.

    Returns ``{"stored": <count>, "skipped": <count>}`` as msgpack or JSON.
    """
    from musehub.services import musehub_reflog as reflog_svc

    repo = await _resolve_repo(session, owner, slug)
    await _assert_writable(repo, claims, session)

    raw = await request.body()
    ct = request.headers.get("Content-Type", "")
    try:
        data = _decode_request_body(raw, ct)
    except (ValueError, Exception):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Request body must be a valid msgpack or JSON object",
        )

    repo_id = repo.repo_id

    entries_raw = data.get("entries", [])
    if not isinstance(entries_raw, list):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="'entries' must be a list",
        )

    entries: list[dict] = [item for item in entries_raw if isinstance(item, dict)]

    stored, skipped = await reflog_svc.store_reflog_entries(session, repo_id, branch, entries)
    await session.commit()
    logger.info(
        "✅ wire: %d reflog entr(y/ies) stored, %d skipped for %s/%s branch %s",
        stored, skipped, owner, slug, branch,
    )
    return _mpack_response({"stored": stored, "skipped": skipped}, request)


@router.get(
    "/{owner}/{slug}/reflog",
    summary="Fetch branch reflog entries for a repo",
    status_code=status.HTTP_200_OK,
)
@limiter.limit(WIRE_FETCH_LIMIT)
async def wire_get_reflog(
    request: Request,
    owner: SlugParam,
    slug: SlugParam,
    branch: BranchQueryParam,
    claims: TokenClaims = Depends(require_valid_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Return all reflog entries stored for one branch.

    Returns ``{"entries": [...]}`` as msgpack or JSON, ordered oldest-first
    (the order the entries were originally appended on the pushing client).
    """
    from musehub.services import musehub_reflog as reflog_svc

    repo = await _resolve_repo(session, owner, slug)
    await _assert_readable(repo, claims, session)

    entries = await reflog_svc.list_reflog_entries(session, repo.repo_id, branch)
    logger.info(
        "✅ wire: returning %d reflog entr(y/ies) for %s/%s branch %s",
        len(entries), owner, slug, branch,
    )
    return _mpack_response({"entries": entries}, request)

# ── content-addressed CDN ──────────────────────────────────────────────────────

@router.get(
    "/o/{object_id:path}",
    summary="Content-addressed object CDN endpoint",
    response_description="Raw binary blob",
    tags=["Objects"],
)
@limiter.limit(OBJECT_LIMIT)
async def get_object(
    request: Request,
    object_id: str,
    repo_id: str | None = None,
    _claims: TokenClaims | None = Depends(optional_token),
    session: AsyncSession = Depends(get_session),
) -> Response:
    """Serve a content-addressed binary object.

    Objects are immutable (ID is derived from content hash), so the response
    carries ``Cache-Control: max-age=31536000, immutable`` — safe to place
    behind CloudFront forever.
    """
    from musehub.storage.backends import read_object_bytes
    backend = get_backend()
    try:
        raw = await backend.get(object_id)
    except ValueError:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid object path")
    if raw is None:
        obj_row = await session.scalar(
            select(MusehubObject).where(MusehubObject.object_id == object_id).limit(1)
        )
        if obj_row is not None:
            raw = await read_object_bytes(obj_row, session=session)
    if raw is None:
        logger.error(
            "OBJECT_404 object_id=%s repo_id=%s backend=%s",
            object_id,
            repo_id,
            type(get_backend()).__name__,
        )
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="object not found")

    # Integrity check: verify stored bytes match the content-addressed ID.
    import hashlib as _hashlib
    actual_hash = _hashlib.sha256(raw).hexdigest()
    expected_hex = object_id.removeprefix("sha256:")
    if actual_hash != expected_hex:
        logger.error(
            "OBJECT_CORRUPT object_id=%s actual_hash=sha256:%s size=%d backend=%s",
            object_id,
            actual_hash,
            len(raw),
            type(get_backend()).__name__,
        )

    return Response(
        content=raw,
        media_type="application/octet-stream",
        headers={
            "Cache-Control": "public, max-age=31536000, immutable",
            "ETag": f'"{object_id}"',
        },
    )