crank/scripts/check-rust-boundaries.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any


@dataclass(frozen=True)
class Package:
    id: str
    name: str
    category: str
    manifest_path: Path


@dataclass(frozen=True)
class Violation:
    source: str
    dependency: str
    reason: str


def load_metadata(metadata_file: Path | None) -> dict[str, Any]:
    if metadata_file is not None:
        return json.loads(metadata_file.read_text(encoding="utf-8"))

    result = subprocess.run(
        ["cargo", "metadata", "--no-deps", "--format-version", "1"],
        check=True,
        text=True,
        stdout=subprocess.PIPE,
    )
    return json.loads(result.stdout)


def package_category(name: str, manifest_path: Path, workspace_root: Path) -> str:
    try:
        rel = manifest_path.parent.relative_to(workspace_root)
    except ValueError:
        rel = manifest_path.parent

    parts = rel.parts
    if parts and parts[0] == "apps":
        return "app"
    if name == "crank-core":
        return "core"
    if name == "crank-registry":
        return "registry"
    if name == "crank-runtime":
        return "runtime"
    if name.startswith("crank-adapter-"):
        return "adapter"
    return "crate"


def workspace_packages(metadata: dict[str, Any]) -> dict[str, Package]:
    workspace_root = Path(metadata["workspace_root"]).resolve()
    workspace_members = set(metadata["workspace_members"])
    packages: dict[str, Package] = {}

    for raw_package in metadata["packages"]:
        package_id = raw_package["id"]
        if package_id not in workspace_members:
            continue

        manifest_path = Path(raw_package["manifest_path"]).resolve()
        name = raw_package["name"]
        packages[package_id] = Package(
            id=package_id,
            name=name,
            category=package_category(name, manifest_path, workspace_root),
            manifest_path=manifest_path,
        )

    return packages


def dependency_package_ids(raw_package: dict[str, Any], packages_by_name: dict[str, str]) -> list[str]:
    dependency_ids: list[str] = []
    for dependency in raw_package.get("dependencies", []):
        dependency_name = dependency["name"]
        dependency_id = packages_by_name.get(dependency_name)
        if dependency_id is not None:
            dependency_ids.append(dependency_id)
    return dependency_ids


def boundary_reason(source: Package, dependency: Package) -> str | None:
    if source.category == "app":
        if dependency.category == "app":
            return "apps must not depend on other apps"
        return None

    if dependency.category == "app":
        return "workspace crates must not depend on apps"

    if source.category == "core" and dependency.category in {"runtime", "registry", "adapter"}:
        return "crank-core must stay below runtime, registry and adapters"

    if source.category == "registry" and dependency.category in {"runtime", "adapter"}:
        return "crank-registry must not depend on runtime or adapters"

    if source.category == "runtime" and dependency.category == "registry":
        return "crank-runtime must not depend on registry"

    if source.category == "adapter" and dependency.category == "app":
        return "adapters must not depend on apps"

    return None


def find_violations(metadata: dict[str, Any]) -> list[Violation]:
    packages = workspace_packages(metadata)
    packages_by_name = {package.name: package.id for package in packages.values()}
    raw_packages_by_id = {package["id"]: package for package in metadata["packages"]}
    violations: list[Violation] = []

    for source in sorted(packages.values(), key=lambda package: package.name):
        raw_package = raw_packages_by_id[source.id]
        for dependency_id in dependency_package_ids(raw_package, packages_by_name):
            dependency = packages[dependency_id]
            reason = boundary_reason(source, dependency)
            if reason is not None:
                violations.append(
                    Violation(
                        source=source.name,
                        dependency=dependency.name,
                        reason=reason,
                    )
                )

    return violations


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Check Rust workspace dependency boundaries.")
    parser.add_argument(
        "--metadata-file",
        type=Path,
        help="Use captured cargo metadata JSON instead of invoking cargo.",
    )
    return parser.parse_args(argv)


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    metadata = load_metadata(args.metadata_file)
    violations = find_violations(metadata)

    if violations:
        for violation in violations:
            print(
                f"error: {violation.source} -> {violation.dependency}: {violation.reason}",
                file=sys.stderr,
            )
        return 1

    print("Rust boundary check passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))