bambuddy/backend/app/api/routes/system.py

"""System information API routes."""

import asyncio
import os
import platform
import time
from collections.abc import Callable
from datetime import datetime, timezone
from pathlib import Path

import psutil
from fastapi import APIRouter, Depends
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.core.auth import RequestPrinterScope, RequirePermissionIfAuthEnabled
from backend.app.core.config import APP_VERSION, settings
from backend.app.core.database import get_db
from backend.app.core.local_config import read_local_toml, read_ntp_gate
from backend.app.core.permissions import Permission
from backend.app.core.printer_scope import PrinterScope
from backend.app.models.archive import PrintArchive
from backend.app.models.filament import Filament
from backend.app.models.print_log import PrintLogEntry
from backend.app.models.printer import Printer
from backend.app.models.project import Project
from backend.app.models.smart_plug import SmartPlug
from backend.app.models.user import User
from backend.app.services.log_health import ScanResult, scan_logs
from backend.app.services.log_reader import collect_sensitive_strings
from backend.app.services.printer_manager import printer_manager

router = APIRouter(prefix="/system", tags=["system"])

STORAGE_USAGE_CACHE_SECONDS = 300
_storage_usage_cache: dict | None = None
_storage_usage_cache_ts: float | None = None
_storage_usage_lock = asyncio.Lock()


def get_directory_size(path: Path) -> int:
    """Calculate total size of a directory in bytes."""
    total = 0
    try:
        for entry in path.rglob("*"):
            if entry.is_file():
                total += entry.stat().st_size
    except (PermissionError, OSError):
        pass  # Return partial total if directory traversal is interrupted
    return total


def format_bytes(bytes_value: int) -> str:
    """Format bytes to human-readable string."""
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if bytes_value < 1024:
            return f"{bytes_value:.1f} {unit}"
        bytes_value /= 1024
    return f"{bytes_value:.1f} PB"


