"""Primary-only fast path for one forge-derived repository projection.""" from __future__ import annotations import asyncio from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from api.config import settings from api.database import get_session from api.models.task import Task from api.models.workplan import Workplan from api.models.workplan_dependency import WorkplanDependency from api.schemas.work_record_projection import ( RepositoryProjectionReconcile, RepositoryProjectionSnapshot, ) from api.services.forge_projection import ( ForgeDeriveError, ForgeUnreadableError, derive_from_forge, reset_repository_projection, ) from api.services.repository_aliases import resolve_repository_slug router = APIRouter(prefix="/repos", tags=["repository-work-record-projection"]) @router.get( "/{slug}/work-record-projection/snapshot", response_model=RepositoryProjectionSnapshot, ) async def repository_work_record_snapshot( slug: str, session: AsyncSession = Depends(get_session), ) -> RepositoryProjectionSnapshot: """Return all consistency-check inputs for one repository in one request.""" resolution = await resolve_repository_slug(session, slug) workplans = list( ( await session.execute( select(Workplan) .where(Workplan.repo_id == resolution.repo.id) .order_by(Workplan.slug) ) ).scalars() ) workplan_ids = [workplan.id for workplan in workplans] tasks: list[Task] = [] dependencies: list[WorkplanDependency] = [] if workplan_ids: tasks = list( ( await session.execute( select(Task) .where(Task.workplan_id.in_(workplan_ids)) .order_by(Task.workplan_id, Task.id) ) ).scalars() ) dependencies = list( ( await session.execute( select(WorkplanDependency) .where(WorkplanDependency.from_workplan_id.in_(workplan_ids)) .order_by( WorkplanDependency.from_workplan_id, WorkplanDependency.id ) ) ).scalars() ) return RepositoryProjectionSnapshot( schema="state-hub.repository-projection-snapshot.v1", repo_slug=resolution.canonical_slug, repo_id=resolution.repo.id, workplans=workplans, tasks=tasks, dependencies=dependencies, ) @router.post("/{slug}/work-record-projection/reconcile") async def reconcile_repository_work_records( slug: str, body: RepositoryProjectionReconcile, session: AsyncSession = Depends(get_session), ) -> dict: """Derive an exact pushed commit centrally and apply it transactionally.""" if settings.state_hub_instance_role != "primary": raise HTTPException( status_code=409, detail={ "message": "repository projection writes require the primary State Hub", "instance_role": settings.state_hub_instance_role, "instance_label": settings.state_hub_instance_label, }, ) try: derived = await asyncio.to_thread(derive_from_forge, slug) except ForgeUnreadableError as exc: raise HTTPException( status_code=424, detail={ "message": "repository is unreadable from the forge", "detail": str(exc)[:300], }, ) from exc except ForgeDeriveError as exc: raise HTTPException( status_code=502, detail={ "message": "forge projection derivation failed", "detail": str(exc)[:300], }, ) from exc if derived.commit.lower() != body.expected_commit: raise HTTPException( status_code=409, detail={ "message": "forge default branch is not at the expected commit", "expected_commit": body.expected_commit, "derived_commit": derived.commit, }, ) outcome = await reset_repository_projection( session, slug, acknowledge_retirements=body.acknowledge_retirements, derived=derived, ) if outcome.status in {"applied", "noop"} or outcome.released: await session.commit() else: await session.rollback() from api.routers.workstreams import _invalidate_workplan_index_cache _invalidate_workplan_index_cache() return { "schema": "state-hub.repository-projection-reconcile.v1", "instance_role": settings.state_hub_instance_role, "instance_label": settings.state_hub_instance_label, "expected_commit": body.expected_commit, "derived_commit": derived.commit, "outcome": outcome.to_dict(), }