From e4f7cfed605226b4ea826265b83dd158ae321c96 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 27 Jun 2026 13:30:30 +0200 Subject: [PATCH] Add Inter-Hub migration bundle importer --- Makefile | 6 +- .../migration/interhub-minimal.bundle.json | 66 +++ .../schemas/migration-bundle.schema.json | 71 +++ docs/specs/data-model.md | 13 + docs/specs/testing-release-and-migration.md | 61 +++ .../20260627_0002_add_migration_runs.py | 39 ++ scripts/core_hub_migrate.py | 52 +++ src/core_hub/migration.py | 419 ++++++++++++++++++ src/core_hub/models.py | 16 +- tests/test_database.py | 1 + tests/test_migration.py | 116 +++++ ...CORE-WP-0005-data-migration-and-cutover.md | 6 +- 12 files changed, 861 insertions(+), 5 deletions(-) create mode 100644 contracts/fixtures/migration/interhub-minimal.bundle.json create mode 100644 contracts/schemas/migration-bundle.schema.json create mode 100644 migrations/versions/20260627_0002_add_migration_runs.py create mode 100644 scripts/core_hub_migrate.py create mode 100644 src/core_hub/migration.py create mode 100644 tests/test_migration.py diff --git a/Makefile b/Makefile index eb5de17..7293de9 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: install test lint run openapi +.PHONY: install test lint run openapi migrate-validate UV ?= /home/worsch/.local/bin/uv PYTHONPATH ?= src @@ -17,3 +17,7 @@ run: openapi: PYTHONPATH=$(PYTHONPATH) $(UV) run python scripts/export_openapi.py contracts/openapi/core-hub.openapi.json + + +migrate-validate: + PYTHONPATH=$(PYTHONPATH) $(UV) run --extra dev python scripts/core_hub_migrate.py validate contracts/fixtures/migration/interhub-minimal.bundle.json diff --git a/contracts/fixtures/migration/interhub-minimal.bundle.json b/contracts/fixtures/migration/interhub-minimal.bundle.json new file mode 100644 index 0000000..d9f1bf1 --- /dev/null +++ b/contracts/fixtures/migration/interhub-minimal.bundle.json @@ -0,0 +1,66 @@ +{ + "schemaVersion": "core-hub.migration.v1", + "source": "inter-hub-haskell", + "exportedAt": "2026-06-27T00:00:00Z", + "records": { + "hubs": [ + { + "id": "11111111-1111-4111-8111-111111111111", + "slug": "ops-hub", + "name": "Ops Hub", + "domain": "ops.coulomb.social", + "hubKind": "domain", + "hubFamily": "vsm", + "vsmFunction": "OPS", + "vsmSystem": "1" + } + ], + "hubCapabilityManifests": [ + { + "id": "22222222-2222-4222-8222-222222222222", + "hubSlug": "ops-hub", + "manifestVersion": "1.0", + "status": "active" + } + ], + "apiConsumers": [ + { + "id": "33333333-3333-4333-8333-333333333333", + "slug": "ops-hub", + "name": "ops-hub", + "hubCapabilityManifestId": "22222222-2222-4222-8222-222222222222", + "keyPrefix": "ch_prefixonly" + } + ], + "apiKeys": [ + { + "id": "44444444-4444-4444-8444-444444444444", + "apiConsumerSlug": "ops-hub", + "keyPrefix": "ch_prefixonly", + "keyHash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "scopes": "framework:read hub:ops-hub:write" + } + ], + "widgets": [ + { + "id": "55555555-5555-4555-8555-555555555555", + "hubSlug": "ops-hub", + "name": "Readiness", + "widgetType": "ops-readiness-gate", + "capabilityRef": "ops:readiness", + "viewContext": "ops-hub/readiness", + "policyScope": "ops-registry" + } + ], + "interactionEvents": [ + { + "id": "66666666-6666-4666-8666-666666666666", + "widgetId": "55555555-5555-4555-8555-555555555555", + "eventType": "ops-endpoint-verified", + "metadata": { + "expectedStatus": 401 + } + } + ] + } +} diff --git a/contracts/schemas/migration-bundle.schema.json b/contracts/schemas/migration-bundle.schema.json new file mode 100644 index 0000000..aad1993 --- /dev/null +++ b/contracts/schemas/migration-bundle.schema.json @@ -0,0 +1,71 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://core-hub.local/contracts/schemas/migration-bundle.schema.json", + "title": "Core Hub Migration Bundle", + "type": "object", + "required": [ + "schemaVersion", + "source", + "records" + ], + "additionalProperties": true, + "properties": { + "schemaVersion": { + "const": "core-hub.migration.v1" + }, + "source": { + "type": "string" + }, + "exportedAt": { + "type": "string" + }, + "records": { + "type": "object", + "additionalProperties": false, + "properties": { + "hubs": { + "type": "array", + "items": { + "type": "object" + }, + "default": [] + }, + "hubCapabilityManifests": { + "type": "array", + "items": { + "type": "object" + }, + "default": [] + }, + "apiConsumers": { + "type": "array", + "items": { + "type": "object" + }, + "default": [] + }, + "apiKeys": { + "type": "array", + "items": { + "type": "object" + }, + "default": [] + }, + "widgets": { + "type": "array", + "items": { + "type": "object" + }, + "default": [] + }, + "interactionEvents": { + "type": "array", + "items": { + "type": "object" + }, + "default": [] + } + } + } + } +} diff --git a/docs/specs/data-model.md b/docs/specs/data-model.md index 89e422e..dd85289 100644 --- a/docs/specs/data-model.md +++ b/docs/specs/data-model.md @@ -41,6 +41,19 @@ Operations: - compatibility fixtures - import/export runs +## Initial Persisted Framework Tables + +The first replacement slice persists the framework resources that current Inter-Hub +consumers need during bootstrap and smoke testing: + +- `hubs` +- `hub_capability_manifests` +- `api_consumers` +- `api_keys` with hash and prefix only, never full key material +- `widgets` +- `interaction_events` +- `migration_runs` for import summaries, bundle hashes, counts, and diagnostics + ## Migration Posture - Import Inter-Hub schema facts through explicit migrations or import tools, not ad hoc SQL editing. diff --git a/docs/specs/testing-release-and-migration.md b/docs/specs/testing-release-and-migration.md index e09e901..6e75d72 100644 --- a/docs/specs/testing-release-and-migration.md +++ b/docs/specs/testing-release-and-migration.md @@ -30,6 +30,67 @@ A Core Hub release cannot replace Inter-Hub until: - Run dual-read or shadow smokes. - Switch traffic only after smoke evidence is recorded. +## Migration Bundle v1 + +Core Hub staging imports use `core-hub.migration.v1` JSON bundles. The bundle is +the handoff contract from the Haskell Inter-Hub export path into the Python/FastAPI +replacement: + +- `schemaVersion`: `core-hub.migration.v1` +- `source`: normally `inter-hub-haskell` +- `exportedAt`: export timestamp for operator traceability +- `records.hubs` +- `records.hubCapabilityManifests` +- `records.apiConsumers` +- `records.apiKeys` +- `records.widgets` +- `records.interactionEvents` + +API key records may contain `keyPrefix`, `keyHash`, scopes, status, and lifecycle +metadata. They must not contain full key material, raw tokens, or secrets. Existing +runtime keys are not recoverable from Inter-Hub unless an approved custody path +provides them; migration preserves only non-secret evidence. + +The reference schema is +`contracts/schemas/migration-bundle.schema.json`, with a minimal fixture at +`contracts/fixtures/migration/interhub-minimal.bundle.json`. + +## Import Procedure + +Validate a bundle without database access: + +```bash +PYTHONPATH=src uv run --extra dev python scripts/core_hub_migrate.py validate \ + contracts/fixtures/migration/interhub-minimal.bundle.json +``` + +Dry-run against the configured target database: + +```bash +CORE_HUB_DATABASE_URL=postgresql+asyncpg://... \ +PYTHONPATH=src uv run --extra dev python scripts/core_hub_migrate.py import \ + /secure/operator/path/interhub-export.bundle.json --dry-run +``` + +Import after the dry-run report is reviewed: + +```bash +CORE_HUB_DATABASE_URL=postgresql+asyncpg://... \ +PYTHONPATH=src uv run --extra dev python scripts/core_hub_migrate.py import \ + /secure/operator/path/interhub-export.bundle.json +``` + +Each successful import writes a `migration_runs` row containing the source, +schema version, bundle SHA-256, row counts, warnings, and relationship checks. + +## Rollback Procedure + +For staging, reset the target database and replay the same bundle after fixing the +export or mapper. For production cutover, take a database snapshot before import, +keep Inter-Hub read-only during verification, and roll traffic back to Inter-Hub if +row counts, relationship checks, or consumer smokes fail. Do not delete or rotate +legacy key custody records during the rollback window. + ## Haskell Retirement Gate Do not retire production Inter-Hub or the haskelseed/Haskell build lane until Core Hub serves the required compatibility surface and production traffic has moved. diff --git a/migrations/versions/20260627_0002_add_migration_runs.py b/migrations/versions/20260627_0002_add_migration_runs.py new file mode 100644 index 0000000..26588e1 --- /dev/null +++ b/migrations/versions/20260627_0002_add_migration_runs.py @@ -0,0 +1,39 @@ +"""add migration run ledger + +Revision ID: 20260627_0002 +Revises: 20260627_0001 +Create Date: 2026-06-27 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260627_0002" +down_revision: str | None = "20260627_0001" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "migration_runs", + sa.Column("id", sa.String(length=36), primary_key=True), + sa.Column("source", sa.String(length=120), nullable=False), + sa.Column("schema_version", sa.String(length=80), nullable=False), + sa.Column("bundle_sha256", sa.String(length=64), nullable=False), + sa.Column("dry_run", sa.Boolean(), nullable=False), + sa.Column("status", sa.String(length=40), nullable=False), + sa.Column("counts_json", sa.Text(), nullable=False), + sa.Column("diagnostics_json", sa.Text(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + op.create_index("ix_migration_runs_source", "migration_runs", ["source"]) + op.create_index("ix_migration_runs_bundle_sha256", "migration_runs", ["bundle_sha256"]) + + +def downgrade() -> None: + op.drop_index("ix_migration_runs_bundle_sha256", table_name="migration_runs") + op.drop_index("ix_migration_runs_source", table_name="migration_runs") + op.drop_table("migration_runs") diff --git a/scripts/core_hub_migrate.py b/scripts/core_hub_migrate.py new file mode 100644 index 0000000..7905c89 --- /dev/null +++ b/scripts/core_hub_migrate.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +import argparse +import asyncio +import json +from pathlib import Path +from typing import Any + +from core_hub.db import get_sessionmaker +from core_hub.migration import import_bundle, validate_bundle + + +def load_bundle(path: Path) -> dict[str, Any]: + with path.open("r", encoding="utf-8") as handle: + value = json.load(handle) + if not isinstance(value, dict): + raise SystemExit("migration bundle must be a JSON object") + return value + + +def print_report(report: dict[str, Any]) -> None: + print(json.dumps(report, indent=2, sort_keys=True)) + + +async def run_import(path: Path, dry_run: bool) -> dict[str, Any]: + bundle = load_bundle(path) + sessionmaker = get_sessionmaker() + async with sessionmaker() as session: + return await import_bundle(session, bundle, dry_run=dry_run) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Validate or import Core Hub migration bundles.") + subparsers = parser.add_subparsers(dest="command", required=True) + + validate_parser = subparsers.add_parser("validate", help="Validate a bundle without DB access.") + validate_parser.add_argument("bundle", type=Path) + + import_parser = subparsers.add_parser("import", help="Import a bundle into the configured DB.") + import_parser.add_argument("bundle", type=Path) + import_parser.add_argument("--dry-run", action="store_true") + + args = parser.parse_args() + if args.command == "validate": + report = validate_bundle(load_bundle(args.bundle)) + else: + report = asyncio.run(run_import(args.bundle, dry_run=args.dry_run)) + print_report(report) + return 0 if report.get("ok") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/core_hub/migration.py b/src/core_hub/migration.py new file mode 100644 index 0000000..970cdcd --- /dev/null +++ b/src/core_hub/migration.py @@ -0,0 +1,419 @@ +import hashlib +import json +from collections.abc import Iterable +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from .models import ( + ApiConsumer, + ApiKey, + Hub, + HubCapabilityManifest, + InteractionEvent, + MigrationRun, + Widget, + uuid_str, +) + +MIGRATION_SCHEMA_VERSION = "core-hub.migration.v1" +COLLECTIONS = ( + "hubs", + "hubCapabilityManifests", + "apiConsumers", + "apiKeys", + "widgets", + "interactionEvents", +) +SECRET_FIELD_NAMES = {"fullKey", "rawKey", "apiKey", "secret", "token"} + + +def encode_body(payload: dict[str, Any]) -> str: + return json.dumps(payload, sort_keys=True) + + +def bundle_digest(bundle: dict[str, Any]) -> str: + canonical = json.dumps(bundle, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def records_for(bundle: dict[str, Any], collection: str) -> list[dict[str, Any]]: + records = bundle.get("records", {}) + if not isinstance(records, dict): + return [] + values = records.get(collection, []) + if not isinstance(values, list): + return [] + return [record for record in values if isinstance(record, dict)] + + +def _input_counts(bundle: dict[str, Any]) -> dict[str, dict[str, int]]: + return { + collection: {"input": len(records_for(bundle, collection))} for collection in COLLECTIONS + } + + +def _work_counts(bundle: dict[str, Any]) -> dict[str, dict[str, int]]: + return { + collection: { + "input": len(records_for(bundle, collection)), + "created": 0, + "updated": 0, + "skipped": 0, + } + for collection in COLLECTIONS + } + + +def _require(record: dict[str, Any], fields: Iterable[str], path: str, errors: list[str]) -> None: + for field in fields: + if record.get(field) in (None, ""): + errors.append(f"{path}.{field} is required") + + +def _note_duplicate(value: str | None, seen: set[str], path: str, errors: list[str]) -> None: + if not value: + return + if value in seen: + errors.append(f"duplicate {path}: {value}") + seen.add(value) + + +def _contains_secret_field(value: Any) -> bool: + if isinstance(value, dict): + for key, child in value.items(): + if key in SECRET_FIELD_NAMES: + return True + if _contains_secret_field(child): + return True + if isinstance(value, list): + return any(_contains_secret_field(child) for child in value) + return False + + +def _relationship(status: str, name: str, message: str) -> dict[str, str]: + return {"name": name, "status": status, "message": message} + + +def validate_bundle(bundle: dict[str, Any]) -> dict[str, Any]: + errors: list[str] = [] + warnings: list[str] = [] + relationships: list[dict[str, str]] = [] + + if bundle.get("schemaVersion") != MIGRATION_SCHEMA_VERSION: + errors.append( + f"schemaVersion must be {MIGRATION_SCHEMA_VERSION}, got {bundle.get('schemaVersion')!r}" + ) + if not isinstance(bundle.get("records", {}), dict): + errors.append("records must be an object") + + hub_ids: set[str] = set() + hub_slugs: set[str] = set() + manifest_ids: set[str] = set() + consumer_ids: set[str] = set() + consumer_slugs: set[str] = set() + widget_ids: set[str] = set() + + for index, record in enumerate(records_for(bundle, "hubs")): + path = f"records.hubs[{index}]" + _require(record, ("slug", "name"), path, errors) + _note_duplicate(record.get("id"), hub_ids, "hub id", errors) + _note_duplicate(record.get("slug"), hub_slugs, "hub slug", errors) + + for index, record in enumerate(records_for(bundle, "hubCapabilityManifests")): + path = f"records.hubCapabilityManifests[{index}]" + _require(record, ("id",), path, errors) + _note_duplicate(record.get("id"), manifest_ids, "manifest id", errors) + if not record.get("hubId") and not record.get("hubSlug"): + errors.append(f"{path} must include hubId or hubSlug") + elif record.get("hubId") not in hub_ids and record.get("hubSlug") not in hub_slugs: + warnings.append(f"{path} references a hub outside this bundle") + + for index, record in enumerate(records_for(bundle, "apiConsumers")): + path = f"records.apiConsumers[{index}]" + _require(record, ("name",), path, errors) + if not record.get("id") and not record.get("slug"): + errors.append(f"{path} must include id or slug for idempotent import") + _note_duplicate(record.get("id"), consumer_ids, "api consumer id", errors) + _note_duplicate(record.get("slug"), consumer_slugs, "api consumer slug", errors) + manifest_id = record.get("hubCapabilityManifestId") + if manifest_id and manifest_id not in manifest_ids: + warnings.append( + f"{path}.hubCapabilityManifestId references a manifest outside this bundle" + ) + + for index, record in enumerate(records_for(bundle, "apiKeys")): + path = f"records.apiKeys[{index}]" + _require(record, ("id", "keyPrefix", "keyHash"), path, errors) + if _contains_secret_field(record): + errors.append( + f"{path} contains a secret-like key field; import only hashes and prefixes" + ) + if not record.get("apiConsumerId") and not record.get("apiConsumerSlug"): + errors.append(f"{path} must include apiConsumerId or apiConsumerSlug") + elif ( + record.get("apiConsumerId") not in consumer_ids + and record.get("apiConsumerSlug") not in consumer_slugs + ): + warnings.append(f"{path} references an API consumer outside this bundle") + + for index, record in enumerate(records_for(bundle, "widgets")): + path = f"records.widgets[{index}]" + _require(record, ("id", "name"), path, errors) + _note_duplicate(record.get("id"), widget_ids, "widget id", errors) + if not record.get("hubId") and not record.get("hubSlug"): + errors.append(f"{path} must include hubId or hubSlug") + elif record.get("hubId") not in hub_ids and record.get("hubSlug") not in hub_slugs: + warnings.append(f"{path} references a hub outside this bundle") + + for index, record in enumerate(records_for(bundle, "interactionEvents")): + path = f"records.interactionEvents[{index}]" + _require(record, ("id", "eventType"), path, errors) + if not record.get("widgetId"): + errors.append(f"{path}.widgetId is required") + elif record.get("widgetId") not in widget_ids: + warnings.append(f"{path}.widgetId references a widget outside this bundle") + + if errors: + relationships.append(_relationship("fail", "bundle_integrity", "validation errors present")) + else: + relationships.append( + _relationship("pass", "bundle_integrity", "required fields are present") + ) + if warnings: + relationships.append( + _relationship( + "warn", "external_references", "some references must resolve in the target DB" + ) + ) + else: + relationships.append( + _relationship("pass", "external_references", "all references resolve within the bundle") + ) + + return { + "ok": not errors, + "schemaVersion": bundle.get("schemaVersion"), + "source": bundle.get("source", "unknown"), + "bundleSha256": bundle_digest(bundle), + "counts": _input_counts(bundle), + "errors": errors, + "warnings": warnings, + "relationshipChecks": relationships, + } + + +async def _one_or_none(session: AsyncSession, statement: Any) -> Any | None: + result = await session.execute(statement) + return result.scalars().first() + + +async def _hub_maps(session: AsyncSession) -> tuple[dict[str, Hub], dict[str, Hub]]: + result = await session.execute(select(Hub)) + rows = list(result.scalars()) + return {row.id: row for row in rows}, {row.slug: row for row in rows} + + +async def _manifest_maps(session: AsyncSession) -> dict[str, HubCapabilityManifest]: + result = await session.execute(select(HubCapabilityManifest)) + rows = list(result.scalars()) + return {row.id: row for row in rows} + + +async def _consumer_maps( + session: AsyncSession, +) -> tuple[dict[str, ApiConsumer], dict[str, ApiConsumer]]: + result = await session.execute(select(ApiConsumer)) + rows = list(result.scalars()) + return {row.id: row for row in rows}, {row.slug: row for row in rows if row.slug} + + +async def _widget_maps(session: AsyncSession) -> dict[str, Widget]: + result = await session.execute(select(Widget)) + rows = list(result.scalars()) + return {row.id: row for row in rows} + + +def _resolved_hub_id( + record: dict[str, Any], hubs_by_id: dict[str, Hub], hubs_by_slug: dict[str, Hub] +) -> str: + if record.get("hubId"): + return str(record["hubId"]) + return hubs_by_slug[str(record["hubSlug"])].id + + +def _resolved_consumer_id(record: dict[str, Any], consumers_by_slug: dict[str, ApiConsumer]) -> str: + if record.get("apiConsumerId"): + return str(record["apiConsumerId"]) + return consumers_by_slug[str(record["apiConsumerSlug"])].id + + +def _mark(counts: dict[str, dict[str, int]], collection: str, existing: Any | None) -> None: + key = "updated" if existing else "created" + counts[collection][key] += 1 + + +async def import_bundle( + session: AsyncSession, bundle: dict[str, Any], *, dry_run: bool = False +) -> dict[str, Any]: + report = validate_bundle(bundle) + report["dryRun"] = dry_run + report["counts"] = _work_counts(bundle) + if not report["ok"]: + return report + + hubs_by_id, hubs_by_slug = await _hub_maps(session) + for record in records_for(bundle, "hubs"): + existing = hubs_by_id.get(record.get("id")) or hubs_by_slug.get(record.get("slug")) + _mark(report["counts"], "hubs", existing) + if dry_run: + continue + row = existing or Hub( + id=record.get("id") or uuid_str(), + slug=record["slug"], + name=record["name"], + ) + row.slug = record["slug"] + row.name = record["name"] + row.domain = record.get("domain") + row.hub_kind = record.get("hubKind") + row.hub_family = record.get("hubFamily") + row.vsm_function = record.get("vsmFunction") + row.vsm_system = record.get("vsmSystem") + row.status = record.get("status", "active") + row.description = record.get("description") + row.body_json = encode_body(record) + session.add(row) + if not dry_run: + await session.flush() + hubs_by_id, hubs_by_slug = await _hub_maps(session) + + manifests_by_id = await _manifest_maps(session) + for record in records_for(bundle, "hubCapabilityManifests"): + existing = manifests_by_id.get(record.get("id")) + _mark(report["counts"], "hubCapabilityManifests", existing) + if dry_run: + continue + hub_id = _resolved_hub_id(record, hubs_by_id, hubs_by_slug) + hub_slug = hubs_by_id[hub_id].slug if hub_id in hubs_by_id else record.get("hubSlug") + row = existing or HubCapabilityManifest(id=record["id"], body_json="{}") + row.hub_id = hub_id + row.hub_slug = hub_slug + row.manifest_version = record.get("manifestVersion", "0.1.0") + row.status = record.get("status", "draft") + row.body_json = encode_body(record) + session.add(row) + if not dry_run: + await session.flush() + + consumers_by_id, consumers_by_slug = await _consumer_maps(session) + for record in records_for(bundle, "apiConsumers"): + existing = consumers_by_id.get(record.get("id")) or consumers_by_slug.get( + record.get("slug") + ) + _mark(report["counts"], "apiConsumers", existing) + if dry_run: + continue + row = existing or ApiConsumer( + id=record.get("id") or uuid_str(), + slug=record.get("slug"), + name=record["name"], + ) + row.slug = record.get("slug") + row.name = record["name"] + row.description = record.get("description") + row.hub_capability_manifest_id = record.get("hubCapabilityManifestId") + row.rate_limit_per_minute = record.get("rateLimitPerMinute") + row.quota_per_day = record.get("quotaPerDay") + row.key_prefix = record.get("keyPrefix") + row.key_hash = record.get("keyHash") + row.status = record.get("status", "active") + row.body_json = encode_body(record) + session.add(row) + if not dry_run: + await session.flush() + consumers_by_id, consumers_by_slug = await _consumer_maps(session) + + for record in records_for(bundle, "apiKeys"): + existing = await session.get(ApiKey, record.get("id")) + if existing is None: + existing = await _one_or_none( + session, select(ApiKey).where(ApiKey.key_hash == record["keyHash"]) + ) + _mark(report["counts"], "apiKeys", existing) + if dry_run: + continue + row = existing or ApiKey( + id=record["id"], + api_consumer_id=_resolved_consumer_id(record, consumers_by_slug), + ) + row.api_consumer_id = _resolved_consumer_id(record, consumers_by_slug) + row.key_prefix = record["keyPrefix"] + row.key_hash = record["keyHash"] + row.scopes = record.get("scopes") + row.status = record.get("status", "active") + session.add(row) + + widgets_by_id = await _widget_maps(session) + for record in records_for(bundle, "widgets"): + existing = widgets_by_id.get(record.get("id")) + _mark(report["counts"], "widgets", existing) + if dry_run: + continue + row = existing or Widget( + id=record["id"], + hub_id=_resolved_hub_id(record, hubs_by_id, hubs_by_slug), + name=record["name"], + ) + row.hub_id = _resolved_hub_id(record, hubs_by_id, hubs_by_slug) + row.name = record["name"] + row.widget_type = record.get("widgetType") + row.capability_ref = record.get("capabilityRef") + row.view_context = record.get("viewContext") + row.policy_scope = record.get("policyScope") + row.status = record.get("status", "active") + row.body_json = encode_body(record) + session.add(row) + if not dry_run: + await session.flush() + + for record in records_for(bundle, "interactionEvents"): + existing = await session.get(InteractionEvent, record.get("id")) + _mark(report["counts"], "interactionEvents", existing) + if dry_run: + continue + row = existing or InteractionEvent( + id=record["id"], + widget_id=record["widgetId"], + event_type=record["eventType"], + ) + row.widget_id = record["widgetId"] + row.event_type = record["eventType"] + row.view_context = record.get("viewContext") + row.metadata_json = encode_body(record.get("metadata", {})) + row.body_json = encode_body(record) + session.add(row) + + if not dry_run: + run = MigrationRun( + source=report["source"], + schema_version=report["schemaVersion"], + bundle_sha256=report["bundleSha256"], + dry_run=False, + status="imported", + counts_json=encode_body(report["counts"]), + diagnostics_json=encode_body( + { + "errors": report["errors"], + "warnings": report["warnings"], + "relationshipChecks": report["relationshipChecks"], + } + ), + ) + session.add(run) + await session.commit() + await session.refresh(run) + report["migrationRunId"] = run.id + + return report diff --git a/src/core_hub/models.py b/src/core_hub/models.py index 1f69d0f..7a99443 100644 --- a/src/core_hub/models.py +++ b/src/core_hub/models.py @@ -1,7 +1,7 @@ import uuid from datetime import datetime -from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func +from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, func from sqlalchemy.orm import Mapped, mapped_column from .db import Base @@ -102,3 +102,17 @@ class InteractionEvent(Base): metadata_json: Mapped[str] = mapped_column(Text(), default="{}") body_json: Mapped[str] = mapped_column(Text(), default="{}") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + +class MigrationRun(Base): + __tablename__ = "migration_runs" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_str) + source: Mapped[str] = mapped_column(String(120), index=True) + schema_version: Mapped[str] = mapped_column(String(80)) + bundle_sha256: Mapped[str] = mapped_column(String(64), index=True) + dry_run: Mapped[bool] = mapped_column(Boolean(), default=False) + status: Mapped[str] = mapped_column(String(40), default="imported") + counts_json: Mapped[str] = mapped_column(Text(), default="{}") + diagnostics_json: Mapped[str] = mapped_column(Text(), default="{}") + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) diff --git a/tests/test_database.py b/tests/test_database.py index 10f06ae..85a654d 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -10,6 +10,7 @@ FOUNDATION_TABLES = { "api_keys", "widgets", "interaction_events", + "migration_runs", } diff --git a/tests/test_migration.py b/tests/test_migration.py new file mode 100644 index 0000000..9e30a43 --- /dev/null +++ b/tests/test_migration.py @@ -0,0 +1,116 @@ +import json + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from core_hub.migration import import_bundle, validate_bundle +from core_hub.models import Hub, MigrationRun, Widget + + +def minimal_bundle(): + return { + "schemaVersion": "core-hub.migration.v1", + "source": "inter-hub-haskell", + "exportedAt": "2026-06-27T00:00:00Z", + "records": { + "hubs": [ + { + "id": "11111111-1111-4111-8111-111111111111", + "slug": "ops-hub", + "name": "Ops Hub", + "domain": "ops.coulomb.social", + "hubKind": "domain", + "hubFamily": "vsm", + "vsmFunction": "OPS", + "vsmSystem": "1", + } + ], + "hubCapabilityManifests": [ + { + "id": "22222222-2222-4222-8222-222222222222", + "hubSlug": "ops-hub", + "manifestVersion": "1.0", + "status": "active", + } + ], + "apiConsumers": [ + { + "id": "33333333-3333-4333-8333-333333333333", + "slug": "ops-hub", + "name": "ops-hub", + "hubCapabilityManifestId": "22222222-2222-4222-8222-222222222222", + "keyPrefix": "ch_prefixonly", + } + ], + "apiKeys": [ + { + "id": "44444444-4444-4444-8444-444444444444", + "apiConsumerSlug": "ops-hub", + "keyPrefix": "ch_prefixonly", + "keyHash": "a" * 64, + "scopes": "framework:read hub:ops-hub:write", + } + ], + "widgets": [ + { + "id": "55555555-5555-4555-8555-555555555555", + "hubSlug": "ops-hub", + "name": "Readiness", + "widgetType": "ops-readiness-gate", + "capabilityRef": "ops:readiness", + "viewContext": "ops-hub/readiness", + "policyScope": "ops-registry", + } + ], + "interactionEvents": [ + { + "id": "66666666-6666-4666-8666-666666666666", + "widgetId": "55555555-5555-4555-8555-555555555555", + "eventType": "ops-endpoint-verified", + "metadata": {"expectedStatus": 401}, + } + ], + }, + } + + +def test_validate_bundle_rejects_secret_key_material(): + bundle = minimal_bundle() + bundle["records"]["apiKeys"][0]["fullKey"] = "do-not-import" + + report = validate_bundle(bundle) + + assert report["ok"] is False + assert any("secret-like" in error for error in report["errors"]) + + +async def test_dry_run_reports_counts_without_writing(sqlite_engine): + sessionmaker = async_sessionmaker(sqlite_engine, expire_on_commit=False) + async with sessionmaker() as session: + report = await import_bundle(session, minimal_bundle(), dry_run=True) + + assert report["ok"] is True + assert report["counts"]["hubs"]["created"] == 1 + assert (await session.execute(select(Hub))).scalars().first() is None + + +async def test_import_bundle_persists_records_and_run_summary(sqlite_engine): + sessionmaker = async_sessionmaker(sqlite_engine, expire_on_commit=False) + async with sessionmaker() as session: + report = await import_bundle(session, minimal_bundle()) + + assert report["ok"] is True + assert report["migrationRunId"] + assert report["counts"]["widgets"]["created"] == 1 + + hub = (await session.execute(select(Hub))).scalars().one() + widget = (await session.execute(select(Widget))).scalars().one() + run = (await session.execute(select(MigrationRun))).scalars().one() + + assert hub.slug == "ops-hub" + assert widget.hub_id == hub.id + assert json.loads(run.counts_json)["interactionEvents"]["created"] == 1 + + second = await import_bundle(session, minimal_bundle()) + assert second["counts"]["hubs"]["updated"] == 1 + assert second["counts"]["interactionEvents"]["updated"] == 1 diff --git a/workplans/CORE-WP-0005-data-migration-and-cutover.md b/workplans/CORE-WP-0005-data-migration-and-cutover.md index e3a27cb..277bd6a 100644 --- a/workplans/CORE-WP-0005-data-migration-and-cutover.md +++ b/workplans/CORE-WP-0005-data-migration-and-cutover.md @@ -20,12 +20,12 @@ Move production framework responsibility from Haskell Inter-Hub to Core Hub only ```task id: CORE-WP-0005-T01 -status: wait +status: done priority: high state_hub_task_id: "01909b75-2229-41e2-a61a-c4c9ef84c232" ``` -Define export/import flow, migration batches, row-count checks, relationship checks, and rollback. Waits on CORE-WP-0002 and CORE-WP-0003. +Result 2026-06-27: Defined and implemented the `core-hub.migration.v1` bundle path with validation, dry-run/import CLI, idempotent framework-record import, row-count reports, relationship diagnostics, non-secret API key handling, persisted `migration_runs`, fixture/schema coverage, and rollback procedure. ## Run Staging Import @@ -36,7 +36,7 @@ priority: high state_hub_task_id: "e0ea0928-a3ea-4ece-bbb7-ee08f8a3279b" ``` -Import production-safe or full approved Inter-Hub data into Core Hub staging. Record counts and discrepancies. +Import production-safe or full approved Inter-Hub data into Core Hub staging. The importer is ready; this now waits on an approved Inter-Hub export bundle and target staging database. Record counts and discrepancies from the generated migration report. ## Dual-Run Smokes