ops-bridge/src/bridge/config.py

"""Config loading for OpsBridge."""
from __future__ import annotations

import os
import warnings
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Optional

import yaml

from bridge.models import ActorInfo, ActorType, HealthCheckConfig, ReconnectPolicy, TunnelConfig


class ConfigError(Exception):
    """Raised when config is invalid or missing."""


@dataclass
class BridgeConfig:
    tunnels: Dict[str, TunnelConfig]
    actors: Dict[str, ActorInfo]
    catalog_path: Optional[Path] = None


def _default_config_path() -> Path:
    return Path.home() / ".config" / "bridge" / "tunnels.yaml"


def load_config() -> BridgeConfig:
    """Load and validate tunnels.yaml. Respects BRIDGE_CONFIG env var."""
    path = Path(os.environ.get("BRIDGE_CONFIG", str(_default_config_path())))

    if not path.exists():
        raise ConfigError(f"Config file not found: {path}")

    try:
        with path.open() as f:
            raw = yaml.safe_load(f)
    except yaml.YAMLError as e:
        raise ConfigError(f"Invalid YAML in {path}: {e}") from e

    if not isinstance(raw, dict):
        raise ConfigError(f"Config must be a YAML mapping, got: {type(raw)}")

    tunnels = _parse_tunnels(raw.get("tunnels") or {})
    actors = _parse_actors(raw.get("actors") or {})

    catalog_path = None
    if "catalog_path" in raw and raw["catalog_path"]:
        catalog_path = Path(os.path.expanduser(str(raw["catalog_path"])))

    return BridgeConfig(tunnels=tunnels, actors=actors, catalog_path=catalog_path)


def _parse_tunnels(raw: dict) -> Dict[str, TunnelConfig]:
    tunnels = {}
    for name, data in raw.items():
        if not isinstance(data, dict):
            raise ConfigError(f"Tunnel '{name}' must be a mapping")
        tunnels[name] = _parse_tunnel(name, data)
    return tunnels


def _parse_tunnel(name: str, data: dict) -> TunnelConfig:
    required = ["host", "remote_port", "local_port", "ssh_user", "ssh_key", "actor"]
    for field in required:
        if field not in data:
            raise ConfigError(f"Tunnel '{name}' missing required field: {field}")

    reconnect = ReconnectPolicy()
    if "reconnect" in data and data["reconnect"]:
        r = data["reconnect"]
        reconnect = ReconnectPolicy(
            max_attempts=r.get("max_attempts", 0),
            backoff_initial=r.get("backoff_initial", 5),
            backoff_max=r.get("backoff_max", 60),
        )

    health_check = None
    if "health_check" in data and data["health_check"]:
        hc = data["health_check"]
        if "url" not in hc:
            raise ConfigError(f"Tunnel '{name}' health_check missing required field: url")
        health_check = HealthCheckConfig(
            url=hc["url"],
            interval_seconds=hc.get("interval_seconds", 30),
            timeout_seconds=hc.get("timeout_seconds", 5),
        )

    direction = str(data.get("direction", "reverse"))
    if direction not in ("reverse", "local"):
        raise ConfigError(f"Tunnel '{name}' direction must be 'reverse' or 'local', got: {direction!r}")

    cert_command = data.get("cert_command") or None
    if cert_command is not None:
        cert_command = str(cert_command)

    return TunnelConfig(
        name=name,
        host=str(data["host"]),
        remote_port=int(data["remote_port"]),
        local_port=int(data["local_port"]),
        ssh_user=str(data["ssh_user"]),
        ssh_key=str(data["ssh_key"]),
        actor=str(data["actor"]),
        reconnect=reconnect,
        health_check=health_check,
        direction=direction,
        cert_command=cert_command,
    )


_LEGACY_CLASS_MAP = {
    "human": ActorType.ADM,
    "automation": ActorType.ATM,
}

_ACTOR_TYPE_PREFIXES = {
    ActorType.ADM: "adm-",
    ActorType.AGT: "agt-",
    ActorType.ATM: "atm-",
}


def _parse_actor_type(name: str, raw_class: str) -> ActorType:
    if raw_class in _LEGACY_CLASS_MAP:
        warnings.warn(
            f"Actor '{name}': class '{raw_class}' is deprecated; "
            f"use '{_LEGACY_CLASS_MAP[raw_class].value}' instead.",
            DeprecationWarning,
            stacklevel=4,
        )
        return _LEGACY_CLASS_MAP[raw_class]
    try:
        return ActorType(raw_class)
    except ValueError:
        raise ConfigError(
            f"Actor '{name}' has unknown class '{raw_class}'; "
            f"must be one of: adm, agt, atm (or legacy: human, automation). "
            f"Run `bridge conventions` for the full naming rules."
        )


def _parse_actors(raw: dict) -> Dict[str, ActorInfo]:
    actors = {}
    for name, data in raw.items():
        if not isinstance(data, dict):
            raise ConfigError(f"Actor '{name}' must be a mapping")
        if "class" not in data:
            raise ConfigError(f"Actor '{name}' missing required field: class")
        actor_type = _parse_actor_type(name, str(data["class"]))
        required_prefix = _ACTOR_TYPE_PREFIXES[actor_type]
        if not name.startswith(required_prefix):
            raise ConfigError(
                f"Actor '{name}' has type '{actor_type.value}' but name must start "
                f"with '{required_prefix}' (got '{name}'). "
                f"Run `bridge conventions` for the full naming rules."
            )
        actors[name] = ActorInfo(
            name=name,
            actor_type=actor_type,
            description=str(data.get("description", "")),
        )
    return actors