admin_api.py

from __future__ import annotations

import hmac
import os
import time
from datetime import timezone
from typing import Any, Callable

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse

from core import PLANS, STORE, human_bytes, parse_dt
from features_v23 import plan_features, update_plan_features, send_admin_notification
from resource_monitor import RESOURCE_MONITOR

ADMIN_ID = "cloudvault-admin"

def _configured_key() -> str:
    return os.getenv("ADMIN_API_KEY", "").strip()


def _authorize(request: Request) -> None:
    expected = _configured_key()
    if not expected:
        raise HTTPException(status_code=503, detail="ADMIN_API_KEY is not configured.")

    provided = request.headers.get("X-CloudVault-Admin-Key", "").strip()
    if not provided:
        auth = request.headers.get("Authorization", "")
        if auth.lower().startswith("bearer "):
            provided = auth[7:].strip()

    if not provided or not hmac.compare_digest(provided, expected):
        raise HTTPException(status_code=401, detail="Invalid admin API key.")


def _user_status(user: dict[str, Any]) -> str:
    if user.get("is_banned"):
        until = parse_dt(user.get("ban_until"))
        if until:
            return f"Banned until {until.astimezone(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}"
        return "Banned permanently"
    if user.get("delete_on_expiry"):
        return "Deletes on package expiry"
    return "Active"


def _system_metrics() -> dict[str, Any]:
    """Return real CloudVault container usage without percentage conversion."""
    snapshot = RESOURCE_MONITOR.snapshot()

    try:
        import psutil  # type: ignore

        process = psutil.Process(os.getpid())
        process_threads = int(process.num_threads())
        uptime_seconds = max(0, int(time.time() - process.create_time()))
    except Exception:
        process_threads = 0
        uptime_seconds = 0

    return {
        # Average CPU cores used during the latest sample. Example: 0.26 vCPU.
        "cpu_used_vcpu": snapshot.cpu_used_vcpu,
        # Current memory used by the whole CloudVault Linux container.
        "ram_used_bytes": snapshot.ram_used_bytes,
        "ram_used_gb": snapshot.ram_used_gb,
        "effective_vcpu": snapshot.effective_vcpu,
        "logical_cpu_visible": snapshot.logical_cpu_visible,
        "sample_seconds": snapshot.sample_seconds,
        "metrics_updated_at": snapshot.updated_at_unix,
        "cpu_source": snapshot.cpu_source,
        "ram_source": snapshot.ram_source,
        "process_threads": process_threads,
        "uptime_seconds": uptime_seconds,
        "metrics_scope": "cloudvault_container",
    }


