rmm-backend/central_api_prototype.py

import os
from dotenv import load_dotenv

# Dynamically select configuration file based on APP_ENV
app_env = os.getenv("APP_ENV", "development")
if app_env == "production":
    load_dotenv(".env.production")
else:
    load_dotenv(".env.development")

# Fallback to load default .env if any variables are not yet defined
load_dotenv()

from fastapi import FastAPI, HTTPException, Depends, Header, Request, BackgroundTasks
from fastapi.responses import PlainTextResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from pydantic import BaseModel
from typing import Dict, Any, List, Optional
import uvicorn
import base64
import hmac
import hashlib
import json
from datetime import datetime, timedelta

app = FastAPI(title="RMM Central API")
# Dashboard polling was ~90% of this server's 428 GB/month egress; JSON
# compresses ~15x. Agents' urllib doesn't send Accept-Encoding, so agent
# traffic (incl. speedtest blobs) is unaffected.
app.add_middleware(GZipMiddleware, minimum_size=1024)

# Enable CORS for frontend dashboard queries (React dev server runs on a separate port)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# JWT-like HMAC Secure Token Utilities
JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY", "seekright_rmm_secret_key_2026_default")

# Per-heartbeat telemetry dumps flooded journald+syslog and helped fill the disk
# (2026-07-23: / hit 100% and mongod fatally aborted). Now opt-in via env.
VERBOSE_TELEMETRY = os.getenv("VERBOSE_TELEMETRY", "false").lower() in ("1", "true", "yes")

def generate_token(username: str) -> str:
    # Expire in 1 day (24 hours)
    expiry = (datetime.utcnow() + timedelta(days=1)).isoformat()
    payload = {
        "username": username,
        "expires": expiry
    }
    payload_json = json.dumps(payload)
    payload_b64 = base64.urlsafe_b64encode(payload_json.encode('utf-8')).decode('utf-8').rstrip('=')
    
    sig = hmac.new(JWT_SECRET_KEY.encode('utf-8'), payload_b64.encode('utf-8'), hashlib.sha256).hexdigest()
    return f"{payload_b64}.{sig}"

def verify_token(token: str) -> bool:
    try:
        parts = token.split(".")
        if len(parts) != 2:
            return False
        payload_b64, sig = parts
        
        # Verify signature
        expected_sig = hmac.new(JWT_SECRET_KEY.encode('utf-8'), payload_b64.encode('utf-8'), hashlib.sha256).hexdigest()
        if not hmac.compare_digest(sig, expected_sig):
            return False
            
        # Decode and check expiry
        padding = 4 - (len(payload_b64) % 4)
        if padding < 4:
            payload_b64 += "=" * padding
        payload_json = base64.urlsafe_b64decode(payload_b64.encode('utf-8')).decode('utf-8')
        payload = json.loads(payload_json)
        
        expires_dt = datetime.fromisoformat(payload["expires"])
        if datetime.utcnow() > expires_dt:
            return False # Expired
            
        return payload["username"] == "root"
    except Exception:
        return False

