kandev/apps/backend/internal/workflow/handlers/handlers.go

package handlers

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"mime"
	"net/http"
	"strings"

	"github.com/gin-gonic/gin"
	"go.uber.org/zap"
	"gopkg.in/yaml.v3"

	"github.com/kandev/kandev/internal/common/logger"
	"github.com/kandev/kandev/internal/events"
	"github.com/kandev/kandev/internal/events/bus"
	taskmodels "github.com/kandev/kandev/internal/task/models"
	"github.com/kandev/kandev/internal/task/repository/repoerrors"
	"github.com/kandev/kandev/internal/workflow/controller"
	"github.com/kandev/kandev/internal/workflow/models"
	"github.com/kandev/kandev/internal/workflow/service"
	"github.com/kandev/kandev/internal/workflow/stepevents"
	ws "github.com/kandev/kandev/pkg/websocket"
)

// stepEventSource labels workflow-step events published by this surface.
const stepEventSource = "workflow-handlers"

// Handlers manages workflow HTTP and WebSocket handlers
type Handlers struct {
	controller *controller.Controller
	eventBus   bus.EventBus
	stepEvents *stepevents.Publisher
	logger     *logger.Logger
}

// NewHandlers creates new workflow handlers
func NewHandlers(ctrl *controller.Controller, eventBus bus.EventBus, log *logger.Logger) *Handlers {
	scoped := log.WithFields(zap.String("component", "workflow-handlers"))
	return &Handlers{
		controller: ctrl,
		eventBus:   eventBus,
		stepEvents: stepevents.NewPublisher(eventBus, stepEventSource, scoped),
		logger:     scoped,
	}
}

// RegisterRoutes registers workflow HTTP and WebSocket handlers
func RegisterRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, ctrl *controller.Controller, eventBus bus.EventBus, log *logger.Logger) {
	handlers := NewHandlers(ctrl, eventBus, log)
	handlers.registerHTTP(router)
	handlers.registerWS(dispatcher)
}

func (h *Handlers) registerHTTP(router *gin.Engine) {
	api := router.Group("/api/v1")

	// Template routes
	api.GET("/workflow/templates", h.httpListTemplates)
	api.GET("/workflow/templates/:id", h.httpGetTemplate)

	// Step routes
	api.GET("/workflows/:id/workflow/steps", h.httpListStepsByWorkflow)
	api.GET("/workspaces/:id/workflow-steps", h.httpListStepsByWorkspace)
	api.GET("/workflow/steps/:id", h.httpGetStep)
	api.POST("/workflows/:id/workflow/steps", h.httpCreateStepsFromTemplate)
	api.POST("/workflow/steps", h.httpCreateStep)
	api.PUT("/workflow/steps/:id", h.httpUpdateStep)
	api.DELETE("/workflow/steps/:id", h.httpDeleteStep)
	api.PUT("/workflows/:id/workflow/steps/reorder", h.httpReorderSteps)

	// Export/Import routes
	api.GET("/workflows/:id/export", h.httpExportWorkflow)
	api.GET("/workspaces/:id/workflows/export", h.httpExportWorkflows)
	api.POST("/workspaces/:id/workflows/import/preview", h.httpPreviewImportWorkflows)
	api.POST("/workspaces/:id/workflows/import", h.httpImportWorkflows)

	// History routes
	api.GET("/sessions/:id/workflow/history", h.httpListHistoryBySession)
}

func (h *Handlers) registerWS(dispatcher *ws.Dispatcher) {
	dispatcher.RegisterFunc(ws.ActionWorkflowTemplateList, h.wsListTemplates)
	dispatcher.RegisterFunc(ws.ActionWorkflowTemplateGet, h.wsGetTemplate)
	dispatcher.RegisterFunc(ws.ActionWorkflowStepList, h.wsListSteps)
	dispatcher.RegisterFunc(ws.ActionWorkflowStepGet, h.wsGetStep)
	dispatcher.RegisterFunc(ws.ActionWorkflowStepCreate, h.wsCreateStepsFromTemplate)
	dispatcher.RegisterFunc(ws.ActionWorkflowHistoryList, h.wsListHistory)
}

// HTTP handlers - Templates

func (h *Handlers) httpListTemplates(c *gin.Context) {
	resp, err := h.controller.ListTemplates(c.Request.Context())
	if err != nil {
		h.logger.Error("failed to list templates", zap.Error(err))
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list templates"})
		return
	}
	c.JSON(http.StatusOK, resp)
}

