service_manager.py

from __future__ import annotations

import subprocess
from dataclasses import dataclass, field

import privileged
import service_assets

SERVICE_CATALOG = [
    {
        "key": "apache",
        "label": "Apache",
        "candidates": ["apache2", "httpd"],
        "ports": [80, 443],
    },
    {
        "key": "nginx",
        "label": "Nginx",
        "candidates": ["nginx"],
        "ports": [80, 443],
    },
    {
        "key": "mysql",
        "label": "MySQL",
        "candidates": ["mysql", "mysqld"],
        "ports": [3306],
    },
    {
        "key": "mariadb",
        "label": "MariaDB",
        "candidates": ["mariadb"],
        "ports": [3306],
    },
    {
        "key": "postgresql",
        "label": "PostgreSQL",
        "candidates": ["postgresql"],
        "ports": [5432],
        "prefix_fallback": "postgresql",
    },
    {
        "key": "redis",
        "label": "Redis",
        "candidates": ["redis", "redis-server"],
        "ports": [6379],
    },
    {
        "key": "mongodb",
        "label": "MongoDB",
        "candidates": ["mongod", "mongodb"],
        "ports": [27017],
    },
    {
        "key": "php",
        "label": "PHP",
        "candidates": [],
        "ports": [],
        "detect": "php",
    },
    {
        "key": "phpmyadmin",
        "label": "phpMyAdmin",
        "candidates": [],
        "ports": [80],
        "detect": "phpmyadmin",
    },
]

_unit_cache: set[str] | None = None
_pending_units: dict[str, subprocess.Popen] = {}


@dataclass
class ServiceInfo:
    key: str
    label: str
    unit: str
    ports: list[int] = field(default_factory=list)
    state: str = "unknown"
    main_pid: int = 0
    available: bool = True
    sub_state: str = ""
    controllable: bool = True


def _run_command(args: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        args,
        capture_output=True,
        text=True,
        check=False,
    )


def _normalize_unit_name(name: str) -> str:
    return name.removesuffix(".service")


def _load_unit_cache() -> set[str]:
    global _unit_cache
    if _unit_cache is not None:
        return _unit_cache
    result = _run_command(
        [
            "systemctl",
            "list-unit-files",
            "--type=service",
            "--no-pager",
            "--no-legend",
        ]
    )
    units: set[str] = set()
    for line in result.stdout.splitlines():
        parts = line.split()
        if parts:
            units.add(_normalize_unit_name(parts[0]))
    _unit_cache = units
    return units


def _resolve_unit(entry: dict) -> str | None:
    units = _load_unit_cache()
    for candidate in entry["candidates"]:
        if candidate in units:
            return candidate
    prefix = entry.get("prefix_fallback")
    if prefix:
        matches = sorted(
            name for name in units if name.startswith(prefix) and "@" not in name
        )
        if matches:
            return matches[0]
    return None


def _parse_show_output(stdout: str) -> tuple[str, int, str]:
    state = "unknown"
    main_pid = 0
    sub_state = ""
    for line in stdout.splitlines():
        if line.startswith("ActiveState="):
            state = line.split("=", 1)[1].strip()
        elif line.startswith("MainPID="):
            try:
                main_pid = int(line.split("=", 1)[1].strip())
            except ValueError:
                main_pid = 0
        elif line.startswith("SubState="):
            sub_state = line.split("=", 1)[1].strip()
    return state, main_pid, sub_state


def get_service_state(unit: str, key: str = "", label: str = "", ports: list[int] | None = None) -> ServiceInfo:
    active_result = _run_command(["systemctl", "is-active", unit])
    active_state = active_result.stdout.strip() or "unknown"
    show_result = _run_command(
        [
            "systemctl",
            "show",
            unit,
            "--property=MainPID,ActiveState,SubState",
        ]
    )
    show_state, main_pid, sub_state = _parse_show_output(show_result.stdout)
    state = show_state if show_state != "unknown" else active_state
    return ServiceInfo(
        key=key or unit,
        label=label or unit,
        unit=unit,
        ports=ports or [],
        state=state,
        main_pid=main_pid,
        available=True,
        sub_state=sub_state,
    )


