DataOps-platform/deployment/compose/product_engineering.py

#!/usr/bin/env python3
"""Evidence-first product engineering controls for P2-WP11.

The commands in this module validate or render evidence.  They do not promote
objects, mutate customer source data, switch traffic, or downgrade databases.
"""

from __future__ import annotations

import argparse
import ast
import hashlib
import json
import re
import subprocess
import sys
from pathlib import Path
from typing import Any


READINESS_CATEGORIES = (
    "service",
    "task",
    "database",
    "graph",
    "object_storage",
    "external_dependency",
)
PROMOTABLE_KINDS = {"asset", "standard", "ontology", "rule", "workflow"}
ENVIRONMENT_ORDER = {"development": 10, "test": 20, "preproduction": 30}
HEALTHY_STATES = {"healthy", "ready", "succeeded", "completed", "available"}
SECRET_KEY_PATTERN = re.compile(
    r"(?:secret|password|token|credential|authorization|api[_-]?key)", re.I
)
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
COMPOSE_SERVICE_CATEGORIES = {
    "backend": "service",
    "frontend": "service",
    "runner": "task",
    "n8n": "task",
    "kestra": "task",
    "kestra-db-init": "task",
    "postgres": "database",
    "neo4j": "graph",
    "minio": "object_storage",
    "minio-init": "object_storage",
    "source-postgres": "external_dependency",
    "source-mysql": "external_dependency",
}


