Nexus/server/api/servers.py

"""Nexus — Servers API Routes (CRUD + heartbeat + health check + push)
Presentation layer — receives HTTP requests, delegates to ServerService/SyncService.

Live server status comes from Redis (real-time), base info from MySQL.
All write operations require JWT authentication and produce audit logs.
"""

import json
import logging
from typing import Optional, List

from fastapi import APIRouter, Depends, HTTPException, Query

from server.api.dependencies import get_server_service, get_sync_service, get_db
from server.api.auth_jwt import get_current_admin
from server.api.schemas import ServerCreate, ServerUpdate, ServerPush, ServerCheck
from server.application.services.server_service import ServerService
from server.application.services.sync_service import SyncService
from server.domain.models import Server, SyncLog, Admin, AuditLog
from server.infrastructure.redis.client import get_redis
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger("nexus.servers")

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

REDIS_KEY_PREFIX = "heartbeat:"


# ── CRUD ──

@router.get("/", response_model=dict)
async def list_servers(
    category: Optional[str] = Query(None, description="Filter by category"),
    page: int = Query(1, ge=1, description="Page number"),
    per_page: int = Query(50, ge=1, le=200, description="Items per page"),
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """List servers with live status from Redis (DB-level pagination)

    Base info (name/IP/category) from MySQL with DB-level pagination,
    real-time status from Redis. If Redis heartbeat key exists,
    override is_online/system_info/last_heartbeat.
    """
    from server.infrastructure.database.server_repo import ServerRepositoryImpl

    offset = (page - 1) * per_page
    repo = ServerRepositoryImpl(db)
    page_servers, total = await repo.get_paginated(category, offset, per_page)
    redis = get_redis()

    result = []
    for server in page_servers:
        server_data = _server_to_dict(server)

        # Overlay Redis heartbeat data (real-time)
        try:
            heartbeat = await redis.hgetall(f"{REDIS_KEY_PREFIX}{server.id}")
            if heartbeat:
                server_data["is_online"] = heartbeat.get("is_online") == "True"
                if heartbeat.get("system_info"):
                    try:
                        server_data["system_info"] = json.loads(heartbeat.get("system_info", "{}"))
                    except (json.JSONDecodeError, TypeError):
                        pass  # Keep MySQL value
                server_data["last_heartbeat"] = heartbeat.get("last_heartbeat", server_data.get("last_heartbeat"))
                server_data["agent_version"] = heartbeat.get("agent_version", server_data.get("agent_version", ""))
                server_data["_source"] = "redis"
        except Exception as e:
            logger.warning(f"Redis read failed for server {server.id}: {e}")

        result.append(server_data)

    return {
        "items": result,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": (total + per_page - 1) // per_page,
    }


# ── Dashboard Stats (must be before /{id} to avoid path collision) ──

@router.get("/stats", response_model=dict)
async def server_stats(
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Aggregated server stats for dashboard — SQL aggregation + Redis alert count"""
    from sqlalchemy import func, text

    # Total + category breakdown in a single query
    result = await db.execute(
        select(Server.category, func.count(Server.id))
        .group_by(Server.category)
    )
    categories = {}
    total = 0
    for cat, cnt in result.all():
        key = cat or "uncategorized"
        categories[key] = cnt
        total += cnt

    # Online count via SQL (fallback; Redis heartbeat overrides this)
    result = await db.execute(
        select(func.count(Server.id)).where(Server.is_online == True)
    )
    db_online = result.scalar() or 0

    # Count active alerts in Redis
    redis = get_redis()
    alerts = 0
    try:
        # SCAN for alert keys (avoid KEYS on large datasets)
        cursor = 0
        while True:
            cursor, keys = await redis.scan(cursor, match="alerts:*", count=200)
            alerts += len(keys)
            if cursor == 0:
                break
    except Exception:
        pass

    return {
        "total": total,
        "online": db_online,
        "offline": total - db_online,
        "alerts": alerts,
        "categories": categories,
    }


@router.get("/{id}", response_model=dict)
async def get_server(
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
):
    """Get a single server by ID with live status from Redis"""
    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    server_data = _server_to_dict(server)

    # Overlay Redis heartbeat data
    redis = get_redis()
    try:
        heartbeat = await redis.hgetall(f"{REDIS_KEY_PREFIX}{id}")
        if heartbeat:
            server_data["is_online"] = heartbeat.get("is_online") == "True"
            if heartbeat.get("system_info"):
                try:
                    server_data["system_info"] = json.loads(heartbeat.get("system_info", "{}"))
                except (json.JSONDecodeError, TypeError):
                    pass
            server_data["last_heartbeat"] = heartbeat.get("last_heartbeat", server_data.get("last_heartbeat"))
            server_data["agent_version"] = heartbeat.get("agent_version", server_data.get("agent_version", ""))
            server_data["_source"] = "redis"
    except Exception as e:
        logger.warning(f"Redis read failed for server {id}: {e}")

    return server_data


@router.post("/", response_model=dict, status_code=201)
async def create_server(
    payload: ServerCreate,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Create a new server (sensitive fields encrypted at rest)"""
    from server.infrastructure.database.crypto import encrypt_value

    data = payload.model_dump(exclude_none=True)
    # Encrypt sensitive fields before storing
    if data.get("password"):
        data["password"] = encrypt_value(data["password"])
    if data.get("ssh_key_private"):
        data["ssh_key_private"] = encrypt_value(data["ssh_key_private"])

    server = Server(**data)
    created = await service.create_server(server)

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="create_server",
        target_type="server",
        target_id=created.id,
        detail=f"name={created.name} domain={created.domain}",
        ip_address="",
    ))

    return _server_to_dict(created)


