stackpilot/backend/services/dashboard_service.py

"""Aggregated dashboard data: stack-health funnel + summary widgets.

Everything here is read-only and cheap by construction: one container
*summary* list (no per-container inspect) feeds the whole funnel, image
freshness comes from the cache the update-service background loop already
maintains, and the daily uptime sample is appended lazily on read.
"""
from __future__ import annotations

import asyncio
import json
import os
import time
from datetime import datetime, timedelta, timezone
from typing import Optional

from sqlmodel import Session, select

from config import settings
from docker_client import DockerError, get_client, safe_call
from models.audit import AuditLog
from models.setting import Webhook
from services import compose_service, update_service

COMPOSE_LABEL = compose_service.COMPOSE_LABEL
DOCKER_TIMEOUT = 5.0  # seconds — a slow daemon must not stall the dashboard
FUNNEL_TTL = 30.0

UPTIME_FILE = os.path.join(settings.DATA_DIR, "uptime.jsonl")
UPTIME_DAYS = 30

_funnel_cache: dict = {"data": None, "ts": 0.0}


# --------------------------------------------------------------------------- #
# Container summary (single Docker round-trip)
# --------------------------------------------------------------------------- #


def _list_containers() -> list[dict]:
    client = get_client()
    return safe_call(client.api.containers, all=True)


async def _containers_with_timeout() -> list[dict]:
    return await asyncio.wait_for(asyncio.to_thread(_list_containers), timeout=DOCKER_TIMEOUT)


def _group_by_project(raw: list[dict]) -> dict[str, list[dict]]:
    by_project: dict[str, list[dict]] = {}
    for c in raw:
        project = (c.get("Labels") or {}).get(COMPOSE_LABEL)
        if project:
            by_project.setdefault(project, []).append(c)
    return by_project


def _is_healthy(containers: list[dict]) -> bool:
    """All containers that *have* a healthcheck report healthy.

    The summary ``Status`` string carries the health suffix — "(healthy)",
    "(unhealthy)" or "(health: starting)" — only for containers with a
    healthcheck configured, so its absence simply means "no healthcheck".
    """
    for c in containers:
        status = c.get("Status", "") or ""
        if "(unhealthy)" in status or "(health:" in status:
            return False
    return True


def _is_updated(containers: list[dict], cache: dict[str, dict]) -> bool:
    for c in containers:
        st = cache.get(c.get("Image", ""))
        if st and st.get("update_available"):
            return False
    return True


def _has_notify_target(session: Session) -> bool:
    if settings.NOTIFY_WEBHOOKS:
        return True
    for wh in session.exec(select(Webhook)).all():
        if wh.enabled:
            return True
    return False


async def compute_funnel(session: Session, refresh: bool = False) -> dict:
    now = time.time()
    if not refresh and _funnel_cache["data"] and now - _funnel_cache["ts"] < FUNNEL_TTL:
        return _funnel_cache["data"]

    discovered_ids = compose_service.discover_stacks()
    try:
        raw = await _containers_with_timeout()
    except (DockerError, asyncio.TimeoutError):
        raw = []
    by_project = _group_by_project(raw)
    update_cache = update_service.get_cache()
    notify_configured = _has_notify_target(session)

    running = healthy = updated = monitored = 0
    for stack_id in discovered_ids:
        containers = by_project.get(stack_id, [])
        states = [c.get("State", "") for c in containers]
        if not states or any(s != "running" for s in states):
            continue
        running += 1
        if not _is_healthy(containers):
            continue
        healthy += 1
        if not _is_updated(containers, update_cache):
            continue
        updated += 1
        if notify_configured:
            monitored += 1

    data = {
        "discovered": len(discovered_ids),
        "running": running,
        "healthy": healthy,
        "updated": updated,
        "monitored": monitored,
        "as_of": datetime.now(timezone.utc).isoformat(),
    }
    _funnel_cache["data"] = data
    _funnel_cache["ts"] = now
    return data


# --------------------------------------------------------------------------- #
# Uptime series (one sample per day, JSONL on disk)
# --------------------------------------------------------------------------- #


def _read_uptime() -> list[dict]:
    if not os.path.isfile(UPTIME_FILE):
        return []
    entries = []
    with open(UPTIME_FILE, "r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                entries.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return entries


def _append_uptime(entry: dict) -> None:
    os.makedirs(settings.DATA_DIR, exist_ok=True)
    with open(UPTIME_FILE, "a", encoding="utf-8") as fh:
        fh.write(json.dumps(entry) + "\n")


def _sample_uptime(raw: list[dict]) -> Optional[dict]:
    """Append today's sample if not yet recorded. Uptime% = share of compose
    containers currently running."""
    today = datetime.now(timezone.utc).date().isoformat()
    entries = _read_uptime()
    if any(e.get("date") == today for e in entries):
        return None
    labelled = [c for c in raw if (c.get("Labels") or {}).get(COMPOSE_LABEL)]
    total = len(labelled)
    running = sum(1 for c in labelled if c.get("State") == "running")
    value = round(running / total * 100, 1) if total else 100.0
    entry = {"date": today, "value": value}
    _append_uptime(entry)
    return entry


def uptime_series(raw: list[dict]) -> list[dict]:
    _sample_uptime(raw)
    entries = _read_uptime()
    by_date = {e["date"]: e for e in entries if "date" in e}
    series = []
    today = datetime.now(timezone.utc).date()
    last_value: Optional[float] = None
    for i in range(UPTIME_DAYS - 1, -1, -1):
        day = (today - timedelta(days=i)).isoformat()
        e = by_date.get(day)
        if e is not None:
            last_value = e.get("value")
        # Days before monitoring started (or gaps) reuse the last known value
        # so the chart doesn't show artificial dips.
        series.append({"date": day, "value": last_value})
    return series


# --------------------------------------------------------------------------- #
# Ops (audit-log) activity
# --------------------------------------------------------------------------- #

_WEEKDAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]


def ops_activity(session: Session) -> tuple[list[dict], Optional[str]]:
    today = datetime.now(timezone.utc).date()
    cutoff = datetime.combine(today - timedelta(days=UPTIME_DAYS - 1), datetime.min.time(), timezone.utc)
    timestamps = session.exec(
        select(AuditLog.timestamp).where(AuditLog.timestamp >= cutoff)
    ).all()

    per_day: dict[str, int] = {}
    per_weekday = [0] * 7
    for ts in timestamps:
        per_day[ts.date().isoformat()] = per_day.get(ts.date().isoformat(), 0) + 1
        per_weekday[ts.weekday()] += 1

    series = []
    for i in range(UPTIME_DAYS - 1, -1, -1):
        day = (today - timedelta(days=i)).isoformat()
        series.append({"date": day, "count": per_day.get(day, 0)})

    peak = _WEEKDAYS[per_weekday.index(max(per_weekday))] if any(per_weekday) else None
    return series, peak


async def compute_summary(session: Session) -> dict:
    try:
        raw = await _containers_with_timeout()
    except (DockerError, asyncio.TimeoutError):
        raw = []
    labelled = [c for c in raw if (c.get("Labels") or {}).get(COMPOSE_LABEL)]
    ops_series, peak = ops_activity(session)
    return {
        "total_containers": sum(1 for c in labelled if c.get("State") == "running"),
        "containers_total": len(labelled),
        "uptime_series": uptime_series(raw),
        "ops_last_30d": ops_series,
        "ops_peak_day": peak,
        "as_of": datetime.now(timezone.utc).isoformat(),
    }