# FastAPI dependency to secure UI endpoints
async def get_current_user(authorization: str = Header(None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing or invalid authentication credentials")
    token = authorization.split(" ")[1]
    if not verify_token(token):
        raise HTTPException(status_code=401, detail="Authentication token is invalid or has expired")
    return "root"

import os
import json
from datetime import datetime

# In-memory storage for client hardware telemetry
client_telemetry = {}

from pymongo import MongoClient

# Database Connection: Configured via environment variable with local fallback
MONGODB_URI = os.getenv("MONGODB_URI", "mongodb://localhost:27017")
DB_NAME = os.getenv("MONGODB_DB", "rmm_db")

mongo_client = MongoClient(MONGODB_URI)
db = mongo_client[DB_NAME]

clients_collection = db["clients"]
logs_collection = db["logs"]
config_collection = db["config"]

# Rolling vitals history: one slim sample per client per minute, expired
# automatically by MongoDB after 7 days (~1.1 MB/week/system, ~35 MB fleet).
history_collection = db["telemetry_history"]
HISTORY_RETENTION_DAYS = 7
try:
    history_collection.create_index("ts", expireAfterSeconds=HISTORY_RETENTION_DAYS * 24 * 3600)
    history_collection.create_index([("client_id", 1), ("ts", 1)])
except Exception as e:
    print(f"[!] Could not create telemetry_history indexes: {e}")

# --- Agent authentication ---
# Shared secret agents present in X-Agent-Token. Auth mode ("grace"|"strict") is
# stored in the config collection so it can be flipped live from the dashboard:
#   grace  = accept authenticated and legacy (tokenless) agents; used during rollout
#   strict = reject any request without a valid token
AGENT_TOKEN = os.getenv("AGENT_TOKEN", "")

def get_agent_auth_mode() -> str:
    doc = config_collection.find_one({"_id": "agent_auth"})
    return (doc or {}).get("mode", "grace")

# During a token rotation both the new and the outgoing token must validate,
# otherwise agents can't authenticate the very download that gives them the new
# one. Set AGENT_TOKEN_PREVIOUS in .env while rotating, then clear it once the
# whole fleet reports the new token.
AGENT_TOKEN_PREVIOUS = os.getenv("AGENT_TOKEN_PREVIOUS", "")

def is_valid_agent_token(token: Optional[str]) -> bool:
    if token is None:
        return False
    for accepted in (AGENT_TOKEN, AGENT_TOKEN_PREVIOUS):
        if accepted and hmac.compare_digest(token, accepted):
            return True
    return False

async def require_agent_token(x_agent_token: str = Header(None)):
    """Dependency guarding every agent-facing endpoint. Enforces only in strict mode;
    in grace mode it lets legacy tokenless agents through so a fleet can migrate."""
    if get_agent_auth_mode() == "strict" and not is_valid_agent_token(x_agent_token):
        raise HTTPException(status_code=401, detail="Invalid or missing agent token")
    return is_valid_agent_token(x_agent_token)

MAX_ENTRIES_PER_CLIENT = 100  # Keep the latest 100 historical logs per client node

ROCKETCHAT_WEBHOOK_URL = os.getenv("ROCKETCHAT_WEBHOOK_URL", "")

# Alert states are now stored directly in the MongoDB client documents 
# to support multi-worker environments safely.

def send_rocketchat_notification(text: str, color: str = "#808080"):
    """
    Dispatches a formatted notification payload to the Rocket.Chat incoming webhook.
    """
    if not ROCKETCHAT_WEBHOOK_URL:
        try:
            print(f"[Rocket.Chat Simulation] {text}")
        except UnicodeEncodeError:
            # Fallback for Windows consoles that do not support printing unicode emojis
            sanitized_text = text.encode('ascii', errors='backslashreplace').decode('ascii')
            print(f"[Rocket.Chat Simulation] {sanitized_text}")
        return

    payload = {
        "text": text,
        "attachments": [
            {
                "color": color,
                "ts": datetime.now().isoformat()
            }
        ]
    }
    
    try:
        import urllib.request
        import json
        req = urllib.request.Request(
            ROCKETCHAT_WEBHOOK_URL,
            data=json.dumps(payload).encode('utf-8'),
            headers={'Content-Type': 'application/json'}
        )
        with urllib.request.urlopen(req, timeout=3.0) as response:
            pass
    except Exception as e:
        print(f"[!] Error sending Rocket.Chat notification: {e}")

def append_to_logs(log_type: str, client_id: str, detail: Any):
    """
    Appends a new event log grouped by client_id in MongoDB logs collection.
    Ensures fair-share log limits per system so noisy clients never overwrite others.
    """
    try:
        entry = {
            "client_id": client_id,
            "timestamp": datetime.now().isoformat(),
            "type": log_type,
            "detail": detail
        }
        logs_collection.insert_one(entry)

        # Enforce fair-share logging limits per system (FIFO circular buffer)
        count = logs_collection.count_documents({"client_id": client_id})
        if count > MAX_ENTRIES_PER_CLIENT:
            oldest_docs = logs_collection.find(
                {"client_id": client_id},
                {"_id": 1}
            ).sort("timestamp", 1).limit(count - MAX_ENTRIES_PER_CLIENT)
            
            ids_to_delete = [doc["_id"] for doc in oldest_docs]
            if ids_to_delete:
                logs_collection.delete_many({"_id": {"$in": ids_to_delete}})
    except Exception as e:
        print(f"[!] Error writing log to MongoDB: {e}")

def record_history_sample(client_id: str, payload):
    """
    Store one slim vitals sample per client per minute. The _id is the
    client+minute bucket so concurrent uvicorn workers can't duplicate a
    sample, and $setOnInsert keeps the first write of each minute.
    """
    try:
        bucket = datetime.now().replace(second=0, microsecond=0)
        gpu_temp = None
        for g in (payload.gpus or []):
            raw = str(g.get("temp", "")).strip().rstrip("C").strip()
            try:
                gpu_temp = float(raw)
                break
            except ValueError:
                continue
        history_collection.update_one(
            {"_id": f"{client_id}|{bucket.isoformat()}"},
            {"$setOnInsert": {
                "client_id": client_id,
                "ts": bucket,
                "cpu": payload.cpu_percent,
                "cpu_temp": payload.cpu_temp,
                "ram": payload.memory_percent,
                "gpu_temp": gpu_temp,
            }},
            upsert=True,
        )
    except Exception as e:
        print(f"[!] History sample failed for {client_id}: {e}")

def mark_client_active(client_id: str):
    try:
        now_str = datetime.now().isoformat()
        # Atomically check if transitioned from offline to online (active is not True)
        old_doc = clients_collection.find_one_and_update(
            {"_id": client_id, "active": {"$ne": True}},
            {"$set": {"active": True, "last_seen": now_str, "last_down_alert_time": None}},
            return_document=False
        )
        if old_doc:
            # Client transitioned from inactive/None to active!
            # Send UP recovery alert only if they had a down alert time logged
            if old_doc.get("last_down_alert_time"):
                send_rocketchat_notification(
                    text=f"✅ **System UP:** Client `{client_id}` has recovered and is back online.",
                    color="#2ecc71"
                )
        else:
            # Client is already active, just update their last_seen timestamp
            clients_collection.update_one(
                {"_id": client_id},
                {"$set": {"last_seen": now_str}}
            )
    except Exception as e:
        print(f"[!] Error marking client active: {e}")

def load_clients() -> Dict[str, Any]:
    try:
        # Clean up any ghost clients with empty or whitespace-only keys
        clients_collection.delete_many({"_id": {"$in": ["", None]}})
        clients_collection.delete_many({"_id": {"$regex": "^\\s*$"}})

        # Load all documents
        cursor = clients_collection.find()
        data = {}
        for doc in cursor:
            cid = doc["_id"]
            
            # Compute active status dynamically for reads, but do NOT write back to database or alert
            last_seen_str = doc.get("last_seen")
            is_active = doc.get("active", False)
            if last_seen_str:
                try:
                    last_seen_dt = datetime.fromisoformat(last_seen_str)
                    if (datetime.now() - last_seen_dt).total_seconds() >= 30:
                        is_active = False
                except Exception:
                    pass
            
            info = {k: v for k, v in doc.items() if k != "_id"}
            info["active"] = is_active
            data[cid] = info
            
        return data
    except Exception as e:
        print(f"[!] Error loading clients from MongoDB: {e}")
        return {}

import asyncio
import shutil

CPU_TEMP_ALERT_C = float(os.getenv("CPU_TEMP_ALERT_C", "85"))
CPU_TEMP_CLEAR_C = float(os.getenv("CPU_TEMP_CLEAR_C", "75"))
SERVER_DISK_ALERT_PERCENT = 85.0
_server_disk_state = {"alerted_at": None}

def check_server_disk():
    # The RMM watches every site's disks but died on its own full disk
    # (2026-07-23: / hit 100%, mongod aborted). Watch our own disk too.
    try:
        usage = shutil.disk_usage("/")
        pct = round(usage.used / usage.total * 100, 1)
        free_gb = round(usage.free / (1024 ** 3), 1)
        now = datetime.now()
        if pct >= SERVER_DISK_ALERT_PERCENT:
            last = _server_disk_state["alerted_at"]
            if last is None or (now - last).total_seconds() >= 3600:
                _server_disk_state["alerted_at"] = now
                send_rocketchat_notification(
                    text=f"🚨 **RMM SERVER DISK:** {pct}% used, only {free_gb} GB free on the central server. At 100% MongoDB dies — clean up now.",
                    color="#e74c3c",
                )
        elif _server_disk_state["alerted_at"] is not None:
            _server_disk_state["alerted_at"] = None
            send_rocketchat_notification(
                text=f"✅ **RMM SERVER DISK recovered:** {pct}% used, {free_gb} GB free.",
                color="#2ecc71",
            )
    except Exception as e:
        print(f"[!] Server disk self-check failed: {e}")

async def monitor_heartbeats_loop():
    while True:
        check_server_disk()
        try:
            now = datetime.now()
            timeout_time = (now - timedelta(seconds=30)).isoformat()
            
            # Find clients that are active in DB but haven't checked in for 30s
            cursor = clients_collection.find({
                "active": True,
                "last_seen": {"$lt": timeout_time}
            })
            for doc in cursor:
                cid = doc["_id"]
                now_str = now.isoformat()
                
                # Atomically update to active = False
                # Ensures only one request/worker triggers the state transition and sends the alert
                old_doc = clients_collection.find_one_and_update(
                    {"_id": cid, "active": True, "last_seen": doc["last_seen"]},
                    {"$set": {"active": False, "last_down_alert_time": now_str}},
                    return_document=False
                )
                if old_doc:
                    send_rocketchat_notification(
                        text=f"🚨 **System DOWN:** Client `{cid}` has missed heartbeats for over 30 seconds.",
                        color="#e74c3c"
                    )
            
            # Hourly reminders for STILL DOWN clients
            reminder_time = (now - timedelta(hours=1)).isoformat()
            cursor_still_down = clients_collection.find({
                "active": False,
                "last_down_alert_time": {"$lt": reminder_time}
            })
            for doc in cursor_still_down:
                cid = doc["_id"]
                old_last_down = doc["last_down_alert_time"]
                now_str = now.isoformat()
                
                # Atomically update reminder timestamp
                old_doc = clients_collection.find_one_and_update(
                    {"_id": cid, "active": False, "last_down_alert_time": old_last_down},
                    {"$set": {"last_down_alert_time": now_str}},
                    return_document=False
                )
                if old_doc:
                    send_rocketchat_notification(
                        text=f"🚨 **System STILL DOWN:** Client `{cid}` remains offline (reminder sent every hour).",
                        color="#e74c3c"
                    )
        except Exception as e:
            print(f"[!] Error in heartbeat monitoring loop: {e}")
            
        await asyncio.sleep(10)

@app.on_event("startup")
async def startup_event():
    asyncio.create_task(monitor_heartbeats_loop())
    print("[*] Heartbeat monitoring background task initialized.")

class TelemetryPayload(BaseModel):
    cpu_percent: float
    memory_percent: float
    memory_total_gb: float
    memory_free_gb: float
    disks: List[Dict[str, Any]]
    gpus: List[Dict[str, Any]]
    cpu_temp: Optional[float] = None
    local_ips: Optional[List[str]] = None
    agent_version: Optional[str] = None
    speed_upload_mbps: Optional[float] = None
    speed_download_mbps: Optional[float] = None

class CommandResultPayload(BaseModel):
    command: str
    returncode: int
    stdout: str
    stderr: str

class SearchResultsPayload(BaseModel):
    query: str
    results: List[Dict[str, Any]]
    searched_path: Optional[str] = None
    error: Optional[str] = None

class LoginPayload(BaseModel):
    username: str
    password: str

DASHBOARD_USERNAME = os.getenv("DASHBOARD_USERNAME", "root")
DASHBOARD_PASSWORD = os.getenv("DASHBOARD_PASSWORD", "seekright159@")

@app.post("/api/login")
async def login(payload: LoginPayload):
    user_ok = hmac.compare_digest(payload.username, DASHBOARD_USERNAME)
    pass_ok = hmac.compare_digest(payload.password, DASHBOARD_PASSWORD)
    if user_ok and pass_ok:
        token = generate_token("root")
        return {"token": token}
    raise HTTPException(status_code=401, detail="Invalid username or password")

class AuthModePayload(BaseModel):
    mode: str

@app.get("/api/agent-auth-mode")
async def read_agent_auth_mode(current_user: str = Depends(get_current_user)):
    return {"mode": get_agent_auth_mode()}

@app.post("/api/set-agent-auth-mode")
async def set_agent_auth_mode(payload: AuthModePayload, current_user: str = Depends(get_current_user)):
    if payload.mode not in ("grace", "strict"):
        raise HTTPException(status_code=400, detail="mode must be 'grace' or 'strict'")
    config_collection.update_one({"_id": "agent_auth"}, {"$set": {"mode": payload.mode}}, upsert=True)
    return {"status": "success", "mode": payload.mode}

@app.get("/api/agent-versions")
async def agent_versions(current_user: str = Depends(get_current_user)):
    """Migration dashboard: which agent version each node reports and whether its token is valid."""
    out = {}
    for doc in clients_collection.find({}, {"agent_version": 1, "agent_token_ok": 1}):
        out[doc["_id"]] = {
            "agent_version": doc.get("agent_version", "unknown"),
            "agent_token_ok": doc.get("agent_token_ok", False),
        }
    return out

@app.get("/api/get-commands")
async def get_whitelisted_commands(platform: str, token_valid: bool = Depends(require_agent_token)):
    """
    Endpoint for remote agents to fetch their OS-specific whitelisted commands.
    """
    commands_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "commands.json")
    try:
        with open(commands_file, "r") as f:
            data = json.load(f)
            return data.get(platform, {})
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to load commands: {e}")

@app.get("/api/get-command")
async def get_command(client_id: str, token_valid: bool = Depends(require_agent_token)):
    """
    Endpoint for clients to poll for pending commands.
    Client Agent hits this endpoint to ask: "Do I have any work to do?"
    """
    if not client_id or not client_id.strip():
        raise HTTPException(status_code=400, detail="client_id cannot be empty")
        
    now_str = datetime.now().isoformat()
    # Dynamic fleet auto-registration
    clients_collection.update_one(
        {"_id": client_id},
        {"$setOnInsert": {"pending_command": "none", "active": True, "last_seen": now_str}},
        upsert=True
    )
    
    # Get and reset command atomically
    updated_doc = clients_collection.find_one_and_update(
        {"_id": client_id},
        {"$set": {"pending_command": "none"}},
        return_document=False # Returns the state before update
    )
    
    cmd = "none"
    if updated_doc:
        cmd = updated_doc.get("pending_command", "none")
        
    if cmd != "none":
        append_to_logs("command_polled", client_id, {"command": cmd})
        
    mark_client_active(client_id)
    return {"command": cmd}

@app.post("/api/schedule-command")
async def schedule_command(client_id: str, command: str, current_user: str = Depends(get_current_user)):
    """
    Endpoint for your Dashboard/UI to schedule a new command.
    Supports a comma-separated list of client IDs for batch fleet updates.
    """
    target_ids = [cid.strip() for cid in client_id.split(",") if cid.strip()]
    if not target_ids:
        raise HTTPException(status_code=400, detail="No target client IDs specified")

    for cid in target_ids:
        clients_collection.update_one(
            {"_id": cid},
            {"$set": {"pending_command": command}},
            upsert=True
        )
        append_to_logs("command_scheduled", cid, {"command": command})
        
    return {"status": "success", "message": f"Command '{command}' scheduled for {', '.join(target_ids)}"}

@app.post("/api/telemetry")
async def receive_telemetry(client_id: str, payload: TelemetryPayload, token_valid: bool = Depends(require_agent_token)):
    """
    Endpoint for agents to push their live hardware telemetry.
    """
    if not client_id or not client_id.strip():
        raise HTTPException(status_code=400, detail="client_id cannot be empty")
        
    now_str = datetime.now().isoformat()
    # Auto-register client if we see it through telemetry first
    clients_collection.update_one(
        {"_id": client_id},
        {"$setOnInsert": {"pending_command": "none", "active": True, "last_seen": now_str}},
        upsert=True
    )
    
    mark_client_active(client_id)
    record_history_sample(client_id, payload)

    # Get current alert states from database to prevent duplicate alerts
    client_doc = clients_collection.find_one({"_id": client_id}) or {}
    storage_alerts = client_doc.get("active_storage_alerts", [])
    gpu_alert_sent = client_doc.get("gpu_alert_sent", False)
    
    storage_modified = False
    gpu_modified = False

    # Storage warning checker
    for disk in payload.disks:
        mount = disk.get("mount", "/")
        percent = disk.get("percent", 0.0)
        
        if percent >= 80.0:
            if mount not in storage_alerts:
                storage_alerts.append(mount)
                storage_modified = True
                send_rocketchat_notification(
                    text=f"⚠️ **Storage Warning:** Client `{client_id}` disk `{mount}` is at **{percent}%** capacity.",
                    color="#f39c12"
                )
        else:
            if mount in storage_alerts:
                storage_alerts.remove(mount)
                storage_modified = True
                send_rocketchat_notification(
                    text=f"✅ **Storage Recovered:** Client `{client_id}` disk `{mount}` has cleared warning state and is at **{percent}%**.",
                    color="#2ecc71"
                )

    # GPU Failure Monitor
    if len(payload.gpus) == 0:
        if not gpu_alert_sent:
            gpu_alert_sent = True
            gpu_modified = True
            send_rocketchat_notification(
                text=f"🚨 **GPU Failure:** Client `{client_id}` is not reporting any GPU data! (nvidia-smi has failed or is missing)",
                color="#e74c3c"
            )
    else:
        if gpu_alert_sent:
            gpu_alert_sent = False
            gpu_modified = True
            send_rocketchat_notification(
                text=f"✅ **GPU Recovered:** Client `{client_id}` is reporting GPU data successfully again.",
                color="#2ecc71"
            )

    # CPU thermal alerting. Hysteresis (alert at 85C, clear at 75C) stops a node
    # hovering on the threshold from spamming Rocket.Chat every 10 seconds.
    temp_alert_sent = client_doc.get("cpu_temp_alert_sent", False)
    temp_modified = False
    cpu_temp = payload.cpu_temp
    if cpu_temp is not None:
        if cpu_temp >= CPU_TEMP_ALERT_C and not temp_alert_sent:
            temp_alert_sent = True
            temp_modified = True
            send_rocketchat_notification(
                text=f"🔥 **CPU OVERHEATING:** Client `{client_id}` CPU at **{cpu_temp}°C** (threshold {CPU_TEMP_ALERT_C}°C). Check cooling — thermal shutdown risk.",
                color="#e74c3c",
            )
        elif cpu_temp <= CPU_TEMP_CLEAR_C and temp_alert_sent:
            temp_alert_sent = False
            temp_modified = True
            send_rocketchat_notification(
                text=f"✅ **CPU temperature normal:** Client `{client_id}` back down to {cpu_temp}°C.",
                color="#2ecc71",
            )

    update_fields = {
        "telemetry": payload.dict()
    }
    if storage_modified:
        update_fields["active_storage_alerts"] = storage_alerts
    if gpu_modified:
        update_fields["gpu_alert_sent"] = gpu_alert_sent
    if temp_modified:
        update_fields["cpu_temp_alert_sent"] = temp_alert_sent
        
    clients_collection.update_one(
        {"_id": client_id},
        {"$set": update_fields}
    )

    # Log telemetry history event
    append_to_logs("telemetry", client_id, payload.dict())

    data = payload.dict()
    client_telemetry[client_id] = data
    
    # Log received stats (opt-in: this was the main journald/syslog flood source)
    if VERBOSE_TELEMETRY:
        print(f"\n[TELEMETRY RECEIVED] from {client_id}:")
        print(f"  CPU:  {data['cpu_percent']}%")
        print(f"  RAM:  {data['memory_percent']}% ({data['memory_free_gb']} GB Free / {data['memory_total_gb']} GB Total)")

        for disk in data['disks']:
            print(f"  Drive {disk['mount']}: {disk['percent']}% Used ({disk['free_gb']} GB Free / {disk['total_gb']} GB Total)")

        for gpu in data['gpus']:
            power_str = f"Power: {gpu.get('power_draw', 'N/A')} / {gpu.get('power_limit', 'N/A')}"
            fan_str = f"Fan: {gpu.get('fan_speed', 'N/A')}"
            print(f"  GPU [{gpu['name']}]: Core: {gpu['utilization']}, Temp: {gpu['temp']}, VRAM: {gpu['memory_used']}/{gpu['memory_total']}, {power_str}, {fan_str}")

    return {"status": "success"}

    data = payload.dict()
    client_telemetry[client_id] = data
    
    # Temporarily print the data to the console for the user to see!
    print(f"\n[TELEMETRY RECEIVED] from {client_id}:")
    print(f"  CPU:  {data['cpu_percent']}%")
    print(f"  RAM:  {data['memory_percent']}% ({data['memory_free_gb']} GB Free / {data['memory_total_gb']} GB Total)")
    
    for disk in data['disks']:
        print(f"  Drive {disk['mount']}: {disk['percent']}% Used ({disk['free_gb']} GB Free / {disk['total_gb']} GB Total)")
        
    for gpu in data['gpus']:
        power_str = f"Power: {gpu.get('power_draw', 'N/A')} / {gpu.get('power_limit', 'N/A')}"
        fan_str = f"Fan: {gpu.get('fan_speed', 'N/A')}"
        print(f"  GPU [{gpu['name']}]: Core: {gpu['utilization']}, Temp: {gpu['temp']}, VRAM: {gpu['memory_used']}/{gpu['memory_total']}, {power_str}, {fan_str}")
        
    return {"status": "success"}

@app.get("/api/telemetry")
async def get_all_telemetry(current_user: str = Depends(get_current_user)):
    """
    Endpoint to view the live dashboard data of all clients.
    """
    clients = load_clients()
    return {cid: info.get("telemetry", {}) for cid, info in clients.items() if info.get("telemetry")}

@app.get("/api/clients")
async def get_clients_api(current_user: str = Depends(get_current_user)):
    """
    Endpoint for the React UI to fetch the live active client registry with computed statuses.
    """
    return load_clients()

@app.get("/api/logs")
async def get_logs_history(current_user: str = Depends(get_current_user)):
    """
    Endpoint to retrieve logs history grouped by client_id.
    """
    try:
        # Newest 25 per client with slim details. The full-fat version shipped
        # every stored telemetry payload (2 MB+ JSON) and the dashboard re-polls
        # this every 3s — it was hanging the browser.
        grouped_logs = {}
        cursor = logs_collection.find().sort("timestamp", -1)
        for doc in cursor:
            cid = doc.get("client_id")
            if not cid:
                continue
            bucket = grouped_logs.setdefault(cid, [])
            if len(bucket) >= 25:
                continue
            dtype = doc.get("type")
            detail = doc.get("detail")
            if dtype == "telemetry" and isinstance(detail, dict):
                detail = {"cpu_percent": detail.get("cpu_percent"),
                          "memory_percent": detail.get("memory_percent")}
            elif isinstance(detail, dict):
                detail = {k: (v[:400] if isinstance(v, str) else v) for k, v in detail.items()}
            bucket.append({
                "timestamp": doc.get("timestamp"),
                "type": dtype,
                "detail": detail
            })
        for cid in grouped_logs:
            grouped_logs[cid].reverse()
        return grouped_logs
    except Exception as e:
        print(f"[!] Error loading logs from MongoDB: {e}")
        return {}

@app.post("/api/command-result")
async def receive_command_result(client_id: str, payload: CommandResultPayload, token_valid: bool = Depends(require_agent_token)):
    """
    Endpoint for remote agents to push execution results back to the central server.
    """
    if not client_id or not client_id.strip():
        raise HTTPException(status_code=400, detail="client_id cannot be empty")
        
    append_to_logs("command_result", client_id, payload.dict())
    return {"status": "success"}

@app.get("/api/get-raw-commands")
async def get_raw_commands(current_user: str = Depends(get_current_user)):
    """
    Endpoint for the UI to load the complete commands.json whitelist file.
    """
    commands_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "commands.json")
    try:
        with open(commands_file, "r") as f:
            return json.load(f)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to read commands: {e}")