@router.put("/{id}", response_model=dict)
async def update_server(
    id: int,
    payload: ServerUpdate,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Update an existing server (sensitive fields encrypted at rest)"""
    from server.infrastructure.database.crypto import encrypt_value

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")
    for key, value in payload.model_dump(exclude_unset=True).items():
        if hasattr(server, key) and key != "id":
            # Encrypt sensitive fields on update
            if key == "password" and value:
                value = encrypt_value(value)
            elif key == "ssh_key_private" and value:
                value = encrypt_value(value)
            setattr(server, key, value)
    updated = await service.update_server(server)

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="update_server",
        target_type="server",
        target_id=id,
        detail=f"name={updated.name}",
        ip_address="",
    ))

    return _server_to_dict(updated)


@router.delete("/{id}", status_code=204)
async def delete_server(
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Delete a server"""
    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")
    result = await service.delete_server(id)
    if not result:
        raise HTTPException(status_code=404, detail="Server not found")

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="delete_server",
        target_type="server",
        target_id=id,
        detail=f"name={server.name}",
        ip_address="",
    ))


# ── Agent API Key ──

@router.post("/{id}/agent-key", response_model=dict)
async def generate_agent_api_key(
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Generate a new per-server Agent API key"""
    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    import secrets
    new_key = f"nxs-{secrets.token_urlsafe(32)}"
    server.agent_api_key = new_key
    await service.update_server(server)

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="generate_agent_key",
        target_type="server",
        target_id=id,
        detail=f"Generated new Agent API key for {server.name}",
        ip_address="",
    ))

    return {"agent_api_key": new_key, "server_id": id}


# ── Health Check ──

@router.post("/check", response_model=dict)
async def check_servers(
    payload: ServerCheck,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
):
    """Check health status of specified servers (Agent-based, no SSH)"""
    return await service.check_all_servers(payload.server_ids)


# ── Push ──

@router.post("/push", response_model=dict)
async def push_to_servers(
    payload: ServerPush,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Push files to multiple servers via SyncEngineV2 (with retry jobs)"""
    from server.application.services.sync_engine_v2 import SyncEngineV2
    from server.infrastructure.database.server_repo import ServerRepositoryImpl
    from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
    from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
    from server.infrastructure.database.push_schedule_repo import PushRetryJobRepositoryImpl

    engine = SyncEngineV2(
        server_repo=ServerRepositoryImpl(db),
        sync_log_repo=SyncLogRepositoryImpl(db),
        audit_repo=AuditLogRepositoryImpl(db),
        retry_repo=PushRetryJobRepositoryImpl(db),
    )
    result = await engine.sync_files(
        server_ids=payload.server_ids,
        source_path=payload.source_path,
        target_path=payload.target_path,
        sync_mode=payload.sync_mode,
        trigger_type="manual",
        operator=admin.username,
    )
    return result


# ── Sync Logs ──

@router.get("/logs", response_model=list)
async def get_all_sync_logs(
    limit: int = Query(100, ge=1, le=500),
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Get recent sync logs across all servers (for dashboard charts)"""
    result = await db.execute(
        select(SyncLog, Server.name).join(Server, SyncLog.server_id == Server.id, isouter=True)
        .order_by(SyncLog.started_at.desc()).limit(limit)
    )
    rows = result.all()
    return [_sync_log_to_dict(log, server_name=name) for log, name in rows]


@router.get("/{id}/logs", response_model=list)
async def get_server_logs(
    id: int,
    limit: int = Query(50, ge=1, le=200),
    admin: Admin = Depends(get_current_admin),
    sync_service: SyncService = Depends(get_sync_service),
):
    """Get sync logs for a specific server"""
    logs = await sync_service.get_sync_logs(id, limit)
    return [_sync_log_to_dict(log) for log in logs]


# ── Helper functions ──

def _server_to_dict(server: Server) -> dict:
    """Convert Server model to API response dict (hide sensitive fields)"""
    return {
        "id": server.id,
        "name": server.name,
        "domain": server.domain,
        "port": server.port,
        "username": server.username,
        "auth_method": server.auth_method,
        "password_set": bool(server.password),
        "ssh_key_path": server.ssh_key_path,
        "ssh_key_private_set": bool(server.ssh_key_private),
        "agent_port": server.agent_port,
        "agent_api_key": (server.agent_api_key[:8] + "...") if server.agent_api_key and len(server.agent_api_key) > 8 else (server.agent_api_key or ""),
        "agent_api_key_set": bool(server.agent_api_key),
        "description": server.description,
        "target_path": server.target_path,
        "category": server.category,
        "platform_id": server.platform_id,
        "node_id": server.node_id,
        "protocols": server.protocols,
        "extra_attrs": server.extra_attrs,
        "connectivity": server.connectivity,
        "ssh_key_configured": server.ssh_key_configured,
        "is_online": server.is_online,
        "last_heartbeat": str(server.last_heartbeat) if server.last_heartbeat else None,
        "system_info": server.system_info,
        "agent_version": server.agent_version,
        "created_at": str(server.created_at) if server.created_at else None,
        "updated_at": str(server.updated_at) if server.updated_at else None,
    }


def _sync_log_to_dict(log, server_name: str = None) -> dict:
    """Convert SyncLog model to API response dict"""
    return {
        "id": log.id,
        "server_id": log.server_id,
        "server_name": server_name,
        "source_path": log.source_path,
        "target_path": log.target_path,
        "trigger_type": log.trigger_type,
        "operator": log.operator,
        "status": log.status,
        "sync_mode": log.sync_mode,
        "files_total": log.files_total,
        "files_transferred": log.files_transferred,
        "duration_seconds": log.duration_seconds,
        "error_message": log.error_message,
        "started_at": str(log.started_at) if log.started_at else None,
        "finished_at": str(log.finished_at) if log.finished_at else None,
    }