#!/usr/bin/env python3
"""Exercise a pinned GovOPlaN runtime image pair on one OCI platform."""

from __future__ import annotations

import argparse
import base64
from datetime import UTC, datetime
import json
import os
from pathlib import Path
import re
import secrets
import subprocess
import time
from typing import Callable, Sequence


PLATFORMS = frozenset({"linux/amd64", "linux/arm64"})
DIGEST_IMAGE = re.compile(r"^[^\s@]+@sha256:[0-9a-f]{64}$")
BASE_MODULES = (
    "tenancy",
    "organizations",
    "identity",
    "idm",
    "access",
    "admin",
    "dashboard",
    "policy",
    "audit",
    "docs",
    "ops",
)


class SmokeError(RuntimeError):
    """A runtime image failed its bounded acceptance drill."""


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--api-metadata", type=Path, required=True)
    parser.add_argument("--web-metadata", type=Path, required=True)
    parser.add_argument("--postgres-image", required=True)
    parser.add_argument("--redis-image", required=True)
    parser.add_argument("--platform", choices=sorted(PLATFORMS), required=True)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--timeout-seconds", type=float, default=600.0)
    return parser


def _utc_now() -> str:
    return datetime.now(UTC).isoformat().replace("+00:00", "Z")


def _digest_image(value: object, label: str) -> str:
    if not isinstance(value, str) or DIGEST_IMAGE.fullmatch(value) is None:
        raise SmokeError(f"{label} must be an exact sha256 image reference")
    return value


def platform_image(path: Path, platform: str, label: str) -> str:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise SmokeError(f"cannot read {label} OCI metadata") from exc
    if not isinstance(payload, dict) or not isinstance(payload.get("platforms"), dict):
        raise SmokeError(f"{label} OCI metadata has no platform map")
    return _digest_image(payload["platforms"].get(platform), f"{label} {platform}")


def _tail(value: str, *, limit: int = 4000) -> str:
    return value[-limit:].strip()