def _canonical_bytes(value: Any) -> bytes:
    return json.dumps(
        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")


def _sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def _sha256_file(path: Path) -> str:
    return _sha256_bytes(path.read_bytes())


def _tree_manifest(root: Path) -> dict[str, Any]:
    files: list[dict[str, str]] = []
    if not root.exists():
        return {"sha256": None, "file_count": 0}
    for path in sorted(root.rglob("*")):
        if not path.is_file():
            continue
        relative = path.relative_to(root)
        if "__pycache__" in relative.parts or path.suffix in {".pyc", ".pyo"}:
            continue
        files.append(
            {"path": relative.as_posix(), "sha256": _sha256_file(path)}
        )
    return {
        "sha256": _sha256_bytes(_canonical_bytes(files)),
        "file_count": len(files),
    }


def _redact(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            str(key): (
                "[REDACTED]"
                if SECRET_KEY_PATTERN.search(str(key))
                else _redact(item)
            )
            for key, item in value.items()
        }
    if isinstance(value, list):
        return [_redact(item) for item in value]
    return value


def _read_json(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError(f"JSON object required: {path}")
    return value


def _write_json(value: Any, output: str | None = None) -> None:
    rendered = json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n"
    if not output or output == "-":
        sys.stdout.write(rendered)
        return
    path = Path(output)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(rendered, encoding="utf-8")


def validate_environment_profiles(profiles_dir: Path) -> dict[str, Any]:
    profiles: list[dict[str, Any]] = []
    seen: dict[tuple[str, str], str] = {}
    collisions: list[dict[str, str]] = []
    expected = set(ENVIRONMENT_ORDER)

    for path in sorted(profiles_dir.glob("*.json")):
        profile = _read_json(path)
        environment = str(profile.get("environment", ""))
        if environment not in ENVIRONMENT_ORDER:
            raise ValueError(f"unknown environment in {path.name}: {environment}")
        if int(profile.get("promotion_order", -1)) != ENVIRONMENT_ORDER[environment]:
            raise ValueError(f"invalid promotion order for {environment}")

        required_text = ("project_name", "network_name", "data_namespace")
        for field in required_text:
            if not str(profile.get(field, "")).strip():
                raise ValueError(f"{field} is required for {environment}")
            key = (field, str(profile[field]))
            if key in seen:
                collisions.append(
                    {
                        "field": field,
                        "value": str(profile[field]),
                        "environments": f"{seen[key]},{environment}",
                    }
                )
            seen[key] = environment

        ports = profile.get("ports")
        if not isinstance(ports, dict) or not ports:
            raise ValueError(f"ports are required for {environment}")
        for name, raw_port in sorted(ports.items()):
            port = int(raw_port)
            if not 1 <= port <= 65535:
                raise ValueError(f"invalid port {name}={port} for {environment}")
            key = ("published_port", str(port))
            if key in seen:
                collisions.append(
                    {
                        "field": "published_port",
                        "value": str(port),
                        "environments": f"{seen[key]},{environment}",
                    }
                )
            seen[key] = environment

        controls = profile.get("controls")
        if not isinstance(controls, dict) or "auth_required" not in controls:
            raise ValueError(f"controls.auth_required is required for {environment}")
        profiles.append(
            {
                "environment": environment,
                "promotion_order": ENVIRONMENT_ORDER[environment],
                "project_name": profile["project_name"],
                "network_name": profile["network_name"],
                "data_namespace": profile["data_namespace"],
                "ports": dict(sorted((str(k), int(v)) for k, v in ports.items())),
                "controls": controls,
                "profile_sha256": _sha256_file(path),
                "path": path.name,
            }
        )

    found = {item["environment"] for item in profiles}
    if collisions:
        raise ValueError(f"environment collision detected: {collisions}")
    if found != expected:
        raise ValueError(
            "environment profile set must be development,test,preproduction; "
            f"found={','.join(sorted(found))}"
        )
    profiles.sort(key=lambda item: int(item["promotion_order"]))
    differences: list[dict[str, Any]] = []
    baseline = profiles[0]
    for profile in profiles[1:]:
        changed = [
            field
            for field in ("project_name", "network_name", "data_namespace", "ports", "controls")
            if profile[field] != baseline[field]
        ]
        differences.append(
            {
                "from": baseline["environment"],
                "to": profile["environment"],
                "changed_fields": changed,
            }
        )
    return {
        "schema_version": 1,
        "status": "ready",
        "promotion_path": [item["environment"] for item in profiles],
        "profiles": profiles,
        "collisions": [],
        "differences": differences,
    }


def build_readiness_report(request: dict[str, Any]) -> dict[str, Any]:
    components = request.get("components")
    if not isinstance(components, list):
        raise ValueError("components must be a list")
    category_reports: dict[str, dict[str, Any]] = {}
    blocking: list[str] = []
    redacted_components: list[dict[str, Any]] = []
    for component in components:
        if not isinstance(component, dict):
            raise ValueError("each component must be an object")
        category = str(component.get("category", ""))
        if category not in READINESS_CATEGORIES:
            raise ValueError(f"unsupported readiness category: {category}")
        name = str(component.get("name", "")).strip()
        if not name:
            raise ValueError("component name is required")
        status = str(component.get("status", "unknown")).lower()
        required = bool(component.get("required", True))
        ready = status in HEALTHY_STATES
        if required and not ready:
            blocking.append(name)
        redacted_components.append(_redact(component))

    for category in READINESS_CATEGORIES:
        selected = [
            item for item in redacted_components if item["category"] == category
        ]
        required = [item for item in selected if item.get("required", True)]
        blocked = [
            item["name"]
            for item in required
            if str(item.get("status", "unknown")).lower() not in HEALTHY_STATES
        ]
        category_reports[category] = {
            "status": "ready" if required and not blocked else (
                "blocked" if blocked else "unconfigured"
            ),
            "component_count": len(selected),
            "required_count": len(required),
            "blocked_components": blocked,
        }

    unconfigured = sorted(
        category
        for category, report in category_reports.items()
        if report["status"] == "unconfigured"
    )
    return {
        "schema_version": 1,
        "environment": str(request.get("environment", "unknown")),
        "status": "blocked" if blocking or unconfigured else "ready",
        "blocking_components": sorted(blocking),
        "unconfigured_categories": unconfigured,
        "categories": category_reports,
        "components": redacted_components,
    }


def build_compose_readiness_report(
    rows: list[dict[str, Any]], *, environment: str
) -> dict[str, Any]:
    components: list[dict[str, Any]] = []
    for row in rows:
        service = str(row.get("Service", ""))
        category = COMPOSE_SERVICE_CATEGORIES.get(service)
        if not category:
            continue
        state = str(row.get("State", "unknown")).lower()
        health = str(row.get("Health", "")).lower()
        exit_code = int(row.get("ExitCode", -1))
        completed = state == "exited" and exit_code == 0 and service.endswith("-init")
        ready = completed or (state == "running" and health in {"", "healthy"})
        components.append(
            {
                "name": service,
                "category": category,
                "status": "completed" if completed else ("healthy" if ready else "unhealthy"),
                "required": True,
                "details": {
                    "state": state,
                    "health": health or "not_reported",
                    "exit_code": exit_code,
                    "container_id": str(row.get("ID", ""))[:12],
                },
            }
        )
    report = build_readiness_report(
        {"environment": environment, "components": components}
    )
    report["collector"] = "docker_compose_ps"
    return report


def collect_compose_readiness(
    compose_file: Path,
    *,
    project_name: str,
    environment: str,
    env_file: Path | None = None,
) -> dict[str, Any]:
    command = ["docker", "compose"]
    if env_file:
        command.extend(["--env-file", str(env_file)])
    command.extend(
        ["-p", project_name, "-f", str(compose_file), "ps", "-a", "--format", "json"]
    )
    completed = subprocess.run(command, check=True, capture_output=True, text=True)
    rendered = completed.stdout.strip()
    rows: list[dict[str, Any]] = []
    if rendered:
        try:
            parsed = json.loads(rendered)
            rows = parsed if isinstance(parsed, list) else [parsed]
        except json.JSONDecodeError:
            rows = [json.loads(line) for line in rendered.splitlines() if line.strip()]
    return build_compose_readiness_report(rows, environment=environment)


def _promotion_digest_payload(bundle: dict[str, Any]) -> dict[str, Any]:
    return {
        "schema_version": bundle["schema_version"],
        "source_environment": bundle["source_environment"],
        "target_environment": bundle["target_environment"],
        "approval_ref": bundle["approval_ref"],
        "objects": bundle["objects"],
    }


def create_promotion_bundle(request: dict[str, Any]) -> dict[str, Any]:
    source = str(request.get("source_environment", ""))
    target = str(request.get("target_environment", ""))
    if source not in ENVIRONMENT_ORDER or target not in ENVIRONMENT_ORDER:
        raise ValueError("known source and target environments are required")
    if ENVIRONMENT_ORDER[target] - ENVIRONMENT_ORDER[source] != 10:
        raise ValueError("promotion must target the adjacent environment")
    approval_ref = str(request.get("approval_ref", "")).strip()
    if not approval_ref:
        raise ValueError("approval_ref is required")
    objects = request.get("objects")
    if not isinstance(objects, list) or not objects:
        raise ValueError("at least one promotion object is required")
    normalized: list[dict[str, Any]] = []
    identities: set[tuple[str, str]] = set()
    for item in objects:
        if not isinstance(item, dict):
            raise ValueError("promotion object must be an object")
        kind = str(item.get("kind", ""))
        uid = str(item.get("uid", "")).strip()
        version = str(item.get("version", "")).strip()
        digest = str(item.get("payload_sha256", "")).lower()
        rollback_ref = str(item.get("rollback_ref", "")).strip()
        evidence_refs = item.get("evidence_refs")
        if kind not in PROMOTABLE_KINDS:
            raise ValueError(f"unsupported promotion kind: {kind}")
        if not uid or not version or not rollback_ref:
            raise ValueError("uid, version and rollback_ref are required")
        if not SHA256_PATTERN.fullmatch(digest):
            raise ValueError(f"invalid payload sha256 for {kind}:{uid}")
        if not isinstance(evidence_refs, list) or not evidence_refs:
            raise ValueError(f"evidence_refs are required for {kind}:{uid}")
        identity = (kind, uid)
        if identity in identities:
            raise ValueError(f"duplicate promotion object: {kind}:{uid}")
        identities.add(identity)
        normalized.append(
            {
                "kind": kind,
                "uid": uid,
                "version": version,
                "payload_sha256": digest,
                "rollback_ref": rollback_ref,
                "evidence_refs": sorted(str(value) for value in evidence_refs),
            }
        )
    normalized.sort(key=lambda item: (item["kind"], item["uid"]))
    bundle: dict[str, Any] = {
        "schema_version": 1,
        "status": "approved",
        "source_environment": source,
        "target_environment": target,
        "approval_ref": approval_ref,
        "objects": normalized,
    }
    bundle["bundle_sha256"] = _sha256_bytes(
        _canonical_bytes(_promotion_digest_payload(bundle))
    )
    return bundle


def verify_promotion_bundle(bundle: dict[str, Any]) -> dict[str, Any]:
    expected = _sha256_bytes(_canonical_bytes(_promotion_digest_payload(bundle)))
    actual = str(bundle.get("bundle_sha256", ""))
    if actual != expected:
        raise ValueError(f"promotion bundle digest mismatch: expected={expected}, actual={actual}")
    return {
        "schema_version": 1,
        "status": "verified",
        "bundle_sha256": actual,
        "object_count": len(bundle.get("objects", [])),
    }


def evaluate_customer_migration(request: dict[str, Any]) -> dict[str, Any]:
    reasons: list[str] = []
    mappings = request.get("mappings")
    if not isinstance(mappings, list):
        raise ValueError("mappings must be a list")
    incomplete = [
        str(item.get("source", "unknown"))
        for item in mappings
        if isinstance(item, dict)
        and item.get("required", True)
        and (not str(item.get("source", "")).strip() or not str(item.get("target", "")).strip())
    ]
    targets = [
        str(item.get("target", ""))
        for item in mappings
        if isinstance(item, dict) and str(item.get("target", "")).strip()
    ]
    duplicates = sorted({item for item in targets if targets.count(item) > 1})
    if incomplete or duplicates:
        reasons.append("mapping_incomplete")

    incremental = request.get("incremental") or {}
    latest = int(incremental.get("latest_sequence", 0))
    caught_up = int(incremental.get("caught_up_sequence", -1))
    lag = max(0, latest - caught_up)
    if lag:
        reasons.append("incremental_not_caught_up")

    reconciliation = request.get("reconciliation") or {}
    source_count = int(reconciliation.get("source_count", -1))
    target_count = int(reconciliation.get("target_count", -2))
    source_digest = str(reconciliation.get("source_digest", ""))
    target_digest = str(reconciliation.get("target_digest", ""))
    reconciled = (
        source_count >= 0
        and source_count == target_count
        and bool(source_digest)
        and source_digest == target_digest
    )
    if not reconciled:
        reasons.append("reconciliation_mismatch")

    rollback = request.get("rollback") or {}
    rollback_ready = bool(rollback.get("backup_ref") and rollback.get("release_ref"))
    if not rollback_ready:
        reasons.append("rollback_not_prepared")
    snapshot = request.get("source_snapshot") or {}
    if not snapshot.get("snapshot_ref"):
        reasons.append("source_snapshot_missing")

    return {
        "schema_version": 1,
        "migration_id": str(request.get("migration_id", "unknown")),
        "mode": "dry_run",
        "source_mutated": False,
        "status": "blocked" if reasons else "ready",
        "blocking_reasons": sorted(set(reasons)),
        "source_snapshot": _redact(snapshot),
        "mapping": {
            "status": "blocked" if incomplete or duplicates else "ready",
            "mapping_count": len(mappings),
            "incomplete_sources": incomplete,
            "duplicate_targets": duplicates,
        },
        "incremental_catchup": {
            "status": "caught_up" if lag == 0 else "lagging",
            "latest_sequence": latest,
            "caught_up_sequence": caught_up,
            "lag": lag,
        },
        "reconciliation": {
            "status": "matched" if reconciled else "mismatch",
            "source_count": source_count,
            "target_count": target_count,
            "source_digest": source_digest,
            "target_digest": target_digest,
        },
        "checkpoints": sorted(str(value) for value in request.get("checkpoints", [])),
        "rollback": {
            "status": "prepared" if rollback_ready else "blocked",
            "backup_ref": rollback.get("backup_ref"),
            "release_ref": rollback.get("release_ref"),
            "strategy": "restore_prior_release_and_pre_migration_backup",
            "database_downgrade": False,
        },
    }


def _git_value(repo_root: Path, *arguments: str, default: str = "unknown") -> str:
    try:
        completed = subprocess.run(
            ["git", "-C", str(repo_root), *arguments],
            check=True,
            capture_output=True,
            text=True,
        )
        return completed.stdout.strip() or default
    except (OSError, subprocess.CalledProcessError):
        return default


def _migration_head(repo_root: Path) -> str:
    revisions: dict[str, set[str]] = {}
    referenced: set[str] = set()
    for path in sorted((repo_root / "migrations/versions").glob("*.py")):
        tree = ast.parse(path.read_text(encoding="utf-8"))
        revision = ""
        parents: set[str] = set()
        for node in tree.body:
            if not isinstance(node, (ast.Assign, ast.AnnAssign)):
                continue
            target = node.targets[0] if isinstance(node, ast.Assign) else node.target
            if not isinstance(target, ast.Name):
                continue
            if target.id not in {"revision", "down_revision"}:
                continue
            value_node = node.value
            try:
                value = ast.literal_eval(value_node)
            except (ValueError, TypeError):
                continue
            if target.id == "revision":
                revision = str(value)
            elif value:
                if isinstance(value, (list, tuple)):
                    parents.update(str(item) for item in value)
                else:
                    parents.add(str(value))
        if revision:
            revisions[revision] = parents
            referenced.update(parents)
    heads = sorted(set(revisions) - referenced)
    if len(heads) != 1:
        raise ValueError(f"exactly one migration head required; found={heads}")
    return heads[0]


def build_release_manifest(
    repo_root: Path, *, version: str, source_date_epoch: int
) -> dict[str, Any]:
    repo_root = repo_root.resolve()
    openapi = repo_root / "docs/architecture/OPENAPI.yaml"
    route_match = re.search(
        r"^x-route-count:\s*(\d+)\s*$",
        openapi.read_text(encoding="utf-8"),
        re.MULTILINE,
    )
    if not route_match:
        raise ValueError("OpenAPI x-route-count is missing")
    source_trees = {
        name: _tree_manifest(repo_root / name)
        for name in ("app", "database", "migrations")
    }
    release_copies: dict[str, dict[str, Any]] = {}
    for name, source in source_trees.items():
        release = _tree_manifest(repo_root / "deployment" / name)
        release_copies[name] = {
            **release,
            "matches_source": release["sha256"] == source["sha256"],
        }
    manifest = {
        "schema_version": 1,
        "version": version,
        "source_date_epoch": int(source_date_epoch),
        "source": {
            "commit": _git_value(repo_root, "rev-parse", "HEAD"),
            "tree": _git_value(repo_root, "rev-parse", "HEAD^{tree}"),
        },
        "source_trees": source_trees,
        "release_copies": release_copies,
        "migration": {
            "head": _migration_head(repo_root),
            "tree_sha256": source_trees["migrations"]["sha256"],
        },
        "openapi": {
            "route_count": int(route_match.group(1)),
            "sha256": _sha256_file(openapi),
        },
        "compose": {
            "sha256": _sha256_file(repo_root / "deploy/docker/docker-compose.yml")
        },
        "environment_profiles": _tree_manifest(repo_root / "deployment/environments"),
        "dependency_inputs": {
            "backend_sha256": _sha256_file(repo_root / "requirements.txt"),
            "frontend_sha256": _sha256_file(repo_root / "frontend/package-lock.json"),
        },
    }
    if not all(item["matches_source"] for item in release_copies.values()):
        raise ValueError("release copies do not match source trees")
    manifest["manifest_sha256"] = _sha256_bytes(_canonical_bytes(manifest))
    return manifest


def _component(name: str, version: str, ecosystem: str) -> dict[str, str]:
    return {
        "type": "library",
        "name": name,
        "version": version,
        "purl": f"pkg:{ecosystem}/{name}@{version}",
    }


def _sbom(name: str, components: list[dict[str, str]], input_digest: str) -> dict[str, Any]:
    components.sort(key=lambda item: (item["name"].lower(), item["version"]))
    return {
        "bomFormat": "CycloneDX",
        "specVersion": "1.5",
        "version": 1,
        "metadata": {
            "component": {"type": "application", "name": name},
            "properties": [
                {"name": "dataops:source-input-sha256", "value": input_digest}
            ],
        },
        "components": components,
    }


def generate_backend_sbom(requirements: Path) -> dict[str, Any]:
    components: list[dict[str, str]] = []
    pattern = re.compile(r"^([A-Za-z0-9_.-]+)\s*(===|==|>=|~=|<=|>|<)\s*([^;\s]+)")
    for raw_line in requirements.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        match = pattern.match(line)
        if not match:
            raise ValueError(f"unsupported requirement line: {line}")
        name, operator, version = match.groups()
        rendered_version = version if operator in {"==", "==="} else operator + version
        components.append(_component(name, rendered_version, "pypi"))
    return _sbom("dataops-platform-backend", components, _sha256_file(requirements))


def generate_frontend_sbom(package_lock: Path) -> dict[str, Any]:
    lock = _read_json(package_lock)
    components: list[dict[str, str]] = []
    packages = lock.get("packages") or {}
    if not isinstance(packages, dict):
        raise ValueError("package-lock packages object is required")
    for package_path, value in packages.items():
        if not package_path or not isinstance(value, dict):
            continue
        name = str(value.get("name", ""))
        if not name and "node_modules/" in package_path:
            name = package_path.rsplit("node_modules/", 1)[-1]
        version = str(value.get("version", ""))
        if name and version:
            components.append(_component(name, version, "npm"))
    return _sbom("dataops-platform-frontend", components, _sha256_file(package_lock))


def generate_sboms(repo_root: Path) -> dict[str, Any]:
    return {
        "backend": generate_backend_sbom(repo_root / "requirements.txt"),
        "frontend": generate_frontend_sbom(repo_root / "frontend/package-lock.json"),
    }


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)

    validate = subparsers.add_parser("validate-environments")
    validate.add_argument("--profiles-dir", type=Path, required=True)
    validate.add_argument("--output")

    readiness = subparsers.add_parser("readiness")
    readiness.add_argument("--input", type=Path, required=True)
    readiness.add_argument("--output")

    compose_readiness = subparsers.add_parser("compose-readiness")
    compose_readiness.add_argument("--compose-file", type=Path, required=True)
    compose_readiness.add_argument("--project-name", required=True)
    compose_readiness.add_argument("--environment", required=True)
    compose_readiness.add_argument("--env-file", type=Path)
    compose_readiness.add_argument("--output")

    promotion = subparsers.add_parser("promotion")
    promotion.add_argument("--input", type=Path, required=True)
    promotion.add_argument("--output")
    promotion.add_argument("--verify", action="store_true")

    migration = subparsers.add_parser("migration-dry-run")
    migration.add_argument("--input", type=Path, required=True)
    migration.add_argument("--output")

    manifest = subparsers.add_parser("manifest")
    manifest.add_argument("--repo-root", type=Path, required=True)
    manifest.add_argument("--version", required=True)
    manifest.add_argument("--source-date-epoch", type=int, required=True)
    manifest.add_argument("--output")

    sbom = subparsers.add_parser("sbom")
    sbom.add_argument("--repo-root", type=Path, required=True)
    sbom.add_argument("--output-dir", required=True)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    if args.command == "validate-environments":
        _write_json(validate_environment_profiles(args.profiles_dir), args.output)
    elif args.command == "readiness":
        _write_json(build_readiness_report(_read_json(args.input)), args.output)
    elif args.command == "compose-readiness":
        _write_json(
            collect_compose_readiness(
                args.compose_file,
                project_name=args.project_name,
                environment=args.environment,
                env_file=args.env_file,
            ),
            args.output,
        )
    elif args.command == "promotion":
        value = _read_json(args.input)
        report = verify_promotion_bundle(value) if args.verify else create_promotion_bundle(value)
        _write_json(report, args.output)
    elif args.command == "migration-dry-run":
        _write_json(evaluate_customer_migration(_read_json(args.input)), args.output)
    elif args.command == "manifest":
        _write_json(
            build_release_manifest(
                args.repo_root,
                version=args.version,
                source_date_epoch=args.source_date_epoch,
            ),
            args.output,
        )
    elif args.command == "sbom":
        sboms = generate_sboms(args.repo_root)
        if args.output_dir == "-":
            _write_json(sboms)
        else:
            output_dir = Path(args.output_dir)
            output_dir.mkdir(parents=True, exist_ok=True)
            _write_json(sboms["backend"], str(output_dir / "backend.cdx.json"))
            _write_json(sboms["frontend"], str(output_dir / "frontend.cdx.json"))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(2)