feat: add migration export and write fencing

src/core_hub/api/v2.py

@@ -3,12 +3,13 @@ from typing import Annotated, Any
 
 from hub_core.utils.slugs import slugify_or_default
 
-from fastapi import APIRouter, Depends, Query, Request, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from fastapi.responses import PlainTextResponse, RedirectResponse
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from core_hub.crypto import new_api_key
+from core_hub.config import Settings, get_settings
 from core_hub.db import session_scope
 from core_hub.models import (
     ApiConsumer,
@@ -32,6 +33,22 @@ TokenDependency = Annotated[str, Depends(require_api_token)]
 SessionDependency = Annotated[AsyncSession, Depends(session_scope)]
 
 
+def require_write_group(group: str):
+    def dependency(settings: Annotated[Settings, Depends(get_settings)]) -> None:
+        if group not in settings.v2_write_groups:
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail={"code": "write_group_disabled", "group": group},
+            )
+
+    return dependency
+
+
+RegistryWrite = Annotated[None, Depends(require_write_group("registry"))]
+CredentialWrite = Annotated[None, Depends(require_write_group("credentials"))]
+InteractionWrite = Annotated[None, Depends(require_write_group("interaction"))]
+
+
 def encode_body(payload: dict[str, Any]) -> str:
     return json.dumps(payload, sort_keys=True)
 