def _run(
    arguments: Sequence[str],
    *,
    check: bool = True,
    timeout: float = 600.0,
    redactions: Sequence[str] = (),
) -> subprocess.CompletedProcess[str]:
    try:
        result = subprocess.run(
            list(arguments),
            check=False,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise SmokeError(f"container command could not complete: {type(exc).__name__}") from exc
    if check and result.returncode != 0:
        detail = _tail(result.stderr or result.stdout or "no diagnostic output")
        for secret in redactions:
            if secret:
                detail = detail.replace(secret, "[redacted]")
        raise SmokeError(f"container command failed: {detail}")
    return result


def _wait_for(
    label: str,
    probe: Callable[[], subprocess.CompletedProcess[str]],
    *,
    timeout: float,
    container: str | None = None,
    redactions: Sequence[str] = (),
) -> None:
    deadline = time.monotonic() + timeout
    last = ""
    while time.monotonic() < deadline:
        if container is not None:
            state = _run(
                ("docker", "inspect", "--format", "{{.State.Running}}", container),
                check=False,
                timeout=30,
            )
            if state.returncode != 0 or state.stdout.strip() != "true":
                logs = _run(
                    ("docker", "logs", "--tail", "100", container),
                    check=False,
                    timeout=30,
                )
                detail = _tail(logs.stdout + logs.stderr, limit=8000)
                for secret in redactions:
                    if secret:
                        detail = detail.replace(secret, "[redacted]")
                raise SmokeError(
                    f"{label} container exited before readiness: "
                    f"{detail or 'no diagnostic output'}"
                )
        result = probe()
        if result.returncode == 0:
            return
        last = _tail(result.stderr or result.stdout)
        time.sleep(2.0)
    raise SmokeError(f"{label} did not become ready: {last or 'probe failed'}")


def _environment(
    *,
    database_password: str,
    master_key: str,
    platform_slug: str,
) -> dict[str, str]:
    database = (
        f"postgresql+psycopg://govoplan:{database_password}@postgres:5432/govoplan"
    )
    return {
        "APP_ENV": "dev",
        "GOVOPLAN_INSTALL_PROFILE": "evaluation",
        "GOVOPLAN_INSTALLATION_ID": f"runtime-smoke-{platform_slug}",
        "GOVOPLAN_STATE_PROFILE": "host-shared",
        "GOVOPLAN_RUNTIME_HEARTBEAT_SECONDS": "5",
        "GOVOPLAN_RUNTIME_STALE_AFTER_SECONDS": "30",
        "GOVOPLAN_EXPECTED_API_REPLICAS": "1",
        "GOVOPLAN_EXPECTED_WORKER_REPLICAS": "1",
        "DATABASE_URL": database,
        "GOVOPLAN_DATABASE_URL_PGTOOLS": (
            f"postgresql://govoplan:{database_password}@postgres:5432/govoplan"
        ),
        "GOVOPLAN_DB_CONNECTION_LIMIT": "100",
        "GOVOPLAN_DB_CONNECTION_RESERVE": "10",
        "REDIS_URL": "redis://redis:6379/0",
        "CELERY_ENABLED": "true",
        "CELERY_QUEUES": "default",
        "CELERY_WORKER_CONCURRENCY": "1",
        "ENABLED_MODULES": ",".join(BASE_MODULES),
        "GOVOPLAN_MIGRATION_TRACK": "release",
        "DEV_AUTO_MIGRATE_ENABLED": "false",
        "DEV_BOOTSTRAP_ENABLED": "false",
        "AUTH_LOGIN_THROTTLE_ENABLED": "true",
        "AUTH_COOKIE_SECURE": "false",
        "CORS_ORIGINS": "http://localhost",
        "GOVOPLAN_TRUSTED_HOSTS": "127.0.0.1,localhost,api",
        "FORWARDED_ALLOW_IPS": "127.0.0.1",
        "MASTER_KEY_B64": master_key,
        "FILE_STORAGE_BACKEND": "local",
        "FILE_STORAGE_LOCAL_ROOT": "/var/lib/govoplan/files",
        "GOVOPLAN_MODULE_LIVE_APPLY_ENABLED": "false",
    }


def _env_arguments(values: dict[str, str], *, role: str, node_id: str) -> list[str]:
    arguments: list[str] = []
    for key, value in sorted(
        {**values, "GOVOPLAN_RUNTIME_ROLE": role, "GOVOPLAN_NODE_ID": node_id}.items()
    ):
        arguments.extend(("--env", f"{key}={value}"))
    return arguments


def run_smoke(
    *,
    api_image: str,
    web_image: str,
    postgres_image: str,
    redis_image: str,
    platform: str,
    timeout: float,
) -> dict[str, object]:
    for label, value in (
        ("API image", api_image),
        ("Web image", web_image),
        ("PostgreSQL image", postgres_image),
        ("Redis image", redis_image),
    ):
        _digest_image(value, label)
    if platform not in PLATFORMS:
        raise SmokeError(f"unsupported runtime smoke platform: {platform}")

    slug = platform.replace("linux/", "").replace("/", "-")
    suffix = secrets.token_hex(4)
    prefix = f"govoplan-runtime-{slug}-{suffix}"
    names = {
        "network": f"{prefix}-network",
        "volume": f"{prefix}-data",
        "postgres": f"{prefix}-postgres",
        "redis": f"{prefix}-redis",
        "api": f"{prefix}-api",
        "web": f"{prefix}-web",
        "worker": f"{prefix}-worker",
    }
    database_password = secrets.token_hex(20)
    master_key = base64.urlsafe_b64encode(os.urandom(32)).decode("ascii")
    redactions = (database_password, master_key)
    environment = _environment(
        database_password=database_password,
        master_key=master_key,
        platform_slug=slug,
    )
    checks: list[dict[str, object]] = []
    started = time.monotonic()

    def record(check_id: str, began: float) -> None:
        duration = round(time.monotonic() - began, 3)
        checks.append(
            {
                "id": check_id,
                "state": "passed",
                "duration_seconds": duration,
            }
        )
        print(f"PASS {platform} {check_id} ({duration}s)", flush=True)

    common_runtime = [
        "--platform",
        platform,
        "--network",
        names["network"],
        "--read-only",
        "--tmpfs",
        "/tmp:rw,noexec,nosuid,size=64m",
        "--security-opt",
        "no-new-privileges:true",
        "--cap-drop",
        "ALL",
        "--mount",
        f"type=volume,source={names['volume']},target=/var/lib/govoplan",
    ]

    try:
        _run(("docker", "network", "create", names["network"]), timeout=timeout)
        _run(("docker", "volume", "create", names["volume"]), timeout=timeout)

        began = time.monotonic()
        _run(
            (
                "docker",
                "run",
                "--detach",
                "--platform",
                platform,
                "--name",
                names["postgres"],
                "--network",
                names["network"],
                "--network-alias",
                "postgres",
                "--env",
                "POSTGRES_DB=govoplan",
                "--env",
                "POSTGRES_USER=govoplan",
                "--env",
                f"POSTGRES_PASSWORD={database_password}",
                "--tmpfs",
                "/var/lib/postgresql/data:rw,noexec,nosuid,size=384m",
                postgres_image,
            ),
            timeout=timeout,
            redactions=redactions,
        )
        _run(
            (
                "docker",
                "run",
                "--detach",
                "--platform",
                platform,
                "--name",
                names["redis"],
                "--network",
                names["network"],
                "--network-alias",
                "redis",
                "--read-only",
                "--tmpfs",
                "/data:rw,noexec,nosuid,size=64m",
                redis_image,
                "redis-server",
                "--save",
                "",
                "--appendonly",
                "no",
            ),
            timeout=timeout,
        )
        _wait_for(
            "PostgreSQL",
            lambda: _run(
                (
                    "docker",
                    "exec",
                    names["postgres"],
                    "pg_isready",
                    "--username",
                    "govoplan",
                    "--dbname",
                    "govoplan",
                ),
                check=False,
                timeout=30,
            ),
            timeout=timeout,
            container=names["postgres"],
            redactions=redactions,
        )
        _wait_for(
            "Redis",
            lambda: _run(
                ("docker", "exec", names["redis"], "redis-cli", "ping"),
                check=False,
                timeout=30,
            ),
            timeout=timeout,
            container=names["redis"],
            redactions=redactions,
        )
        record("managed_dependencies_ready", began)

        began = time.monotonic()
        _run(
            (
                "docker",
                "run",
                "--rm",
                "--name",
                f"{prefix}-migrate",
                *common_runtime,
                *_env_arguments(
                    environment,
                    role="migration",
                    node_id=f"runtime-smoke-{slug}-migration",
                ),
                api_image,
                "python",
                "-m",
                "govoplan_core.commands.init_db",
                "--migration-track",
                "release",
            ),
            timeout=timeout,
            redactions=redactions,
        )
        record("release_migrations", began)

        began = time.monotonic()
        _run(
            (
                "docker",
                "run",
                "--rm",
                "--name",
                f"{prefix}-schema",
                *common_runtime,
                *_env_arguments(
                    environment,
                    role="migration",
                    node_id=f"runtime-smoke-{slug}-schema",
                ),
                api_image,
                "python",
                "-c",
                (
                    "import os;"
                    "from sqlalchemy import create_engine,inspect;"
                    "engine=create_engine(os.environ['DATABASE_URL']);"
                    "tables=set(inspect(engine).get_table_names());"
                    "required={'alembic_version','core_scopes','core_system_settings',"
                    "'core_runtime_nodes'};"
                    "missing=required-tables;"
                    "assert not missing, f'missing release tables: {sorted(missing)}';"
                    "engine.dispose()"
                ),
            ),
            timeout=timeout,
            redactions=redactions,
        )
        record("release_schema_contract", began)

        began = time.monotonic()
        _run(
            (
                "docker",
                "run",
                "--detach",
                "--name",
                names["api"],
                "--network-alias",
                "api",
                "--network-alias",
                "load-balancer",
                *common_runtime,
                *_env_arguments(
                    environment,
                    role="api",
                    node_id=f"runtime-smoke-{slug}-api",
                ),
                api_image,
            ),
            timeout=timeout,
            redactions=redactions,
        )
        _wait_for(
            "GovOPlaN API",
            lambda: _run(
                (
                    "docker",
                    "exec",
                    names["api"],
                    "python",
                    "-c",
                    (
                        "import urllib.request;"
                        "r=urllib.request.Request('http://127.0.0.1:8000/health/ready',"
                        "headers={'Host':'127.0.0.1'});"
                        "assert urllib.request.urlopen(r,timeout=3).status==200"
                    ),
                ),
                check=False,
                timeout=30,
            ),
            timeout=timeout,
            container=names["api"],
            redactions=redactions,
        )
        _run(
            (
                "docker",
                "exec",
                names["api"],
                "python",
                "-c",
                "import os; assert os.getuid() == 10001",
            ),
            timeout=30,
        )
        record("api_non_root_readiness", began)

        began = time.monotonic()
        _run(
            (
                "docker",
                "run",
                "--detach",
                "--platform",
                platform,
                "--name",
                names["web"],
                "--network",
                names["network"],
                "--network-alias",
                "web",
                "--read-only",
                "--tmpfs",
                "/tmp:rw,noexec,nosuid,size=64m",
                "--security-opt",
                "no-new-privileges:true",
                "--cap-drop",
                "ALL",
                web_image,
            ),
            timeout=timeout,
        )
        _wait_for(
            "GovOPlaN WebUI",
            lambda: _run(
                (
                    "docker",
                    "exec",
                    names["api"],
                    "python",
                    "-c",
                    (
                        "import urllib.request;"
                        "assert urllib.request.urlopen('http://web:8080/health',timeout=3).status==200;"
                        "assert urllib.request.urlopen('http://web:8080/',timeout=3).status==200"
                    ),
                ),
                check=False,
                timeout=30,
            ),
            timeout=timeout,
            container=names["web"],
            redactions=redactions,
        )
        _run(
            ("docker", "exec", names["web"], "sh", "-c", "test \"$(id -u)\" = 101"),
            timeout=30,
        )
        record("web_non_root_readiness", began)

        began = time.monotonic()
        _run(
            (
                "docker",
                "run",
                "--detach",
                "--name",
                names["worker"],
                *common_runtime,
                *_env_arguments(
                    environment,
                    role="worker",
                    node_id=f"runtime-smoke-{slug}-worker",
                ),
                api_image,
                "python",
                "-m",
                "celery",
                "-A",
                "govoplan_core.celery_app:celery",
                "worker",
                "--queues",
                "default",
                "--pool",
                "solo",
                "--concurrency",
                "1",
                "--hostname",
                f"runtime-smoke-{slug}@%h",
                "--loglevel",
                "WARNING",
            ),
            timeout=timeout,
            redactions=redactions,
        )
        _wait_for(
            "GovOPlaN worker",
            lambda: _run(
                (
                    "docker",
                    "exec",
                    names["api"],
                    "python",
                    "-c",
                    (
                        "from govoplan_core.celery_app import celery;"
                        "result=celery.send_task('govoplan.ping',queue='default');"
                        "assert result.get(timeout=10)=='pong'"
                    ),
                ),
                check=False,
                timeout=30,
            ),
            timeout=timeout,
            container=names["worker"],
            redactions=redactions,
        )
        _run(("docker", "stop", "--time", "20", names["worker"]), timeout=30)
        record("worker_delivery_and_shutdown", began)
    except SmokeError as exc:
        for role in ("api", "web", "worker", "postgres", "redis"):
            result = _run(
                ("docker", "logs", "--tail", "100", names[role]),
                check=False,
                timeout=30,
            )
            if result.stdout or result.stderr:
                detail = _tail(result.stdout + result.stderr, limit=8000)
                for secret in redactions:
                    detail = detail.replace(secret, "[redacted]")
                print(f"--- {role} logs ---\n{detail}")
        raise exc
    finally:
        for role in ("worker", "web", "api", "redis", "postgres"):
            _run(
                ("docker", "rm", "--force", names[role]),
                check=False,
                timeout=30,
            )
        _run(("docker", "volume", "rm", "--force", names["volume"]), check=False)
        _run(("docker", "network", "rm", names["network"]), check=False)

    return {
        "schema_version": "1",
        "evidence_kind": "govoplan.runtime-image-smoke",
        "captured_at": _utc_now(),
        "platform": platform,
        "images": {
            "api": api_image,
            "web": web_image,
            "postgres": postgres_image,
            "redis": redis_image,
        },
        "result": {"state": "passed"},
        "checks": checks,
        "duration_seconds": round(time.monotonic() - started, 3),
    }


def main() -> int:
    args = build_parser().parse_args()
    try:
        api_image = platform_image(args.api_metadata, args.platform, "API")
        web_image = platform_image(args.web_metadata, args.platform, "Web")
        evidence = run_smoke(
            api_image=api_image,
            web_image=web_image,
            postgres_image=args.postgres_image,
            redis_image=args.redis_image,
            platform=args.platform,
            timeout=args.timeout_seconds,
        )
    except (OSError, SmokeError, ValueError) as exc:
        print(f"runtime image smoke failed: {exc}")
        return 1
    args.output.parent.mkdir(parents=True, exist_ok=True)
    temporary = args.output.with_suffix(args.output.suffix + ".tmp")
    temporary.write_text(json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    temporary.chmod(0o644)
    temporary.replace(args.output)
    print(f"Runtime image smoke evidence written to {args.output}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())