#!/usr/bin/env python3
"""Fail-closed drift check for the generated runtime configuration contract."""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path

MAX_INPUT_BYTES = 4 * 1024 * 1024
OWNED_PREFIXES = ("CRANK_", "POSTGRES_", "OTEL_")
ENV_FILES = (
    ".env.example",
    "deploy/community/.env.example",
    "deploy/community/.env.images.example",
)
COMPOSE_FILES = (
    "docker-compose.yml",
    "deploy/community/docker-compose.yml",
    "deploy/community/docker-compose.images.yml",
)
BEGIN = "# BEGIN GENERATED CRANK RUNTIME CONFIG"
END = "# END GENERATED CRANK RUNTIME CONFIG"
ENV_NAME = re.compile(r"^[A-Z_][A-Z0-9_]*$")
INTERPOLATION_OPERATORS = (":-", ":?", ":+", "-", "?", "+")


class ContractError(Exception):
    pass


def reject_duplicates(pairs: list[tuple[str, object]]) -> dict[str, object]:
    result: dict[str, object] = {}
    for key, value in pairs:
        if key in result:
            raise ContractError("duplicate JSON key")
        result[key] = value
    return result


def read_text(path: Path) -> str:
    if not path.is_file() or path.is_symlink():
        raise ContractError(f"missing regular file: {path.name}")
    data = path.read_bytes()
    if len(data) > MAX_INPUT_BYTES:
        raise ContractError(f"input too large: {path.name}")
    try:
        return data.decode("utf-8")
    except UnicodeDecodeError as error:
        raise ContractError(f"invalid UTF-8: {path.name}") from error


def load_contract(
    root: Path,
) -> tuple[set[str], dict[str, dict[str, object]], set[str]]:
    path = root / "docs/schemas/runtime-config.schema.json"
    try:
        payload = json.loads(read_text(path), object_pairs_hook=reject_duplicates)
    except (json.JSONDecodeError, ValueError, RecursionError) as error:
        raise ContractError("invalid runtime configuration contract JSON") from error
    fields = payload.get("fields")
    if payload.get("schema_version") != 1 or not isinstance(fields, list):
        raise ContractError("invalid runtime configuration contract shape")
    effective: set[str] = set()
    specifications: dict[str, dict[str, object]] = {}
    for field in fields:
        if not isinstance(field, dict):
            raise ContractError("invalid field entry")
        name = field.get("env_name")
        scope = field.get("process")
        mode = field.get("mode")
        if not isinstance(name, str) or scope not in {"shared", "admin_api", "mcp_server"}:
            raise ContractError("invalid field identity")
        if name in specifications:
            raise ContractError("duplicate runtime field")
        required = field.get("required")
        sensitivity = field.get("sensitivity")
        default = field.get("default")
        if not isinstance(required, bool) or sensitivity not in {
            "public",
            "internal",
            "secret",
        }:
            raise ContractError("invalid field semantics")
        if default is not None and not isinstance(default, str):
            raise ContractError("invalid field default")
        specifications[name] = field
        if mode == "effective":
            effective.add(name)
        elif mode != "deprecated_no_effect":
            raise ContractError("unknown runtime field mode")
    deployment = payload.get("deployment_only_fields")
    if (
        not isinstance(deployment, list)
        or not all(isinstance(name, str) and ENV_NAME.fullmatch(name) for name in deployment)
        or len(set(deployment)) != len(deployment)
    ):
        raise ContractError("invalid deployment-only fields")
    return effective, specifications, set(deployment)


def env_entries(content: str) -> dict[str, str]:
    if content.count(BEGIN) != 1 or content.count(END) != 1:
        raise ContractError("generated environment markers are missing or duplicated")
    section = content.split(BEGIN, 1)[1].split(END, 1)[0]
    entries: dict[str, str] = {}
    for line in section.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        name, separator, _ = line.partition("=")
        if not separator or not ENV_NAME.fullmatch(name) or name in entries:
            raise ContractError("invalid or duplicate generated environment entry")
        entries[name] = line.partition("=")[2]
    return entries


