ServiceMap-Public/backend/app/health.py

from __future__ import annotations

from .models import HealthStatus, Service
from .store import Store

SEVERITY = {
    HealthStatus.MAINTENANCE: 0,
    HealthStatus.UNMONITORED: 0,
    HealthStatus.UNKNOWN: 0,
    HealthStatus.HEALTHY: 1,
    HealthStatus.RECOVERING: 2,
    HealthStatus.DEGRADED: 3,
    HealthStatus.DOWN: 4,
}


def _direct_status(service: Service, store: Store) -> HealthStatus:
    return store.service_health(service.id)


def _dependency_status(service: Service, store: Store, visiting: set[str]) -> HealthStatus:
    if not service.depends_on:
        return HealthStatus.HEALTHY

    states: list[HealthStatus] = []
    for dependency_id in service.depends_on:
        dependency = store.get_service(dependency_id)
        if dependency is None:
            states.append(HealthStatus.UNKNOWN)
            continue
        states.append(_effective_status(dependency, store, visiting))
    return max(states, key=lambda status: SEVERITY[status])


def _effective_status(service: Service, store: Store, visiting: set[str]) -> HealthStatus:
    if service.id in visiting:
        return HealthStatus.UNKNOWN
    next_visiting = set(visiting)
    next_visiting.add(service.id)
    own = _direct_status(service, store)
    if own == HealthStatus.MAINTENANCE:
        return own
    dependency = _dependency_status(service, store, next_visiting)
    if own == HealthStatus.DOWN:
        return own
    if dependency == HealthStatus.DOWN:
        return HealthStatus.DEGRADED
    if dependency in {HealthStatus.DEGRADED, HealthStatus.RECOVERING} and own in {
        HealthStatus.HEALTHY,
        HealthStatus.RECOVERING,
    }:
        return HealthStatus.DEGRADED
    return own


def dependency_status(service: Service, store: Store) -> HealthStatus:
    return _dependency_status(service, store, {service.id})


def effective_status(service: Service, store: Store) -> HealthStatus:
    return _effective_status(service, store, set())


def status_maps(
    services: list[Service],
    *,
    direct_statuses: dict[str, HealthStatus] | None = None,
) -> tuple[dict[str, HealthStatus], dict[str, HealthStatus]]:
    """Calculate the complete topology in memory for one bounded API read."""

    by_id = {service.id: service for service in services}
    def legacy_direct_state(service: Service) -> HealthStatus:
        runtime_state = (service.runtime_state or "").casefold()
        runtime_health = (service.runtime_health or "").casefold()
        if not service.enabled:
            return HealthStatus.UNKNOWN
        if service.maintenance_enabled:
            return HealthStatus.MAINTENANCE
        if runtime_health == "unhealthy" or runtime_state in {"exited", "dead", "stopped"}:
            return HealthStatus.DOWN
        if runtime_state == "restarting":
            return HealthStatus.DEGRADED
        if not service.monitoring_enabled:
            if runtime_health == "healthy":
                return HealthStatus.HEALTHY
            return HealthStatus.UNMONITORED if service.monitoring_state.value in {"unmonitored", "observed"} else HealthStatus.UNKNOWN
        return service.state

    direct = {
        service.id: (direct_statuses.get(service.id, HealthStatus.UNKNOWN) if direct_statuses else legacy_direct_state(service))
        for service in services
    }
    effective: dict[str, HealthStatus] = {}
    dependency: dict[str, HealthStatus] = {}

    def visit(service_id: str, visiting: set[str]) -> HealthStatus:
        if service_id in effective:
            return effective[service_id]
        if service_id in visiting:
            return HealthStatus.UNKNOWN
        service = by_id.get(service_id)
        if service is None:
            return HealthStatus.UNKNOWN
        next_visiting = {*visiting, service_id}
        own = direct[service_id]
        if not service.depends_on:
            dependency[service_id] = HealthStatus.HEALTHY
            effective[service_id] = own
            return own
        dependencies = [visit(item, next_visiting) for item in service.depends_on]
        dependency_state = max(dependencies, key=lambda status: SEVERITY[status])
        dependency[service_id] = dependency_state
        if own == HealthStatus.MAINTENANCE or own == HealthStatus.DOWN:
            result = own
        elif dependency_state == HealthStatus.DOWN or (
            dependency_state in {HealthStatus.DEGRADED, HealthStatus.RECOVERING}
            and own in {HealthStatus.HEALTHY, HealthStatus.RECOVERING}
        ):
            result = HealthStatus.DEGRADED
        else:
            result = own
        effective[service_id] = result
        return result

    for service in services:
        visit(service.id, set())
    return effective, dependency


def validate_dependencies(
    depends_on: list[str],
    store: Store,
    *,
    service_id: str | None = None,
) -> None:
    if len(depends_on) != len(set(depends_on)):
        raise ValueError("Dependencies must be unique")
    if service_id and service_id in depends_on:
        raise ValueError("A service cannot depend on itself")

    services = store.list_services()
    known_ids = {service.id for service in services}
    missing = [dependency_id for dependency_id in depends_on if dependency_id not in known_ids]
    if missing:
        raise ValueError(f"Unknown dependency ids: {', '.join(missing)}")

    if service_id is None:
        return

    graph = {service.id: list(service.depends_on) for service in services}
    graph[service_id] = list(depends_on)
    visiting: set[str] = set()
    visited: set[str] = set()

    def visit(node: str) -> None:
        if node in visiting:
            raise ValueError("Dependency graph contains a cycle")
        if node in visited:
            return
        visiting.add(node)
        for dependency_id in graph.get(node, []):
            visit(dependency_id)
        visiting.remove(node)
        visited.add(node)

    for node in graph:
        visit(node)