live/backend/routes/nodes.py

from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from typing import Optional
from config import settings
from services.settings_store import save_settings
from routes.auth import require_auth
from services.load_balancer import lb
from services.multicast_detector import multicast_detector
from services.node_store import save_nodes

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


class NodeUpdate(BaseModel):
    weight: int | None = None
    enabled: bool | None = None
    subnet: str | None = None
    name: str | None = None


class StrategyUpdate(BaseModel):
    strategy: str
    multicast_enabled: bool | None = None


class NodeHeartbeat(BaseModel):
    node_id: str
    cpu_percent: float | None = None
    memory_percent: float | None = None
    bandwidth_mbps: float | None = None
    active_connections: int | None = None
    hls_readers: int | None = None
    network_in_mbps: float | None = None
    network_out_mbps: float | None = None
    viewer_count: int | None = None
    frame_drop_indicators: int | None = None
    packet_loss_indicators: int | None = None
    media_metrics_available: bool = False


class NodeRegister(BaseModel):
    node_id: str
    name: str
    host: str
    hls_port: int
    api_port: int
    weight: int = 100
    subnet: str = "0.0.0.0/0"


@router.get("")
async def list_nodes(_: bool = Depends(require_auth)):
    nodes = []
    for n in settings.edge_nodes:
        nid = n["id"]
        nodes.append({
            "id": nid,
            "name": n["name"],
            "host": n["host"],
            "hls_port": n["hls_port"],
            "api_port": n["api_port"],
            "weight": n["weight"],
            "enabled": n.get("enabled", True),
            "subnet": n.get("subnet", "0.0.0.0/0"),
            "is_origin": n.get("is_origin", False),
            "online": lb.is_online(nid),
            "metrics": lb._node_health.get(nid),
        })
    return nodes


@router.get("/strategy")
async def get_strategy(_: bool = Depends(require_auth)):
    return {
        "strategy": settings.load_balance_strategy,
        "multicast_enabled": settings.multicast_enabled,
        "multicast_available": multicast_detector.is_available(),
    }


@router.put("/strategy")
async def update_strategy(body: StrategyUpdate, _: bool = Depends(require_auth)):
    if body.strategy in ("round-robin", "weighted", "least-conn", "subnet"):
        settings.load_balance_strategy = body.strategy
    if body.multicast_enabled is not None:
        settings.multicast_enabled = body.multicast_enabled
    # Persisted for the same reason as the brand: in-memory only meant a restart
    # silently restored the defaults.
    save_settings()
    return {"ok": True}


@router.put("/{node_id}")
async def update_node(node_id: str, body: NodeUpdate, _: bool = Depends(require_auth)):
    found = False
    for n in settings.edge_nodes:
        if n["id"] == node_id:
            if body.weight is not None:
                n["weight"] = body.weight
            if body.enabled is not None:
                n["enabled"] = body.enabled
            if body.subnet is not None:
                n["subnet"] = body.subnet
            if body.name is not None:
                name = body.name.strip()
                if not name:
                    raise HTTPException(status_code=422, detail="Node name cannot be empty")
                n["name"] = name
            found = True
            break
    if not found:
        raise HTTPException(status_code=404, detail="Node not found")
    save_nodes()
    return {"ok": True}


@router.post("/register")
async def register_node(body: NodeRegister):
    for n in settings.edge_nodes:
        if n["id"] == body.node_id:
            n.update({
                "host": body.host,
                "hls_port": body.hls_port,
                "api_port": body.api_port,
                "weight": body.weight,
                "subnet": body.subnet,
            })
            save_nodes()
            return {"ok": True, "action": "updated"}
    settings.edge_nodes.append({
        "id": body.node_id,
        "name": body.name,
        "host": body.host,
        "hls_port": body.hls_port,
        "api_port": body.api_port,
        "weight": body.weight,
        "enabled": True,
        "subnet": body.subnet,
        "is_origin": False,
    })
    save_nodes()
    return {"ok": True, "action": "registered"}


@router.post("/heartbeat")
async def node_heartbeat(body: NodeHeartbeat):
    lb.report_health(body.node_id, {
        "cpu_percent": body.cpu_percent,
        "memory_percent": body.memory_percent,
        "bandwidth_mbps": body.bandwidth_mbps,
        "active_connections": body.active_connections,
        "hls_readers": body.hls_readers,
        "network_in_mbps": body.network_in_mbps,
        "network_out_mbps": body.network_out_mbps,
        "viewer_count": body.viewer_count,
        "frame_drop_indicators": body.frame_drop_indicators,
        "packet_loss_indicators": body.packet_loss_indicators,
        "media_metrics_available": body.media_metrics_available,
    })
    return {"ok": True}


@router.get("/health")
async def get_all_health():
    return lb.get_health_summary()