def interpolations(content: str) -> list[tuple[str, str, str]]:
    """Parse bounded, non-nested Compose ${NAME<operator>payload} expressions."""
    results: list[tuple[str, str, str]] = []
    cursor = 0
    while True:
        start = content.find("${", cursor)
        if start < 0:
            return results
        end = content.find("}", start + 2)
        if end < 0:
            raise ContractError("unterminated Compose interpolation")
        expression = content[start + 2 : end]
        if "${" in expression or len(expression) > 8192:
            raise ContractError("invalid nested or oversized Compose interpolation")
        name_end = 0
        while name_end < len(expression) and (
            expression[name_end].isalnum() or expression[name_end] == "_"
        ):
            name_end += 1
        name = expression[:name_end]
        remainder = expression[name_end:]
        if not ENV_NAME.fullmatch(name):
            raise ContractError("invalid Compose interpolation name")
        operator = ""
        payload = ""
        if remainder:
            operator = next(
                (candidate for candidate in INTERPOLATION_OPERATORS if remainder.startswith(candidate)),
                "",
            )
            if not operator:
                raise ContractError("invalid Compose interpolation operator")
            payload = remainder[len(operator) :]
        results.append((name, operator, payload))
        cursor = end + 1


def parse_runtime_reference(value: str) -> tuple[str, str, str]:
    references = interpolations(value)
    if len(references) != 1 or value.strip() != "${" + "".join(references[0]) + "}":
        raise ContractError("runtime Compose value must be one direct interpolation")
    return references[0]


def compose_environment(content: str, service: str) -> dict[str, str]:
    lines = content.splitlines()
    in_service = False
    in_environment = False
    entries: dict[str, str] = {}
    for line in lines:
        if line == f"  {service}:":
            in_service = True
            in_environment = False
            continue
        if in_service and line.startswith("  ") and not line.startswith("    "):
            break
        if in_service and line == "    environment:":
            in_environment = True
            continue
        if in_environment:
            if not line.startswith("      "):
                break
            stripped = line.strip()
            name, separator, _ = stripped.partition(":")
            if separator and name.startswith(OWNED_PREFIXES):
                if name in entries:
                    raise ContractError("duplicate Compose runtime field")
                entries[name] = stripped.partition(":")[2].strip()
    return entries


def validate_runtime_reference(
    name: str,
    value: str,
    specification: dict[str, object],
    declared_values: set[str],
) -> None:
    reference, operator, payload = parse_runtime_reference(value)
    if reference != name:
        raise ContractError(f"wrong Compose interpolation reference: {name}")
    required = specification["required"]
    sensitivity = specification["sensitivity"]
    if required:
        if operator not in {"", ":?"} or (operator == ":?" and not payload):
            raise ContractError(f"required runtime field has fallback: {name}")
        return
    if operator != ":-":
        raise ContractError(f"optional runtime field must use empty-aware default: {name}")
    if sensitivity != "secret" and payload not in declared_values:
        raise ContractError(f"divergent Compose inline default: {name}")


def compose_optional(specification: dict[str, object]) -> bool:
    compatibility = specification.get("compatibility")
    return (
        isinstance(compatibility, str)
        and "deprecated startup-bootstrap password" in compatibility
    )


def validate(root: Path) -> None:
    effective, specifications, deployment = load_contract(root)
    declared_values: dict[str, set[str]] = {name: set() for name in effective}
    for relative in ENV_FILES:
        entries = env_entries(read_text(root / relative))
        if set(entries) != effective:
            raise ContractError(f"generated environment drift: {Path(relative).name}")
        for name, value in entries.items():
            declared_values[name].add(value)
    for name in effective:
        default = specifications[name].get("default")
        if isinstance(default, str):
            declared_values[name].add(default)

    for relative in COMPOSE_FILES:
        content = read_text(root / relative)
        known = effective | deployment
        for name, _, _ in interpolations(content):
            if name not in known:
                raise ContractError(f"unknown Compose interpolation: {name}")
        admin = compose_environment(content, "admin-api")
        mcp = compose_environment(content, "mcp-server")
        required_admin = {
            name
            for name in effective
            if specifications[name]["process"] in {"shared", "admin_api"}
            and not compose_optional(specifications[name])
        }
        required_mcp = {
            name
            for name in effective
            if specifications[name]["process"] in {"shared", "mcp_server"}
        }
        if set(admin) != required_admin:
            raise ContractError(f"admin Compose runtime drift: {Path(relative).name}")
        if set(mcp) != required_mcp:
            raise ContractError(f"MCP Compose runtime drift: {Path(relative).name}")
        for entries in (admin, mcp):
            for name, value in entries.items():
                validate_runtime_reference(
                    name,
                    value,
                    specifications[name],
                    declared_values[name],
                )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, default=Path("."))
    args = parser.parse_args()
    try:
        validate(args.root.resolve())
    except (ContractError, OSError) as error:
        print(f"runtime configuration contract check failed: {error}", file=sys.stderr)
        return 1
    print("runtime configuration contract check passed")
    return 0


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