spark-mcp/internal/admin/router.go

// Package admin exposes the admin HTTP API for cluster management and audit
// log access. All routes are protected by middleware.AdminAuth.
package admin

import (
	"errors"
	"net/http"
	"strconv"

	"github.com/gin-gonic/gin"

	"spark-mcp-go/internal/audit"
	"spark-mcp-go/internal/cluster"
	"spark-mcp-go/internal/middleware"
	"spark-mcp-go/internal/storage"
)

// Mount attaches the /admin sub-router to r, protecting every route with
// bearer-token admin authentication.
func Mount(r *gin.Engine, repo *storage.ClusterRepo, auditRepo *audit.Repo, adminTokens []string) {
	g := r.Group("/admin", middleware.AdminAuth(adminTokens))
	g.GET("", webHandler)
	g.GET("/", webHandler)
	g.GET("/clusters", listClusters(repo))
	g.POST("/clusters", createCluster(repo, auditRepo))
	g.GET("/clusters/:id", getCluster(repo))
	g.PUT("/clusters/:id", updateCluster(repo, auditRepo))
	g.DELETE("/clusters/:id", deleteCluster(repo, auditRepo))
	g.GET("/audit", listAudit(auditRepo))
	g.GET("/docs", DocsHandler)
	g.GET("/docs/spec", OpenAPISpecHandler)
}

func listClusters(repo *storage.ClusterRepo) gin.HandlerFunc {
	return func(c *gin.Context) {
		clusters, err := repo.List(c.Request.Context())
		if err != nil {
			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
			return
		}
		c.JSON(http.StatusOK, clusters)
	}
}

func createCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.HandlerFunc {
	return func(c *gin.Context) {
		var cl cluster.Cluster
		if err := c.ShouldBindJSON(&cl); err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
			return
		}
		if err := validateClusterCreate(&cl); err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
			return
		}

		ctx := c.Request.Context()
		if err := repo.Create(ctx, &cl); err != nil {
			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
			return
		}

		details, err := audit.MarshalDetails(clusterFields(&cl))
		if err == nil {
			_ = auditRepo.Insert(ctx, &audit.Entry{
				Actor:     actor(c),
				Action:    audit.ActionClusterCreate,
				ClusterID: cl.ID,
				Details:   details,
			})
		}

		c.JSON(http.StatusCreated, &cl)
	}
}

func getCluster(repo *storage.ClusterRepo) gin.HandlerFunc {
	return func(c *gin.Context) {
		cl, err := repo.Get(c.Request.Context(), c.Param("id"))
		if err != nil {
			if errors.Is(err, storage.ErrNotFound) {
				c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"})
				return
			}
			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
			return
		}
		c.JSON(http.StatusOK, cl)
	}
}

func updateCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.HandlerFunc {
	return func(c *gin.Context) {
		id := c.Param("id")
		ctx := c.Request.Context()

		old, err := repo.Get(ctx, id)
		if err != nil {
			if errors.Is(err, storage.ErrNotFound) {
				c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"})
				return
			}
			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
			return
		}

		var cl cluster.Cluster
		if err := c.ShouldBindJSON(&cl); err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
			return
		}

		cl.ID = id
		if cl.AuthPassword == "" {
			cl.AuthPassword = old.AuthPassword
		}

		if err := repo.Update(ctx, &cl); err != nil {
			if errors.Is(err, storage.ErrNotFound) {
				c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"})
				return
			}
			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
			return
		}

		details, err := audit.MarshalDetails(map[string]any{
			"before": clusterFields(old),
			"after":  clusterFields(&cl),
		})
		if err == nil {
			_ = auditRepo.Insert(ctx, &audit.Entry{
				Actor:     actor(c),
				Action:    audit.ActionClusterUpdate,
				ClusterID: cl.ID,
				Details:   details,
			})
		}

		c.JSON(http.StatusOK, &cl)
	}
}

func deleteCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.HandlerFunc {
	return func(c *gin.Context) {
		id := c.Param("id")
		ctx := c.Request.Context()

		old, err := repo.Get(ctx, id)
		if err != nil {
			if errors.Is(err, storage.ErrNotFound) {
				c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"})
				return
			}
			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
			return
		}

		if err := repo.Delete(ctx, id); err != nil {
			if errors.Is(err, storage.ErrNotFound) {
				c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"})
				return
			}
			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
			return
		}

		details, err := audit.MarshalDetails(map[string]any{
			"id":   id,
			"name": old.Name,
		})
		if err == nil {
			_ = auditRepo.Insert(ctx, &audit.Entry{
				Actor:     actor(c),
				Action:    audit.ActionClusterDelete,
				ClusterID: id,
				Details:   details,
			})
		}

		c.Status(http.StatusNoContent)
	}
}

func listAudit(auditRepo *audit.Repo) gin.HandlerFunc {
	return func(c *gin.Context) {
		limitStr := c.DefaultQuery("limit", "100")
		limit, err := strconv.Atoi(limitStr)
		if err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid limit"})
			return
		}

		entries, err := auditRepo.List(c.Request.Context(), limit)
		if err != nil {
			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
			return
		}
		c.JSON(http.StatusOK, entries)
	}
}

func actor(c *gin.Context) string {
	tok, _ := c.Get("admin_token")
	s, _ := tok.(string)
	if len(s) > 8 {
		s = s[:8]
	}
	return "admin:" + s
}

func validateClusterCreate(c *cluster.Cluster) error {
	if c.ID == "" {
		return errors.New("id is required")
	}
	if c.Name == "" {
		return errors.New("name is required")
	}
	if c.RMURL == "" {
		return errors.New("rm_url is required")
	}
	if c.SHSURL == "" {
		return errors.New("shs_url is required")
	}
	if c.SparkSubmitExecuteBin == "" {
		return errors.New("spark_submit_execute_bin is required")
	}
	switch c.AuthType {
	case cluster.AuthNone, cluster.AuthSimple, cluster.AuthBasic:
	default:
		return errors.New("auth_type must be one of none, simple, basic")
	}
	if c.RateLimitPerMin < 0 {
		return errors.New("rate_limit_per_min must be >= 0")
	}
	return nil
}

func clusterFields(c *cluster.Cluster) map[string]any {
	return map[string]any{
		"id":                       c.ID,
		"name":                     c.Name,
		"rm_url":                   c.RMURL,
		"shs_url":                  c.SHSURL,
		"spark_submit_execute_bin": c.SparkSubmitExecuteBin,
		"is_active":                c.IsActive,
		"auth_type":                c.AuthType,
		"auth_username":            c.AuthUsername,
		"ssl_verify":               c.SSLVerify,
		"ssl_ca_bundle":            c.SSLCABundle,
		"url_allowlist":            c.URLAllowlist,
		"default_submit_args":      c.DefaultSubmitArgs,
		"rate_limit_per_min":       c.RateLimitPerMin,
	}
}