@app.post("/api/save-commands")
async def save_commands(commands: Dict[str, Any], current_user: str = Depends(get_current_user)):
    """
    Endpoint for the UI to save changes back to commands.json.
    """
    commands_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "commands.json")
    try:
        with open(commands_file, "w") as f:
            json.dump(commands, f, indent=2)
        return {"status": "success", "message": "commands.json saved successfully"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to save commands: {e}")

@app.get("/metrics", response_class=PlainTextResponse)
async def get_prometheus_metrics():
    """
    Exposes the latest client telemetry in standard Prometheus text format.
    """
    clients = load_clients()
    metrics_lines = []
    
    for client_id, data in clients.items():
        # Active status (1.0 for active/online, 0.0 for inactive/offline)
        active_val = 1.0 if data.get("active", False) else 0.0
        metrics_lines.append(f'rmm_client_active{{client_id="{client_id}"}} {active_val}')
        
        # Telemetry stats
        telemetry = data.get("telemetry")
        if telemetry:
            cpu = telemetry.get("cpu_percent", 0.0)
            ram = telemetry.get("memory_percent", 0.0)
            ram_free = telemetry.get("memory_free_gb", 0.0)
            ram_total = telemetry.get("memory_total_gb", 0.0)
            
            metrics_lines.append(f'rmm_cpu_utilization{{client_id="{client_id}"}} {cpu}')
            metrics_lines.append(f'rmm_memory_utilization{{client_id="{client_id}"}} {ram}')
            metrics_lines.append(f'rmm_memory_free_bytes{{client_id="{client_id}"}} {ram_free * (1024**3)}')
            metrics_lines.append(f'rmm_memory_total_bytes{{client_id="{client_id}"}} {ram_total * (1024**3)}')
            
            # Disk mounts
            for disk in telemetry.get("disks", []):
                mount = disk.get("mount", "/")
                disk_percent = disk.get("percent", 0.0)
                metrics_lines.append(f'rmm_disk_utilization{{client_id="{client_id}",mount="{mount}"}} {disk_percent}')
            
            # GPU utilization and temps
            for gpu in telemetry.get("gpus", []):
                gpu_name = gpu.get("name", "GPU")
                
                # Utilization percent parsing
                gpu_util_str = str(gpu.get("utilization", "0"))
                gpu_util = float(gpu_util_str.replace("%", "").strip())
                
                # Temperature parsing
                gpu_temp_str = str(gpu.get("temp", "0"))
                gpu_temp = float(gpu_temp_str.replace("C", "").strip())
                
                metrics_lines.append(f'rmm_gpu_utilization{{client_id="{client_id}",gpu_name="{gpu_name}"}} {gpu_util}')
                metrics_lines.append(f'rmm_gpu_temperature{{client_id="{client_id}",gpu_name="{gpu_name}"}} {gpu_temp}')
                
    return "\n".join(metrics_lines) + "\n"

if __name__ == "__main__":
    print("[*] Starting FastAPI Central Server...")
    # uvicorn runs the FastAPI app on port 8000
    uvicorn.run("central_api_prototype:app", host="0.0.0.0", port=8000, reload=True)

# ============================================================
# Remote File Transfer (video fetch from agent TAKELEAP folders)
# ============================================================
import os
FILE_TRANSFER_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "file_transfers")
os.makedirs(FILE_TRANSFER_DIR, exist_ok=True)

