pobsync/src/pobsync_backend/self_check.py

from __future__ import annotations

import os
import shutil
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

from django.conf import settings
from django.db import connection

from .models import GlobalConfig


CheckStatus = Literal["ok", "warning", "failed", "skipped"]


@dataclass(frozen=True)
class SelfCheck:
    name: str
    status: CheckStatus
    message: str
    detail: str = ""


def collect_self_checks() -> list[SelfCheck]:
    checks: list[SelfCheck] = []
    checks.extend(_django_checks())
    checks.extend(_path_checks())
    checks.extend(_binary_checks())
    checks.extend(_database_checks())
    checks.extend(_config_checks())
    checks.extend(_systemd_checks())
    return checks


def summarize_self_checks(checks: list[SelfCheck]) -> dict[str, int]:
    return {
        "ok": sum(1 for check in checks if check.status == "ok"),
        "warning": sum(1 for check in checks if check.status == "warning"),
        "failed": sum(1 for check in checks if check.status == "failed"),
        "skipped": sum(1 for check in checks if check.status == "skipped"),
    }


def _django_checks() -> list[SelfCheck]:
    checks = [
        SelfCheck(
            "Django debug",
            "warning" if settings.DEBUG else "ok",
            "DEBUG is enabled." if settings.DEBUG else "DEBUG is disabled.",
        ),
        SelfCheck(
            "Django secret key",
            "failed" if settings.SECRET_KEY == "dev-only-change-me" else "ok",
            "Default development secret key is still active."
            if settings.SECRET_KEY == "dev-only-change-me"
            else "Secret key is configured.",
        ),
        SelfCheck(
            "Allowed hosts",
            "ok" if settings.ALLOWED_HOSTS else "failed",
            ", ".join(settings.ALLOWED_HOSTS) if settings.ALLOWED_HOSTS else "No allowed hosts configured.",
        ),
    ]
    return checks


def _path_checks() -> list[SelfCheck]:
    checks = []
    checks.append(_path_check("POBSYNC_HOME", Path(settings.POBSYNC_HOME), must_be_absolute=True, must_be_writable=True))
    checks.append(
        _path_check(
            "POBSYNC_BACKUP_ROOT",
            Path(settings.POBSYNC_BACKUP_ROOT),
            must_be_absolute=True,
            must_exist=True,
            must_be_writable=True,
        )
    )
    checks.append(
        _path_check(
            "Static root",
            Path(settings.STATIC_ROOT),
            must_be_absolute=True,
            must_exist=False,
            must_be_writable=True,
        )
    )
    db_settings = settings.DATABASES["default"]
    if db_settings["ENGINE"] == "django.db.backends.sqlite3":
        checks.append(
            _path_check(
                "SQLite directory",
                Path(str(db_settings["NAME"])).parent,
                must_be_absolute=True,
                must_exist=True,
                must_be_writable=True,
            )
        )
    return checks


def _path_check(
    name: str,
    path: Path,
    *,
    must_be_absolute: bool,
    must_exist: bool = False,
    must_be_writable: bool,
) -> SelfCheck:
    if must_be_absolute and not path.is_absolute():
        return SelfCheck(name, "failed", f"{path} is not absolute.")
    if must_exist and not path.exists():
        return SelfCheck(name, "failed", f"{path} does not exist.")

    target = path if path.exists() else path.parent
    if not target.exists():
        return SelfCheck(name, "failed", f"{target} does not exist.")
    if must_be_writable and not os.access(target, os.W_OK):
        return SelfCheck(name, "failed", f"{target} is not writable by this process.")
    return SelfCheck(name, "ok", str(path))


def _binary_checks() -> list[SelfCheck]:
    checks = []
    for binary in ("rsync", "ssh", "ssh-keygen", "gunicorn"):
        path = shutil.which(binary)
        checks.append(
            SelfCheck(
                f"Binary: {binary}",
                "ok" if path else "failed",
                path or f"{binary} was not found in PATH.",
            )
        )
    return checks


def _database_checks() -> list[SelfCheck]:
    try:
        with connection.cursor() as cursor:
            cursor.execute("SELECT 1")
            cursor.fetchone()
    except Exception as exc:
        return [SelfCheck("Database connection", "failed", f"{type(exc).__name__}: {exc}")]
    return [SelfCheck("Database connection", "ok", settings.DATABASES["default"]["ENGINE"])]


def _config_checks() -> list[SelfCheck]:
    try:
        global_config = GlobalConfig.objects.get(name="default")
    except GlobalConfig.DoesNotExist:
        return [SelfCheck("Global config", "warning", "Default global config has not been created yet.")]

    status: CheckStatus = "ok"
    message = "Default global config exists."
    if global_config.backup_root != settings.POBSYNC_BACKUP_ROOT:
        status = "warning"
        message = "Global config backup root differs from runtime POBSYNC_BACKUP_ROOT."
    return [
        SelfCheck(
            "Global config",
            status,
            message,
            f"database={global_config.backup_root} runtime={settings.POBSYNC_BACKUP_ROOT}",
        )
    ]


def _systemd_checks() -> list[SelfCheck]:
    if not Path("/run/systemd/system").exists() or shutil.which("systemctl") is None:
        return [
            SelfCheck(
                "Systemd services",
                "skipped",
                "systemd is not available in this runtime.",
                "This is expected inside Docker.",
            )
        ]

    checks = []
    for service in ("pobsync-web.service", "pobsync-worker.service", "pobsync-scheduler.service"):
        result = subprocess.run(
            ["systemctl", "is-active", service],
            check=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=5,
        )
        active_state = result.stdout.strip() or result.stderr.strip()
        checks.append(
            SelfCheck(
                service,
                "ok" if result.returncode == 0 else "failed",
                active_state,
            )
        )
    return checks