Agent-System/internal/server/admin_workflow_handlers.go

package server

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

	"chat-agent-system/internal/auth"
	"chat-agent-system/internal/workflowdef"
)

func (s *Server) requireWorkflowAdmin(w http.ResponseWriter, r *http.Request) (*auth.UserAccount, bool) {
	user, ok := s.requireSystemAdmin(w, r)
	if !ok {
		return nil, false
	}
	if s.wfDefSvc == nil || !s.wfDefSvc.Available() {
		http.Error(w, "工作流定义服务需要数据库 IAM", http.StatusServiceUnavailable)
		return nil, false
	}
	return user, true
}

func (s *Server) handleAdminWorkflows(w http.ResponseWriter, r *http.Request) {
	user, ok := s.requireWorkflowAdmin(w, r)
	if !ok {
		return
	}
	switch r.Method {
	case http.MethodGet:
		defs, err := s.wfDefSvc.ListDefinitions(r.Context())
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		writeJSONResponse(w, http.StatusOK, map[string]any{"workflows": defs})
	case http.MethodPost:
		var req workflowdef.CreateDefinitionInput
		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
			http.Error(w, "无效的 JSON", http.StatusBadRequest)
			return
		}
		def, ver, err := s.wfDefSvc.CreateDefinition(r.Context(), req, user.ID)
		if err != nil {
			status := http.StatusBadRequest
			if errors.Is(err, workflowdef.ErrValidationFailed) {
				status = http.StatusUnprocessableEntity
			}
			http.Error(w, err.Error(), status)
			return
		}
		writeJSONResponse(w, http.StatusCreated, map[string]any{"definition": def, "version": ver})
	default:
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
	}
}

func (s *Server) handleAdminWorkflowByID(w http.ResponseWriter, r *http.Request) {
	if _, ok := s.requireWorkflowAdmin(w, r); !ok {
		return
	}
	id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
	if err != nil || id <= 0 {
		http.Error(w, "invalid workflow id", http.StatusBadRequest)
		return
	}
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	def, err := s.wfDefSvc.GetDefinition(r.Context(), id)
	if err != nil {
		http.Error(w, err.Error(), http.StatusNotFound)
		return
	}
	versions, err := s.wfDefSvc.ListVersions(r.Context(), id)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	writeJSONResponse(w, http.StatusOK, map[string]any{"definition": def, "versions": versions})
}

func (s *Server) handleAdminWorkflowVersion(w http.ResponseWriter, r *http.Request) {
	user, ok := s.requireWorkflowAdmin(w, r)
	if !ok {
		return
	}
	workflowID, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
	if err != nil || workflowID <= 0 {
		http.Error(w, "invalid workflow id", http.StatusBadRequest)
		return
	}
	version, err := strconv.Atoi(r.PathValue("version"))
	if err != nil || version <= 0 {
		http.Error(w, "invalid version", http.StatusBadRequest)
		return
	}

	switch r.Method {
	case http.MethodPatch, http.MethodPut:
		var body struct {
			Definition workflowdef.VersionDefinition `json:"definition"`
		}
		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
			http.Error(w, "invalid json", http.StatusBadRequest)
			return
		}
		updated, err := s.wfDefSvc.UpdateDraft(r.Context(), workflowID, version, body.Definition)
		if err != nil {
			status := http.StatusBadRequest
			if errors.Is(err, workflowdef.ErrNotDraft) {
				status = http.StatusConflict
			}
			http.Error(w, err.Error(), status)
			return
		}
		writeJSONResponse(w, http.StatusOK, updated)
	default:
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
	}
	_ = user
}

func (s *Server) handleAdminWorkflowValidate(w http.ResponseWriter, r *http.Request) {
	if _, ok := s.requireWorkflowAdmin(w, r); !ok {
		return
	}
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	workflowID, _ := strconv.ParseInt(r.PathValue("id"), 10, 64)
	version, _ := strconv.Atoi(r.PathValue("version"))
	res, err := s.wfDefSvc.ValidateVersion(r.Context(), workflowID, version)
	if err != nil {
		http.Error(w, err.Error(), http.StatusNotFound)
		return
	}
	writeJSONResponse(w, http.StatusOK, res)
}

func (s *Server) handleAdminWorkflowPublish(w http.ResponseWriter, r *http.Request) {
	user, ok := s.requireWorkflowAdmin(w, r)
	if !ok {
		return
	}
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	workflowID, _ := strconv.ParseInt(r.PathValue("id"), 10, 64)
	version, _ := strconv.Atoi(r.PathValue("version"))
	ver, err := s.wfDefSvc.PublishVersion(r.Context(), workflowID, version, user.ID)
	if err != nil {
		status := http.StatusBadRequest
		if errors.Is(err, workflowdef.ErrValidationFailed) {
			status = http.StatusUnprocessableEntity
		}
		http.Error(w, err.Error(), status)
		return
	}
	if dbCat, ok := s.pwSvc.Catalog().(*workflowdef.DBCatalog); ok {
		_ = dbCat.Refresh(r.Context())
	}
	writeJSONResponse(w, http.StatusOK, ver)
}

func (s *Server) handleAdminWorkflowRetire(w http.ResponseWriter, r *http.Request) {
	if _, ok := s.requireWorkflowAdmin(w, r); !ok {
		return
	}
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	workflowID, _ := strconv.ParseInt(r.PathValue("id"), 10, 64)
	version, _ := strconv.Atoi(r.PathValue("version"))
	ver, err := s.wfDefSvc.RetireVersion(r.Context(), workflowID, version)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if dbCat, ok := s.pwSvc.Catalog().(*workflowdef.DBCatalog); ok {
		_ = dbCat.Refresh(r.Context())
	}
	writeJSONResponse(w, http.StatusOK, ver)
}