class FileStatusPayload(BaseModel):
    filename: str
    status: str
    message: Optional[str] = ""
    progress_percent: Optional[float] = None

def set_file_transfer_state(client_id: str, filename: str, status: str, message: str = "", extra: Optional[Dict[str, Any]] = None):
    state = {
        "filename": filename,
        "status": status,
        "message": message,
        "updated": datetime.now().isoformat()
    }
    if extra:
        state.update(extra)
    clients_collection.update_one(
        {"_id": client_id},
        {"$set": {"file_transfer": state}},
        upsert=True
    )

@app.post("/api/request-file")
async def request_file(client_id: str, filename: str, current_user: str = Depends(get_current_user)):
    safe_name = os.path.basename(filename.strip())
    if not client_id.strip() or not safe_name:
        raise HTTPException(status_code=400, detail="client_id and filename are required")

    clients_collection.update_one(
        {"_id": client_id},
        {"$set": {"pending_file_request": safe_name}},
        upsert=True
    )
    set_file_transfer_state(client_id, safe_name, "requested", "Waiting for agent to poll...")
    append_to_logs("file_requested", client_id, {"filename": safe_name})
    return {"status": "success", "message": f"File '{safe_name}' requested from {client_id}"}

@app.post("/api/set-shift-path")
async def set_shift_path(client_id: str, shift_path: str = "", current_user: str = Depends(get_current_user)):
    """
    Store where this node keeps its SHIFT folder (varies per site:
    /mnt/<disk-uuid>/SR/SHIFT or /mnt/<disk-uuid>/TAKELEAP/SHIFT).
    Delivered to the agent in every heartbeat response.
    """
    path = shift_path.strip()
    result = clients_collection.update_one({"_id": client_id}, {"$set": {"shift_path": path}})
    if result.matched_count == 0:
        raise HTTPException(status_code=404, detail="Unknown client_id")
    append_to_logs("shift_path_set", client_id, {"shift_path": path})
    return {"status": "success", "shift_path": path}

@app.post("/api/request-search")
async def request_search(client_id: str, query: str, current_user: str = Depends(get_current_user)):
    """
    Queue a filename search in the node's SHIFT folder. The agent picks the
    query up on its next heartbeat and posts matches to /api/search-results.
    """
    q = query.strip()
    if not q:
        raise HTTPException(status_code=400, detail="Empty search query")
    now_str = datetime.now().isoformat()
    result = clients_collection.update_one(
        {"_id": client_id},
        {"$set": {"pending_search": q,
                  "file_search": {"query": q, "status": "searching", "results": [], "requested_at": now_str}}}
    )
    if result.matched_count == 0:
        raise HTTPException(status_code=404, detail="Unknown client_id")
    append_to_logs("search_requested", client_id, {"query": q})
    return {"status": "success"}

@app.get("/api/history")
async def get_history(client_id: str, hours: int = 6, current_user: str = Depends(get_current_user)):
    """Vitals history for one node — powers the dashboard sparklines."""
    hours = min(max(hours, 1), HISTORY_RETENTION_DAYS * 24)
    since = datetime.now() - timedelta(hours=hours)
    try:
        cursor = history_collection.find(
            {"client_id": client_id, "ts": {"$gte": since}},
            {"_id": 0, "client_id": 0},
        ).sort("ts", 1)
        return [{
            "ts": d["ts"].isoformat(),
            "cpu": d.get("cpu"),
            "cpu_temp": d.get("cpu_temp"),
            "ram": d.get("ram"),
            "gpu_temp": d.get("gpu_temp"),
        } for d in cursor]
    except Exception as e:
        print(f"[!] Error loading history for {client_id}: {e}")
        return []

@app.get("/api/agent-token")
async def get_agent_token(current_user: str = Depends(get_current_user)):
    """Current agent token, for building the install command in the dashboard.
    Logged-in operators only — the installer download itself is now authenticated."""
    return {"token": AGENT_TOKEN}

@app.get("/api/history-bulk")
async def get_history_bulk(points: int = 60, current_user: str = Depends(get_current_user)):
    """
    Recent vitals for every node in one request — the dashboard sparklines poll
    this instead of one request per card.
    """
    points = min(max(points, 5), 240)
    since = datetime.now() - timedelta(minutes=points)
    out = {}
    try:
        cursor = history_collection.find(
            {"ts": {"$gte": since}},
            {"_id": 0, "ts": 1, "client_id": 1, "cpu": 1, "cpu_temp": 1, "ram": 1},
        ).sort("ts", 1)
        for d in cursor:
            out.setdefault(d["client_id"], []).append({
                "ts": d["ts"].isoformat(),
                "cpu": d.get("cpu"),
                "cpu_temp": d.get("cpu_temp"),
                "ram": d.get("ram"),
            })
    except Exception as e:
        print(f"[!] Error loading bulk history: {e}")
    return out

@app.post("/api/clear-search")
async def clear_search(client_id: str, current_user: str = Depends(get_current_user)):
    """Drop a node's stored search result so the dashboard stops showing it."""
    clients_collection.update_one({"_id": client_id}, {"$unset": {"file_search": ""}})
    return {"status": "success"}

@app.post("/api/search-results")
async def receive_search_results(client_id: str, payload: SearchResultsPayload, token_valid: bool = Depends(require_agent_token)):
    now_str = datetime.now().isoformat()
    clients_collection.update_one(
        {"_id": client_id},
        {"$set": {"file_search": {
            "query": payload.query,
            "status": "error" if payload.error else "done",
            "results": payload.results[:200],
            "searched_path": payload.searched_path,
            "error": payload.error,
            "completed_at": now_str,
        }}}
    )
    append_to_logs("search_results", client_id, {"query": payload.query, "count": len(payload.results), "error": payload.error})
    return {"status": "success"}

@app.get("/api/get-file-request")
async def get_file_request(client_id: str, token_valid: bool = Depends(require_agent_token)):
    if not client_id or not client_id.strip():
        raise HTTPException(status_code=400, detail="client_id cannot be empty")

    doc = clients_collection.find_one_and_update(
        {"_id": client_id},
        {"$set": {"pending_file_request": "none"}},
        return_document=False
    )
    fname = doc.get("pending_file_request", "none") if doc else "none"
    return {"filename": fname}

@app.post("/api/file-transfer-status")
async def update_file_transfer_status(client_id: str, payload: FileStatusPayload, token_valid: bool = Depends(require_agent_token)):
    if not client_id or not client_id.strip():
        raise HTTPException(status_code=400, detail="client_id cannot be empty")
    set_file_transfer_state(client_id, payload.filename, payload.status, payload.message or "", {"progress_percent": payload.progress_percent})
    append_to_logs("file_transfer_status", client_id, payload.dict())
    return {"status": "success"}

@app.post("/api/cancel-upload")
async def cancel_upload(client_id: str, filename: str):
    safe_name = os.path.basename(filename.strip())
    set_file_transfer_state(client_id, safe_name, "cancelled", "Upload cancelled by user")
    
    client_dir = os.path.join(FILE_TRANSFER_DIR, os.path.basename(client_id.strip()))
    if os.path.exists(client_dir):
        for f in os.listdir(client_dir):
            if f.startswith(f"{safe_name}.part"):
                try:
                    os.remove(os.path.join(client_dir, f))
                except:
                    pass
    return {"status": "success"}

