ctfy.sdk.admin_resources.observability

  1"""``client.admin.observability`` — dashboard stats, health, audit log (admin)."""
  2
  3from __future__ import annotations
  4
  5from typing import Any
  6
  7from ctfy.sdk._helpers import _extract_items, _raise_for_status
  8from ctfy.sdk.base import BaseHttpClient
  9from ctfy.server.models import (
 10    Activity,
 11    ActivityTimeseries,
 12    AdminHealthFlags,
 13    AdminOverview,
 14    AdminSolveMatrix,
 15    AdminTimeseries,
 16    AdminTrafficSummary,
 17)
 18
 19
 20class AdminObservabilityResource:
 21    """The admin dashboard's numbers: overview snapshot, solve grid, health
 22    flags, metric timeseries, traffic tally, and the unfiltered audit log."""
 23
 24    def __init__(self, http: BaseHttpClient) -> None:
 25        self._http = http
 26
 27    def overview(self) -> AdminOverview:
 28        """Dashboard snapshot: counts of users / teams / competitions /
 29        instances / solves, plus the latest 24h of activity."""
 30        resp = self._http.request("GET", "/admin/overview")
 31        _raise_for_status(resp)
 32        return AdminOverview.model_validate(resp.json())
 33
 34    def solve_matrix(self) -> AdminSolveMatrix:
 35        """Per-team × per-challenge solve grid for the admin scoreboard."""
 36        resp = self._http.request("GET", "/admin/solve-matrix")
 37        _raise_for_status(resp)
 38        return AdminSolveMatrix.model_validate(resp.json())
 39
 40    def health_flags(self) -> AdminHealthFlags:
 41        """Boolean flags surfaced on the admin top-bar (e.g. node down,
 42        OAuth not configured, secret-key ephemeral)."""
 43        resp = self._http.request("GET", "/admin/health-flags")
 44        _raise_for_status(resp)
 45        return AdminHealthFlags.model_validate(resp.json())
 46
 47    def timeseries(
 48        self,
 49        *,
 50        metric: str,
 51        window: int = 86400,
 52        bucket: int = 3600,
 53    ) -> AdminTimeseries:
 54        """Bucketed counts for one metric. ``metric`` ∈ ``"launches"``,
 55        ``"submissions"``, ``"solves"``, ``"running_instances"``,
 56        ``"teams"``. ``window`` and ``bucket`` are in seconds."""
 57        resp = self._http.request(
 58            "GET",
 59            "/admin/timeseries",
 60            params={"metric": metric, "window": window, "bucket": bucket},
 61        )
 62        _raise_for_status(resp)
 63        return AdminTimeseries.model_validate(resp.json())
 64
 65    def traffic_summary(self) -> AdminTrafficSummary:
 66        """Per-team and per-instance HTTP request counts (from archived
 67        instance records' mitmproxy flow tally)."""
 68        resp = self._http.request("GET", "/admin/traffic/summary")
 69        _raise_for_status(resp)
 70        return AdminTrafficSummary.model_validate(resp.json())
 71
 72    def activities(
 73        self,
 74        *,
 75        actor_id: str = "",
 76        team_id: str = "",
 77        challenge_id: str = "",
 78        event: str = "",
 79        since: str = "",
 80        until: str = "",
 81        offset: int = 0,
 82        limit: int = 50,
 83    ) -> list[Activity]:
 84        """Admin activity log — the unfiltered audit surface (every
 85        actor, including admin-only events). ``since`` / ``until`` are
 86        ISO-8601 timestamps; omit to leave unbounded."""
 87        params: dict[str, Any] = {"offset": offset, "limit": limit}
 88        params.update(
 89            {
 90                k: v
 91                for k, v in (
 92                    ("actor_id", actor_id),
 93                    ("team_id", team_id),
 94                    ("challenge_id", challenge_id),
 95                    ("event", event),
 96                    ("since", since),
 97                    ("until", until),
 98                )
 99                if v
100            }
101        )
102        resp = self._http.request("GET", "/admin/activities", params=params)
103        _raise_for_status(resp)
104        return _extract_items(resp.json(), Activity)
105
106    def activities_timeseries(
107        self, *, window: int = 86400, bucket: int = 3600
108    ) -> ActivityTimeseries:
109        """Bucketed admin activity-event counts. ``window`` + ``bucket``
110        are in seconds."""
111        resp = self._http.request(
112            "GET",
113            "/admin/activities/timeseries",
114            params={"window": window, "bucket": bucket},
115        )
116        _raise_for_status(resp)
117        return ActivityTimeseries.model_validate(resp.json())