feat(api): add admin workflow and template management endpoints

Expose draft/validate/publish/retire workflow APIs and template upload/list routes wired to workflowdef service with DBCatalog refresh on publish. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-12 02:53:09 +00:00
parent 7951b0f0b5
commit 79ce1eb56f
3 changed files with 286 additions and 0 deletions
@@ -0,0 +1,83 @@
+package server
+
+import (
+	"io"
+	"net/http"
+	"strings"
+
+	"chat-agent-system/internal/workflowdef"
+)
+
+func (s *Server) handleAdminTemplates(w http.ResponseWriter, r *http.Request) {
+	user, ok := s.requireWorkflowAdmin(w, r)
+	if !ok {
+		return
+	}
+
+	switch r.Method {
+	case http.MethodGet:
+		assets, err := s.wfDefSvc.ListTemplates(r.Context())
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		writeJSONResponse(w, http.StatusOK, map[string]any{"templates": assets})
+	case http.MethodPost:
+		if err := r.ParseMultipartForm(32 << 20); err != nil {
+			http.Error(w, "invalid multipart", http.StatusBadRequest)
+			return
+		}
+		assetKey := strings.TrimSpace(r.FormValue("asset_key"))
+		name := strings.TrimSpace(r.FormValue("name"))
+		if assetKey == "" {
+			assetKey = strings.TrimSpace(r.FormValue("key"))
+		}
+		if name == "" {
+			name = assetKey
+		}
+		file, header, err := r.FormFile("file")
+		if err != nil || assetKey == "" {
+			http.Error(w, "asset_key and file required", http.StatusBadRequest)
+			return
+		}
+		defer file.Close()
+		data, err := io.ReadAll(io.LimitReader(file, 32<<20))
+		if err != nil {
+			http.Error(w, "read file failed", http.StatusBadRequest)
+			return
+		}
+		tv, err := s.wfDefSvc.UploadTemplate(r.Context(), assetKey, name, data, header.Filename, user.ID)
+		if err != nil {
+			status := http.StatusBadRequest
+			if err == workflowdef.ErrNotAvailable {
+				status = http.StatusServiceUnavailable
+			}
+			http.Error(w, err.Error(), status)
+			return
+		}
+		writeJSONResponse(w, http.StatusCreated, tv)
+	default:
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+	}
+}
+
+func (s *Server) handleAdminTemplateVersions(w http.ResponseWriter, r *http.Request) {
+	if _, ok := s.requireWorkflowAdmin(w, r); !ok {
+		return
+	}
+	if r.Method != http.MethodGet {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	assetKey := strings.TrimSpace(r.PathValue("key"))
+	if assetKey == "" {
+		http.Error(w, "invalid asset key", http.StatusBadRequest)
+		return
+	}
+	versions, err := s.wfDefSvc.ListTemplateVersions(r.Context(), assetKey)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	writeJSONResponse(w, http.StatusOK, map[string]any{"versions": versions})
+}
@@ -0,0 +1,189 @@
+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)
+}
@@ -22,6 +22,7 @@ import (
 	"chat-agent-system/internal/observability"
 	"chat-agent-system/internal/projectworkflow"
 	"chat-agent-system/internal/taskrun"
+	"chat-agent-system/internal/workflowdef"
 	"chat-agent-system/internal/workspace"
 )

@@ -38,6 +39,7 @@ type Server struct {
 	taskRuns   *taskrun.Manager
 	authorizer *authz.Authorizer
 	authzSvc   *authz.Service
+	wfDefSvc   *workflowdef.Service
 	resolver   *workspace.Resolver
 	httpServer *http.Server
 }
@@ -73,6 +75,7 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
 	}

 	authzSvc := authz.NewService(pool)
+	wfDefSvc := workflowdef.NewService(pool, svc.CAS())
 	s := &Server{
 		cfg:        cfg,
 		svc:        svc,
@@ -84,6 +87,7 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
 		taskRuns:   taskrun.NewManager(pwSvc, svc.ContextManager()),
 		authorizer: authz.NewAuthorizer(pool),
 		authzSvc:   authzSvc,
+		wfDefSvc:   wfDefSvc,
 		resolver:   workspace.NewResolver(cfg.DataDir, pool),
 	}
 	s.initTaskRunExecutors()
@@ -214,6 +218,16 @@ func (s *Server) Routes() http.Handler {
 	mux.HandleFunc("POST /api/v1/admin/roles", s.handleAdminRoles)
 	mux.HandleFunc("PATCH /api/v1/admin/roles/{id}", s.handleAdminRoleByID)
 	mux.HandleFunc("DELETE /api/v1/admin/roles/{id}", s.handleAdminRoleByID)
+	mux.HandleFunc("GET /api/v1/admin/workflows", s.handleAdminWorkflows)
+	mux.HandleFunc("POST /api/v1/admin/workflows", s.handleAdminWorkflows)
+	mux.HandleFunc("GET /api/v1/admin/workflows/{id}", s.handleAdminWorkflowByID)
+	mux.HandleFunc("PATCH /api/v1/admin/workflows/{id}/versions/{version}", s.handleAdminWorkflowVersion)
+	mux.HandleFunc("POST /api/v1/admin/workflows/{id}/versions/{version}/validate", s.handleAdminWorkflowValidate)
+	mux.HandleFunc("POST /api/v1/admin/workflows/{id}/versions/{version}/publish", s.handleAdminWorkflowPublish)
+	mux.HandleFunc("POST /api/v1/admin/workflows/{id}/versions/{version}/retire", s.handleAdminWorkflowRetire)
+	mux.HandleFunc("GET /api/v1/admin/templates", s.handleAdminTemplates)
+	mux.HandleFunc("POST /api/v1/admin/templates", s.handleAdminTemplates)
+	mux.HandleFunc("GET /api/v1/admin/templates/{key}/versions", s.handleAdminTemplateVersions)
 	mux.HandleFunc("GET /api/v1/admin/users", s.handleAdminUsers)
 	mux.HandleFunc("POST /api/v1/admin/users", s.handleAdminUsers)
 	mux.HandleFunc("PUT /api/v1/admin/users/{id}/role", s.handleAdminUpdateRole)