@app.get("/api/received-chunks")
async def received_chunks(client_id: str, filename: str, chunk_size: int, file_size: int, token_valid: bool = Depends(require_agent_token)):
    """Resume support: report which chunk indexes are already stored for this file.
    Parts whose size doesn't match the current chunking scheme (stale from an
    earlier run with a different chunk size) are deleted so they can't corrupt
    the final stitch."""
    safe_name = os.path.basename(filename.strip())
    if not client_id.strip() or not safe_name or chunk_size <= 0 or file_size < 0:
        raise HTTPException(status_code=400, detail="invalid parameters")

    total_chunks = 1 if file_size == 0 else (file_size + chunk_size - 1) // chunk_size
    last_expected = file_size - (total_chunks - 1) * chunk_size

    client_dir = os.path.join(FILE_TRANSFER_DIR, os.path.basename(client_id.strip()))
    received = []
    if os.path.isdir(client_dir):
        prefix = f"{safe_name}.part"
        for entry in os.listdir(client_dir):
            if not entry.startswith(prefix) or not entry[len(prefix):].isdigit():
                continue
            idx = int(entry[len(prefix):])
            path = os.path.join(client_dir, entry)
            expected = chunk_size if idx < total_chunks - 1 else last_expected
            try:
                if idx < total_chunks and os.path.getsize(path) == expected:
                    received.append(idx)
                else:
                    os.remove(path)
            except OSError:
                pass  # part is mid-write or locked; report as not received
    return {"received": sorted(received)}

@app.post("/api/upload-chunk")
async def upload_chunk(client_id: str, filename: str, chunk_index: int, request: Request, token_valid: bool = Depends(require_agent_token)):
    safe_name = os.path.basename(filename.strip())
    if not client_id.strip() or not safe_name:
        raise HTTPException(status_code=400, detail="client_id and filename are required")

    doc = clients_collection.find_one({"_id": client_id})
    if doc and doc.get("file_transfer", {}).get("filename") == safe_name and doc.get("file_transfer", {}).get("status") == "cancelled":
        return {"status": "cancelled"}

    client_dir = os.path.join(FILE_TRANSFER_DIR, os.path.basename(client_id.strip()))
    os.makedirs(client_dir, exist_ok=True)
    dest_path = os.path.join(client_dir, f"{safe_name}.part{chunk_index}")

    try:
        with open(dest_path, "wb") as f:
            async for chunk in request.stream():
                f.write(chunk)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to store chunk: {e}")

    return {"status": "success", "chunk_index": chunk_index}

@app.get("/api/server-files")
async def list_server_files(current_user: str = Depends(get_current_user)):
    """
    All videos currently stored on the server (per client), so the dashboard
    can show and manage what is occupying the 20 GB transfer store.
    """
    data = {}
    try:
        for cid in sorted(os.listdir(FILE_TRANSFER_DIR)):
            client_dir = os.path.join(FILE_TRANSFER_DIR, cid)
            if not os.path.isdir(client_dir):
                continue
            files = []
            for name in sorted(os.listdir(client_dir)):
                if ".part" in name:
                    continue
                path = os.path.join(client_dir, name)
                if os.path.isfile(path):
                    st = os.stat(path)
                    files.append({
                        "name": name,
                        "size_mb": round(st.st_size / (1024 * 1024), 2),
                        "modified": datetime.fromtimestamp(st.st_mtime).isoformat(timespec="seconds"),
                    })
            if files:
                data[cid] = files
    except Exception as e:
        print(f"[!] Error listing server files: {e}")
    return data

@app.post("/api/delete-server-file")
async def delete_server_file(client_id: str, filename: str, current_user: str = Depends(get_current_user)):
    safe_client = os.path.basename(client_id.strip())
    safe_name = os.path.basename(filename.strip())
    if not safe_client or not safe_name:
        raise HTTPException(status_code=400, detail="client_id and filename are required")
    path = os.path.join(FILE_TRANSFER_DIR, safe_client, safe_name)
    if not os.path.isfile(path):
        raise HTTPException(status_code=404, detail="File not found on server")
    os.remove(path)
    # Drop the stale 'ready' transfer card if it points at the deleted file
    doc = clients_collection.find_one({"_id": client_id}, {"file_transfer": 1})
    if doc and (doc.get("file_transfer") or {}).get("filename") == safe_name:
        clients_collection.update_one({"_id": client_id}, {"$unset": {"file_transfer": ""}})
    append_to_logs("server_file_deleted", client_id, {"filename": safe_name})
    return {"status": "success", "message": f"Deleted '{safe_name}' from server"}

@app.get("/api/speedtest-blob")
async def speedtest_blob(size_mb: int = 8, token_valid: bool = Depends(require_agent_token)):
    # Agents time this download to estimate their downlink to the server.
    size_mb = max(1, min(int(size_mb), 32))
    def gen():
        chunk = b"\0" * (1024 * 1024)
        for _ in range(size_mb):
            yield chunk
    return StreamingResponse(gen(), media_type="application/octet-stream")

@app.post("/api/speedtest-sink")
async def speedtest_sink(request: Request, token_valid: bool = Depends(require_agent_token)):
    # Agents time this upload; the body is read and discarded.
    total = 0
    async for chunk in request.stream():
        total += len(chunk)
    return {"status": "success", "received_bytes": total}

def enforce_storage_limit(max_bytes: int = 20 * 1024**3):
    """Scan FILE_TRANSFER_DIR recursively and delete oldest completed files if total size > max_bytes."""
    all_files = []
    total_size = 0
    for root, dirs, files in os.walk(FILE_TRANSFER_DIR):
        for name in files:
            if ".part" in name:
                continue
            path = os.path.join(root, name)
            try:
                stat = os.stat(path)
                all_files.append((stat.st_mtime, path, stat.st_size))
                total_size += stat.st_size
            except FileNotFoundError:
                pass
    
    if total_size > max_bytes:
        all_files.sort(key=lambda x: x[0])
        for mtime, path, size in all_files:
            if total_size <= max_bytes:
                break
            try:
                os.remove(path)
                total_size -= size
                print(f"Deleted old file {path} to free space.")
            except Exception as e:
                print(f"Error deleting file {path}: {e}")

@app.post("/api/upload-complete")
async def upload_complete(client_id: str, filename: str, total_chunks: int, background_tasks: BackgroundTasks, token_valid: bool = Depends(require_agent_token)):
    safe_name = os.path.basename(filename.strip())
    client_dir = os.path.join(FILE_TRANSFER_DIR, os.path.basename(client_id.strip()))
    final_path = os.path.join(client_dir, safe_name)
    
    try:
        with open(final_path, "wb") as outfile:
            for i in range(total_chunks):
                part_path = os.path.join(client_dir, f"{safe_name}.part{i}")
                if not os.path.exists(part_path):
                    raise HTTPException(status_code=400, detail=f"Missing chunk {i}")
                with open(part_path, "rb") as infile:
                    outfile.write(infile.read())
                os.remove(part_path)
    except Exception as e:
        set_file_transfer_state(client_id, safe_name, "error", f"Upload finalize failed: {e}")
        raise HTTPException(status_code=500, detail=f"Failed to finalize file: {e}")

    size_mb = round(os.path.getsize(final_path) / (1024 * 1024), 2)
    set_file_transfer_state(client_id, safe_name, "ready", f"File received ({size_mb} MB)", {"size_mb": size_mb, "progress_percent": 100.0})
    append_to_logs("file_received", client_id, {"filename": safe_name, "size_mb": size_mb})
    send_rocketchat_notification(
        text=f"📥 **File Received:** `{safe_name}` ({size_mb} MB) uploaded from client `{client_id}`.",
        color="#2ecc71"
    )
    background_tasks.add_task(enforce_storage_limit)
    return {"status": "success", "size_mb": size_mb}

@app.get("/api/file-transfers")
async def get_file_transfers(current_user: str = Depends(get_current_user)):
    try:
        cursor = clients_collection.find({}, {"file_transfer": 1})
        return {doc["_id"]: doc.get("file_transfer") for doc in cursor if doc.get("file_transfer")}
    except Exception as e:
        print(f"[!] Error loading file transfers: {e}")
        return {}

from fastapi.responses import FileResponse, Response, StreamingResponse

@app.get("/deploy_agent.sh")
async def get_agent_installer(x_agent_token: str = Header(None), token: Optional[str] = None,
                              authorization: str = Header(None)):
    # The served installer carries the LIVE agent token, so this download is
    # authenticated in EVERY mode — gating it on strict mode published the
    # fleet's token to the internet. Accepted proof: an agent token (header for
    # self-update, ?token= for a human running the install by hand) or a
    # logged-in dashboard session.
    dashboard_ok = bool(authorization and authorization.startswith("Bearer ")
                        and verify_token(authorization.split(" ")[1]))
    if not (is_valid_agent_token(x_agent_token) or is_valid_agent_token(token) or dashboard_ok):
        raise HTTPException(status_code=401, detail="Invalid or missing agent token")
    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "deploy_agent.sh")
    if not os.path.isfile(path):
        raise HTTPException(status_code=404, detail="Installer script not found on server")
    # Normalize CRLF -> LF (edited on Windows, executed by bash on Linux) and inject
    # the live agent token so a freshly downloaded installer bakes it into the agent.
    with open(path, "rb") as f:
        content = f.read().replace(b"\r\n", b"\n")
    # Replace only the assignment, not the "was I injected?" sentinel comparison
    # a few lines below it — a blanket replace turns that check into an
    # always-true self-comparison and the installer discards the injected token.
    content = content.replace(
        b'AGENT_TOKEN="__AGENT_TOKEN__"',
        b'AGENT_TOKEN="' + AGENT_TOKEN.encode("utf-8") + b'"',
        1,
    )
    return Response(
        content=content,
        media_type="text/x-shellscript",
        headers={"Content-Disposition": 'attachment; filename="deploy_agent.sh"'},
    )