def format_uptime(seconds: float) -> str:
    """Format uptime in seconds to human-readable string."""
    days = int(seconds // 86400)
    hours = int((seconds % 86400) // 3600)
    minutes = int((seconds % 3600) // 60)

    parts = []
    if days > 0:
        parts.append(f"{days}d")
    if hours > 0:
        parts.append(f"{hours}h")
    if minutes > 0:
        parts.append(f"{minutes}m")

    return " ".join(parts) if parts else "< 1m"


def _is_under(path: Path, root: Path) -> bool:
    try:
        path.resolve().relative_to(root.resolve())
        return True
    except ValueError:
        return False


def _get_database_paths() -> list[Path]:
    from backend.app.core.db_dialect import is_sqlite

    if not is_sqlite():
        return []  # PostgreSQL — no local DB files
    candidates = [settings.base_dir / "bambuddy.db", settings.base_dir / "bambutrack.db"]
    return [path for path in candidates if path.exists()]


def _get_database_items() -> list[dict]:
    items: list[dict] = []
    for path in _get_database_paths():
        try:
            size = path.stat().st_size
        except OSError:
            continue
        items.append(
            {
                "name": path.name,
                "path": str(path),
                "bytes": size,
                "formatted": format_bytes(size),
            }
        )
    items.sort(key=lambda item: item["bytes"], reverse=True)
    return items


def _get_app_dir() -> Path:
    return settings.static_dir.parent


def _get_data_dirs() -> list[Path]:
    return [
        settings.archive_dir,
        settings.log_dir,
        settings.plate_calibration_dir,
        settings.base_dir / "virtual_printer",
        settings.base_dir / "firmware",
    ]


def _is_system_path(path: Path) -> bool:
    app_dir = _get_app_dir()
    if not _is_under(path, app_dir):
        return False
    return all(not _is_under(path, data_dir) for data_dir in _get_data_dirs())


def _get_storage_rules() -> list[tuple[str, str, Callable]]:
    base_dir = settings.base_dir
    archive_dir = settings.archive_dir
    library_dir = archive_dir / "library"
    virtual_printer_dir = base_dir / "virtual_printer"
    upload_dir = virtual_printer_dir / "uploads"

    db_paths = set(_get_database_paths())

    return [
        (
            "database",
            "Database",
            lambda path: path in db_paths,
        ),
        (
            "library_thumbnails",
            "Library Thumbnails",
            lambda path: _is_under(path, library_dir / "thumbnails"),
        ),
        (
            "library_files",
            "Library Files",
            lambda path: _is_under(path, library_dir / "files"),
        ),
        (
            "library_other",
            "Library Other",
            lambda path: _is_under(path, library_dir),
        ),
        (
            "archive_timelapses",
            "Timelapses",
            lambda path: _is_under(path, archive_dir) and "timelapse" in path.name.lower(),
        ),
        (
            "archive_thumbnails",
            "Thumbnails",
            lambda path: _is_under(path, archive_dir) and path.name.lower().startswith("thumbnail"),
        ),
        (
            "archive_files",
            "Archives",
            lambda path: _is_under(path, archive_dir),
        ),
        (
            "virtual_printer_upload_cache",
            "Virtual Printer Upload Cache",
            lambda path: _is_under(path, upload_dir / "cache"),
        ),
        (
            "virtual_printer_uploads",
            "Virtual Printer Uploads",
            lambda path: _is_under(path, upload_dir),
        ),
        (
            "virtual_printer_certs",
            "Virtual Printer Certs",
            lambda path: _is_under(path, virtual_printer_dir / "certs"),
        ),
        (
            "virtual_printer_other",
            "Virtual Printer Other",
            lambda path: _is_under(path, virtual_printer_dir),
        ),
        (
            "downloads",
            "Downloads",
            lambda path: _is_under(path, base_dir / "firmware"),
        ),
        (
            "plate_calibration",
            "Plate Calibration",
            lambda path: _is_under(path, settings.plate_calibration_dir),
        ),
        (
            "logs",
            "Logs",
            lambda path: _is_under(path, settings.log_dir),
        ),
    ]


def _classify_file(path: Path, rules: list[tuple[str, str, Callable]]) -> tuple[str, str]:
    for key, label, matcher in rules:
        try:
            if matcher(path):
                return key, label
        except OSError:
            continue
    return "other_data", "Other"


def _format_percentage(part: int, total: int) -> float:
    if total <= 0:
        return 0.0
    return round((part / total) * 100, 2)


def _get_other_bucket(path: Path, base_dir: Path) -> str:
    try:
        relative = path.resolve().relative_to(base_dir.resolve())
    except ValueError:
        return path.parent.name or path.name

    parts = relative.parts
    return parts[0] if parts else path.name


def _walk_files(roots: list[Path]) -> list[Path]:
    files: list[Path] = []
    stack = [root for root in roots if root.exists()]
    while stack:
        current = stack.pop()
        try:
            with os.scandir(current) as entries:
                for entry in entries:
                    try:
                        if entry.is_symlink():
                            continue
                        if entry.is_dir(follow_symlinks=False):
                            stack.append(Path(entry.path))
                        elif entry.is_file(follow_symlinks=False):
                            files.append(Path(entry.path))
                    except OSError:
                        continue
        except OSError:
            continue
    return files


def _scan_storage_usage() -> dict:
    base_dir = settings.base_dir
    rules = _get_storage_rules()

    roots = _get_data_dirs()

    seen_roots = set()
    unique_roots = []
    for root in roots:
        resolved = root.resolve()
        if resolved not in seen_roots:
            seen_roots.add(resolved)
            unique_roots.append(root)

    total_bytes = 0
    error_count = 0
    category_sizes: dict[str, dict] = {}
    other_breakdown: dict[tuple[str, str], int] = {}
    database_items = _get_database_items()

    files = _walk_files(unique_roots)
    for file_path in files:
        try:
            size = file_path.stat().st_size
        except OSError:
            error_count += 1
            continue

        total_bytes += size

        key, label = _classify_file(file_path, rules)
        if key not in category_sizes:
            category_sizes[key] = {"key": key, "label": label, "bytes": 0}
        category_sizes[key]["bytes"] += size

        if key == "other_data":
            bucket = _get_other_bucket(file_path, base_dir)
            kind = "system" if _is_system_path(file_path) else "data"
            other_breakdown[(bucket, kind)] = other_breakdown.get((bucket, kind), 0) + size

    for item in database_items:
        total_bytes += item["bytes"]
        key = "database"
        label = "Database"
        if key not in category_sizes:
            category_sizes[key] = {"key": key, "label": label, "bytes": 0}
        category_sizes[key]["bytes"] += item["bytes"]

    categories = []
    for item in category_sizes.values():
        bytes_value = item["bytes"]
        categories.append(
            {
                "key": item["key"],
                "label": item["label"],
                "bytes": bytes_value,
                "formatted": format_bytes(bytes_value),
                "percent_of_total": _format_percentage(bytes_value, total_bytes),
            }
        )

    categories.sort(key=lambda entry: entry["bytes"], reverse=True)

    other_items = []
    for (bucket, kind), size in other_breakdown.items():
        other_items.append(
            {
                "bucket": bucket,
                "label": bucket,
                "kind": kind,
                "deletable": kind != "system",
                "bytes": size,
                "formatted": format_bytes(size),
                "percent_of_total": _format_percentage(size, total_bytes),
            }
        )

    other_items.sort(key=lambda entry: entry["bytes"], reverse=True)

    return {
        "roots": [str(root) for root in unique_roots],
        "total_bytes": total_bytes,
        "total_formatted": format_bytes(total_bytes),
        "categories": categories,
        "other_breakdown": other_items,
        "scan_errors": error_count,
    }


async def _get_storage_usage_cached(refresh: bool, max_age_seconds: int) -> dict:
    global _storage_usage_cache
    global _storage_usage_cache_ts

    now = time.time()
    if not refresh and _storage_usage_cache and _storage_usage_cache_ts is not None:
        age = now - _storage_usage_cache_ts
        if age < max_age_seconds:
            return {
                **_storage_usage_cache,
                "cache": {
                    "hit": True,
                    "age_seconds": round(age, 2),
                    "max_age_seconds": max_age_seconds,
                },
            }

    async with _storage_usage_lock:
        now = time.time()
        if not refresh and _storage_usage_cache and _storage_usage_cache_ts is not None:
            age = now - _storage_usage_cache_ts
            if age < max_age_seconds:
                return {
                    **_storage_usage_cache,
                    "cache": {
                        "hit": True,
                        "age_seconds": round(age, 2),
                        "max_age_seconds": max_age_seconds,
                    },
                }

        snapshot = await asyncio.to_thread(_scan_storage_usage)
        _storage_usage_cache = {
            **snapshot,
            "generated_at": datetime.now(timezone.utc).isoformat(),
        }
        _storage_usage_cache_ts = time.time()
        return {
            **_storage_usage_cache,
            "cache": {
                "hit": False,
                "age_seconds": 0,
                "max_age_seconds": max_age_seconds,
            },
        }


@router.get("/info")
async def get_system_info(
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SYSTEM_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Get comprehensive system information."""

    # Database stats
    archive_count = await db.scalar(select(func.count(PrintArchive.id)))
    # Printers are counted and listed as the caller's printer list shows them.
    printer_count_query = select(func.count(Printer.id))
    if (printer_clause := printer_scope.where_strict(Printer.id)) is not None:
        printer_count_query = printer_count_query.where(printer_clause)
    printer_count = await db.scalar(printer_count_query)
    filament_count = await db.scalar(select(func.count(Filament.id)))
    project_count = await db.scalar(select(func.count(Project.id)))
    smart_plug_count = await db.scalar(select(func.count(SmartPlug.id)))

    # Archive stats by status
    completed_count = await db.scalar(select(func.count(PrintArchive.id)).where(PrintArchive.status == "completed"))
    failed_count = await db.scalar(select(func.count(PrintArchive.id)).where(PrintArchive.status == "failed"))
    printing_count = await db.scalar(select(func.count(PrintArchive.id)).where(PrintArchive.status == "printing"))

    # System-wide totals aggregate per-run from ``print_log_entries`` so
    # reprints contribute each run and multi-plate sums are pulled from the
    # measured per-run actuals — same source the per-archive stats and the
    # project rollup use (#1593). Pre-fix this summed ``PrintArchive`` directly,
    # which under-reported the same way the project page did (3 reprints of
    # one file showed as one file's worth of filament/time).
    total_print_time = (
        await db.scalar(
            select(func.sum(PrintLogEntry.duration_seconds)).where(PrintLogEntry.duration_seconds.isnot(None))
        )
        or 0
    )

    total_filament = (
        await db.scalar(
            select(func.sum(PrintLogEntry.filament_used_grams)).where(PrintLogEntry.filament_used_grams.isnot(None))
        )
        or 0
    )

    # Connected printers
    connected_printers = []
    for printer_id, client in printer_manager._clients.items():
        if not printer_scope.allows(printer_id):
            continue
        state = client.state
        if state and state.connected:
            # Get printer name and model from database
            result = await db.execute(select(Printer.name, Printer.model).where(Printer.id == printer_id))
            row = result.first()
            name = row[0] if row else f"Printer {printer_id}"
            model = row[1] if row else "unknown"
            connected_printers.append(
                {
                    "id": printer_id,
                    "name": name,
                    "state": state.state,
                    "model": model,
                }
            )

    # Storage info
    archive_dir = settings.archive_dir
    archive_size = get_directory_size(archive_dir) if archive_dir.exists() else 0

    # Database info (engine type, version, size)
    from backend.app.core.db_dialect import is_postgres, is_sqlite

    db_engine_info: dict = {"engine": "unknown", "version": "unknown"}
    db_size = 0
    try:
        if is_postgres():
            from sqlalchemy import text

            result = await db.execute(text("SELECT version()"))
            pg_version_full = result.scalar() or "unknown"
            # e.g. "PostgreSQL 16.2 on x86_64..." → "PostgreSQL 16.2"
            pg_version = " ".join(pg_version_full.split()[:2])
            result = await db.execute(text("SELECT pg_database_size(current_database())"))
            db_size = result.scalar() or 0
            db_engine_info = {
                "engine": "PostgreSQL",
                "version": pg_version,
            }
        elif is_sqlite():
            from sqlalchemy import text

            result = await db.execute(text("SELECT sqlite_version()"))
            sqlite_ver = result.scalar() or "unknown"
            db_path = settings.base_dir / "bambuddy.db"
            db_size = db_path.stat().st_size if db_path.exists() else 0
            db_engine_info = {
                "engine": "SQLite",
                "version": f"SQLite {sqlite_ver}",
            }
    except Exception:
        pass

    # Disk usage
    disk = psutil.disk_usage(str(settings.base_dir))

    # System info
    memory = psutil.virtual_memory()
    # PID 1's create_time is the right uptime anchor in containerised installs
    # (Docker, LXC) — psutil.boot_time() reads /proc/stat:btime which on a
    # shared-kernel container is the host's boot time, not the container's
    # (#1690). On bare metal / VMs PID 1 is the host init, which starts at
    # boot, so the value matches psutil.boot_time() within a sub-second.
    try:
        boot_time = datetime.fromtimestamp(psutil.Process(1).create_time(), tz=timezone.utc)
    except (psutil.Error, OSError):
        boot_time = datetime.fromtimestamp(psutil.boot_time(), tz=timezone.utc)
    uptime_seconds = (datetime.now(timezone.utc) - boot_time).total_seconds()

    # Python and system info
    import sys

    return {
        "app": {
            "version": APP_VERSION,
            "base_dir": str(settings.base_dir),
            "archive_dir": str(archive_dir),
        },
        "database": {
            "engine": db_engine_info["engine"],
            "version": db_engine_info["version"],
            "archives": archive_count,
            "archives_completed": completed_count,
            "archives_failed": failed_count,
            "archives_printing": printing_count,
            "printers": printer_count,
            "filaments": filament_count,
            "projects": project_count,
            "smart_plugs": smart_plug_count,
            "total_print_time_seconds": total_print_time,
            "total_print_time_formatted": format_uptime(total_print_time),
            "total_filament_grams": round(total_filament, 1),
            "total_filament_kg": round(total_filament / 1000, 2),
        },
        "printers": {
            "total": printer_count,
            "connected": len(connected_printers),
            "connected_list": connected_printers,
        },
        "storage": {
            "archive_size_bytes": archive_size,
            "archive_size_formatted": format_bytes(archive_size),
            "database_size_bytes": db_size,
            "database_size_formatted": format_bytes(db_size),
            "disk_total_bytes": disk.total,
            "disk_total_formatted": format_bytes(disk.total),
            "disk_used_bytes": disk.used,
            "disk_used_formatted": format_bytes(disk.used),
            "disk_free_bytes": disk.free,
            "disk_free_formatted": format_bytes(disk.free),
            "disk_percent_used": disk.percent,
        },
        "system": {
            "platform": platform.system(),
            "platform_release": platform.release(),
            "platform_version": platform.version(),
            "architecture": platform.machine(),
            "hostname": platform.node(),
            "python_version": sys.version.split()[0],
            "uptime_seconds": uptime_seconds,
            "uptime_formatted": format_uptime(uptime_seconds),
            "boot_time": boot_time.isoformat(),
        },
        "memory": {
            "total_bytes": memory.total,
            "total_formatted": format_bytes(memory.total),
            "available_bytes": memory.available,
            "available_formatted": format_bytes(memory.available),
            "used_bytes": memory.used,
            "used_formatted": format_bytes(memory.used),
            "percent_used": memory.percent,
        },
        "cpu": {
            "count": psutil.cpu_count(),
            "count_logical": psutil.cpu_count(logical=True),
            "percent": psutil.cpu_percent(interval=0.1),
        },
    }


@router.get("/storage-usage")
async def get_storage_usage(
    refresh: bool = False,
    max_age_seconds: int = STORAGE_USAGE_CACHE_SECONDS,
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SYSTEM_READ),
):
    """Get storage usage breakdown for Bambuddy data directories."""
    max_age_seconds = max(0, min(max_age_seconds, 3600))
    return await _get_storage_usage_cached(refresh=refresh, max_age_seconds=max_age_seconds)


@router.get("/health", response_model=ScanResult)
async def get_system_health(
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SYSTEM_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Scan the recent application log against the known-issue catalog.

    Powers the self-service triage surfaces (System page + bug reporter).
    Sample lines are sanitized before they leave the process.
    """
    sensitive_strings = await collect_sensitive_strings(db)
    result = await asyncio.to_thread(scan_logs, sensitive_strings=sensitive_strings)
    # A log line can be about any printer; a caller limited to some gets the
    # findings without them.
    if not printer_scope.is_unrestricted:
        for finding in result.findings:
            finding.sample = ""
    return result


@router.get("/db-pool")
async def get_db_pool(
    _: User | None = RequirePermissionIfAuthEnabled(Permission.SYSTEM_READ),
):
    """Live database connection-pool gauges for large-farm diagnostics (#2572).

    Reports the resolved pool configuration plus current checked-out /
    checked-in / overflow counts. Deliberately takes no DB session — reading
    the pool's own counters must not itself consume a connection, so this stays
    truthful even when the pool is saturated. On a healthy install ``checked_out``
    sits well below ``config.pool_size + config.max_overflow``; sustained
    saturation points at connections held across slow I/O (see #2572).
    """
    from backend.app.core.database import get_pool_status

    return get_pool_status()


@router.get("/appliance")
async def get_appliance_defaults():
    """Expose appliance-set state for the SPA's bootstrap surface.

    Two file sources, both optional and silently degraded when absent:

    - ``/etc/bambuddy/local.toml`` — hostname / timezone / locale the
      firstboot wizard collected.
    - ``/run/bambuddy/time-synced`` — chrony NTP gate state. The RPi 5 has
      no battery-backed RTC, so on a fresh boot the clock is wrong until
      ntp-gate.sh writes "ok" (or "warning" if 3-minute timeout elapsed).
      A warning state means JWT expiries and TLS validity windows may be
      misaligned; the UI should surface this.

    No auth required — the frontend bootstrap reads this BEFORE auth might
    be set up, and the contents are user-set defaults plus a public sync
    flag (no secrets).
    """
    config = read_local_toml()
    return {
        "hostname": config.get("hostname"),
        "timezone": config.get("timezone"),
        "locale": config.get("locale"),
        "time_synced": read_ntp_gate(),
    }