diff --git a/internal/server/admin_template_handlers.go b/internal/server/admin_template_handlers.go new file mode 100644 index 00000000..81a184f2 --- /dev/null +++ b/internal/server/admin_template_handlers.go @@ -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}) +} diff --git a/internal/server/admin_workflow_handlers.go b/internal/server/admin_workflow_handlers.go new file mode 100644 index 00000000..4c9e6e22 --- /dev/null +++ b/internal/server/admin_workflow_handlers.go @@ -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) +} diff --git a/internal/server/server.go b/internal/server/server.go index 4fc79724..628c86ba 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -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)