@app.get("/api/download-file")
async def download_file(client_id: str, filename: str, token: Optional[str] = None, inline: bool = False, authorization: str = Header(None)):
    authed = False
    if authorization and authorization.startswith("Bearer ") and verify_token(authorization.split(" ")[1]):
        authed = True
    if token and verify_token(token):
        authed = True
    if not authed:
        raise HTTPException(status_code=401, detail="Authentication token is invalid or has expired")

    safe_name = os.path.basename(filename.strip())
    path = os.path.join(FILE_TRANSFER_DIR, os.path.basename(client_id.strip()), safe_name)
    if not os.path.isfile(path):
        raise HTTPException(status_code=404, detail="File not found on server")

    media_type = "video/mp4" if safe_name.lower().endswith(".mp4") else "application/octet-stream"
    if inline:
        return FileResponse(path, media_type=media_type)
    return FileResponse(path, media_type=media_type, filename=safe_name)

@app.post("/api/heartbeat")
async def receive_heartbeat(client_id: str, payload: TelemetryPayload, request: Request,
                            token_valid: bool = Depends(require_agent_token)):
    await receive_telemetry(client_id, payload)
    # Record migration status so the dashboard can confirm the fleet is upgraded
    # and every node is authenticated before auth is flipped to strict.
    # During rotation both tokens validate, so also record whether the agent
    # holds the CURRENT one — that is the signal for "safe to drop the previous
    # token and flip strict".
    presented = request.headers.get("x-agent-token")
    token_is_current = bool(AGENT_TOKEN and presented
                            and hmac.compare_digest(presented, AGENT_TOKEN))
    set_fields = {"agent_version": payload.agent_version or "unknown",
                  "speed_upload_mbps": payload.speed_upload_mbps,
                  "speed_download_mbps": payload.speed_download_mbps,
                  "agent_token_ok": token_valid,
                  "agent_token_current": token_is_current}
    # The Synology reverse proxy forwards the site's real public IP — record it
    # so every node's WAN address is visible without any agent-side lookup.
    xff = (request.headers.get("x-forwarded-for") or "").split(",")[0].strip()
    if xff:
        set_fields["public_ip"] = xff
    clients_collection.update_one({"_id": client_id}, {"$set": set_fields})
    doc = clients_collection.find_one_and_update(
        {"_id": client_id},
        {"$set": {"pending_command": "none", "pending_file_request": "none", "pending_search": "none"}},
        return_document=False
    )
    cmd = "none"
    fname = "none"
    search = "none"
    shift_path = None
    if doc:
        cmd = doc.get("pending_command", "none")
        fname = doc.get("pending_file_request", "none")
        search = doc.get("pending_search", "none") or "none"
        shift_path = doc.get("shift_path") or None

    if cmd != "none":
        append_to_logs("command_polled", client_id, {"command": cmd})

    return {
        "status": "success",
        "command": cmd,
        "filename": fname,
        "search": search,
        "shift_path": shift_path
    }


# (Phase-1 sync-monitor proxy removed 2026-08-17 — replaced by the native
#  /api/sync-monitor/* implementation below, same URL contract.)


# =============================================================================
# Sync-monitor NATIVE ingest — Phase 1 of absorbing the rclone collector
# (2026-08-17). The v3.6-sync RMM agent POSTs parsed sync-log events here.
# ADDITIVE: own sync_* collections in rmm_db, own endpoint, existing agent
# token auth. Runs in PARALLEL with the standalone collector until Phase 4.
# Logic ported from R-clone monitor/collector/main.py incl. its fixes:
# skip-aware pairs, restart-safe round summaries, transfer-byte accounting.
# =============================================================================
import re
from datetime import datetime as _sdt, timedelta as _std, timezone as _stz

sync_machines = db["sync_machines"]
sync_pairs = db["sync_pairs"]
sync_events = db["sync_events"]
sync_events.create_index([("machine_id", 1), ("received_at", -1)])
sync_events.create_index([("machine_id", 1), ("type", 1), ("direction", 1),
                          ("received_at", -1)])
sync_pairs.create_index([("machine_id", 1), ("pair", 1)], unique=True)
try:
    sync_events.create_index([("received_at", 1)], name="sync_events_ttl",
                             expireAfterSeconds=30 * 86400)
except Exception:
    pass  # window changed — harmless, old TTL keeps working

_SYNC_KEEP = {"round_start", "round_end", "pair_ok", "pair_fail",
              "file_synced", "error"}
_SYNC_LOG_TS = "%Y/%m/%d %H:%M:%S"
_SYNC_SIZES = {"": 1, "K": 1e3, "Ki": 1024, "M": 1e6, "Mi": 1024 ** 2,
               "G": 1e9, "Gi": 1024 ** 3, "T": 1e12, "Ti": 1024 ** 4}


def _sync_size_bytes(s):
    m = re.match(r"([\d.]+)\s*([KMGT]i?)?B$", (s or "").strip())
    return float(m.group(1)) * _SYNC_SIZES[m.group(2) or ""] if m else None


def _sync_summarise_round(mid, end_ev, inventory, ts):
    rnd = end_ev.get("round")
    if db is None or rnd is None:
        return
    # dedup on round AND log_ts: container restarts reset the round counter
    if sync_events.find_one({"machine_id": mid, "type": "round_summary",
                             "round": rnd, "log_ts": end_ev.get("log_ts")}):
        return
    start_ev = sync_events.find_one(
        {"machine_id": mid, "type": "round_start", "round": rnd},
        sort=[("received_at", -1)])
    since = start_ev["received_at"] if start_ev else ts - _std(minutes=30)
    synced = list(sync_events.find(
        {"machine_id": mid, "type": "file_synced", "round": rnd,
         "received_at": {"$gte": since}}, {"file": 1, "direction": 1}))
    up = [e for e in synced if e.get("direction") == "up"]
    sizes = {f["name"]: f.get("size", 0)
             for files in (inventory or {}).values() for f in files}
    bytes_up = sum(sizes.get(e.get("file"), 0) for e in up)
    duration = None
    try:
        t0 = _sdt.strptime(start_ev["log_ts"].strip(), _SYNC_LOG_TS)
        t1 = _sdt.strptime(end_ev["log_ts"].strip(), _SYNC_LOG_TS)
        duration = (t1 - t0).total_seconds()
    except Exception:
        pass
    summary = {"type": "round_summary", "machine_id": mid, "round": rnd,
               "log_ts": end_ev.get("log_ts"), "files": len(synced),
               "files_up": len(up), "files_down": len(synced) - len(up),
               "bytes_up": bytes_up, "duration_s": duration,
               "avg_bps": (bytes_up / duration) if duration and bytes_up else None,
               "received_at": ts}
    sync_events.insert_one(dict(summary))
    if up:
        summary.pop("_id", None)
        sync_machines.update_one({"machine_id": mid},
                                 {"$set": {"last_transfer": summary}})


@app.post("/api/sync-ingest")
async def sync_ingest(payload: Dict[str, Any], request: Request,
                      _auth: None = Depends(require_agent_token)):
    """Heartbeat + parsed sync-log events from the RMM agent's sync module."""
    mid = payload.get("machine_id")
    if not mid:
        raise HTTPException(status_code=400, detail="machine_id required")
    ts = _sdt.now(_stz.utc)
    mset = {"machine_id": mid, "site": payload.get("site"), "last_seen": ts,
            "no_container": bool(payload.get("no_container"))}
    if "inventory" in payload:
        mset["inventory"] = payload["inventory"]
    if payload.get("events"):
        mset["last_activity"] = ts

    to_insert = []
    for ev in payload.get("events", []):
        etype = ev.get("type")
        if etype == "round_start":
            mset["current_round"] = ev.get("round")
            mset["progress"] = None
        elif etype == "round_end":
            mset["last_round"] = ev.get("round")
            mset["last_round_at"] = ts
            mset["progress"] = None
        elif etype == "progress":
            mset["progress"] = {k: ev.get(k) for k in
                                ("pair", "direction", "percent", "speed",
                                 "eta", "done", "total")}
        elif etype == "pair_skip":
            sync_pairs.update_one(
                {"machine_id": mid, "pair": ev.get("pair")},
                {"$set": {"direction": ev.get("direction"),
                          "last_skip_at": ts, "last_idle_s": ev.get("idle_s")},
                 "$setOnInsert": {"last_status": "idle"}}, upsert=True)
        elif etype in ("pair_start", "pair_ok", "pair_fail"):
            status = {"pair_start": "running", "pair_ok": "ok",
                      "pair_fail": "fail"}[etype]
            ev["transferred_bytes"] = _sync_size_bytes(ev.get("transferred"))
            upd = {"machine_id": mid, "pair": ev.get("pair"),
                   "direction": ev.get("direction"), "last_status": status,
                   "updated_at": ts}
            if etype == "pair_fail":
                upd["last_exit"] = ev.get("exit_code")
                upd["last_fail_at"] = ts
            elif etype == "pair_ok":
                upd["last_exit"] = 0
                upd["last_ok_at"] = ts
            sync_pairs.update_one({"machine_id": mid, "pair": ev.get("pair")},
                                  {"$set": upd}, upsert=True)
        if etype in _SYNC_KEEP:
            to_insert.append({**ev, "machine_id": mid, "received_at": ts})

    sync_machines.update_one({"machine_id": mid},
                             {"$set": mset,
                              "$setOnInsert": {"first_seen": ts}}, upsert=True)
    if to_insert:
        sync_events.insert_many(to_insert)
    for ev in payload.get("events", []):
        if ev.get("type") == "round_end":
            inv = payload.get("inventory")
            if inv is None:
                inv = (sync_machines.find_one({"machine_id": mid},
                                              {"inventory": 1}) or {}).get("inventory")
            _sync_summarise_round(mid, ev, inv, ts)
    return {"ok": True, "stored": len(to_insert)}


# =============================================================================
# Sync-monitor NATIVE engine (Phase 2, 2026-08-17): verification against the
# NAS, NAS health, fleet view/alerts, stats & timeseries — same URL contract
# the SyncMonitor UI already uses. Listings/health live in Mongo so the 4
# uvicorn workers share one verification effort instead of quadrupling it.
# rclone is invoked with env-injected config (no config file on disk).
# =============================================================================
import subprocess
import threading
import time