func (h *Handlers) httpGetTemplate(c *gin.Context) {
	resp, err := h.controller.GetTemplate(c.Request.Context(), c.Param("id"))
	if err != nil {
		h.logger.Error("failed to get template", zap.Error(err))
		c.JSON(http.StatusNotFound, gin.H{"error": "Template not found"})
		return
	}
	c.JSON(http.StatusOK, resp)
}

// HTTP handlers - Steps

func (h *Handlers) httpListStepsByWorkflow(c *gin.Context) {
	resp, err := h.controller.ListStepsByWorkflow(c.Request.Context(), controller.ListStepsRequest{
		WorkflowID: c.Param("id"),
	})
	if err != nil {
		h.logger.Error("failed to list steps", zap.Error(err))
		if writeNotVisible(c, err, "Workflow not found") {
			return
		}
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list steps"})
		return
	}
	c.JSON(http.StatusOK, resp)
}

func (h *Handlers) httpListStepsByWorkspace(c *gin.Context) {
	resp, err := h.controller.ListStepsByWorkspace(c.Request.Context(), c.Param("id"))
	if err != nil {
		h.logger.Error("failed to list steps by workspace", zap.Error(err))
		if writeNotVisible(c, err, "Workspace not found") {
			return
		}
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list steps"})
		return
	}
	c.JSON(http.StatusOK, resp)
}

func (h *Handlers) httpGetStep(c *gin.Context) {
	resp, err := h.controller.GetStep(c.Request.Context(), c.Param("id"))
	if err != nil {
		h.logger.Error("failed to get step", zap.Error(err))
		c.JSON(http.StatusNotFound, gin.H{"error": "Step not found"})
		return
	}
	c.JSON(http.StatusOK, resp)
}

type httpCreateStepsRequest struct {
	TemplateID string `json:"template_id"`
}

func (h *Handlers) httpCreateStepsFromTemplate(c *gin.Context) {
	var req httpCreateStepsRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request"})
		return
	}
	if err := h.controller.CreateStepsFromTemplate(c.Request.Context(), controller.CreateStepsFromTemplateRequest{
		WorkflowID: c.Param("id"),
		TemplateID: req.TemplateID,
	}); err != nil {
		h.logger.Error("failed to create steps from template", zap.Error(err))
		h.writeStepMutationError(c, err)
		return
	}
	c.JSON(http.StatusCreated, gin.H{"success": true})
}

func (h *Handlers) httpCreateStep(c *gin.Context) {
	var req controller.CreateStepRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request: " + err.Error()})
		return
	}
	if req.WorkflowID == "" || req.Name == "" {
		c.JSON(http.StatusBadRequest, gin.H{"error": "workflow_id and name are required"})
		return
	}
	ctx := c.Request.Context()
	resp, err := h.controller.CreateStep(ctx, req)
	if err != nil {
		h.logger.Error("failed to create step", zap.Error(err))
		h.writeStepMutationError(c, err)
		return
	}
	h.publishWorkflowStepEvents(ctx, events.WorkflowStepUpdated, resp.DemotedStartSteps)
	h.publishWorkflowStepEvent(ctx, events.WorkflowStepCreated, resp.Step)
	c.JSON(http.StatusCreated, resp.Step)
}

func (h *Handlers) httpUpdateStep(c *gin.Context) {
	var req controller.UpdateStepRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request: " + err.Error()})
		return
	}
	req.ID = c.Param("id")
	ctx := c.Request.Context()
	resp, err := h.controller.UpdateStep(ctx, req)
	if err != nil {
		h.logger.Error("failed to update step", zap.Error(err))
		h.writeStepMutationError(c, err)
		return
	}
	h.publishWorkflowStepEvents(ctx, events.WorkflowStepUpdated, resp.DemotedStartSteps)
	h.publishWorkflowStepEvent(ctx, events.WorkflowStepUpdated, resp.Step)
	c.JSON(http.StatusOK, resp.Step)
}

// writeNotVisible answers 404 for a workflow, workspace or step the caller may
// not see, and reports whether it handled the error. The same 404 covers a
// resource that does not exist, so neither response reveals which it was.
func writeNotVisible(c *gin.Context, err error, message string) bool {
	if !errors.Is(err, service.ErrNotVisible) {
		return false
	}
	c.JSON(http.StatusNotFound, gin.H{"error": message})
	return true
}