def register_admin_api(demo: Any, active_users_provider: Callable[[], int] | None = None) -> None:
    """Register secure CloudVault desktop-admin endpoints on a Gradio app."""

    def active_count() -> int:
        try:
            return max(0, int(active_users_provider() if active_users_provider else 0))
        except Exception:
            return 0

    async def overview(request: Request) -> JSONResponse:
        _authorize(request)

        overview_data = STORE.admin_overview(ADMIN_ID)
        users = []
        for user in STORE.list_users(ADMIN_ID):
            plan_key = str(user.get("plan", "free"))
            used = STORE.used_bytes(user["id"])
            expires = parse_dt(user.get("plan_expires_at"))
            users.append({
                "id": user["id"],
                "username": user.get("username", "User"),
                "plan": plan_key,
                "plan_name": PLANS.get(plan_key, PLANS["free"]).name,
                "used_bytes": used,
                "used_text": human_bytes(used),
                "status": _user_status(user),
                "is_banned": bool(user.get("is_banned")),
                "ban_until": user.get("ban_until"),
                "expires_at": expires.isoformat() if expires else None,
                "created_at": user.get("created_at"),
            })

        room_state = STORE.chat_room_state(ADMIN_ID)
        response = {
            "ok": True,
            "timestamp": time.time(),
            "active_users": active_count(),
            "system": _system_metrics(),
            "storage": {
                "limit_bytes": overview_data["limit"],
                "used_bytes": overview_data["total_used"],
                "remaining_bytes": overview_data["remaining"],
                "reserved_bytes": overview_data["reserved"],
                "used_by_plan": overview_data["used_by_plan"],
            },
            "accounts": {
                "registered": overview_data["registered"],
                "counts": overview_data["counts"],
                "users": users,
            },
            "config": overview_data["config"],
            "plan_features": {k: plan_features(k) for k in ("free","plus","pro")},
            "support": {"runtime": STORE.support_runtime(), "tickets": STORE.admin_support_tickets(ADMIN_ID)},
            "chat": {
                "messages": overview_data.get("chat_messages", 0),
                "mode": room_state.get("mode", "single"),
                "capacity": room_state.get("capacity", 200),
                "rooms": room_state.get("rooms", []),
            },
        }
        return JSONResponse(response, headers={"Cache-Control": "no-store"})


    async def chat(request: Request) -> JSONResponse:
        _authorize(request)
        room_id = str(request.query_params.get("room_id", "global-1")).strip() or "global-1"
        messages = STORE.chat_messages(ADMIN_ID, room_id)
        output = []
        for message in messages:
            created = parse_dt(message.get("created_at"))
            output.append({
                "id": message.get("id"),
                "room_id": message.get("room_id", "global-1"),
                "sender_name": message.get("sender_name", "User"),
                "sender_plan": message.get("sender_plan", "plus"),
                "sender_label": "👑 admin" if message.get("sender_plan") == "admin" else str(message.get("sender_name", "Kullanıcı")),
                "text": message.get("text", ""),
                "created_at": message.get("created_at"),
                "time_text": created.astimezone(timezone.utc).strftime("%d.%m.%Y %H:%M UTC") if created else "",
            })
        return JSONResponse(
            {"ok": True, "room_id": room_id, "messages": output},
            headers={"Cache-Control": "no-store"},
        )

    async def support(request: Request) -> JSONResponse:
        _authorize(request)
        ticket_id = str(request.query_params.get("ticket_id", "")).strip()
        if ticket_id:
            ticket = STORE.get_support_ticket(ADMIN_ID, ticket_id)
            return JSONResponse({"ok": bool(ticket), "ticket": ticket}, headers={"Cache-Control": "no-store"})
        return JSONResponse({"ok": True, "tickets": STORE.admin_support_tickets(ADMIN_ID), "runtime": STORE.support_runtime()}, headers={"Cache-Control": "no-store"})

    async def action(request: Request) -> JSONResponse:
        _authorize(request)
        payload = await request.json()
        name = str(payload.get("action", "")).strip()
        result: Any = None

        if name == "ban_permanent":
            STORE.ban_user(ADMIN_ID, str(payload.get("user_id", "")), None)
        elif name == "ban_temporary":
            STORE.ban_user(ADMIN_ID, str(payload.get("user_id", "")), float(payload.get("hours", 24)))
        elif name == "unban":
            STORE.unban_user(ADMIN_ID, str(payload.get("user_id", "")))
        elif name == "delete_user":
            STORE.admin_delete_user(ADMIN_ID, str(payload.get("user_id", "")))
        elif name == "delete_all_free":
            result = STORE.delete_all_free(ADMIN_ID)
        elif name == "delete_all_users":
            result = STORE.delete_all_users(ADMIN_ID)
        elif name == "toggle_system":
            cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, system_enabled=not cfg.get("system_enabled", True))
        elif name == "toggle_chat":
            cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, chat_enabled=not cfg.get("chat_enabled", True))
        elif name == "toggle_plus_chat":
            cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, plus_chat_enabled=not cfg.get("plus_chat_enabled", True))
        elif name == "toggle_pro_chat":
            cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, pro_chat_enabled=not cfg.get("pro_chat_enabled", True))
        elif name == "toggle_purchases":
            cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, purchases_enabled=not cfg.get("purchases_enabled", True))
        elif name == "toggle_registrations":
            cfg = STORE.config(); result = STORE.set_config(ADMIN_ID, registrations_enabled=not cfg.get("registrations_enabled", True))
        elif name == "clear_all_chat":
            STORE.clear_chat(ADMIN_ID)
        elif name == "begin_shutdown":
            result = STORE.begin_shutdown(ADMIN_ID)
        elif name == "cancel_shutdown":
            STORE.cancel_shutdown(ADMIN_ID)
        elif name == "configure_rooms":
            result = STORE.set_chat_configuration(
                ADMIN_ID,
                str(payload.get("mode", "single")),
                int(payload.get("capacity", 200)),
            )
        elif name == "toggle_room":
            result = STORE.toggle_chat_room(ADMIN_ID, str(payload.get("room_id", "")))
        elif name == "clear_room":
            result = STORE.clear_chat_room(ADMIN_ID, str(payload.get("room_id", "")))
        elif name == "send_admin_chat":
            text = str(payload.get("text", "")).strip()
            room_id = str(payload.get("room_id", "global-1")).strip() or "global-1"
            result = STORE.send_chat(ADMIN_ID, text, room_id)
        elif name == "support_reply":
            result = STORE.reply_support_ticket(
                ADMIN_ID,
                str(payload.get("ticket_id", "")),
                str(payload.get("reply", "")),
            )
        elif name == "support_toggle":
            cfg = STORE.config()
            result = STORE.set_config(
                ADMIN_ID,
                support_enabled=not cfg.get("support_enabled", True),
            )
        elif name == "support_force_on":
            result = STORE.set_config(
                ADMIN_ID,
                support_enabled=True,
                support_force_open=True,
            )
        elif name == "support_force_off":
            result = STORE.set_config(ADMIN_ID, support_force_open=False)
        elif name == "send_notification":
            result = send_admin_notification(ADMIN_ID, str(payload.get("title", "")), str(payload.get("message", "")), list(payload.get("audience", [])))
        elif name == "update_plan_features":
            result = update_plan_features(ADMIN_ID, str(payload.get("plan", "")), dict(payload.get("features", {})))
        else:
            raise HTTPException(status_code=400, detail="Unknown admin action.")

        return JSONResponse({"ok": True, "result": result}, headers={"Cache-Control": "no-store"})

    existing = {getattr(route, "path", "") for route in demo.app.routes}
    overview_path = "/gradio_api/cloudvault-admin/overview"
    action_path = "/gradio_api/cloudvault-admin/action"
    support_path = "/gradio_api/cloudvault-admin/support"
    chat_path = "/gradio_api/cloudvault-admin/chat"

    if overview_path not in existing:
        demo.app.add_api_route(overview_path, overview, methods=["GET"], include_in_schema=False)
    if action_path not in existing:
        demo.app.add_api_route(action_path, action, methods=["POST"], include_in_schema=False)
    if support_path not in existing:
        demo.app.add_api_route(support_path, support, methods=["GET"], include_in_schema=False)
    if chat_path not in existing:
        demo.app.add_api_route(chat_path, chat, methods=["GET"], include_in_schema=False)