_SYNC_NAS_HOST = os.getenv("REAL_SYNOLOGY_HOST", "")
_SYNC_NAS_PORT = os.getenv("REAL_SYNOLOGY_PORT", "22")
_SYNC_NAS_USER = os.getenv("REAL_SYNOLOGY_USER", "")
_SYNC_NAS_PASS = os.getenv("REAL_SYNOLOGY_PASS", "")
_SYNC_VERIFY_IVL = int(os.getenv("SYNC_VERIFY_INTERVAL", "300"))
_SYNC_GRACE_S = int(os.getenv("SYNC_MISSING_GRACE_S", "600"))
_SYNC_NAS_MIN_FREE = float(os.getenv("NAS_MIN_FREE_PCT", "10"))
_SYNC_ONLINE_S = 60
_SYNC_STALLED_S = 900
_sync_listings = db["sync_listings"]   # {_id: remote_path, files: [...], at}
_sync_rclone_env = None


def _sync_now():
    # their Mongo client is NOT tz_aware: reads come back naive-UTC, so all
    # read-path arithmetic uses naive-UTC now
    return _sdt.utcnow()


def _sync_rclone(args, timeout):
    """Run rclone against the NAS remote via env-injected config."""
    global _sync_rclone_env
    if not _SYNC_NAS_HOST:
        return None
    if _sync_rclone_env is None:
        obscured = subprocess.run(["rclone", "obscure", _SYNC_NAS_PASS],
                                  capture_output=True, text=True, timeout=15)
        env = dict(os.environ)
        env.update({"RCLONE_CONFIG_SYNOREAL_TYPE": "sftp",
                    "RCLONE_CONFIG_SYNOREAL_HOST": _SYNC_NAS_HOST,
                    "RCLONE_CONFIG_SYNOREAL_PORT": _SYNC_NAS_PORT,
                    "RCLONE_CONFIG_SYNOREAL_USER": _SYNC_NAS_USER,
                    "RCLONE_CONFIG_SYNOREAL_PASS": obscured.stdout.strip()})
        _sync_rclone_env = env
    return subprocess.run(["rclone"] + args, capture_output=True,
                          timeout=timeout, env=_sync_rclone_env)


def _sync_verify_loop():
    time.sleep(60 + (os.getpid() % 180))  # stagger the 4 workers
    while True:
        try:
            # NAS health first — reachability + capacity, shared via Mongo
            nas_doc = _sync_listings.find_one({"_id": "@nas"})
            if not nas_doc or (_sync_now() - nas_doc["at"]).total_seconds() > _SYNC_VERIFY_IVL:
                # per-VOLUME capacity: `about` on the SFTP root reports the
                # chrooted home (2.4 GB!), not the data shares the uploads
                # land on — each share sits on its own volume
                health = {"_id": "@nas", "at": _sync_now(), "ok": False,
                          "error": None, "volumes": []}
                try:
                    for vol in [v for v in os.getenv(
                            "NAS_VOLUMES", "").split(",") if v.strip()] or [""]:
                        out = _sync_rclone(["about", f"synoreal:{vol.strip()}",
                                            "--json"], 60)
                        if out is None:
                            health["error"] = "NAS creds not configured"
                            break
                        if out.returncode == 0:
                            about = json.loads(out.stdout)
                            health["ok"] = True   # at least one volume answered
                            health["volumes"].append(
                                {"path": vol.strip() or "/",
                                 "total": about.get("total"),
                                 "free": about.get("free")})
                        else:
                            health["error"] = out.stderr.decode()[:200]
                except FileNotFoundError:
                    health["error"] = "rclone binary not installed on server"
                except Exception as e:
                    health["error"] = str(e)[:200]
                _sync_listings.update_one({"_id": "@nas"}, {"$set": health},
                                          upsert=True)

            # listings for every known up-pair remote path
            for p in sync_pairs.find({"direction": "up"}):
                label = p.get("pair", "")
                if "server:" not in label:
                    continue
                rpath = label.split("server:", 1)[1].strip()
                doc = _sync_listings.find_one({"_id": rpath})
                if doc and (_sync_now() - doc["at"]).total_seconds() < _SYNC_VERIFY_IVL:
                    continue
                try:
                    out = _sync_rclone(["lsjson", "--recursive", "--files-only",
                                        f"synoreal:{rpath}"], 300)
                    if out is not None and out.returncode == 0:
                        files = [{"n": f["Path"], "s": f["Size"]}
                                 for f in json.loads(out.stdout)]
                        _sync_listings.update_one(
                            {"_id": rpath},
                            {"$set": {"files": files, "at": _sync_now()}},
                            upsert=True)
                except FileNotFoundError:
                    break  # no rclone binary — health doc already says so
                except Exception:
                    pass
        except Exception:
            pass
        time.sleep(_SYNC_VERIFY_IVL)


threading.Thread(target=_sync_verify_loop, daemon=True).start()


def _sync_annotate(m, pairs):
    inv = m.get("inventory") or {}
    total_pending = total_bytes = 0
    verified_any = False
    for p in pairs:
        label = p.get("pair", "")
        if p.get("direction") != "up" or "server:" not in label:
            continue
        local = label.split("→")[0].strip()
        rpath = label.split("server:", 1)[1].strip()
        listing = _sync_listings.find_one({"_id": rpath})
        files = inv.get(local)
        if listing is None or files is None:
            continue
        verified_any = True
        server_names = {f["n"] for f in listing.get("files", [])}
        pending = [f for f in files if f["name"] not in server_names]
        p["pending_count"] = len(pending)
        p["pending_bytes"] = sum(f.get("size", 0) for f in pending)
        p["verified_at"] = listing["at"]
        p["missing"] = [f["name"] for f in pending
                        if time.time() - f["mtime"] > _SYNC_GRACE_S][:20]
        total_pending += len(pending)
        total_bytes += p["pending_bytes"]
    if not verified_any:
        return None, None
    return total_pending, total_bytes


def _sync_machine_view(m, ts):
    mid = m["machine_id"]
    last_seen = m.get("last_seen")
    online = bool(last_seen) and (ts - last_seen).total_seconds() < _SYNC_ONLINE_S
    pairs = [dict(p, _id=None) for p in sync_pairs.find({"machine_id": mid})]
    for p in pairs:
        p.pop("_id", None)
    pending_total, pending_bytes = _sync_annotate(m, pairs)
    hour_ago = ts - _std(hours=1)
    up_pairs = [p for p in pairs if p.get("direction") == "up"]
    progress = m.get("progress")
    return {
        "machine_id": mid, "site": m.get("site"), "online": online,
        "last_seen": last_seen, "last_activity": m.get("last_activity"),
        "first_seen": m.get("first_seen"),
        "no_container": m.get("no_container", False),
        "current_round": m.get("current_round"),
        "last_round": m.get("last_round"),
        "last_round_at": m.get("last_round_at"),
        "progress": progress if (progress or {}).get("direction") == "up" else None,
        "last_transfer": m.get("last_transfer"),
        "pairs": pairs, "up_pairs": up_pairs,
        "pending_total": pending_total, "pending_bytes": pending_bytes,
        "in_progress": len([p for p in up_pairs if p.get("last_status") == "running"]),
        "failed_up": len([p for p in up_pairs if p.get("last_status") == "fail"]),
        "uploads_1h": sync_events.count_documents(
            {"machine_id": mid, "type": "file_synced", "direction": "up",
             "received_at": {"$gt": hour_ago}}),
        "uploads_24h": sync_events.count_documents(
            {"machine_id": mid, "type": "file_synced", "direction": "up",
             "received_at": {"$gt": ts - _std(hours=24)}}),
        "errors_1h": sync_events.count_documents(
            {"machine_id": mid, "type": "error", "direction": "up",
             "received_at": {"$gt": hour_ago}}),
    }


def _sync_fleet():
    ts = _sync_now()
    machines = sorted((_sync_machine_view(m, ts) for m in sync_machines.find()),
                      key=lambda x: x["machine_id"])
    alerts = []
    nas = _sync_listings.find_one({"_id": "@nas"}) or {}
    nas.pop("_id", None)
    if nas and not nas.get("ok"):
        alerts.append({"severity": "critical", "machine_id": "NAS",
                       "message": f"NAS check failed: {nas.get('error')}"})
    for vol in nas.get("volumes") or []:
        if vol.get("total"):
            vol["free_pct"] = round(100.0 * (vol.get("free") or 0) / vol["total"], 1)
            if vol["free_pct"] < _SYNC_NAS_MIN_FREE:
                alerts.append({"severity": "critical", "machine_id": "NAS",
                               "message": f"NAS volume {vol['path']} nearly full — "
                                          f"{vol['free_pct']}% free"})
    for m in machines:
        if m["no_container"]:
            continue  # roster-style visibility, but no sync rules apply
        if not m["online"]:
            alerts.append({"severity": "critical", "machine_id": m["machine_id"],
                           "message": "machine offline — no heartbeat"})
        if (m["online"] and not m.get("last_round_at") and not m.get("last_activity")
                and m.get("first_seen")
                and (ts - m["first_seen"]).total_seconds() > _SYNC_STALLED_S):
            alerts.append({"severity": "serious", "machine_id": m["machine_id"],
                           "message": "agent alive but sync log SILENT since install "
                                      "— is the sync container running?"})
        for p in m["up_pairs"]:
            if p.get("last_status") == "fail":
                alerts.append({"severity": "serious", "machine_id": m["machine_id"],
                               "message": f"upload pair failing (exit {p.get('last_exit')}): {p['pair']}"})
        if m["online"] and m["errors_1h"] > 0:
            alerts.append({"severity": "warning", "machine_id": m["machine_id"],
                           "message": f"{m['errors_1h']} upload error(s) in the last hour"})
        busy_now = (m.get("last_activity")
                    and (ts - m["last_activity"]).total_seconds() < 120)
        for p in m["up_pairs"]:
            if p.get("missing") and not (p.get("last_status") == "running" and busy_now):
                alerts.append({"severity": "serious", "machine_id": m["machine_id"],
                               "message": f"{len(p['missing'])} file(s) NOT on server after "
                                          f"{_SYNC_GRACE_S // 60}min grace: "
                                          f"{', '.join(p['missing'][:3])}…"})
        if m["online"] and m.get("last_round_at"):
            marks = [m["last_round_at"]]
            if m.get("last_activity"):
                marks.append(m["last_activity"])
            stalled = (ts - max(marks)).total_seconds()
            if stalled > _SYNC_STALLED_S:
                age = f"{stalled / 3600:.1f}h" if stalled >= 3600 else f"{int(stalled // 60)}min"
                alerts.append({"severity": "serious", "machine_id": m["machine_id"],
                               "message": f"agent alive but NO sync round completed in {age}"})
    return ts, machines, alerts, nas