func (h *Handlers) writeStepMutationError(c *gin.Context, err error) {
	if errors.Is(err, service.ErrWorkflowReadOnly) || errors.Is(err, service.ErrWorkflowWorkspaceReadOnly) {
		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
		return
	}
	// A step the caller may not see reads exactly like one that is not there.
	if writeNotVisible(c, err, "Step not found") {
		return
	}
	msg := strings.ToLower(err.Error())
	switch {
	case isStepValidationError(msg):
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
	case strings.Contains(msg, "not found"):
		c.JSON(http.StatusNotFound, gin.H{"error": "Step not found"})
	default:
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to save step"})
	}
}

func isStepValidationError(msg string) bool {
	return strings.Contains(msg, "wip_limit must be non-negative") ||
		strings.Contains(msg, "pull_from_step_id") ||
		strings.Contains(msg, "same workflow") ||
		strings.Contains(msg, "pull cycle")
}

func (h *Handlers) httpDeleteStep(c *gin.Context) {
	ctx := c.Request.Context()
	stepID := c.Param("id")
	stepResp, getErr := h.controller.GetStep(ctx, stepID)
	// A step the caller may not see is the ordinary rejection below, not a
	// failure to read one — warning about it would file every unauthorized
	// delete under infrastructure trouble.
	if getErr != nil && !errors.Is(getErr, service.ErrNotVisible) {
		h.logger.Warn("failed to fetch step before delete; workflow_step.deleted event will not be published",
			zap.String("step_id", stepID), zap.Error(getErr))
	}

	if err := h.controller.DeleteStep(ctx, stepID); err != nil {
		h.logger.Error("failed to delete step", zap.Error(err))
		h.writeStepMutationError(c, err)
		return
	}
	if stepResp != nil {
		h.publishWorkflowStepEvent(ctx, events.WorkflowStepDeleted, stepResp.Step)
	}
	c.JSON(http.StatusOK, gin.H{"success": true})
}

func (h *Handlers) publishWorkflowStepEvent(ctx context.Context, eventType string, step *models.WorkflowStep) {
	h.stepEvents.Publish(ctx, eventType, step)
}

func (h *Handlers) publishWorkflowStepEvents(ctx context.Context, eventType string, steps []*models.WorkflowStep) {
	h.stepEvents.PublishAll(ctx, eventType, steps)
}

type httpReorderStepsRequest struct {
	StepIDs []string `json:"step_ids"`
}

func (h *Handlers) httpReorderSteps(c *gin.Context) {
	var req httpReorderStepsRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request: " + err.Error()})
		return
	}
	if err := h.controller.ReorderSteps(c.Request.Context(), controller.ReorderStepsRequest{
		WorkflowID: c.Param("id"),
		StepIDs:    req.StepIDs,
	}); err != nil {
		h.logger.Error("failed to reorder steps", zap.Error(err))
		h.writeStepMutationError(c, err)
		return
	}
	c.JSON(http.StatusOK, gin.H{"success": true})
}

// HTTP handlers - History

func (h *Handlers) httpListHistoryBySession(c *gin.Context) {
	resp, err := h.controller.ListHistoryBySession(c.Request.Context(), controller.ListHistoryRequest{
		SessionID: c.Param("id"),
	})
	if err != nil {
		h.logger.Error("failed to list history", zap.Error(err))
		// Session IDs are opaque, so denied/missing access is reported as
		// not-found rather than leaking whether another workspace owns the
		// session. Any other error (e.g. a repository read failure) is a
		// genuine server error and must not be masked as not-found.
		if errors.Is(err, taskmodels.ErrTaskSessionNotFound) || errors.Is(err, repoerrors.ErrTaskNotFound) {
			c.JSON(http.StatusNotFound, gin.H{"error": "session not found"})
			return
		}
		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list history"})
		return
	}
	c.JSON(http.StatusOK, resp)
}

// HTTP handlers - Export/Import

func (h *Handlers) httpExportWorkflow(c *gin.Context) {
	resp, err := h.controller.ExportWorkflow(c.Request.Context(), c.Param("id"))
	if err != nil {
		h.logger.Error("failed to export workflow", zap.Error(err))
		if writeNotVisible(c, err, "Workflow not found") {
			return
		}
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to export workflow"})
		return
	}
	h.respondYAML(c, resp)
}

