state-hub/api/services/repository_aliases.py

"""Canonical repository identity resolution across current and prior slugs.

The slug registry is the lookup boundary.  Historical records deliberately keep
the slug they recorded; callers use ``slug_values`` when they need an identity-
wide read and ``canonical_slug`` when they create a new reference.
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass

from fastapi import HTTPException
from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import noload

from api.models.fabric_graph import FabricGraphEdge, FabricGraphImport, FabricGraphNode
from api.models.managed_repo import ManagedRepo
from api.models.repository_rename import RepositorySlug


@dataclass(frozen=True)
class RepositorySlugResolution:
    repo: ManagedRepo
    requested_slug: str
    canonical_slug: str
    slug_status: str
    aliases: tuple[str, ...]
    source_operation_id: uuid.UUID | None = None

    @property
    def slug_values(self) -> tuple[str, ...]:
        return (self.canonical_slug, *self.aliases)


async def resolve_repository_slug(
    session: AsyncSession,
    slug: str,
    *,
    required: bool = True,
) -> RepositorySlugResolution | None:
    """Resolve a current or protected prior slug to one repository UUID."""

    record = (
        await session.execute(
            select(RepositorySlug).where(RepositorySlug.slug == slug)
        )
    ).scalar_one_or_none()

    # Compatibility for databases upgraded before the identity backfill has run.
    if record is None:
        repo = (
            await session.execute(select(ManagedRepo).where(ManagedRepo.slug == slug))
        ).scalar_one_or_none()
        if repo is None:
            if required:
                raise HTTPException(status_code=404, detail=f"Repo '{slug}' not found")
            return None
        return RepositorySlugResolution(
            repo=repo,
            requested_slug=slug,
            canonical_slug=repo.slug,
            slug_status="canonical",
            aliases=(),
        )

    repo = await session.get(ManagedRepo, record.repo_id)
    if repo is None:
        raise HTTPException(
            status_code=409,
            detail=f"Repository slug registry entry '{slug}' has no repository",
        )
    records = list(
        (
            await session.execute(
                select(RepositorySlug)
                .where(RepositorySlug.repo_id == repo.id)
                .order_by(RepositorySlug.kind.desc(), RepositorySlug.slug)
            )
        ).scalars()
    )
    canonicals = [item for item in records if item.kind == "canonical"]
    if len(canonicals) != 1 or canonicals[0].slug != repo.slug:
        raise HTTPException(
            status_code=409,
            detail=f"Repository '{repo.id}' has inconsistent canonical slug state",
        )
    return RepositorySlugResolution(
        repo=repo,
        requested_slug=slug,
        canonical_slug=repo.slug,
        slug_status=record.kind,
        aliases=tuple(item.slug for item in records if item.kind == "alias"),
        source_operation_id=record.source_operation_id,
    )


async def repository_resolutions_for_repositories(
    session: AsyncSession,
    repositories: list[ManagedRepo],
) -> dict[uuid.UUID, RepositorySlugResolution]:
    """Resolve a repository collection with one slug-registry query.

    Collection callers already hold canonical ``ManagedRepo`` rows. Resolving
    every row through :func:`resolve_repository_slug` would issue two identity
    queries per repository. Load the complete alias sets once instead while
    retaining the pre-registry compatibility behavior.
    """

    if not repositories:
        return {}

    repository_ids = [repository.id for repository in repositories]
    records = list(
        (
            await session.execute(
                select(RepositorySlug)
                .options(
                    noload(RepositorySlug.repo),
                    noload(RepositorySlug.source_operation),
                )
                .where(RepositorySlug.repo_id.in_(repository_ids))
                .order_by(RepositorySlug.repo_id, RepositorySlug.kind.desc(), RepositorySlug.slug)
            )
        ).scalars()
    )
    records_by_repository: dict[uuid.UUID, list[RepositorySlug]] = {
        repository_id: [] for repository_id in repository_ids
    }
    for record in records:
        records_by_repository[record.repo_id].append(record)

    resolutions: dict[uuid.UUID, RepositorySlugResolution] = {}
    for repository in repositories:
        repository_records = records_by_repository[repository.id]
        requested_record = next(
            (record for record in repository_records if record.slug == repository.slug),
            None,
        )
        if requested_record is None:
            # Compatibility for databases upgraded before the identity backfill.
            resolutions[repository.id] = RepositorySlugResolution(
                repo=repository,
                requested_slug=repository.slug,
                canonical_slug=repository.slug,
                slug_status="canonical",
                aliases=(),
            )
            continue

        canonicals = [record for record in repository_records if record.kind == "canonical"]
        if len(canonicals) != 1 or canonicals[0].slug != repository.slug:
            raise HTTPException(
                status_code=409,
                detail=f"Repository '{repository.id}' has inconsistent canonical slug state",
            )
        resolutions[repository.id] = RepositorySlugResolution(
            repo=repository,
            requested_slug=repository.slug,
            canonical_slug=repository.slug,
            slug_status=requested_record.kind,
            aliases=tuple(
                record.slug for record in repository_records if record.kind == "alias"
            ),
            source_operation_id=requested_record.source_operation_id,
        )

    return resolutions


async def canonicalize_repository_slug(session: AsyncSession, value: str) -> str:
    """Canonicalize a value only when it is a registered repository identity."""

    resolution = await resolve_repository_slug(session, value, required=False)
    return resolution.canonical_slug if resolution is not None else value


async def repository_resolution_for_id(
    session: AsyncSession, repo: ManagedRepo
) -> RepositorySlugResolution:
    return await resolve_repository_slug(session, repo.slug)  # type: ignore[return-value]


async def stale_external_references(
    session: AsyncSession,
    resolution: RepositorySlugResolution,
) -> list[dict[str, object]]:
    """Name external projections that still contain an historical slug.

    These are handoffs to the projection owner, never implicit rewrite targets.
    """

    if not resolution.aliases:
        return []
    aliases = list(resolution.aliases)
    checks = (
        (FabricGraphImport, "source_repo_slug"),
        (FabricGraphNode, "source_repo_slug"),
        (FabricGraphNode, "repo_slug"),
        (FabricGraphEdge, "source_repo_slug"),
    )
    stale: list[dict[str, object]] = []
    for model, field_name in checks:
        field = getattr(model, field_name)
        rows = (
            await session.execute(
                select(field, func.count()).where(field.in_(aliases)).group_by(field)
            )
        ).all()
        for value, count in rows:
            stale.append(
                {
                    "owner": "railiance-fabric",
                    "surface": model.__tablename__,
                    "field": field_name,
                    "value": value,
                    "count": count,
                    "status": "stale",
                    "handoff": "owner update or re-ingest required",
                }
            )
    return stale


def affected_slug_predicate(field: object, slugs: tuple[str, ...]):
    """Match a JSONB slug-list against any name in one repository lineage."""

    return or_(*(field.contains([slug]) for slug in slugs))