"""MuseHub object (artifact) route handlers. Endpoint summary: GET /repos/{repo_id}/objects — list artifact metadata GET /repos/{repo_id}/objects/{object_id}/content — serve raw artifact bytes GET /repos/{repo_id}/blob/{ref}/{path} — blob metadata + text content for UI GET /repos/{repo_id}/tree/{ref} — list repo root directory at a ref GET /repos/{repo_id}/tree/{ref}/{path} — list subdirectory at a ref All endpoints require a valid MSign token. """ import logging import mimetypes from pathlib import Path from fastapi import APIRouter, Depends, HTTPException, status from fastapi.responses import Response from sqlalchemy.ext.asyncio import AsyncSession from musehub.api.validation import FilePathParam, SlugParam from musehub.auth.dependencies import TokenClaims, optional_token from musehub.db import get_db from musehub.models.musehub import BlobMetaResponse, ObjectMetaListResponse, TreeListResponse from musehub.types.json_types import StrDict from musehub.services import musehub_repository from musehub.storage.backends import get_backend as _get_storage_backend logger = logging.getLogger(__name__) router = APIRouter() # Additional MIME types not always in the system mimetypes database _EXTRA_MIME: StrDict = { ".webp": "image/webp", } def _content_type(path: str) -> str: """Resolve MIME type from path extension; fall back to octet-stream.""" ext = Path(path).suffix.lower() if ext in _EXTRA_MIME: return _EXTRA_MIME[ext] guessed, _ = mimetypes.guess_type(path) return guessed or "application/octet-stream" # Maximum text file size to embed in BlobMetaResponse.content_text (256 KB). _MAX_TEXT_EMBED_BYTES = 256 * 1024 _FILE_TYPE_MAP: StrDict = { ".webp": "image", ".png": "image", ".jpg": "image", ".jpeg": "image", ".json": "json", ".xml": "xml", } def _detect_file_type(path: str) -> str: """Return a rendering hint based on file extension. Values: 'json' | 'image' | 'xml' | 'other' """ ext = Path(path).suffix.lower() return _FILE_TYPE_MAP.get(ext, "other") @router.get( "/repos/{repo_id}/blob/{ref}/{path:path}", response_model=BlobMetaResponse, operation_id="getBlobMeta", summary="Get blob metadata and optional text content for the file viewer", ) async def get_blob_meta( repo_id: str, ref: str, path: FilePathParam, db: AsyncSession = Depends(get_db), claims: TokenClaims | None = Depends(optional_token), ) -> BlobMetaResponse: """Return metadata for a single file in the blob viewer. The response includes the file's size, SHA, creation timestamp, and rendering hint (``file_type``). For text-based files (JSON, XML) up to 256 KB, ``content_text`` is populated so the viewer can render them inline without a second request. Binary and oversized files omit ``content_text``; consumers should stream bytes from ``raw_url`` instead. The ``ref`` parameter accepts branch names or commit SHAs and is used for URL construction (raw_url). Object resolution always returns the most-recently-pushed object at ``path``, consistent with the raw and tree endpoints at MVP scope. Returns 404 if the repo is not found or no object exists at that path. Returns 401 if the repo is private and no valid token is supplied. """ repo = await musehub_repository.get_repo(db, repo_id) if repo is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Repo not found") if repo.visibility != "public" and claims is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required to access private repos.", headers={"WWW-Authenticate": 'MSign realm="musehub"'}, ) file_meta = await musehub_repository.get_file_at_ref(db, repo_id, ref, path) if file_meta is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"No object at path '{path}' in ref '{ref}'", ) object_id = str(file_meta["object_id"]) norm_path = str(file_meta["path"]) file_type = _detect_file_type(norm_path) raw_url = f"/repos/{repo_id}/raw/{ref}/{path}" obj_row = await musehub_repository.get_object_row(db, repo_id, object_id) if obj_row is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Object '{object_id}' not found in storage", ) storage = _get_storage_backend() raw_bytes = await storage.get(object_id) size_bytes = len(raw_bytes) if raw_bytes is not None else obj_row.size_bytes content_text: str | None = None if file_type in ("json", "xml") and raw_bytes is not None and size_bytes <= _MAX_TEXT_EMBED_BYTES: try: content_text = raw_bytes.decode("utf-8", errors="replace") except Exception: logger.warning("⚠️ Could not decode text content for blob %s", object_id) return BlobMetaResponse( object_id=object_id, path=norm_path, filename=Path(norm_path).name, size_bytes=size_bytes, sha=object_id, created_at=obj_row.created_at, raw_url=raw_url, file_type=file_type, content_text=content_text, ) @router.get( "/repos/{repo_id}/objects", response_model=ObjectMetaListResponse, operation_id="listObjects", summary="List artifact metadata for a MuseHub repo", ) async def list_objects( repo_id: str, db: AsyncSession = Depends(get_db), claims: TokenClaims | None = Depends(optional_token), ) -> ObjectMetaListResponse: """Return metadata (path, size, object_id) for all objects in the repo. Binary content is excluded from this response — use the ``/content`` sub-resource to download individual artifacts. Results are ordered by path. Returns 404 if the repo does not exist. """ repo = await musehub_repository.get_repo(db, repo_id) if repo is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Repo not found") if repo.visibility != "public" and claims is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required to access private repos.", headers={"WWW-Authenticate": 'MSign realm="musehub"'}, ) objects = await musehub_repository.list_objects(db, repo_id) return ObjectMetaListResponse(objects=objects) @router.get( "/repos/{repo_id}/objects/{object_id}/content", operation_id="getObjectContent", summary="Download raw artifact bytes", ) async def get_object_content( repo_id: str, object_id: str, db: AsyncSession = Depends(get_db), claims: TokenClaims | None = Depends(optional_token), ) -> Response: """Return the raw bytes of a stored artifact from the blob store. Content-Type is inferred from the stored ``path`` extension (.webp → image/webp). Returns 404 if the repo or object is not found, or 410 if the bytes are not present in the blob store. """ repo = await musehub_repository.get_repo(db, repo_id) if repo is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Repo not found") if repo.visibility != "public" and claims is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required to access private repos.", headers={"WWW-Authenticate": 'MSign realm="musehub"'}, ) obj = await musehub_repository.get_object_row(db, repo_id, object_id) if obj is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Object not found") data = await _get_storage_backend().get(object_id) if data is None: logger.warning("⚠️ Object %s exists in DB but not in blob store", object_id) raise HTTPException( status_code=status.HTTP_410_GONE, detail="Object has been removed from storage", ) filename = Path(obj.path).name media_type = _content_type(obj.path) return Response( content=data, media_type=media_type, headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) @router.get( "/repos/{repo_id}/tree/{ref_and_path:path}", response_model=TreeListResponse, summary="List a directory of a Muse repo at a given ref and optional sub-path", ) async def list_tree( repo_id: str, ref_and_path: str, owner: SlugParam, repo_slug: SlugParam, db: AsyncSession = Depends(get_db), claims: TokenClaims | None = Depends(optional_token), ) -> TreeListResponse: """Return a directory listing for the given ref and optional sub-path. Branch names may contain slashes (e.g. ``feat/my-thing``). This endpoint accepts the full tail after ``/tree/`` as ``ref_and_path`` and resolves the boundary between the ref and the sub-path by fetching all known branch names and selecting the longest prefix match — the same strategy as the UI route. Query params ``owner`` and ``repo_slug`` are passed through for breadcrumb generation; ``repo_id`` is the authoritative key. Returns entries sorted directories-first, then files, both alphabetically. """ repo = await musehub_repository.get_repo(db, repo_id) if repo is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Repo not found") if repo.visibility != "public" and claims is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required to access private repos.", headers={"WWW-Authenticate": 'MSign realm="musehub"'}, ) # Disambiguate ref vs sub-path using longest branch-name prefix match. branches = await musehub_repository.list_branches(db, repo_id) branch_names: list[str] = sorted( (b.name for b in branches), key=len, reverse=True ) ref = ref_and_path dir_path = "" for name in branch_names: if ref_and_path == name: ref = name dir_path = "" break if ref_and_path.startswith(f"{name}/"): ref = name dir_path = ref_and_path[len(name) + 1:] break ref_valid = await musehub_repository.resolve_ref_for_tree(db, repo_id, ref) if not ref_valid: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Ref '{ref}' not found in repo", ) resolved_ref = ( await musehub_repository.resolve_head_ref(db, repo_id) if ref == "HEAD" else ref ) return await musehub_repository.list_tree(db, repo_id, owner, repo_slug, resolved_ref, dir_path)