func (h *Handlers) httpExportWorkflows(c *gin.Context) {
	resp, err := h.controller.ExportWorkflows(c.Request.Context(), c.Param("id"), parseExportIDs(c))
	if err != nil {
		h.logger.Error("failed to export workflows", zap.Error(err))
		if writeNotVisible(c, err, "Workspace not found") {
			return
		}
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to export workflows"})
		return
	}
	h.respondYAML(c, resp)
}

// parseExportIDs reads the optional comma-separated `ids` query param. It
// returns nil when the param is absent (export all, back-compat) and a non-nil
// slice (possibly empty) when present, restricting the export to those IDs.
func parseExportIDs(c *gin.Context) []string {
	raw, ok := c.GetQuery("ids")
	if !ok {
		return nil
	}
	ids := []string{}
	for _, part := range strings.Split(raw, ",") {
		if trimmed := strings.TrimSpace(part); trimmed != "" {
			ids = append(ids, trimmed)
		}
	}
	return ids
}

func (h *Handlers) httpImportWorkflows(c *gin.Context) {
	body, tooLarge, err := readImportBody(c)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Failed to read request body"})
		return
	}
	if tooLarge {
		c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": "Import request exceeds the 1 MiB limit"})
		return
	}

	request, err := parseImportWorkflowsRequest(c, body)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": capitalizeImportError(err.Error())})
		return
	}

	resp, err := h.controller.ImportWorkflows(c.Request.Context(), request)
	if err != nil {
		h.writeImportError(c, err)
		return
	}
	c.JSON(http.StatusOK, resp)
}

func parseImportWorkflowsRequest(c *gin.Context, body []byte) (controller.ImportWorkflowsRequest, error) {
	request := controller.ImportWorkflowsRequest{WorkspaceID: c.Param("id")}
	if importMediaType(c) != "application/json" {
		var data models.WorkflowExport
		if err := yaml.Unmarshal(body, &data); err != nil {
			return request, fmt.Errorf("invalid YAML: %w", err)
		}
		request.Data = &data
		return request, nil
	}

	var envelope struct {
		YAML                string                         `json:"yaml"`
		StepProfileBindings []service.ImportProfileBinding `json:"step_profile_bindings"`
	}
	if err := json.Unmarshal(body, &envelope); err != nil {
		return request, fmt.Errorf("invalid JSON: %w", err)
	}
	if strings.TrimSpace(envelope.YAML) == "" {
		return request, errors.New("the import YAML is required")
	}
	var data models.WorkflowExport
	if err := yaml.Unmarshal([]byte(envelope.YAML), &data); err != nil {
		return request, fmt.Errorf("invalid YAML: %w", err)
	}
	if envelope.StepProfileBindings == nil {
		envelope.StepProfileBindings = []service.ImportProfileBinding{}
	}
	request.Data = &data
	request.StepProfileBindings = envelope.StepProfileBindings
	return request, nil
}

func capitalizeImportError(message string) string {
	if message == "" {
		return message
	}
	return strings.ToUpper(message[:1]) + message[1:]
}

func (h *Handlers) httpPreviewImportWorkflows(c *gin.Context) {
	body, tooLarge, err := readImportBody(c)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Failed to read request body"})
		return
	}
	if tooLarge {
		c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": "Import request exceeds the 1 MiB limit"})
		return
	}
	var data models.WorkflowExport
	if err := yaml.Unmarshal(body, &data); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid YAML: " + err.Error()})
		return
	}
	resp, err := h.controller.PreviewImportWorkflows(c.Request.Context(), c.Param("id"), &data)
	if err != nil {
		h.writeImportError(c, err)
		return
	}
	c.JSON(http.StatusOK, resp)
}

const maxImportSize = 1 << 20

func readImportBody(c *gin.Context) ([]byte, bool, error) {
	body, err := io.ReadAll(io.LimitReader(c.Request.Body, maxImportSize+1))
	if err != nil {
		return nil, false, err
	}
	return body, len(body) > maxImportSize, nil
}

func importMediaType(c *gin.Context) string {
	mediaType, _, err := mime.ParseMediaType(c.GetHeader("Content-Type"))
	if err != nil {
		return strings.ToLower(strings.TrimSpace(strings.Split(c.GetHeader("Content-Type"), ";")[0]))
	}
	return strings.ToLower(mediaType)
}