def _apache_is_active() -> bool:
    for candidate in ("apache2", "httpd"):
        result = _run_command(["systemctl", "is-active", candidate])
        if result.stdout.strip() == "active":
            return True
    return False


def _detect_local_entry(entry: dict) -> ServiceInfo | None:
    detect = entry.get("detect")
    if detect == "php":
        if not service_assets.php_is_installed():
            return None
        assets = service_assets.get_assets("php")
        if not assets.configs:
            return None
        return ServiceInfo(
            key=entry["key"],
            label=entry["label"],
            unit="",
            ports=list(entry["ports"]),
            state="active",
            main_pid=0,
            available=True,
            controllable=False,
        )
    if detect == "phpmyadmin":
        if not service_assets.phpmyadmin_is_installed():
            return None
        state = "active" if _apache_is_active() else "inactive"
        return ServiceInfo(
            key=entry["key"],
            label=entry["label"],
            unit="",
            ports=list(entry["ports"]),
            state=state,
            main_pid=0,
            available=True,
            controllable=False,
        )
    return None


def _catalog_entry(entry: dict, unit: str | None) -> ServiceInfo:
    local = _detect_local_entry(entry)
    if local is not None:
        return local
    if entry.get("detect") and unit is None:
        return ServiceInfo(
            key=entry["key"],
            label=entry["label"],
            unit="",
            ports=list(entry["ports"]),
            state="missing",
            main_pid=0,
            available=False,
            controllable=False,
        )
    if unit is None:
        return ServiceInfo(
            key=entry["key"],
            label=entry["label"],
            unit="",
            ports=list(entry["ports"]),
            state="missing",
            main_pid=0,
            available=False,
            controllable=False,
        )
    info = get_service_state(
        unit,
        key=entry["key"],
        label=entry["label"],
        ports=list(entry["ports"]),
    )
    info.controllable = True
    return info


def discover_services() -> list[ServiceInfo]:
    services: list[ServiceInfo] = []
    for entry in SERVICE_CATALOG:
        unit = _resolve_unit(entry)
        if unit is None:
            continue
        services.append(_catalog_entry(entry, unit))
    return services


def list_catalog_services() -> list[ServiceInfo]:
    services: list[ServiceInfo] = []
    for entry in SERVICE_CATALOG:
        unit = _resolve_unit(entry)
        services.append(_catalog_entry(entry, unit))
    return services


def refresh_all(services: list[ServiceInfo]) -> list[ServiceInfo]:
    refreshed: list[ServiceInfo] = []
    for service in services:
        if service.key == "phpmyadmin" and service.available:
            service.state = "active" if _apache_is_active() else "inactive"
            refreshed.append(service)
            continue
        if not service.available or not service.unit:
            refreshed.append(service)
            continue
        updated = get_service_state(
            service.unit,
            key=service.key,
            label=service.label,
            ports=service.ports,
        )
        updated.controllable = True
        refreshed.append(updated)
    return refreshed


def is_action_pending(unit: str) -> bool:
    process = _pending_units.get(unit)
    if process is None:
        return False
    if process.poll() is None:
        return True
    _pending_units.pop(unit, None)
    return False


def run_service_action(unit: str, action: str) -> subprocess.Popen:
    if action not in {"start", "stop", "restart"}:
        raise ValueError(action)
    existing = _pending_units.get(unit)
    if existing is not None and existing.poll() is None:
        return existing
    process = privileged.popen(["systemctl", action, unit])
    _pending_units[unit] = process
    return process


def clear_pending(unit: str) -> None:
    _pending_units.pop(unit, None)


def invalidate_unit_cache() -> None:
    global _unit_cache
    _unit_cache = None