diff --git a/api/routers/fabric.py b/api/routers/fabric.py index bb7c921..98e704b 100644 --- a/api/routers/fabric.py +++ b/api/routers/fabric.py @@ -1,5 +1,6 @@ from __future__ import annotations +import uuid from typing import Any import httpx @@ -10,6 +11,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from api.database import get_session from api.models.fabric_graph import FabricGraphEdge, FabricGraphImport, FabricGraphNode from api.schemas.fabric_graph import ( + FabricGraphActivationRequest, + FabricGraphActivationResult, FabricGraphEdgeRead, FabricGraphImportRead, FabricGraphIngestResult, @@ -19,6 +22,7 @@ from api.schemas.fabric_graph import ( ) from api.services.fabric_graph import ( FabricGraphValidationError, + activate_fabric_graph_import, ingest_fabric_graph_export, record_fabric_graph_error, split_graph_ingest_body, @@ -132,6 +136,34 @@ async def latest_graph_import( return FabricGraphImportRead.model_validate(import_run) +@router.post( + "/graph-exports/{import_id}/activate", + response_model=FabricGraphActivationResult, +) +async def activate_graph_import( + import_id: uuid.UUID, + body: FabricGraphActivationRequest | None = None, + session: AsyncSession = Depends(get_session), +) -> FabricGraphActivationResult: + request = body or FabricGraphActivationRequest() + try: + import_run, previous_import_id, activated = await activate_fabric_graph_import( + session, + import_id, + requested_by=request.requested_by, + ) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + if import_run is None: + raise HTTPException(status_code=404, detail=f"Fabric graph import '{import_id}' not found") + return FabricGraphActivationResult( + import_run=FabricGraphImportRead.model_validate(import_run), + previous_import_id=previous_import_id, + activated=activated, + idempotent=not activated, + ) + + @router.get("/graph/nodes", response_model=list[FabricGraphNodeRead]) async def list_graph_nodes( source_repo_slug: str = "railiance-fabric", diff --git a/api/schemas/fabric_graph.py b/api/schemas/fabric_graph.py index 3c9e98e..6afe417 100644 --- a/api/schemas/fabric_graph.py +++ b/api/schemas/fabric_graph.py @@ -261,6 +261,17 @@ class FabricGraphIngestResult(BaseModel): edge_count: int +class FabricGraphActivationRequest(BaseModel): + requested_by: str = "operator" + + +class FabricGraphActivationResult(BaseModel): + import_run: FabricGraphImportRead + previous_import_id: uuid.UUID | None = None + activated: bool + idempotent: bool + + class FabricGraphNodeRead(BaseModel): model_config = ConfigDict(from_attributes=True) diff --git a/api/services/fabric_graph.py b/api/services/fabric_graph.py index 8a98bf3..48f1fef 100644 --- a/api/services/fabric_graph.py +++ b/api/services/fabric_graph.py @@ -221,6 +221,55 @@ async def ingest_fabric_graph_export( return import_run, True, False +async def activate_fabric_graph_import( + session: AsyncSession, + import_id: Any, + *, + requested_by: str, +) -> tuple[FabricGraphImport | None, Any | None, bool]: + """Select a retained valid import as the active read model. + + Imports are immutable. Activation only moves the per-source ``is_latest`` + marker, which makes cutover rollback fast and avoids re-fetching an older + authority payload during an incident. + """ + result = await session.execute( + select(FabricGraphImport).where(FabricGraphImport.id == import_id) + ) + import_run = result.scalar_one_or_none() + if import_run is None: + return None, None, False + if import_run.validation_status != "valid": + raise ValueError("Only a valid Fabric graph import can be activated.") + + previous_result = await session.execute( + select(FabricGraphImport).where( + FabricGraphImport.source_repo_slug == import_run.source_repo_slug, + FabricGraphImport.is_latest.is_(True), + ) + ) + previous = previous_result.scalars().first() + if previous is not None and previous.id == import_run.id: + return import_run, previous.id, False + + await _mark_latest(session, import_run) + import_run.last_seen_at = datetime.now(timezone.utc) + await _record_progress( + session, + "Fabric graph read model activation changed.", + { + "source_repo_slug": import_run.source_repo_slug, + "import_id": str(import_run.id), + "previous_import_id": str(previous.id) if previous else None, + "content_hash": import_run.content_hash, + "requested_by": requested_by, + }, + ) + await session.commit() + await session.refresh(import_run) + return import_run, previous.id if previous else None, True + + def validate_fabric_graph_export(payload: dict[str, Any]) -> FabricGraphExportPayload: try: export = FabricGraphExportPayload.model_validate(payload) diff --git a/docs/fabric-graph-read-model.md b/docs/fabric-graph-read-model.md index 15ddcec..da83e4c 100644 --- a/docs/fabric-graph-read-model.md +++ b/docs/fabric-graph-read-model.md @@ -61,6 +61,20 @@ curl -s http://127.0.0.1:8000/fabric/graph-exports/latest curl -s http://127.0.0.1:8000/fabric/graph/summary ``` +Imports are immutable and retained. To roll the read model back to a previously +validated import without depending on the source service, activate its import +record; repeat the operation with the newer import id to roll forward again: + +```bash +curl -s -X POST \ + http://127.0.0.1:8000/fabric/graph-exports//activate \ + -H "Content-Type: application/json" \ + -d '{"requested_by":"operator-rollback"}' +``` + +Activation only moves the per-source latest marker. It does not alter or delete +either graph snapshot, and activating the current import is idempotent. + Representative relationship queries: ```bash diff --git a/tests/test_routers_core.py b/tests/test_routers_core.py index 49ed399..42ef6ea 100644 --- a/tests/test_routers_core.py +++ b/tests/test_routers_core.py @@ -1545,6 +1545,33 @@ class TestFabricGraphReadModel: assert latest_flags[first_id] is False assert latest_flags[second_id] is True + async def test_retained_valid_import_can_be_activated_for_rollback(self, client): + r = await client.post("/fabric/graph-exports", json=_fabric_graph_export()) + first_id = r.json()["import_run"]["id"] + r = await client.post("/fabric/graph-exports", json=_fabric_graph_export(extra_node=True)) + second_id = r.json()["import_run"]["id"] + + r = await client.post( + f"/fabric/graph-exports/{first_id}/activate", + json={"requested_by": "cutover-test"}, + ) + assert r.status_code == 200, r.text + assert r.json()["activated"] is True + assert r.json()["idempotent"] is False + assert r.json()["previous_import_id"] == second_id + assert r.json()["import_run"]["id"] == first_id + assert (await client.get("/fabric/graph/summary")).json()["node_count"] == 3 + + r = await client.post(f"/fabric/graph-exports/{first_id}/activate") + assert r.status_code == 200, r.text + assert r.json()["activated"] is False + assert r.json()["idempotent"] is True + + r = await client.post(f"/fabric/graph-exports/{second_id}/activate") + assert r.status_code == 200, r.text + assert r.json()["previous_import_id"] == first_id + assert (await client.get("/fabric/graph/summary")).json()["node_count"] == 4 + async def test_read_only_queries_filter_graph_without_mutating_state_hub_entities(self, client): await _create_domain(client) topic = await _create_topic(client)