func (h *Handlers) writeImportError(c *gin.Context, err error) {
	if writeNotVisible(c, err, "Workspace not found") {
		return
	}
	var resolutionErr *service.ImportProfileResolutionError
	if errors.As(err, &resolutionErr) {
		c.JSON(http.StatusConflict, gin.H{
			"code":  "workflow_import_profiles_required",
			"error": resolutionErr.Error(),
			"steps": resolutionErr.Conflicts,
		})
		return
	}
	var bindingsErr *service.InvalidImportProfileBindingsError
	if errors.As(err, &bindingsErr) {
		c.JSON(http.StatusBadRequest, gin.H{"code": "invalid_workflow_import_profile_bindings", "error": bindingsErr.Error()})
		return
	}
	if errors.Is(err, service.ErrImportProfileCatalogUnavailable) {
		h.logger.Error("failed to import workflows", zap.Error(err))
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Import profile catalog is unavailable"})
		return
	}
	h.logger.Error("failed to import workflows", zap.Error(err))
	c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
}

// respondYAML marshals the value as YAML and writes it to the response.
func (h *Handlers) respondYAML(c *gin.Context, v any) {
	data, err := yaml.Marshal(v)
	if err != nil {
		h.logger.Error("failed to marshal YAML", zap.Error(err))
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to marshal response"})
		return
	}
	c.Data(http.StatusOK, "application/x-yaml", data)
}

// WS handlers - Templates

func (h *Handlers) wsListTemplates(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
	resp, err := h.controller.ListTemplates(ctx)
	if err != nil {
		h.logger.Error("failed to list templates", zap.Error(err))
		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list templates", nil)
	}
	return ws.NewResponse(msg.ID, msg.Action, resp)
}

func (h *Handlers) wsGetTemplate(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
	return h.wsGetByID(ctx, msg, "failed to get template", "Template not found",
		func(ctx context.Context, id string) (any, error) {
			return h.controller.GetTemplate(ctx, id)
		})
}

// WS handlers - Steps

type wsListStepsRequest struct {
	WorkflowID string `json:"workflow_id"`
}

func (h *Handlers) wsListSteps(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
	var req wsListStepsRequest
	if err := msg.ParsePayload(&req); err != nil {
		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
	}
	return h.wsHandleStringField(ctx, msg, req.WorkflowID, "workflow_id",
		"failed to list steps", "Failed to list steps",
		func(ctx context.Context, workflowID string) (any, error) {
			return h.controller.ListStepsByWorkflow(ctx, controller.ListStepsRequest{WorkflowID: workflowID})
		})
}

func (h *Handlers) wsGetStep(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
	return h.wsGetByID(ctx, msg, "failed to get step", "Step not found",
		func(ctx context.Context, id string) (any, error) {
			return h.controller.GetStep(ctx, id)
		})
}

type wsCreateStepsRequest struct {
	WorkflowID string `json:"workflow_id"`
	TemplateID string `json:"template_id"`
}

func (h *Handlers) wsCreateStepsFromTemplate(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
	var req wsCreateStepsRequest
	if err := msg.ParsePayload(&req); err != nil {
		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
	}
	if req.WorkflowID == "" || req.TemplateID == "" {
		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "workflow_id and template_id are required", nil)
	}
	if err := h.controller.CreateStepsFromTemplate(ctx, controller.CreateStepsFromTemplateRequest{
		WorkflowID: req.WorkflowID,
		TemplateID: req.TemplateID,
	}); err != nil {
		h.logger.Error("failed to create steps", zap.Error(err))
		if errors.Is(err, service.ErrNotVisible) {
			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, notFoundMessage, nil)
		}
		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to create steps", nil)
	}
	return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"success": true})
}

// WS handlers - History

type wsListHistoryRequest struct {
	SessionID string `json:"session_id"`
}

func (h *Handlers) wsListHistory(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
	var req wsListHistoryRequest
	if err := msg.ParsePayload(&req); err != nil {
		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
	}
	return h.wsHandleStringField(ctx, msg, req.SessionID, "session_id",
		"failed to list history", "Failed to list history",
		func(ctx context.Context, sessionID string) (any, error) {
			return h.controller.ListHistoryBySession(ctx, controller.ListHistoryRequest{SessionID: sessionID})
		})
}