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, List, 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)
    _reject_duplicate_local_ports(tunnels)
    return tunnels


def _reject_duplicate_local_ports(tunnels: Dict[str, TunnelConfig]) -> None:
    """Two `direction: local` tunnels must never share a local_port.

    Only local forwards bind a port on this machine
    (``-L local_port:remote_host:remote_port``). If two claim the same one,
    whichever binds first wins and the losers sit in an endless reconnect loop —
    the map reads healthy while the port answers from a different machine than
    the one you asked for. Silent, and worse than an outage. Fail at load
    instead (BRIDGE-WP-0007).

    Reverse tunnels are deliberately exempt. There, local_port is the
    *destination* on this workstation
    (``-R remote_port:remote_host:local_port``) and the listener is on the
    remote, so sharing is the intended design: every ``state-hub-*`` reverse
    tunnel targets 8000 so each remote box reaches this hub at its own 18000.
    Rejecting that would be rejecting the point of the tool.
    """
    by_port: Dict[int, List[str]] = {}
    for name, tunnel in tunnels.items():
        if tunnel.direction != "local":
            continue
        by_port.setdefault(tunnel.local_port, []).append(name)
    collisions = {port: names for port, names in by_port.items() if len(names) > 1}
    if not collisions:
        return
    detail = "; ".join(
        f"port {port}: {', '.join(sorted(names))}" for port, names in sorted(collisions.items())
    )
    raise ConfigError(
        f"Duplicate local_port among 'direction: local' tunnels ({detail}). "
        f"Each local forward binds that port here, so they would race for it — "
        f"give the non-canonical one a different port."
    )


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,
        remote_host=str(data.get("remote_host", "127.0.0.1")),
        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