@@ -147,7 +164,7 @@ async def list_hubs(_: TokenDependency, session: SessionDependency) -> dict[str,
 
 @router.post("/hubs", status_code=status.HTTP_201_CREATED)
 async def create_hub(
-    body: dict[str, Any], _: TokenDependency, session: SessionDependency
+    body: dict[str, Any], _: TokenDependency, session: SessionDependency, __: RegistryWrite
 ) -> dict[str, Any]:
     row = Hub(
         slug=body["slug"],
@@ -182,7 +199,7 @@ async def list_hub_capability_manifests(
 
 @router.post("/hub-capability-manifests", status_code=status.HTTP_201_CREATED)
 async def create_hub_capability_manifest(
-    body: dict[str, Any], _: TokenDependency, session: SessionDependency
+    body: dict[str, Any], _: TokenDependency, session: SessionDependency, __: RegistryWrite
 ) -> dict[str, Any]:
     hub_id = body.get("hubId")
     hub_slug = None
@@ -204,7 +221,8 @@ async def create_hub_capability_manifest(
 
 @router.patch("/hub-capability-manifests/{manifest_id}")
 async def update_hub_capability_manifest(
-    manifest_id: str, body: dict[str, Any], _: TokenDependency, session: SessionDependency
+    manifest_id: str, body: dict[str, Any], _: TokenDependency, session: SessionDependency,
+    __: RegistryWrite,
 ) -> dict[str, Any]:
     row = await session.get(HubCapabilityManifest, manifest_id)
     if row is None:
@@ -220,7 +238,7 @@ async def update_hub_capability_manifest(
 
 @router.post("/hub-capability-manifests/{manifest_id}/activate")
 async def activate_hub_capability_manifest(
-    manifest_id: str, _: TokenDependency, session: SessionDependency
+    manifest_id: str, _: TokenDependency, session: SessionDependency, __: RegistryWrite
 ) -> dict[str, Any]:
     row = await session.get(HubCapabilityManifest, manifest_id)
     if row is None:
@@ -239,7 +257,7 @@ async def list_api_consumers(_: TokenDependency, session: SessionDependency) ->
 
 @router.post("/api-consumers", status_code=status.HTTP_201_CREATED)
 async def create_api_consumer(
-    body: dict[str, Any], _: TokenDependency, session: SessionDependency
+    body: dict[str, Any], _: TokenDependency, session: SessionDependency, __: CredentialWrite
 ) -> dict[str, Any]:
     row = ApiConsumer(
         slug=body.get("slug") or slugify_or_default(body["name"]),
@@ -259,7 +277,8 @@ async def create_api_consumer(
 
 @router.post("/api-consumers/{consumer_id}/api-keys", status_code=status.HTTP_201_CREATED)
 async def create_api_key(
-    consumer_id: str, body: dict[str, Any], _: TokenDependency, session: SessionDependency
+    consumer_id: str, body: dict[str, Any], _: TokenDependency, session: SessionDependency,
+    __: CredentialWrite,
 ) -> dict[str, Any]:
     full_key, key_prefix, key_hash = new_api_key()
     row = ApiKey(
@@ -294,7 +313,7 @@ async def list_widgets(_: TokenDependency, session: SessionDependency) -> dict[s
 
 @router.post("/widgets", status_code=status.HTTP_201_CREATED)
 async def create_widget(
-    body: dict[str, Any], _: TokenDependency, session: SessionDependency
+    body: dict[str, Any], _: TokenDependency, session: SessionDependency, __: InteractionWrite
 ) -> dict[str, Any]:
     row = Widget(
         hub_id=body["hubId"],
@@ -320,7 +339,8 @@ async def list_interaction_events(_: TokenDependency, session: SessionDependency
 
 @router.post("/interaction-events", status_code=status.HTTP_201_CREATED)
 async def create_interaction_event(
-    body: dict[str, Any], _: TokenDependency, session: SessionDependency
+    body: dict[str, Any], _: TokenDependency, session: SessionDependency,
+    __: InteractionWrite,
 ) -> dict[str, Any]:
     row = InteractionEvent(
         widget_id=body["widgetId"],
@@ -352,12 +372,21 @@ async def hub_registry(_: TokenDependency, session: SessionDependency) -> dict[s
 @router.post("/deployment-records", status_code=status.HTTP_201_CREATED)
 @router.get("/outcome-signals")
 @router.post("/outcome-signals", status_code=status.HTTP_201_CREATED)
-async def deferred_collection(_: TokenDependency) -> dict[str, list]:
+async def deferred_collection(
+    request: Request,
+    _: TokenDependency,
+    settings: Annotated[Settings, Depends(get_settings)],
+) -> dict[str, list]:
+    if request.method == "POST" and "deferred" not in settings.v2_write_groups:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail={"code": "write_group_disabled", "group": "deferred"},
+        )
     return {"data": []}
 
 
 @router.post("/token", tags=["api-v2-protected"], operation_id="post_token")
-async def issue_token(_: TokenDependency) -> dict[str, str]:
+async def issue_token(_: TokenDependency, __: CredentialWrite) -> dict[str, str]:
     return {"access_token": "already-authenticated", "token_type": "bearer"}
 
 

src/core_hub/config.py

@@ -4,6 +4,12 @@ from functools import lru_cache
 from pydantic import BaseModel, Field
 
 
+def _env_set(name: str, default: str) -> frozenset[str]:
+    return frozenset(
+        value.strip() for value in os.getenv(name, default).split(",") if value.strip()
+    )
+
+
 class Settings(BaseModel):
     app_name: str = "Core Hub"
     environment: str = Field(default_factory=lambda: os.getenv("CORE_HUB_ENV", "development"))
@@ -17,6 +23,11 @@ class Settings(BaseModel):
     auto_create_tables: bool = Field(
         default_factory=lambda: os.getenv("CORE_HUB_AUTO_CREATE_TABLES", "0") == "1"
     )
+    v2_write_groups: frozenset[str] = Field(
+        default_factory=lambda: _env_set(
+            "CORE_HUB_V2_WRITE_GROUPS", "registry,credentials,interaction,deferred"
+        )
+    )
 
     @property
     def auth_configured(self) -> bool:

src/core_hub/migration.py

@@ -1,6 +1,7 @@
 import hashlib
 import json
 from collections.abc import Iterable
+from datetime import datetime, timezone
 from typing import Any
 
 from sqlalchemy import select
@@ -34,10 +35,179 @@ def encode_body(payload: dict[str, Any]) -> str:
 
 
 def bundle_digest(bundle: dict[str, Any]) -> str:
-    canonical = json.dumps(bundle, sort_keys=True, separators=(",", ":"))
+    digestable = {key: value for key, value in bundle.items() if key != "bundleSha256"}
+    canonical = json.dumps(digestable, sort_keys=True, separators=(",", ":"))
     return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
 
 
+def _timestamp(value: datetime) -> str:
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _decoded(raw: str | None) -> dict[str, Any]:
+    value = json.loads(raw or "{}")
+    return value if isinstance(value, dict) else {}
+
+
+def _with_timestamps(
+    body: str | None,
+    *,
+    created_at: datetime,
+    updated_at: datetime | None = None,
+) -> dict[str, Any]:
+    value = _decoded(body)
+    value["createdAt"] = _timestamp(created_at)
+    if updated_at is not None:
+        value["updatedAt"] = _timestamp(updated_at)
+    return value
+
+
+async def export_bundle(
+    session: AsyncSession, *, source_revision: str | None = None
+) -> dict[str, Any]:
+    """Export all seven durable Core Hub tables without raw credential material."""
+    hubs = list((await session.execute(select(Hub).order_by(Hub.id))).scalars())
+    manifests = list(
+        (await session.execute(select(HubCapabilityManifest).order_by(HubCapabilityManifest.id)))
+        .scalars()
+    )
+    consumers = list(
+        (await session.execute(select(ApiConsumer).order_by(ApiConsumer.id))).scalars()
+    )
+    keys = list((await session.execute(select(ApiKey).order_by(ApiKey.id))).scalars())
+    widgets = list((await session.execute(select(Widget).order_by(Widget.id))).scalars())
+    events = list(
+        (await session.execute(select(InteractionEvent).order_by(InteractionEvent.id))).scalars()
+    )
+    runs = list(
+        (await session.execute(select(MigrationRun).order_by(MigrationRun.id))).scalars()
+    )
+
+    records: dict[str, list[dict[str, Any]]] = {
+        "hubs": [],
+        "hubCapabilityManifests": [],
+        "apiConsumers": [],
+        "apiKeys": [],
+        "widgets": [],
+        "interactionEvents": [],
+        "migrationRuns": [],
+    }
+    for row in hubs:
+        value = _with_timestamps(
+            row.body_json, created_at=row.created_at, updated_at=row.updated_at
+        )
+        value.update(
+            id=row.id,
+            slug=row.slug,
+            name=row.name,
+            domain=row.domain,
+            hubKind=row.hub_kind,
+            hubFamily=row.hub_family,
+            vsmFunction=row.vsm_function,
+            vsmSystem=row.vsm_system,
+            status=row.status,
+            description=row.description,
+        )
+        records["hubs"].append({key: child for key, child in value.items() if child is not None})
+    for row in manifests:
+        value = _with_timestamps(
+            row.body_json, created_at=row.created_at, updated_at=row.updated_at
+        )
+        value.update(
+            id=row.id,
+            hubId=row.hub_id,
+            hubSlug=row.hub_slug,
+            manifestVersion=row.manifest_version,
+            status=row.status,
+        )
+        records["hubCapabilityManifests"].append(
+            {key: child for key, child in value.items() if child is not None}
+        )
+    for row in consumers:
+        value = _with_timestamps(row.body_json, created_at=row.created_at)
+        value.update(
+            id=row.id,
+            slug=row.slug,
+            name=row.name,
+            description=row.description,
+            hubCapabilityManifestId=row.hub_capability_manifest_id,
+            rateLimitPerMinute=row.rate_limit_per_minute,
+            quotaPerDay=row.quota_per_day,
+            keyPrefix=row.key_prefix,
+            status=row.status,
+        )
+        records["apiConsumers"].append(
+            {key: child for key, child in value.items() if child is not None}
+        )
+    records["apiKeys"] = [
+        {
+            "id": row.id,
+            "apiConsumerId": row.api_consumer_id,
+            "keyPrefix": row.key_prefix,
+            "keyHash": row.key_hash,
+            "scopes": row.scopes,
+            "status": row.status,
+            "createdAt": _timestamp(row.created_at),
+        }
+        for row in keys
+    ]
+    for row in widgets:
+        value = _with_timestamps(row.body_json, created_at=row.created_at)
+        value.update(
+            id=row.id,
+            hubId=row.hub_id,
+            name=row.name,
+            widgetType=row.widget_type,
+            capabilityRef=row.capability_ref,
+            viewContext=row.view_context,
+            policyScope=row.policy_scope,
+            status=row.status,
+        )
+        records["widgets"].append(
+            {key: child for key, child in value.items() if child is not None}
+        )
+    for row in events:
+        value = _with_timestamps(row.body_json, created_at=row.created_at)
+        value.update(
+            id=row.id,
+            widgetId=row.widget_id,
+            eventType=row.event_type,
+            viewContext=row.view_context,
+            metadata=_decoded(row.metadata_json),
+        )
+        records["interactionEvents"].append(
+            {key: child for key, child in value.items() if child is not None}
+        )
+    records["migrationRuns"] = [
+        {
+            "id": row.id,
+            "source": row.source,
+            "schemaVersion": row.schema_version,
+            "bundleSha256": row.bundle_sha256,
+            "dryRun": row.dry_run,
+            "status": row.status,
+            "counts": _decoded(row.counts_json),
+            "diagnostics": _decoded(row.diagnostics_json),
+            "createdAt": _timestamp(row.created_at),
+        }
+        for row in runs
+    ]
+    high_water = max((_timestamp(row.created_at) for row in events), default=None)
+    bundle: dict[str, Any] = {
+        "schemaVersion": MIGRATION_SCHEMA_VERSION,
+        "source": "core-hub",
+        "exportedAt": _timestamp(datetime.now(timezone.utc)),
+        "highWaterMark": high_water,
+        "records": records,
+    }
+    if source_revision:
+        bundle["sourceRevision"] = source_revision
+    bundle["bundleSha256"] = bundle_digest(bundle)
+    return bundle
+
+
 def records_for(bundle: dict[str, Any], collection: str) -> list[dict[str, Any]]:
     records = bundle.get("records", {})
     if not isinstance(records, dict):

src/core_hub/operator_cli.py

@@ -10,7 +10,7 @@ from typing import Any
 import httpx
 
 from core_hub.db import get_sessionmaker
-from core_hub.migration import import_bundle, validate_bundle
+from core_hub.migration import export_bundle, import_bundle, validate_bundle
 from core_hub.smoke import (
     DEFAULT_OPERATOR_TOKEN_ENV,
     DEFAULT_RUNTIME_TOKEN_ENV,
@@ -292,6 +292,12 @@ async def run_import(path: Path, dry_run: bool) -> dict[str, Any]:
         return await import_bundle(session, bundle, dry_run=dry_run)
 
 
+async def run_export(source_revision: str | None) -> dict[str, Any]:
+    sessionmaker = get_sessionmaker()
+    async with sessionmaker() as session:
+        return await export_bundle(session, source_revision=source_revision)
+
+
 def add_token_args(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("--base-url", default=os.environ.get("CORE_HUB_BASE_URL"))
     parser.add_argument(
@@ -338,6 +344,8 @@ def build_parser() -> argparse.ArgumentParser:
     import_parser = migration_subparsers.add_parser("import")
     import_parser.add_argument("bundle", type=Path)
     import_parser.add_argument("--dry-run", action="store_true")
+    export_parser = migration_subparsers.add_parser("export")
+    export_parser.add_argument("--source-revision")
 
     readiness_parser = subparsers.add_parser("readiness-summary", help="Summarize cutover gates.")
     readiness_parser.add_argument("--deployed-smoke-report", type=Path)
@@ -398,8 +406,10 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "migration":
         if args.migration_command == "validate":
             report = validate_bundle(load_bundle(args.bundle))
-        else:
+        elif args.migration_command == "import":
             report = asyncio.run(run_import(args.bundle, dry_run=args.dry_run))
+        else:
+            report = asyncio.run(run_export(args.source_revision))
         print_report(report, args.output)
         return 0 if report.get("ok") else 1