@app.get("/api/sync-monitor/health")
def sync_native_health():
    return {"ok": True, "time": _sync_now(), "native": True}


@app.get("/api/sync-monitor/overview")
def sync_native_overview(current_user: str = Depends(get_current_user)):
    ts, machines, alerts, nas = _sync_fleet()
    return {"generated_at": ts, "machines": machines, "alerts": alerts, "nas": nas}


@app.get("/api/sync-monitor/alerts")
def sync_native_alerts(current_user: str = Depends(get_current_user)):
    ts, _, alerts, _nas = _sync_fleet()
    return {"generated_at": ts, "count": len(alerts), "alerts": alerts}


_SYNC_WINDOWS = [("1h", 3600), ("24h", 86400), ("3d", 259200),
                 ("7d", 604800), ("30d", 2592000)]
_SYNC_BUCKET = {"1h": 300, "24h": 3600, "3d": 10800, "7d": 21600, "30d": 86400}


@app.get("/api/sync-monitor/stats")
def sync_native_stats(current_user: str = Depends(get_current_user)):
    ts = _sync_now()
    cuts = {lbl: ts - _std(seconds=s) for lbl, s in _SYNC_WINDOWS}
    counters = {"uploaded": {"type": "file_synced", "direction": "up"},
                "downloaded": {"type": "file_synced", "direction": "down"},
                "errors": {"type": "error", "direction": "up"},
                "failures": {"type": "pair_fail", "direction": "up"},
                "rounds": {"type": "round_summary"}}
    sums = {"bytes": ({"type": "round_summary"}, "bytes_up"),
            "seconds": ({"type": "round_summary"}, "duration_s"),
            "up_bytes": ({"type": "pair_ok", "direction": "up"}, "transferred_bytes"),
            "down_bytes": ({"type": "pair_ok", "direction": "down"}, "transferred_bytes")}

    def cond(cut, conds):
        cl = [{"$gte": ["$received_at", cut]}]
        cl += [{"$eq": [{"$ifNull": [f"${k}", None]}, v]} for k, v in conds.items()]
        return {"$and": cl}

    group = {"_id": "$machine_id"}
    for lbl, _s in _SYNC_WINDOWS:
        for name, conds in counters.items():
            group[f"{name}|{lbl}"] = {"$sum": {"$cond": [cond(cuts[lbl], conds), 1, 0]}}
        for name, (conds, field) in sums.items():
            group[f"{name}|{lbl}"] = {"$sum": {"$cond": [
                cond(cuts[lbl], conds), {"$ifNull": [f"${field}", 0]}, 0]}}

    per_machine = {}
    for row in sync_events.aggregate([
            {"$match": {"received_at": {"$gte": cuts["30d"]}}},
            {"$group": group}]):
        windows = {}
        for lbl, _s in _SYNC_WINDOWS:
            byts = row.get(f"bytes|{lbl}") or 0
            secs = row.get(f"seconds|{lbl}") or 0
            windows[lbl] = {k: row.get(f"{k}|{lbl}", 0) for k in
                            ("uploaded", "downloaded", "errors", "failures",
                             "rounds", "up_bytes", "down_bytes")}
            windows[lbl]["bytes"] = byts
            windows[lbl]["seconds"] = secs
            windows[lbl]["avg_bps"] = (byts / secs) if secs > 0 and byts else None
        per_machine[row["_id"]] = windows

    fleet = {}
    for lbl, _s in _SYNC_WINDOWS:
        agg = {k: 0 for k in ("uploaded", "downloaded", "errors", "failures",
                              "rounds", "bytes", "up_bytes", "down_bytes", "seconds")}
        for wins in per_machine.values():
            w = wins.get(lbl) or {}
            for k in agg:
                agg[k] += w.get(k) or 0
        agg["avg_bps"] = (agg["bytes"] / agg["seconds"]) if agg["seconds"] else None
        fleet[lbl] = agg
    return {"generated_at": ts, "windows": [w for w, _ in _SYNC_WINDOWS],
            "retention_days": 30, "fleet": fleet, "machines": per_machine}


@app.get("/api/sync-monitor/timeseries")
def sync_native_timeseries(range: str = "24h", machine_id: str = None,
                           current_user: str = Depends(get_current_user)):
    if range not in _SYNC_BUCKET:
        raise HTTPException(status_code=400, detail="bad range")
    step = _SYNC_BUCKET[range]
    ts = _sync_now()
    start = ts - _std(seconds=dict(_SYNC_WINDOWS)[range])
    match = {"received_at": {"$gte": start},
             "type": {"$in": ["file_synced", "error"]}}
    volm = {"received_at": {"$gte": start}, "type": "pair_ok"}
    if machine_id:
        match["machine_id"] = machine_id
        volm["machine_id"] = machine_id
    bucket = {"$toDate": {"$subtract": [{"$toLong": "$received_at"},
              {"$mod": [{"$toLong": "$received_at"}, step * 1000]}]}}
    is_up = {"$eq": [{"$ifNull": ["$direction", ""]}, "up"]}
    is_down = {"$eq": [{"$ifNull": ["$direction", ""]}, "down"]}
    series = {"$switch": {"branches": [
        {"case": {"$and": [{"$eq": ["$type", "error"]}, is_up]}, "then": "errors"},
        {"case": {"$and": [{"$eq": ["$type", "file_synced"]}, is_up]}, "then": "uploaded"},
        {"case": {"$and": [{"$eq": ["$type", "file_synced"]}, is_down]}, "then": "downloaded"},
    ], "default": "skip"}}
    counts, volumes = {}, {}
    for row in sync_events.aggregate([{"$match": match},
            {"$group": {"_id": {"b": bucket, "s": series}, "n": {"$sum": 1}}}]):
        if row["_id"]["s"] != "skip":
            counts[(int(row["_id"]["b"].replace(tzinfo=_stz.utc).timestamp()), row["_id"]["s"])] = row["n"]
    for row in sync_events.aggregate([{"$match": volm},
            {"$group": {"_id": {"b": bucket, "d": "$direction"},
                        "bytes": {"$sum": {"$ifNull": ["$transferred_bytes", 0]}}}}]):
        volumes[(int(row["_id"]["b"].replace(tzinfo=_stz.utc).timestamp()), row["_id"]["d"])] = row["bytes"]
    first = int(start.replace(tzinfo=_stz.utc).timestamp()) // step * step
    last = int(ts.replace(tzinfo=_stz.utc).timestamp()) // step * step
    points = []
    b = first
    while b <= last:
        points.append({"t": _sdt.fromtimestamp(b, _stz.utc),
                       "uploaded": counts.get((b, "uploaded"), 0),
                       "downloaded": counts.get((b, "downloaded"), 0),
                       "errors": counts.get((b, "errors"), 0),
                       "up_bytes": volumes.get((b, "up"), 0),
                       "down_bytes": volumes.get((b, "down"), 0)})
        b += step
    return {"generated_at": ts, "range": range, "bucket_s": step,
            "machine_id": machine_id, "points": points}


@app.get("/api/sync-monitor/machines/{machine_id}/events")
def sync_native_events(machine_id: str, type: str = None, direction: str = None,
                       q: str = None, since: _sdt = None, until: _sdt = None,
                       busy: bool = False, limit: int = 50, skip: int = 0,
                       current_user: str = Depends(get_current_user)):
    query = {"machine_id": machine_id}
    if type:
        query["type"] = {"$in": type.split(",")}
    if direction:
        query["direction"] = direction
    if busy:
        query.setdefault("$and", []).append(
            {"$or": [{"files_up": {"$gt": 0}}, {"files": {"$gt": 0}},
                     {"type": {"$ne": "round_summary"}}]})
    if since or until:
        rng = {}
        if since:
            rng["$gte"] = since.replace(tzinfo=None) if since.tzinfo else since
        if until:
            rng["$lt"] = until.replace(tzinfo=None) if until.tzinfo else until
        query["received_at"] = rng
    if q:
        rx = {"$regex": re.escape(q), "$options": "i"}
        query.setdefault("$and", []).append(
            {"$or": [{"file": rx}, {"message": rx}, {"pair": rx}]})
    limit = max(1, min(limit, 500))
    total = sync_events.count_documents(query)
    evs = list(sync_events.find(query, {"_id": 0})
               .sort("received_at", -1).skip(max(0, skip)).limit(limit))
    return {"machine_id": machine_id, "total": total, "skip": skip,
            "limit": limit, "events": evs}


@app.delete("/api/sync-monitor/machines/{machine_id}")
def sync_native_forget(machine_id: str,
                       current_user: str = Depends(get_current_user)):
    sync_machines.delete_one({"machine_id": machine_id})
    pairs = sync_pairs.delete_many({"machine_id": machine_id}).deleted_count
    events = sync_events.delete_many({"machine_id": machine_id}).deleted_count
    return {"ok": True, "machine_id": machine_id, "pairs": pairs, "events": events}