"""MuseHub persistence adapter — single point of DB access for Hub entities. This module is the ONLY place that touches the musehub_* tables. Route handlers delegate here; no business logic lives in routes. Boundary rules: - Must NOT import state stores, SSE queues, or LLM clients. - May import ORM models from musehub.db domain-specific modules. - May import Pydantic response models from musehub.models.musehub. """ from datetime import datetime, timezone import logging import re from collections import deque from sqlalchemy import desc, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from sqlalchemy.sql.elements import ColumnElement from musehub.services.musehub_snapshot import get_snapshot_manifests_batch GENERIC_DOMAIN = "generic" from musehub.core.genesis import compute_branch_id, compute_fork_id, compute_identity_id, compute_repo_id, compute_session_id from musehub.db.musehub_identity_models import MusehubIdentity from musehub.db.musehub_intel_models import MusehubFileLastCommit, MusehubSymbolHistoryEntry from musehub.db.musehub_repo_models import ( MusehubBranch, MusehubCommit, MusehubCommitRef, MusehubObject, MusehubObjectRef, MusehubRepo, MusehubSession, ) from musehub.db.musehub_social_models import MusehubFork from musehub.db import musehub_collaborator_models as collab_db from musehub.db.utils import escape_like from musehub.models.musehub import ( SessionListResponse, SessionResponse, BranchDetailListResponse, BranchDetailResponse, BranchDivergenceScores, BranchResponse, CommitListResponse, CommitResponse, GlobalSearchCommitMatch, GlobalSearchRepoGroup, GlobalSearchResult, DagEdge, DagGraphResponse, DagNode, MuseHubContextCommitInfo, MuseHubContextHistoryEntry, MuseHubContextMusicalState, MuseHubContextResponse, ObjectMetaResponse, RepoListResponse, RepoResponse, RepoSettingsPatch, RepoSettingsResponse, TimelineCommitEvent, TimelineResponse, TreeEntryResponse, TreeListResponse, ForkNetworkNode, ForkNetworkResponse, ForkRepoRequest, UserForkedRepoEntry, UserForksResponse, ) from musehub.types.json_types import IntDict, JSONObject, StrDict type FileLastCommits = dict[str, StrDict] logger = logging.getLogger(__name__) def _generate_slug(name: str) -> str: """Derive a URL-safe slug from a human-readable repo name. Rules: lowercase, non-alphanumeric chars collapsed to single hyphens, leading/trailing hyphens stripped, max 64 chars. If the result is empty (e.g. name was all symbols) we fall back to "repo". """ slug = name.lower() slug = re.sub(r"[^a-z0-9]+", "-", slug) slug = slug.strip("-") slug = slug[:64].strip("-") return slug or "repo" def _repo_clone_url(owner: str, slug: str) -> str: """Derive the canonical clone URL from owner and slug. Returns a plain HTTPS URL using the configured public_url so that `muse clone ` works without any extra flags. Override the host via the PUBLIC_URL environment variable (e.g. https://staging.musehub.ai). """ from musehub.config import settings return f"{settings.public_url.rstrip('/')}/{owner}/{slug}" def _to_repo_response(row: MusehubRepo, domain: str = GENERIC_DOMAIN) -> RepoResponse: return RepoResponse( repo_id=row.repo_id, name=row.name, owner=row.owner, slug=row.slug, visibility=row.visibility, owner_user_id=row.owner_user_id, clone_url=_repo_clone_url(row.owner, row.slug), description=row.description, tags=list(row.tags or []), domain_id=getattr(row, "domain_id", None), domain=domain, default_branch=row.default_branch, created_at=row.created_at, updated_at=row.updated_at, pushed_at=row.pushed_at, ) def _to_branch_response(row: MusehubBranch) -> BranchResponse: return BranchResponse( branch_id=row.branch_id, name=row.name, head_commit_id=row.head_commit_id, ) def _to_commit_response(row: MusehubCommit) -> CommitResponse: return CommitResponse( commit_id=row.commit_id, branch=row.branch, parent_ids=list(row.parent_ids or []), message=row.message, author=row.author, timestamp=row.timestamp, snapshot_id=row.snapshot_id, ) async def create_repo( session: AsyncSession, *, name: str, owner: str, visibility: str, owner_user_id: str, owner_identity_id: str = "", description: str = "", tags: list[str] | None = None, domain: str = "", marketplace_domain_id: str | None = None, # ── Wizard extensions ──────────────────────────────────────── license: str | None = None, topics: list[str] | None = None, initialize: bool = False, default_branch: str = "main", template_repo_id: str | None = None, ) -> RepoResponse: """Persist a new remote repo and return its wire representation. ``slug`` is auto-generated from ``name``. The ``(owner, slug)`` pair must be unique — callers should catch ``IntegrityError`` and surface a 409. Wizard behaviors: - When ``template_repo_id`` is set, the template's description and topics are copied into the new repo (template must be public; silently skipped when it doesn't exist or is private). - When ``initialize=True``, an empty "Initial commit" is written plus the default branch pointer so the repo is immediately browsable. - ``license`` is stored in the settings JSON blob under the ``license`` key. - ``topics`` are merged with ``tags`` into a single unified tag list. Marketplace domain link (musehub#120): ``domain`` is a plain VCS-plugin category string (e.g. "code") carrying no author context, so it can never by itself identify a specific marketplace ``MusehubDomain``. - ``marketplace_domain_id`` explicit and non-``None`` — validated via ``get_domain_by_id`` (raises ``ValueError("marketplace_domain_not_found")`` if it doesn't exist, same contract ``update_repo_settings`` already has) and stored as-is. - Omitted — auto-resolved via ``resolve_unambiguous_domain_id_by_category(session, domain)``, which only links when exactly one non-deprecated marketplace domain matches the category; ambiguous or unmatched categories are left ``None``, never guessed. Either way, when the repo ends up linked, ``record_domain_install`` is called once for ``owner_user_id`` so the target domain's ``install_count`` is correct from the moment of creation. """ from musehub.services import musehub_domains as _musehub_domains # Merge topics into tags (deduplicated, stable order). combined_tags: list[str] = list(dict.fromkeys((tags or []) + (topics or []))) # Copy template metadata when a template repo is supplied. if template_repo_id is not None: tmpl = await session.get(MusehubRepo, template_repo_id) if tmpl is not None and tmpl.visibility == "public": if not description: description = tmpl.description # Prepend template tags; deduplicate preserving order. combined_tags = list(dict.fromkeys(list(tmpl.tags or []) + combined_tags)) # Build the settings JSON blob with optional license field. settings: JSONObject = {} if license is not None: settings["license"] = license slug = _generate_slug(name) _created_at = datetime.now(timezone.utc) # Canonical hash input: empty/absent domain maps to "muse/generic" — never "". _domain_hash_input = domain or "muse/generic" # Stored domain: default to "code" — the column must never be NULL for new repos. _domain_id = domain or "code" # ── Marketplace domain link (musehub#120) — never guess ───────────────── _marketplace_domain_id: str | None if marketplace_domain_id is not None: target = await _musehub_domains.get_domain_by_id(session, marketplace_domain_id) if target is None: raise ValueError("marketplace_domain_not_found") _marketplace_domain_id = marketplace_domain_id else: _marketplace_domain_id = await _musehub_domains.resolve_unambiguous_domain_id_by_category( session, _domain_id ) repo = MusehubRepo( repo_id=compute_repo_id(owner_identity_id, slug, _domain_hash_input, _created_at.isoformat()), name=name, owner=owner, slug=slug, visibility=visibility, owner_user_id=owner_user_id, description=description, tags=combined_tags, settings=settings or None, domain_id=_domain_id, marketplace_domain_id=_marketplace_domain_id, default_branch=default_branch, ) session.add(repo) await session.flush() # populate default columns before reading await session.refresh(repo) if _marketplace_domain_id is not None: await _musehub_domains.record_domain_install(session, owner_user_id, _marketplace_domain_id) # Wizard initialisation: create default branch + empty initial commit. if initialize: from muse.core.types import blob_id as _blob_id init_commit_id = _blob_id(f"init:{repo.repo_id}".encode()) now = datetime.now(tz=timezone.utc) branch = MusehubBranch( branch_id=compute_branch_id(repo.repo_id, default_branch), repo_id=repo.repo_id, name=default_branch, head_commit_id=init_commit_id, ) session.add(branch) init_commit = MusehubCommit( commit_id=init_commit_id, branch=default_branch, parent_ids=[], message="Initial commit", author=owner_user_id, timestamp=now, ) session.add(init_commit) session.add(MusehubCommitRef(repo_id=repo.repo_id, commit_id=init_commit_id)) await session.flush() logger.info( "✅ Created MuseHub repo %s (%s/%s) for user %s (initialize=%s)", repo.repo_id, owner, slug, owner_user_id, initialize, ) return _to_repo_response(repo) async def get_repo(session: AsyncSession, repo_id: str) -> RepoResponse | None: """Return repo metadata by internal ID, or None if not found.""" result = await session.get(MusehubRepo, repo_id) if result is None: return None return _to_repo_response(result) async def check_write_access( session: AsyncSession, repo_id: str, actor: str, repo_owner: str, ) -> bool: """Return True if *actor* has write-level access to the repo. Write access is granted when the actor is the repository owner, or when they are an accepted write/admin collaborator. This mirrors the check performed by ``_guard_repo_owner`` in the REST route layer. Args: session: Active async DB session. repo_id: ID of the repository. actor: Identity handle of the caller. repo_owner: Owner handle of the repository (from ``RepoResponse.owner``). Returns: ``True`` when the caller has write access; ``False`` otherwise. """ if actor == repo_owner: return True collab = (await session.execute( select(collab_db.MusehubCollaborator).where( collab_db.MusehubCollaborator.repo_id == repo_id, collab_db.MusehubCollaborator.identity_handle == actor, collab_db.MusehubCollaborator.accepted_at.isnot(None), collab_db.MusehubCollaborator.permission.in_(["write", "admin"]), ) )).scalar_one_or_none() return collab is not None async def delete_repo(session: AsyncSession, repo_id: str) -> bool: """Hard-delete a repo and all its cascade-deleted dependents. Returns True when the repo existed and was deleted; False when not found. The caller is responsible for committing the session. """ row = await session.get(MusehubRepo, repo_id) if row is None: return False await session.delete(row) await session.flush() logger.info("✅ Hard-deleted MuseHub repo %s", repo_id) return True async def transfer_repo_ownership( session: AsyncSession, repo_id: str, new_owner_user_id: str ) -> RepoResponse | None: """Transfer repo ownership to a new user. Only touches ``owner_user_id`` — the public ``owner`` username slug is intentionally NOT changed here; the owner username update (if desired) is a settings-level change the new owner makes separately. Returns the updated RepoResponse, or None when the repo is not found. The caller is responsible for committing the session. """ row = await session.get(MusehubRepo, repo_id) if row is None: return None row.owner_user_id = new_owner_user_id await session.flush() await session.refresh(row) logger.info("✅ Transferred MuseHub repo %s ownership to user %s", repo_id, new_owner_user_id) return _to_repo_response(row) async def get_identity_id_for_handle(session: AsyncSession, handle: str) -> str: """Return the genesis-addressed identity_id for a MSign handle, or '' if not found.""" row = (await session.execute( select(MusehubIdentity.identity_id).where(MusehubIdentity.handle == handle) )).scalar_one_or_none() return row or "" async def get_repo_row_by_owner_slug( session: AsyncSession, owner: str, slug: str ) -> MusehubRepo | None: """Return the raw ORM row for owner/slug, or None if not found. Use this when you need access to internal fields (e.g. ``owner_user_id``, ``visibility``) that are not exposed by :class:`RepoResponse`. """ stmt = select(MusehubRepo).where( MusehubRepo.owner == owner, MusehubRepo.slug == slug, ) return (await session.execute(stmt)).scalars().first() async def get_repo_by_owner_slug( session: AsyncSession, owner: str, slug: str ) -> RepoResponse | None: """Return repo metadata by owner+slug canonical URL pair, or None if not found. This is the primary resolver for all external /{owner}/{slug} routes. """ row = await get_repo_row_by_owner_slug(session, owner, slug) if row is None: return None return _to_repo_response(row) _PAGE_SIZE = 20 async def list_repos_for_user( session: AsyncSession, user_id: str, *, limit: int = _PAGE_SIZE, cursor: str | None = None, ) -> RepoListResponse: """Return repos owned by or collaborated on by ``user_id``. Results are ordered by ``created_at`` descending (newest first). Pagination uses an opaque cursor encoding the ``created_at`` ISO timestamp of the last item on the current page — pass it back as ``?cursor=`` to advance. Args: session: Active async DB session. user_id: MSign handle of the authenticated caller. limit: Maximum repos per page (default 20). cursor: Opaque pagination cursor from a previous response. Returns: ``RepoListResponse`` with the page of repos, total count, and next cursor. """ # Correlated subquery: repo IDs the user has accepted collaborator access to. # Using a subquery (not a Python list) avoids fetching all IDs into memory # and avoids large IN() clauses for users with many collaboration repos. collab_subq = ( select(collab_db.MusehubCollaborator.repo_id) .where( collab_db.MusehubCollaborator.identity_handle == user_id, collab_db.MusehubCollaborator.accepted_at.is_not(None), ) ) # Base filter: repos the caller owns OR collaborates on. base_filter = or_( MusehubRepo.owner == user_id, MusehubRepo.repo_id.in_(collab_subq), ) # Total count across all pages. count_stmt = select(func.count()).select_from(MusehubRepo).where(base_filter) total: int = (await session.execute(count_stmt)).scalar_one() # Apply cursor: skip repos created at or after the cursor timestamp. page_filter = base_filter if cursor is not None: try: # Normalise 'Z' suffix so fromisoformat works on all Python versions. _cursor = cursor.replace("Z", "+00:00") cursor_dt = datetime.fromisoformat(_cursor) page_filter = base_filter & (MusehubRepo.created_at < cursor_dt) except ValueError: pass # malformed cursor — ignore and return from the beginning stmt = ( select(MusehubRepo) .where(page_filter) .order_by(desc(MusehubRepo.created_at)) .limit(limit) ) rows = (await session.execute(stmt)).scalars().all() repos = [_to_repo_response(r) for r in rows] # Build next cursor from the last item's created_at when there may be more. # Use 'Z' suffix (not '+00:00') so the cursor is URL-safe in query strings. next_cursor: str | None = None if len(rows) == limit: _ts = rows[-1].created_at.astimezone(timezone.utc) next_cursor = f"{_ts.strftime('%Y-%m-%dT%H:%M:%S.%f')}Z" return RepoListResponse(repos=repos, next_cursor=next_cursor, total=total) async def get_repo_orm_by_owner_slug( session: AsyncSession, owner: str, slug: str ) -> MusehubRepo | None: """Return the raw ORM repo row by owner+slug, or None if not found. Used internally when the route needs the repo_id for downstream calls. """ stmt = select(MusehubRepo).where( MusehubRepo.owner == owner, MusehubRepo.slug == slug, ) return (await session.execute(stmt)).scalars().first() async def list_branches(session: AsyncSession, repo_id: str) -> list[BranchResponse]: """Return all branches for a repo, ordered by name.""" stmt = ( select(MusehubBranch) .where(MusehubBranch.repo_id == repo_id) .order_by(MusehubBranch.name) ) rows = (await session.execute(stmt)).scalars().all() return [_to_branch_response(r) for r in rows] async def get_branch_head_commit_id( session: AsyncSession, repo_id: str, branch_name: str, ) -> str | None: """Return the head commit ID of ``branch_name`` in ``repo_id``, or ``None``.""" stmt = select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == branch_name, ) row = (await session.execute(stmt)).scalar_one_or_none() return row.head_commit_id if row is not None else None async def list_branches_with_detail( session: AsyncSession, repo_id: str ) -> BranchDetailListResponse: """Return branches enriched with ahead/behind counts vs the default branch. The default branch is whichever branch is named "main"; if no "main" branch exists, the first branch alphabetically is used. Ahead/behind counts are computed by comparing the set of commit IDs on each branch vs the default branch — a set-difference approximation suitable for display purposes. Musical divergence scores are not yet computable server-side (they require audio snapshots), so all divergence fields are returned as ``None`` (placeholder). """ branch_stmt = ( select(MusehubBranch) .where(MusehubBranch.repo_id == repo_id) .order_by(MusehubBranch.name) ) branch_rows = (await session.execute(branch_stmt)).scalars().all() if not branch_rows: return BranchDetailListResponse(branches=[], default_branch="main") # Determine default branch name: prefer "main", fall back to first alphabetically. branch_names = [r.name for r in branch_rows] default_branch_name = "main" if "main" in branch_names else branch_names[0] # Load commit IDs per branch in one query. commit_stmt = ( select(MusehubCommit.commit_id, MusehubCommit.branch) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where(MusehubCommitRef.repo_id == repo_id) ) commit_rows = (await session.execute(commit_stmt)).all() commits_by_branch = {} for commit_id, branch_name in commit_rows: commits_by_branch.setdefault(branch_name, set()).add(commit_id) default_commits: set[str] = commits_by_branch.get(default_branch_name, set()) results: list[BranchDetailResponse] = [] for row in branch_rows: is_default = row.name == default_branch_name branch_commits: set[str] = commits_by_branch.get(row.name, set()) ahead = len(branch_commits - default_commits) if not is_default else 0 behind = len(default_commits - branch_commits) if not is_default else 0 results.append( BranchDetailResponse( branch_id=row.branch_id, name=row.name, head_commit_id=row.head_commit_id, is_default=is_default, ahead_count=ahead, behind_count=behind, divergence=BranchDivergenceScores( melodic=None, harmonic=None, rhythmic=None, structural=None, dynamic=None ), ) ) return BranchDetailListResponse(branches=results, default_branch=default_branch_name) def _to_object_meta_response(row: MusehubObject) -> ObjectMetaResponse: return ObjectMetaResponse( object_id=row.object_id, path=row.path, size_bytes=row.size_bytes, created_at=row.created_at, ) async def get_commit( session: AsyncSession, repo_id: str, commit_id: str ) -> CommitResponse | None: """Return a single commit by ID, or None if not found in this repo.""" ref_row = await session.get(MusehubCommitRef, (repo_id, commit_id)) if ref_row is None: return None row = await session.get(MusehubCommit, commit_id) if row is None: return None return _to_commit_response(row) async def list_objects( session: AsyncSession, repo_id: str ) -> list[ObjectMetaResponse]: """Return all object metadata for a repo (no binary content), ordered by path.""" stmt = ( select(MusehubObject) .join(MusehubObjectRef, MusehubObject.object_id == MusehubObjectRef.object_id) .where(MusehubObjectRef.repo_id == repo_id) .order_by(MusehubObject.path) ) rows = (await session.execute(stmt)).scalars().all() return [_to_object_meta_response(r) for r in rows] async def get_object_row( session: AsyncSession, repo_id: str, object_id: str ) -> MusehubObject | None: """Return the raw ORM object row for content delivery, or None if not found.""" stmt = ( select(MusehubObject) .join(MusehubObjectRef, MusehubObject.object_id == MusehubObjectRef.object_id) .where( MusehubObjectRef.repo_id == repo_id, MusehubObject.object_id == object_id, ) ) return (await session.execute(stmt)).scalars().first() async def list_commits( session: AsyncSession, repo_id: str, *, branch: str | None = None, cursor: str | None = None, limit: int = 50, ) -> CommitListResponse: """Return commits for a repo with cursor-based keyset pagination (newest first). ``branch`` restricts results to a specific branch when given. ``cursor`` is the ISO 8601 ``timestamp`` of the last seen commit (opaque to callers — pass ``nextCursor`` from a previous response verbatim). Omit to start from the most recent commit. """ base_conditions = [MusehubCommitRef.repo_id == repo_id] if branch: base_conditions.append(MusehubCommit.branch == branch) count_stmt = ( select(func.count(MusehubCommitRef.commit_id)) .join(MusehubCommit, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where(*base_conditions) ) total: int = (await session.execute(count_stmt)).scalar_one() data_conditions = list(base_conditions) if cursor is not None: data_conditions.append( MusehubCommit.timestamp < datetime.fromisoformat(cursor) ) rows = list( ( await session.execute( select(MusehubCommit) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where(*data_conditions) .order_by(desc(MusehubCommit.timestamp)) .limit(limit + 1) ) ).scalars() ) next_cursor: str | None = None if len(rows) == limit + 1: next_cursor = rows[limit - 1].timestamp.isoformat() rows = rows[:limit] return CommitListResponse( commits=[_to_commit_response(r) for r in rows], total=total, next_cursor=next_cursor, ) async def get_timeline_events( session: AsyncSession, repo_id: str, *, limit: int = 200, ) -> TimelineResponse: """Return a chronological timeline of commits for a repo. Fetches up to ``limit`` commits (oldest-first for temporal rendering) and derives two event streams: - commits: every commit as a timeline marker - emotion: deterministic emotion vectors from commit SHAs Callers must verify the repo exists before calling this function. Returns an empty timeline when the repo has no commits. """ total_stmt = ( select(func.count()) .select_from(MusehubCommitRef) .where(MusehubCommitRef.repo_id == repo_id) ) total: int = (await session.execute(total_stmt)).scalar_one() rows_stmt = ( select(MusehubCommit) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where(MusehubCommitRef.repo_id == repo_id) .order_by(MusehubCommit.timestamp) # oldest-first for temporal rendering .limit(limit) ) rows = (await session.execute(rows_stmt)).scalars().all() commit_events = [ TimelineCommitEvent( commit_id=row.commit_id, branch=row.branch, message=row.message, author=row.author, timestamp=row.timestamp, parent_ids=list(row.parent_ids or []), ) for row in rows ] return TimelineResponse( commits=commit_events, total_commits=total, ) async def global_search( session: AsyncSession, *, query: str, mode: str = "keyword", cursor: str | None = None, limit: int = 10, ) -> GlobalSearchResult: """Search commit messages across all public MuseHub repos. Only ``visibility='public'`` repos are searched — private repos are never exposed regardless of caller identity. This enforces the public-only contract at the persistence layer so no route handler can accidentally bypass it. ``mode`` controls matching strategy: - ``keyword``: OR-match of whitespace-split query terms against message and repo name using LIKE (case-insensitive via lower()). - ``pattern``: raw SQL LIKE pattern applied to commit message only. Results are grouped by repo and cursor-paginated by repo-group (``limit`` controls how many repo-groups per page). Within each group, up to 20 matching commits are returned newest-first. An audio preview object ID is attached when the repo contains any .mp3, .ogg, or .wav artifact — the first one found by path ordering is used. Audio previews are resolved in a single batched query across all matching repos (not N per-repo queries) to avoid the N+1 pattern. Args: session: Active async DB session. query: Raw search string from the user or agent. mode: "keyword" or "pattern". Defaults to "keyword". cursor: Opaque cursor from a previous nextCursor field (integer offset encoded as string). limit: Number of repo-groups per page (1–50). Returns: GlobalSearchResult with groups, cursor pagination metadata, and counts. """ # ── 1. Collect all public repos ───────────────────────────────────────── public_repos_stmt = ( select(MusehubRepo) .where( MusehubRepo.visibility == "public", ) .order_by(MusehubRepo.created_at) ) public_repo_rows = (await session.execute(public_repos_stmt)).scalars().all() total_repos_searched = len(public_repo_rows) if not public_repo_rows or not query.strip(): return GlobalSearchResult( query=query, mode=mode, groups=[], total_repos_searched=total_repos_searched, ) repo_ids = [r.repo_id for r in public_repo_rows] repo_map = {r.repo_id: r for r in public_repo_rows} # ── 2. Build commit filter predicate ──────────────────────────────────── predicate: ColumnElement[bool] if mode == "pattern": predicate = MusehubCommit.message.like(query) else: # keyword: OR-match each whitespace-split term against message (lower) terms = [t for t in query.lower().split() if t] if not terms: return GlobalSearchResult( query=query, mode=mode, groups=[], total_repos_searched=total_repos_searched, ) term_predicates = [ or_( func.lower(MusehubCommit.message).ilike(f"%{escape_like(term)}%", escape="\\"), func.lower(MusehubRepo.name).ilike(f"%{escape_like(term)}%", escape="\\"), ) for term in terms ] predicate = or_(*term_predicates) # ── 3. Query matching commits joined to their repo ─────────────────────── commits_stmt = ( select(MusehubCommit, MusehubRepo) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .join(MusehubRepo, MusehubCommitRef.repo_id == MusehubRepo.repo_id) .where( MusehubCommitRef.repo_id.in_(repo_ids), predicate, ) .order_by(desc(MusehubCommit.timestamp)) ) commit_pairs = (await session.execute(commits_stmt)).all() # ── 4. Group commits by repo ───────────────────────────────────────────── groups_map = {} for commit_row, _repo_row in commit_pairs: groups_map.setdefault(_repo_row.repo_id, []).append(commit_row) # ── 5. Cursor-paginate repo-groups ─────────────────────────────────────── # Cursor is the repo_id of the last item returned on the previous page. # Find its position and start the next page immediately after it. sorted_repo_ids = list(groups_map.keys()) page_start = 0 if cursor: try: idx = sorted_repo_ids.index(cursor) page_start = idx + 1 except ValueError: page_start = 0 page_repo_ids = sorted_repo_ids[page_start : page_start + limit] has_more = (page_start + limit) < len(sorted_repo_ids) next_cursor_val = page_repo_ids[-1] if has_more and page_repo_ids else None groups: list[GlobalSearchRepoGroup] = [] for rid in page_repo_ids: repo_row = repo_map[rid] all_matches = groups_map[rid] commit_matches = [ GlobalSearchCommitMatch( commit_id=c.commit_id, message=c.message, author=c.author, branch=c.branch, timestamp=c.timestamp, repo_id=rid, repo_name=repo_row.name, repo_owner=repo_row.owner_user_id, repo_visibility=repo_row.visibility, ) for c in all_matches[:20] ] groups.append( GlobalSearchRepoGroup( repo_id=rid, repo_name=repo_row.name, repo_owner=repo_row.owner_user_id, repo_slug=repo_row.slug, repo_visibility=repo_row.visibility, matches=commit_matches, total_matches=len(all_matches), ) ) return GlobalSearchResult( query=query, mode=mode, groups=groups, total_repos_searched=total_repos_searched, next_cursor=next_cursor_val, ) async def list_commits_dag( session: AsyncSession, repo_id: str, ) -> DagGraphResponse: """Return the full commit graph for a repo as a topologically sorted DAG. Fetches every commit for the repo (no limit — required for correct DAG traversal). Applies Kahn's algorithm to produce a topological ordering from oldest ancestor to newest commit, which graph renderers can consume directly without additional sorting. Edges flow child → parent (source = child, target = parent) following the standard directed graph convention where arrows point toward ancestors. Branch head commits are identified by querying the branches table. The highest-timestamp commit across all branches is designated as HEAD for display purposes when no explicit HEAD ref exists. Agent use case: call this to reason about the project's branching topology, find common ancestors, or identify which branches contain a given commit. """ # Fetch all commits for this repo stmt = ( select(MusehubCommit) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where(MusehubCommitRef.repo_id == repo_id) ) all_rows = (await session.execute(stmt)).scalars().all() if not all_rows: return DagGraphResponse(nodes=[], edges=[], head_commit_id=None) # Build lookup map row_map = {r.commit_id: r for r in all_rows} # Fetch all branches to identify HEAD candidates and branch labels branch_stmt = select(MusehubBranch).where(MusehubBranch.repo_id == repo_id) branch_rows = (await session.execute(branch_stmt)).scalars().all() # Map commit_id → branch names pointing at it branch_label_map = {} for br in branch_rows: if br.head_commit_id and br.head_commit_id in row_map: branch_label_map.setdefault(br.head_commit_id, []).append(br.name) # Identify HEAD: the branch head with the most recent timestamp, or the # most recent commit overall when no branches exist head_commit_id: str | None = None if branch_rows: latest_ts = None for br in branch_rows: if br.head_commit_id and br.head_commit_id in row_map: ts = row_map[br.head_commit_id].timestamp if latest_ts is None or ts > latest_ts: latest_ts = ts head_commit_id = br.head_commit_id if head_commit_id is None: head_commit_id = max(all_rows, key=lambda r: r.timestamp).commit_id # Kahn's topological sort (oldest → newest). # in_degree[c] = number of c's parents that are present in this repo's commit set. # Commits with in_degree == 0 are roots (no parents) — they enter the queue first, # producing a parent-before-child ordering (oldest ancestor → newest commit). in_degree: IntDict = {r.commit_id: 0 for r in all_rows} # children_map[parent_id] = list of commit IDs whose parent_ids contains parent_id children_map = {r.commit_id: [] for r in all_rows} edges: list[DagEdge] = [] for row in all_rows: for parent_id in (row.parent_ids or []): if parent_id in row_map: edges.append(DagEdge(source=row.commit_id, target=parent_id)) children_map.setdefault(parent_id, []).append(row.commit_id) in_degree[row.commit_id] += 1 # Kahn's algorithm: start from commits with no parents (roots) queue: deque[str] = deque( cid for cid, deg in in_degree.items() if deg == 0 ) topo_order: list[str] = [] while queue: cid = queue.popleft() topo_order.append(cid) for child_id in children_map.get(cid, []): in_degree[child_id] -= 1 if in_degree[child_id] == 0: queue.append(child_id) # Handle cycles or disconnected commits (append remaining in timestamp order) remaining = set(row_map.keys()) - set(topo_order) if remaining: sorted_remaining = sorted(remaining, key=lambda c: row_map[c].timestamp) topo_order.extend(sorted_remaining) _conv_re = re.compile(r'^(\w+)(\([^)]*\))?(!)?\s*:') nodes: list[DagNode] = [] for cid in topo_order: row = row_map[cid] # Extract conventional-commit prefix from the message m = _conv_re.match((row.message or "").strip()) commit_type = m.group(1).lower() if m else "" sem_ver_bump = str(row.sem_ver_bump or "none").lower() # Breaking change: bang suffix OR breaking_changes column is_breaking = bool((m and m.group(3)) or row.breaking_changes) is_agent = bool(row.agent_id) sym_added = 0 sym_removed = 0 delta = row.structured_delta if isinstance(row.structured_delta, dict) else {} for file_op in (delta.get("ops") or []): for child_op in (file_op.get("child_ops") or []) if isinstance(file_op, dict) else []: if not isinstance(child_op, dict): continue if child_op.get("op") == "insert": sym_added += 1 elif child_op.get("op") == "delete": sym_removed += 1 nodes.append( DagNode( commit_id=row.commit_id, message=row.message, author=row.author, timestamp=row.timestamp, branch=row.branch, parent_ids=list(row.parent_ids or []), is_head=(row.commit_id == head_commit_id), branch_labels=branch_label_map.get(row.commit_id, []), tag_labels=[], commit_type=commit_type, sem_ver_bump=sem_ver_bump, is_breaking=is_breaking, is_agent=is_agent, sym_added=sym_added, sym_removed=sym_removed, ) ) logger.debug("✅ Built DAG for repo %s: %d nodes, %d edges", repo_id, len(nodes), len(edges)) return DagGraphResponse(nodes=nodes, edges=edges, head_commit_id=head_commit_id) # --------------------------------------------------------------------------- # Context document builder # --------------------------------------------------------------------------- _CONTEXT_HISTORY_DEPTH = 5 async def _get_commit_by_id( session: AsyncSession, repo_id: str, commit_id: str ) -> MusehubCommit | None: """Fetch a raw MusehubCommit ORM row by (repo_id, commit_id).""" ref_row = await session.get(MusehubCommitRef, (repo_id, commit_id)) if ref_row is None: return None return await session.get(MusehubCommit, commit_id) async def _build_hub_history( session: AsyncSession, repo_id: str, start_commit: MusehubCommit, depth: int, ) -> list[MuseHubContextHistoryEntry]: """Walk the parent chain, returning up to *depth* ancestor entries. The *start_commit* (the context target) is NOT included — it is surfaced separately as ``head_commit`` in the result. Entries are newest-first. """ entries: list[MuseHubContextHistoryEntry] = [] parent_ids: list[str] = list(start_commit.parent_ids or []) while parent_ids and len(entries) < depth: parent_id = parent_ids[0] commit = await _get_commit_by_id(session, repo_id, parent_id) if commit is None: logger.warning("⚠️ Hub history chain broken at %s", parent_id) break entries.append( MuseHubContextHistoryEntry( commit_id=commit.commit_id, message=commit.message, author=commit.author, timestamp=commit.timestamp, active_tracks=[], ) ) parent_ids = list(commit.parent_ids or []) return entries async def get_context_for_commit( session: AsyncSession, repo_id: str, ref: str, ) -> MuseHubContextResponse | None: """Build a context document for a MuseHub commit. Traverses the commit's parent chain (up to 5 ancestors). Args: session: Open async DB session. Read-only — no writes performed. repo_id: Hub repo identifier. ref: Target commit ID. Must belong to this repo. Returns: ``MuseHubContextResponse`` ready for JSON serialisation, or None if the commit does not exist in this repo. """ commit = await _get_commit_by_id(session, repo_id, ref) if commit is None: return None head_commit_info = MuseHubContextCommitInfo( commit_id=commit.commit_id, message=commit.message, author=commit.author, branch=commit.branch, timestamp=commit.timestamp, ) musical_state = MuseHubContextMusicalState(active_tracks=[]) history = await _build_hub_history( session, repo_id, commit, _CONTEXT_HISTORY_DEPTH ) logger.info("✅ MuseHub context built for repo %s commit %s", repo_id, ref) return MuseHubContextResponse( repo_id=repo_id, current_branch=commit.branch, head_commit=head_commit_info, musical_state=musical_state, history=history, missing_elements=[], suggestions={}, ) def _to_session_response(s: MusehubSession) -> SessionResponse: """Compute derived fields and return a SessionResponse.""" duration: float | None = None if s.ended_at is not None: # Normalize to offset-naive UTC before subtraction ended = s.ended_at.replace(tzinfo=None) if s.ended_at.tzinfo else s.ended_at started = s.started_at.replace(tzinfo=None) if s.started_at.tzinfo else s.started_at duration = (ended - started).total_seconds() return SessionResponse( session_id=s.session_id, started_at=s.started_at, ended_at=s.ended_at, duration_seconds=duration, participants=s.participants or [], commits=list(s.commits) if s.commits else [], notes=s.notes or "", intent=s.intent, location=s.location, is_active=s.is_active, created_at=s.created_at, ) async def create_session( session: AsyncSession, repo_id: str, started_at: datetime | None, participants: list[str], intent: str, location: str, *, author_identity_id: str = "", ) -> SessionResponse: """Create and persist a new recording session.""" _started_at = started_at or datetime.now(timezone.utc) new_session = MusehubSession( session_id=compute_session_id(repo_id, author_identity_id, _started_at.isoformat()), repo_id=repo_id, started_at=_started_at, participants=participants, intent=intent, location=location, is_active=True, ) session.add(new_session) await session.flush() return _to_session_response(new_session) async def stop_session( session: AsyncSession, repo_id: str, session_id: str, ended_at: datetime | None, ) -> SessionResponse | None: """Mark a session as ended; idempotent if already stopped. Returns None if not found.""" from sqlalchemy import select result = await session.execute( select(MusehubSession).where( MusehubSession.session_id == session_id, MusehubSession.repo_id == repo_id, ) ) row = result.scalar_one_or_none() if row is None: return None if row.is_active: row.ended_at = ended_at or datetime.now(timezone.utc) row.is_active = False await session.flush() return _to_session_response(row) async def list_sessions( session: AsyncSession, repo_id: str, limit: int = 50, cursor: str | None = None, ) -> tuple[list[SessionResponse], int, str | None]: """Return sessions for a repo, newest first, with total count and next cursor. Ordered by ``is_active DESC, started_at DESC``. ``cursor`` is an opaque ISO-8601 ``started_at`` timestamp received from a previous response. When provided, only sessions with ``started_at`` strictly before the cursor instant are returned (ties broken by ``is_active`` sort ordering, which means active sessions always float to the top of page 1). Returns ``(sessions, total, next_cursor)`` where ``next_cursor`` is ``None`` on the last page. """ import datetime as _dt from sqlalchemy import func, select total_result = await session.execute( select(func.count(MusehubSession.session_id)).where( MusehubSession.repo_id == repo_id ) ) total = total_result.scalar_one() stmt = ( select(MusehubSession) .where(MusehubSession.repo_id == repo_id) .order_by(MusehubSession.is_active.desc(), MusehubSession.started_at.desc()) .limit(limit + 1) ) if cursor: cursor_dt = _dt.datetime.fromisoformat(cursor) # Active sessions always sort first; cursor only filters the started_at dimension # so we skip rows already seen by checking started_at < cursor_dt. stmt = stmt.where(MusehubSession.started_at < cursor_dt) result = await session.execute(stmt) rows = result.scalars().all() has_more = len(rows) > limit page_rows = rows[:limit] next_cursor: str | None = None if has_more and page_rows: next_cursor = page_rows[-1].started_at.isoformat() return [_to_session_response(s) for s in page_rows], total, next_cursor async def get_session( session: AsyncSession, repo_id: str, session_id: str, ) -> SessionResponse | None: """Fetch a single session by id.""" from sqlalchemy import select result = await session.execute( select(MusehubSession).where( MusehubSession.session_id == session_id, MusehubSession.repo_id == repo_id, ) ) row = result.scalar_one_or_none() if row is None: return None return _to_session_response(row) async def resolve_head_ref(session: AsyncSession, repo_id: str) -> str: """Resolve the symbolic "HEAD" ref to the repo's default branch name. Prefers "main" when that branch exists; otherwise returns the lexicographically first branch name, and falls back to "main" when the repo has no branches yet. """ branch_stmt = ( select(MusehubBranch) .where(MusehubBranch.repo_id == repo_id) .order_by(MusehubBranch.name) ) branches = (await session.execute(branch_stmt)).scalars().all() if not branches: return "main" names = [b.name for b in branches] return "main" if "main" in names else names[0] async def resolve_ref_for_tree( session: AsyncSession, repo_id: str, ref: str ) -> bool: """Return True if ref resolves to a known branch or commit in this repo. The ref can be: - ``"HEAD"`` — always valid; resolves to the default branch. - A branch name (e.g. "main", "feature/groove") — validated via the musehub_branches table. - A commit ID prefix or full SHA — validated via musehub_commits. Returns False if the ref is unknown, which the caller should surface as a 404. This is a lightweight existence check; callers that need the full commit object should call ``get_commit()`` separately. """ if ref == "HEAD": return True branch_stmt = select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == ref, ) branch_row = (await session.execute(branch_stmt)).scalars().first() if branch_row is not None: return True ref_row = await session.get(MusehubCommitRef, (repo_id, ref)) return ref_row is not None async def _get_head_snapshot_manifest( session: AsyncSession, repo_id: str, ref: str, ) -> StrDict: """Return the ``{path: object_id}`` manifest for the HEAD commit on *ref*. Falls back to an empty dict when no snapshot exists (e.g. new empty repo). """ # Resolve branch head → commit_id branch_row = ( await session.execute( select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == ref, ) ) ).scalar_one_or_none() if branch_row is None or not branch_row.head_commit_id: return {} head_commit = ( await session.execute( select(MusehubCommit).where( MusehubCommit.commit_id == branch_row.head_commit_id ) ) ).scalar_one_or_none() if head_commit is None or head_commit.snapshot_id is None: return {} from musehub.services.musehub_snapshot import get_snapshot_manifest return await get_snapshot_manifest(session, head_commit.snapshot_id) def _manifest_to_tree( manifest: StrDict, dir_path: str, ) -> tuple[list[TreeEntryResponse], list[TreeEntryResponse]]: """Build sorted (dirs, files) tree entries from a snapshot manifest. ``dir_path`` is the directory prefix to list (empty = repo root). Returns (dirs, files) each sorted alphabetically. """ prefix = f"{dir_path.strip('/')}/" if dir_path.strip("/") else "" seen_dirs: set[str] = set() dirs: list[TreeEntryResponse] = [] files: list[TreeEntryResponse] = [] for path, object_id in manifest.items(): norm = path.lstrip("/") if not norm.startswith(prefix): continue remainder = norm[len(prefix):] if not remainder: continue slash_pos = remainder.find("/") if slash_pos == -1: files.append( TreeEntryResponse( type="file", name=remainder, path=norm, size_bytes=None, object_id=object_id, ) ) else: dir_name = remainder[:slash_pos] if dir_name not in seen_dirs: seen_dirs.add(dir_name) dirs.append( TreeEntryResponse( type="dir", name=dir_name, path=prefix + dir_name, size_bytes=None, object_id=None, ) ) dirs.sort(key=lambda e: e.name) files.sort(key=lambda e: e.name) return dirs, files async def list_tree( session: AsyncSession, repo_id: str, owner: str, repo_slug: str, ref: str, dir_path: str, manifest: StrDict | None = None, ) -> TreeListResponse: """Build a directory listing for the tree browser via the snapshot manifest. Resolves: ref → HEAD commit → snapshot manifest → directory entries. Returns an empty listing when no snapshot manifest exists for the ref. Pass ``manifest`` to skip the fetch when the caller already has it. """ if manifest is None: manifest = await _get_head_snapshot_manifest(session, repo_id, ref) dirs, files = _manifest_to_tree(manifest or {}, dir_path) return TreeListResponse( owner=owner, repo_slug=repo_slug, ref=ref, dir_path=dir_path.strip("/"), entries=dirs + files, ) async def _resolve_ref_to_commit( session: AsyncSession, repo_id: str, ref: str ) -> MusehubCommit | None: """Resolve a branch name or commit SHA to a commit row. Tries branch lookup first, then falls back to direct commit_id lookup so both ``main`` and full/partial SHAs work. """ # 1. Try branch branch_row = ( await session.execute( select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == ref, ) ) ).scalar_one_or_none() if branch_row and branch_row.head_commit_id: return await session.get(MusehubCommit, branch_row.head_commit_id) # 2. Try exact commit_id match ref_row = await session.get(MusehubCommitRef, (repo_id, ref)) if ref_row is not None: row = await session.get(MusehubCommit, ref) if row: return row # 3. Prefix match (short SHA) if len(ref) >= 7: row = ( await session.execute( select(MusehubCommit) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where( MusehubCommitRef.repo_id == repo_id, MusehubCommit.commit_id.like(f"{ref}%"), ).limit(1) ) ).scalar_one_or_none() if row: return row return None async def get_file_at_ref( session: AsyncSession, repo_id: str, ref: str, file_path: str, ) -> JSONObject | None: """Resolve a file path at a given ref via the snapshot manifest. Looks up: ref → commit → snapshot → manifest[file_path] → object_id, then returns metadata. Content bytes are intentionally NOT returned here (callers fetch via the storage backend directly to avoid loading into memory unless needed). Returns a dict with: - ``object_id``: content-addressed SHA - ``snapshot_id``: the snapshot this file belongs to - ``commit_id``: resolved commit SHA - ``path``: normalised file path Returns None when the ref or file is not found. """ commit = await _resolve_ref_to_commit(session, repo_id, ref) if commit is None or commit.snapshot_id is None: return None from musehub.services.musehub_snapshot import get_snapshot_manifest manifest = await get_snapshot_manifest(session, commit.snapshot_id) norm_path = file_path.lstrip("/") object_id = manifest.get(norm_path) if object_id is None: return None return { "object_id": object_id, "snapshot_id": commit.snapshot_id, "commit_id": commit.commit_id, "path": norm_path, "manifest_size": len(manifest), } async def get_last_commit_for_file( session: AsyncSession, repo_id: str, file_path: str, current_commit_id: str, ) -> MusehubCommit | None: """Return the most recent commit that changed ``file_path``. Fast path: query musehub_symbol_history_entries by (repo_id, address) using the ix_symbol_history_repo_address index — O(1) regardless of repo size. Matches both bare file entries (address == path) and symbol-level entries (address starts with path::). Fallback: when no history entries exist for the file (e.g. the file predates indexing), walks up to 200 snapshot manifests as before. """ norm = file_path.lstrip("/") # ── Fast path: symbol history index ────────────────────────────────────── history_stmt = ( select(MusehubSymbolHistoryEntry) .where( MusehubSymbolHistoryEntry.repo_id == repo_id, (MusehubSymbolHistoryEntry.address == norm) | MusehubSymbolHistoryEntry.address.like(f"{norm}::%"), ) .order_by(desc(MusehubSymbolHistoryEntry.committed_at)) .limit(1) ) history_row = (await session.execute(history_stmt)).scalars().first() if history_row is not None: return await session.get(MusehubCommit, history_row.commit_id) # ── Fallback: snapshot manifest scan ───────────────────────────────────── current_commit = await session.get(MusehubCommit, current_commit_id) if current_commit is None or current_commit.snapshot_id is None: return current_commit stmt = ( select(MusehubCommit) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where( MusehubCommitRef.repo_id == repo_id, MusehubCommit.branch == (current_commit.branch or "main"), MusehubCommit.timestamp <= current_commit.timestamp, ) .order_by(desc(MusehubCommit.timestamp)) .limit(200) ) rows = (await session.execute(stmt)).scalars().all() snapshot_ids = [r.snapshot_id for r in rows if r.snapshot_id] manifests: dict[str, dict] = {} for i in range(0, len(snapshot_ids), 100): chunk = await get_snapshot_manifests_batch(session, snapshot_ids[i:i + 100]) manifests.update(chunk) current_oid = manifests.get(current_commit.snapshot_id, {}).get(norm) if current_oid is None: return None prev_commit: MusehubCommit | None = current_commit for row in rows: if row.snapshot_id is None: continue oid = manifests.get(row.snapshot_id, {}).get(norm) if oid != current_oid: break prev_commit = row return prev_commit async def get_snapshot_diff( session: AsyncSession, repo_id: str, commit_snapshot_id: str | None, parent_snapshot_id: str | None, ) -> JSONObject: """Diff two snapshot manifests, returning file-level change lists. Returns a dict with: - ``added``: files present in the new snapshot but not the parent - ``removed``: files present in the parent but not the new snapshot - ``modified``: files present in both but with different object IDs - ``unchanged``: count only (not listed, to keep payload small) """ from musehub.services.musehub_snapshot import get_snapshot_manifest new_manifest: StrDict = {} old_manifest: StrDict = {} if commit_snapshot_id: new_manifest = await get_snapshot_manifest(session, commit_snapshot_id) if parent_snapshot_id: old_manifest = await get_snapshot_manifest(session, parent_snapshot_id) added: list[str] = sorted(p for p in new_manifest if p not in old_manifest) removed: list[str] = sorted(p for p in old_manifest if p not in new_manifest) modified: list[str] = sorted( p for p in new_manifest if p in old_manifest and new_manifest[p] != old_manifest[p] ) return { "added": added, "removed": removed, "modified": modified, "total_files": len(new_manifest), } async def get_repo_home_stats( session: AsyncSession, repo_id: str, ref: str, manifest: StrDict | None = None, ) -> JSONObject: """Return aggregate stats for the repo home page. Returns a dict with: - ``total_commits``: int — total commit count across all branches - ``total_objects``: int — number of stored objects - ``total_size_bytes``: int — sum of all object sizes - ``commit_activity``: list[int] — daily commit counts for the last 14 days (oldest first) """ from datetime import timedelta total_commits: int = ( await session.execute( select(func.count()).select_from(MusehubCommitRef).where(MusehubCommitRef.repo_id == repo_id) ) ).scalar_one() or 0 obj_agg = ( await session.execute( select( func.count().label("cnt"), func.coalesce(func.sum(MusehubObject.size_bytes), 0).label("sz"), ) .join( MusehubObjectRef, MusehubObject.object_id == MusehubObjectRef.object_id, ) .where(MusehubObjectRef.repo_id == repo_id) ) ).one() total_objects = int(obj_agg.cnt or 0) total_size_bytes = int(obj_agg.sz or 0) # File count from HEAD snapshot manifest if manifest is None: manifest = await _get_head_snapshot_manifest(session, repo_id, ref) # Daily commit activity for last 14 days now = datetime.now(tz=timezone.utc) fourteen_days_ago = now - timedelta(days=14) recent_rows = ( await session.execute( select(MusehubCommit.timestamp) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where( MusehubCommitRef.repo_id == repo_id, MusehubCommit.timestamp >= fourteen_days_ago, ) .order_by(MusehubCommit.timestamp) ) ).scalars().all() # Bucket into 14 daily bins daily: list[int] = [0] * 14 for ts in recent_rows: t = ts if ts.tzinfo else ts.replace(tzinfo=timezone.utc) day_idx = (now - t).days if 0 <= day_idx < 14: daily[13 - day_idx] += 1 return { "total_commits": total_commits, "total_objects": total_objects, "total_size_bytes": total_size_bytes, "commit_activity": daily, "total_files": len(manifest), } async def get_file_last_commits( session: AsyncSession, repo_id: str, paths: list[str], ref: str = "", max_commits: int = 60, ) -> FileLastCommits: """Return the last-touching commit for each file/directory path. Reads from the materialized musehub_file_last_commits table (populated at push time). Falls back to the old blob-walk if the table has no rows for this repo/branch (e.g. repos pushed before the migration). ``ref`` should be the branch name. When empty, falls back to reading the repo's default branch. """ import re as _re from datetime import timezone as _tz if not paths: return {} branch = ref or "" def _row_to_record(row: MusehubFileLastCommit) -> StrDict: ts = row.commit_timestamp if ts.tzinfo is None: ts = ts.replace(tzinfo=_tz.utc) model_id: str = row.model_id or "" parts = model_id.replace("claude-", "").split("-") model_label = parts[0] if parts and parts[0] else model_id msg = (row.commit_message or "").split("\n")[0][:72] _ct = _re.match(r"^(feat|fix|docs|style|refactor|perf|test|chore|build|ci|revert)(\([^)]+\))?(!)?:", msg.strip()) commit_type = _ct.group(1) if _ct else "" return { "sha": row.commit_id, "message": msg, "author": row.commit_author or "", "timestamp": ts.isoformat(), "agentId": row.agent_id or "", "modelId": model_id, "modelLabel": model_label, "commitType": commit_type, } # Fetch all rows for this repo+branch in one query, then filter in Python. rows_result = await session.execute( select(MusehubFileLastCommit).where( MusehubFileLastCommit.repo_id == repo_id, MusehubFileLastCommit.branch == branch, ) ) all_rows: list[MusehubFileLastCommit] = list(rows_result.scalars().all()) if all_rows: # Build path→row map for exact lookups. by_path: dict[str, MusehubFileLastCommit] = {r.path: r for r in all_rows} result: FileLastCommits = {} for p in paths: if p in by_path: result[p] = _row_to_record(by_path[p]) else: # Directory: find the most recently touched file under this prefix. prefix = f"{p.rstrip('/')}/" best: MusehubFileLastCommit | None = None for row in all_rows: if row.path.startswith(prefix): if best is None or row.commit_timestamp > best.commit_timestamp: best = row if best is not None: result[p] = _row_to_record(best) return result # --- Fallback: no materialized data yet — walk blobs (old behaviour). --- commits_result = await session.execute( select(MusehubCommit) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where(MusehubCommitRef.repo_id == repo_id) .order_by(MusehubCommit.timestamp.desc()) .limit(max_commits) ) commits = list(commits_result.scalars().all()) from musehub.services.musehub_snapshot import get_snapshot_manifests_batch snap_ids = [c.snapshot_id for c in commits if c.snapshot_id] if not snap_ids: return {} snap_by_id = await get_snapshot_manifests_batch(session, snap_ids) def _commit_record(c: MusehubCommit) -> StrDict: ts = c.timestamp if ts.tzinfo is None: ts = ts.replace(tzinfo=_tz.utc) agent_id: str = c.agent_id or "" model_id: str = c.model_id or "" parts = model_id.replace("claude-", "").split("-") model_label = parts[0] if parts and parts[0] else model_id _ct = _re.match(r"^(feat|fix|docs|style|refactor|perf|test|chore|build|ci|revert)(\([^)]+\))?(!)?:", c.message.strip()) commit_type = _ct.group(1) if _ct else "" return { "sha": c.commit_id, "message": c.message.split("\n")[0][:72], "author": c.author, "timestamp": ts.isoformat(), "agentId": agent_id, "modelId": model_id, "modelLabel": model_label, "commitType": commit_type, } fb_result: FileLastCommits = {} remaining = set(paths) prev_manifest: StrDict = {} prev_commit: MusehubCommit | None = None for commit in commits: if not remaining: break cur_manifest = snap_by_id.get(commit.snapshot_id or "", {}) if prev_commit is not None: claimed: set[str] = set() for p in list(remaining): cur_oid = cur_manifest.get(p) prev_oid = prev_manifest.get(p) if prev_oid and cur_oid != prev_oid: fb_result[p] = _commit_record(prev_commit) claimed.add(p) remaining -= claimed prev_manifest = cur_manifest prev_commit = commit if prev_commit is not None: for p in list(remaining): if prev_manifest.get(p): fb_result[p] = _commit_record(prev_commit) unclaimed_dirs = set(paths) - set(fb_result.keys()) for dir_path in unclaimed_dirs: prefix = f"{dir_path.rstrip('/')}/" dir_prev_manifest: StrDict = {} dir_prev_commit: MusehubCommit | None = None for commit in commits: cur_m = snap_by_id.get(commit.snapshot_id or "", {}) if dir_prev_commit is not None: if any( fp.startswith(prefix) and cur_m.get(fp) != dir_prev_manifest.get(fp) for fp in dir_prev_manifest if fp.startswith(prefix) ): fb_result[dir_path] = _commit_record(dir_prev_commit) break dir_prev_manifest = cur_m dir_prev_commit = commit if dir_path not in fb_result and dir_prev_commit is not None: if any(fp.startswith(prefix) for fp in dir_prev_manifest): fb_result[dir_path] = _commit_record(dir_prev_commit) return fb_result async def get_recently_pushed_branches( session: AsyncSession, repo_id: str, current_ref: str, within_hours: int = 72, ) -> list[StrDict]: """Return branches (other than current_ref) whose head commit is recent. Used to render GitHub-style "branch had recent pushes N minutes ago" banners. Returns list of ``{name, sha, message, timestamp}`` sorted newest-first. """ from datetime import timezone as _tz, timedelta branches_result = await session.execute( select(MusehubBranch).where(MusehubBranch.repo_id == repo_id) ) branches = [ b for b in branches_result.scalars().all() if b.name != current_ref and b.head_commit_id ] if not branches: return [] head_ids = [b.head_commit_id for b in branches if b.head_commit_id] commits_result = await session.execute( select(MusehubCommit).where(MusehubCommit.commit_id.in_(head_ids)) ) commit_by_id = { c.commit_id: c for c in commits_result.scalars().all() } cutoff = datetime.now(_tz.utc) - timedelta(hours=within_hours) recent = [] for branch in branches: commit = commit_by_id.get(branch.head_commit_id or "") if not commit: continue ts = commit.timestamp if ts.tzinfo is None: ts = ts.replace(tzinfo=_tz.utc) if ts >= cutoff: recent.append({ "name": branch.name, "sha": commit.commit_id, "message": commit.message.split("\n")[0][:72], "timestamp": ts.isoformat(), }) recent.sort(key=lambda x: x["timestamp"], reverse=True) return recent async def _unique_slug_for_owner( db_session: AsyncSession, owner: str, base_slug: str, ) -> str: """Return a slug that does not collide with any existing (non-deleted) repo for ``owner``. If ``base_slug`` is taken it appends ``-2``, ``-3``, … (up to ``-100``) until it finds an available name. This mirrors GitHub's fork-naming behaviour and prevents the duplicate-slug ``IntegrityError`` from being misreported as a duplicate-fork 409 at the route layer. Raises ``ValueError("cannot_generate_unique_slug")`` if all 100 suffixes are already taken — an extreme edge case that should never occur in practice. """ slug = base_slug for suffix in range(2, 101): exists = ( await db_session.execute( select(MusehubRepo.repo_id).where( MusehubRepo.owner == owner, MusehubRepo.slug == slug, ) ) ).scalar_one_or_none() if exists is None: return slug slug = f"{base_slug[:60]}-{suffix}" raise ValueError("cannot_generate_unique_slug") async def fork_repo( db_session: AsyncSession, *, source_repo_id: str, forked_by_handle: str, request: ForkRepoRequest, ) -> UserForkedRepoEntry: """Fork a repo — creates a new public repo owned by ``forked_by_handle`` and records the relationship in ``musehub_forks``. Rules enforced here (callers must also enforce via HTTP layer): - Source repo must exist and not be soft-deleted. - A handle cannot fork a repo they already own. - The same handle cannot fork the same source repo twice (unique constraint). Slug collisions with repos the caller already owns are resolved automatically by appending a numeric suffix (``-2``, ``-3``, …), matching GitHub behaviour. Returns the newly created fork entry with source attribution. Raises ``ValueError`` for business-rule violations; callers map these to HTTP errors. Raises ``sqlalchemy.exc.IntegrityError`` on duplicate fork (unique constraint). """ # Resolve source repo. source_row = ( await db_session.execute( select(MusehubRepo).where( MusehubRepo.repo_id == source_repo_id, ) ) ).scalar_one_or_none() if source_row is None: raise ValueError("source_repo_not_found") if source_row.visibility == "private": raise ValueError("source_repo_private") if source_row.owner == forked_by_handle: raise ValueError("cannot_fork_own_repo") # Pre-check: reject duplicate fork before touching the repo table so that # any IntegrityError that reaches the caller is unambiguously a duplicate # fork, not a slug collision. duplicate = ( await db_session.execute( select(MusehubFork.fork_id).where( MusehubFork.source_repo_id == source_repo_id, MusehubFork.forked_by == forked_by_handle, ) ) ).scalar_one_or_none() if duplicate is not None: raise ValueError("duplicate_fork") # Determine fork repo name/description. Auto-suffix slug to avoid collision # with repos the caller already owns (mirrors GitHub behaviour). fork_name = request.name or source_row.name fork_description = ( request.description or f"Fork of {source_row.owner}/{source_row.slug}: {source_row.description}".rstrip(": ") ) base_slug = _generate_slug(fork_name) fork_slug = await _unique_slug_for_owner(db_session, forked_by_handle, base_slug) # Create the fork repo. from datetime import datetime, timezone _fork_created_at = datetime.now(timezone.utc) _fork_owner_id = compute_identity_id(forked_by_handle.encode()) _fork_domain_id = source_row.domain_id or "code" fork_repo_row = MusehubRepo( repo_id=compute_repo_id(_fork_owner_id, fork_slug, _fork_domain_id, _fork_created_at.isoformat()), name=fork_name, owner=forked_by_handle, slug=fork_slug, visibility=request.visibility or "public", owner_user_id=forked_by_handle, description=fork_description, tags=list(source_row.tags or []), settings=None, created_at=_fork_created_at, updated_at=_fork_created_at, domain_id=_fork_domain_id, default_branch="main", ) db_session.add(fork_repo_row) await db_session.flush() await db_session.refresh(fork_repo_row) # Create the fork relationship record. _fork_now = _fork_created_at fork_record = MusehubFork( fork_id=compute_fork_id(source_repo_id, fork_repo_row.repo_id, _fork_now.isoformat()), source_repo_id=source_repo_id, fork_repo_id=fork_repo_row.repo_id, forked_by=forked_by_handle, created_at=_fork_now, ) db_session.add(fork_record) await db_session.flush() await db_session.refresh(fork_record) logger.info( "✅ Forked repo %s (%s/%s) → %s (%s/%s) by %s", source_repo_id, source_row.owner, source_row.slug, fork_repo_row.repo_id, forked_by_handle, fork_slug, forked_by_handle, ) return UserForkedRepoEntry( fork_id=fork_record.fork_id, fork_repo=_to_repo_response(fork_repo_row), source_owner=source_row.owner, source_slug=source_row.slug, forked_at=fork_record.created_at, ) async def get_user_forks( db_session: AsyncSession, username: str, visible_to_user: str | None = None, ) -> UserForksResponse: """Return repos that ``username`` has forked, with source attribution. Joins ``musehub_forks`` (where ``forked_by`` matches the given username) with ``musehub_repos`` twice — once for the fork repo and once for the source repo's owner/slug — so callers can render "forked from {source_owner}/{source_slug}" on each card. Private forks are only visible when ``visible_to_user == username`` (the fork owner can see their own private forks; unauthenticated or third-party callers see only public forks). Returns forks ordered newest-first. Soft-deleted repos on either side of the relationship are excluded. """ SourceRepo = aliased(MusehubRepo, name="source_repo") ForkRepo = aliased(MusehubRepo, name="fork_repo") base_stmt = ( select(MusehubFork, ForkRepo, SourceRepo) .join(ForkRepo, MusehubFork.fork_repo_id == ForkRepo.repo_id) .join(SourceRepo, MusehubFork.source_repo_id == SourceRepo.repo_id) .where( MusehubFork.forked_by == username, ) .order_by(desc(MusehubFork.created_at)) ) # Unauthenticated callers and third parties see only public forks. # The fork owner (visible_to_user == username) sees all their forks. stmt = ( base_stmt if visible_to_user == username else base_stmt.where(ForkRepo.visibility == "public") ) rows = (await db_session.execute(stmt)).all() entries: list[UserForkedRepoEntry] = [ UserForkedRepoEntry( fork_id=fork_rec.fork_id, fork_repo=_to_repo_response(fork_row), source_owner=src_row.owner, source_slug=src_row.slug, forked_at=fork_rec.created_at, ) for fork_rec, fork_row, src_row in rows ] return UserForksResponse(forks=entries, total=len(entries)) async def list_repo_forks_flat( db_session: AsyncSession, repo_id: str, ) -> UserForksResponse: """Return a flat list of all public direct forks of ``repo_id``. Each entry contains full fork repo metadata plus source owner/slug attribution. Ordered newest-first. Soft-deleted fork repos and private forks are excluded — private forks are hidden from the source repo's fork list unconditionally (a fork owner's private fork is discoverable only via their own forks list). """ source_row = ( await db_session.execute( select(MusehubRepo).where( MusehubRepo.repo_id == repo_id, ) ) ).scalar_one_or_none() if source_row is None: return UserForksResponse(forks=[], total=0) ForkRepoAlias = aliased(MusehubRepo) rows = ( await db_session.execute( select(MusehubFork, ForkRepoAlias) .join(ForkRepoAlias, MusehubFork.fork_repo_id == ForkRepoAlias.repo_id) .where( MusehubFork.source_repo_id == repo_id, ForkRepoAlias.visibility == "public", ) .order_by(desc(MusehubFork.created_at)) ) ).all() entries: list[UserForkedRepoEntry] = [ UserForkedRepoEntry( fork_id=fork_rec.fork_id, fork_repo=_to_repo_response(fork_row), source_owner=source_row.owner, source_slug=source_row.slug, forked_at=fork_rec.created_at, ) for fork_rec, fork_row in rows ] return UserForksResponse(forks=entries, total=len(entries)) async def list_repo_forks( db_session: AsyncSession, repo_id: str, ) -> ForkNetworkResponse: """Return the fork network tree rooted at the given repo. The root node represents the source repo. Its ``children`` are direct forks; each child carries its own ``children`` for second-level forks, and so on. ``divergence_commits`` is always 0 at this time — commit-graph divergence counting is deferred to a future index. Private forks are excluded unconditionally — private fork owners' forks are discoverable only via their own forks list, not via the source repo's network. Returns an empty-root ``ForkNetworkResponse`` when the repo does not exist. """ source_row = ( await db_session.execute( select(MusehubRepo).where( MusehubRepo.repo_id == repo_id, ) ) ).scalar_one_or_none() if source_row is None: return ForkNetworkResponse( root=ForkNetworkNode( owner="", repo_slug="", repo_id=repo_id, divergence_commits=0, forked_by="", forked_at=None, ), total_forks=0, ) # Load all forks in a single query and build the tree in Python. # For most repos the fork count is small; if it ever grows large a # recursive CTE would replace this approach. ForkRepoAlias = aliased(MusehubRepo) fork_rows = ( await db_session.execute( select(MusehubFork, ForkRepoAlias) .join(ForkRepoAlias, MusehubFork.fork_repo_id == ForkRepoAlias.repo_id) .where( MusehubFork.source_repo_id == repo_id, ForkRepoAlias.visibility == "public", ) .order_by(MusehubFork.created_at) ) ).all() children: list[ForkNetworkNode] = [ ForkNetworkNode( owner=fork_repo.owner, repo_slug=fork_repo.slug, repo_id=fork_repo.repo_id, divergence_commits=0, forked_by=fork_rec.forked_by, forked_at=fork_rec.created_at, children=[], ) for fork_rec, fork_repo in fork_rows ] root = ForkNetworkNode( owner=source_row.owner, repo_slug=source_row.slug, repo_id=source_row.repo_id, divergence_commits=0, forked_by="", forked_at=None, children=children, ) return ForkNetworkResponse(root=root, total_forks=len(children)) # ── Repo settings helpers ───────────────────────────────────────────────────── _SETTINGS_DEFAULTS: JSONObject = { "default_branch": "main", "has_issues": True, "has_projects": False, "has_wiki": False, "license": None, "homepage_url": None, "allow_merge_commit": True, "allow_squash_merge": True, "allow_rebase_merge": False, "delete_branch_on_merge": True, } def _merge_settings(stored: JSONObject | None) -> JSONObject: """Return a complete settings dict by filling missing keys with defaults. ``stored`` may be None (new repos) or a partial dict (old rows that predate individual flag additions). Defaults are applied for any absent key so callers always receive a fully-populated dict. """ base = dict(_SETTINGS_DEFAULTS) if stored: base.update(stored) return base async def get_repo_settings( session: AsyncSession, repo_id: str ) -> RepoSettingsResponse | None: """Return the mutable settings for a repo, or None if the repo does not exist. Combines dedicated column values (name, description, visibility, tags) with feature-flag values from the ``settings`` JSON blob. Missing flags are back-filled with ``_SETTINGS_DEFAULTS`` so new and legacy repos both return a complete response. Called by ``GET /api/repos/{repo_id}/settings``. """ row = await session.get(MusehubRepo, repo_id) if row is None: return None flags = _merge_settings(row.settings) # Derive default_branch from stored flag; fall back to "main" default_branch = str(flags.get("default_branch") or "main") return RepoSettingsResponse( name=row.name, description=row.description, visibility=row.visibility, default_branch=default_branch, has_issues=bool(flags.get("has_issues", True)), has_projects=bool(flags.get("has_projects", False)), has_wiki=bool(flags.get("has_wiki", False)), topics=list(row.tags or []), license=str(flags["license"]) if flags.get("license") is not None else None, homepage_url=str(flags["homepage_url"]) if flags.get("homepage_url") is not None else None, allow_merge_commit=bool(flags.get("allow_merge_commit", True)), allow_squash_merge=bool(flags.get("allow_squash_merge", True)), allow_rebase_merge=bool(flags.get("allow_rebase_merge", False)), delete_branch_on_merge=bool(flags.get("delete_branch_on_merge", True)), domain_id=row.domain_id, marketplace_domain_id=row.marketplace_domain_id, ) async def update_repo_settings( session: AsyncSession, repo_id: str, patch: RepoSettingsPatch, *, caller_user_id: str | None = None, ) -> RepoSettingsResponse | None: """Apply a partial settings update to a repo and return the updated settings. Only non-None fields in ``patch`` are written. Dedicated columns (name, description, visibility, tags) are updated directly on the ORM row; feature flags are merged into the ``settings`` JSON blob. Returns None if the repo does not exist. The caller is responsible for committing the session after a successful return. Raises ``ValueError("marketplace_domain_not_found")`` if ``patch.marketplace_domain_id`` is a non-empty string that doesn't match any registered ``MusehubDomain`` — callers must map this to a 404. ``caller_user_id`` is required whenever ``marketplace_domain_id`` is set in the patch (musehub#117 DOM_12/DOM_13 — install/uninstall bookkeeping is tracked per user). Called by ``PATCH /api/repos/{repo_id}/settings``. """ row = await session.get(MusehubRepo, repo_id) if row is None: return None # ── Dedicated column fields ────────────────────────────────────────────── if patch.name is not None: row.name = patch.name if patch.description is not None: row.description = patch.description if patch.visibility is not None: row.visibility = patch.visibility if patch.topics is not None: row.tags = patch.topics if patch.domain_id is not None: row.domain_id = patch.domain_id # ── Marketplace domain link (musehub#117 Phase 3) ──────────────────────── if patch.marketplace_domain_id is not None: from musehub.services import musehub_domains as _musehub_domains previous_domain_id = row.marketplace_domain_id new_domain_id = patch.marketplace_domain_id or None # "" clears the link if new_domain_id is not None: target = await _musehub_domains.get_domain_by_id(session, new_domain_id) if target is None: raise ValueError("marketplace_domain_not_found") if new_domain_id != previous_domain_id: row.marketplace_domain_id = new_domain_id if caller_user_id is not None: if previous_domain_id is not None: await _musehub_domains.record_domain_uninstall( session, caller_user_id, previous_domain_id ) if new_domain_id is not None: await _musehub_domains.record_domain_install( session, caller_user_id, new_domain_id ) # ── Feature-flag JSON blob ─────────────────────────────────────────────── current_flags = _merge_settings(row.settings) flag_updates: JSONObject = {} if patch.default_branch is not None: flag_updates["default_branch"] = patch.default_branch if patch.has_issues is not None: flag_updates["has_issues"] = patch.has_issues if patch.has_projects is not None: flag_updates["has_projects"] = patch.has_projects if patch.has_wiki is not None: flag_updates["has_wiki"] = patch.has_wiki if patch.license is not None: flag_updates["license"] = patch.license if patch.homepage_url is not None: flag_updates["homepage_url"] = patch.homepage_url if patch.allow_merge_commit is not None: flag_updates["allow_merge_commit"] = patch.allow_merge_commit if patch.allow_squash_merge is not None: flag_updates["allow_squash_merge"] = patch.allow_squash_merge if patch.allow_rebase_merge is not None: flag_updates["allow_rebase_merge"] = patch.allow_rebase_merge if patch.delete_branch_on_merge is not None: flag_updates["delete_branch_on_merge"] = patch.delete_branch_on_merge if flag_updates: current_flags.update(flag_updates) row.settings = current_flags logger.info("✅ Updated settings for repo %s", repo_id) return await get_repo_settings(session, repo_id)