import json
import asyncio

from sqlalchemy.ext.asyncio import create_async_engine

from hub_core.runtime.migration import (
    bundle_digest,
    export_bundle,
    import_bundle,
    validate_bundle,
)
from hub_core.runtime.postgres_store import PostgresPortStore
from hub_core.runtime.tables import runtime_metadata


def bundle() -> dict:
    value = {
        "schemaVersion": "core-hub.migration.v1",
        "source": "core-hub",
        "sourceRevision": "abc123",
        "exportedAt": "2026-08-21T12:00:00Z",
        "highWaterMark": "2026-08-21T11:00:00Z",
        "records": {
            "hubs": [{"id": "hub-1", "slug": "ops-hub", "name": "Ops Hub", "createdAt": "2026-08-20T10:00:00Z", "updatedAt": "2026-08-20T11:00:00Z"}],
            "hubCapabilityManifests": [{"id": "manifest-1", "hubId": "hub-1", "manifestVersion": "1.0", "status": "active"}],
            "apiConsumers": [{"id": "consumer-1", "slug": "ops-hub", "name": "ops-hub", "hubCapabilityManifestId": "manifest-1"}],
            "apiKeys": [{"id": "key-1", "apiConsumerId": "consumer-1", "keyPrefix": "ch_prefix", "keyHash": "a" * 64, "scopes": "framework:read", "status": "active", "createdAt": "2026-08-21T10:00:00Z"}],
            "widgets": [{"id": "widget-1", "hubId": "hub-1", "name": "Readiness", "widgetType": "ops-readiness-gate"}],
            "interactionEvents": [{"id": "event-1", "widgetId": "widget-1", "eventType": "ops-endpoint-verified", "metadata": {"status": 401}, "createdAt": "2026-08-21T11:00:00Z"}],
            "migrationRuns": [{"id": "run-1", "source": "inter-hub", "schemaVersion": "core-hub.migration.v1", "bundleSha256": "b" * 64, "dryRun": False, "status": "imported", "counts": {}, "diagnostics": {}, "createdAt": "2026-08-19T09:00:00Z"}],
        },
    }
    value["bundleSha256"] = bundle_digest(value)
    return value


async def make_store(tmp_path) -> PostgresPortStore:
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'migration.db'}")
    async with engine.begin() as connection:
        await connection.run_sync(runtime_metadata.create_all)
    return PostgresPortStore(engine)


def test_validation_rejects_secret_material_and_digest_drift():
    value = bundle()
    value["records"]["apiKeys"][0]["fullKey"] = "must-not-move"

    report = validate_bundle(value)

    assert report["ok"] is False
    assert any("secret-shaped" in error for error in report["errors"])
    assert any("bundleSha256" in error for error in report["errors"])


def test_numeric_legacy_timestamps_have_deterministic_content_hashes():
    numeric = bundle()
    numeric["records"]["interactionEvents"][0]["metadata"]["recordedAt"] = 1
    numeric["bundleSha256"] = bundle_digest(numeric)
    textual = bundle()
    textual["records"]["interactionEvents"][0]["metadata"]["recordedAt"] = (
        "1970-01-01T00:00:01Z"
    )
    textual["bundleSha256"] = bundle_digest(textual)

    assert (
        validate_bundle(numeric)["contentHashes"]["interactionEvents"]
        == validate_bundle(textual)["contentHashes"]["interactionEvents"]
    )


def test_import_is_idempotent_and_exports_all_seven_collections(tmp_path):
    async def run():
        store = await make_store(tmp_path)
        try:
            return (
                await import_bundle(store, bundle()),
                await import_bundle(store, bundle()),
                await export_bundle(store, source_revision="target-revision"),
            )
        finally:
            await store.aclose()

    first, second, exported = asyncio.run(run())

    assert first["ok"] is True
    assert first["counts"]["migrationRuns"]["created"] == 1
    assert second["idempotent"] is True
    assert second["migrationRunId"] == first["migrationRunId"]
    assert set(exported["records"]) == {
        "hubs", "hubCapabilityManifests", "apiConsumers", "apiKeys",
        "widgets", "interactionEvents", "migrationRuns",
    }
    assert exported["records"]["apiKeys"][0]["keyHash"] == "a" * 64
    assert "fullKey" not in json.dumps(exported)
    assert validate_bundle(exported)["ok"] is True
    assert (
        validate_bundle(exported)["contentHashes"]
        == validate_bundle(bundle())["contentHashes"]
    )


def test_dry_run_does_not_record_import(tmp_path):
    async def run():
        store = await make_store(tmp_path)
        try:
            return (
                await import_bundle(store, bundle(), dry_run=True),
                await import_bundle(store, bundle()),
            )
        finally:
            await store.aclose()

    report, followup = asyncio.run(run())

    assert report["dryRun"] is True
    assert "migrationRunId" not in report
    assert followup["idempotent"] is False