package handlers import ( "context" "encoding/json" "fmt" "log" "net/http" "time" "github.com/gin-gonic/gin" "chat-switchboard/events" "chat-switchboard/models" "chat-switchboard/notifications" "chat-switchboard/sandbox" "chat-switchboard/store" "chat-switchboard/tools" "chat-switchboard/workflow" ) // ── Workflow Instance Handler ─────────────── // Manages the runtime lifecycle of workflow channels: starting instances, // advancing/rejecting stages, and querying status. type WorkflowInstanceHandler struct { stores store.Stores hub *events.Hub notifSvc *notifications.Service runner *sandbox.Runner } func NewWorkflowInstanceHandler(stores store.Stores, hub *events.Hub, notifSvc *notifications.Service, runner *sandbox.Runner) *WorkflowInstanceHandler { return &WorkflowInstanceHandler{stores: stores, hub: hub, notifSvc: notifSvc, runner: runner} } // ── Start ─────────────────────────────────── // Start creates a new workflow channel from a published workflow version. // POST /api/v1/workflows/:id/start func (h *WorkflowInstanceHandler) Start(c *gin.Context) { ctx := c.Request.Context() wfID := c.Param("id") userID := c.GetString("user_id") wf, err := h.stores.Workflows.GetByID(ctx, wfID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"}) return } if !wf.IsActive { c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is not active"}) return } ver, err := h.stores.Workflows.GetLatestVersion(ctx, wfID) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "workflow has no published version — publish first"}) return } stages, err := h.stores.Workflows.ListStages(ctx, wfID) if err != nil || len(stages) == 0 { c.JSON(http.StatusBadRequest, gin.H{"error": "workflow has no stages"}) return } // Create the workflow channel ch := &models.Channel{ UserID: userID, Title: wf.Name, Description: wf.Description, Type: "workflow", TeamID: wf.TeamID, } if err := h.stores.Channels.Create(ctx, ch); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create channel: " + err.Error()}) return } // Set workflow-specific columns (not part of base Channel.Create) allowAnon := wf.EntryMode == "public_link" err = h.stores.Channels.SetWorkflowInstance(ctx, ch.ID, wfID, ver.VersionNumber, []byte("{}"), "active") if err != nil { log.Printf("Failed to set workflow columns on channel %s: %v", ch.ID, err) } _ = h.stores.Channels.Update(ctx, ch.ID, map[string]interface{}{ "allow_anonymous": allowAnon, "ai_mode": "auto", }) // Add caller as channel owner _ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{ ChannelID: ch.ID, ParticipantType: "user", ParticipantID: userID, Role: "owner", }) // Bind stage 0 persona as participant firstStage := stages[0] if firstStage.PersonaID != nil { _ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{ ChannelID: ch.ID, ParticipantType: "persona", ParticipantID: *firstStage.PersonaID, Role: "member", }) } c.JSON(http.StatusCreated, gin.H{ "channel_id": ch.ID, "workflow_id": wfID, "workflow_version": ver.VersionNumber, "current_stage": 0, "stage": firstStage, }) } // ── Status ────────────────────────────────── // GetStatus returns the workflow state for a channel. // GET /api/v1/channels/:id/workflow/status func (h *WorkflowInstanceHandler) GetStatus(c *gin.Context) { channelID := c.Param("id") ws, err := h.stores.Channels.GetWorkflowStatus(c.Request.Context(), channelID) if err != nil || ws == nil { c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"}) return } c.JSON(http.StatusOK, ws) } // ── Advance ───────────────────────────────── // Advance moves the workflow to the next stage. // POST /api/v1/channels/:id/workflow/advance func (h *WorkflowInstanceHandler) Advance(c *gin.Context) { ctx := c.Request.Context() channelID := c.Param("id") workflowID, currentStage, status, err := h.readWorkflowState(ctx, channelID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"}) return } if status != "active" { c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + status + ", cannot advance"}) return } stages, err := h.stores.Workflows.ListStages(ctx, workflowID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load stages"}) return } var body struct { Data json.RawMessage `json:"data,omitempty"` } _ = c.ShouldBindJSON(&body) mergedData := tools.MergeWorkflowStageData(ctx, h.stores.Channels, channelID, body.Data) nextStage, err := workflow.ResolveNextStage(stages, currentStage, json.RawMessage(mergedData)) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "routing error: " + err.Error()}) return } if nextStage >= len(stages) { // Workflow complete err = h.stores.Channels.CompleteWorkflow(ctx, channelID, nextStage, json.RawMessage(mergedData)) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to complete workflow"}) return } tools.CreateWorkflowStageNote(ctx, h.stores, channelID, currentStage, body.Data, "") // v0.27.0: Emit workflow.completed WS event h.emitWorkflowEvent("workflow.completed", channelID, map[string]any{ "channel_id": channelID, "workflow_id": workflowID, "stage": nextStage, }) // v0.27.0: on_complete chaining — trigger target workflow if configured h.triggerOnComplete(ctx, workflowID, channelID, mergedData) c.JSON(http.StatusOK, gin.H{"status": "completed", "current_stage": nextStage}) return } // Advance to next stage err = h.stores.Channels.AdvanceWorkflowStage(ctx, channelID, nextStage, json.RawMessage(mergedData)) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to advance stage"}) return } tools.CreateWorkflowStageNote(ctx, h.stores, channelID, currentStage, body.Data, "") // v0.35.0: Fire on_advance hook (can enrich stage_data) currentStageDef := stages[currentStage] if hookResult := FireOnAdvanceHook(ctx, h.stores, h.runner, currentStageDef.TransitionRules, channelID, currentStage, nextStage, json.RawMessage(mergedData)); hookResult != nil { if hookResult.Error != "" { log.Printf("[workflow] on_advance hook rejected: %s", hookResult.Error) } else if hookResult.EnrichedData != nil { _ = h.stores.Channels.AdvanceWorkflowStage(ctx, channelID, nextStage, hookResult.EnrichedData) } } nextStageDef := stages[nextStage] // Bind next persona (skip for form_only — no LLM needed) if nextStageDef.StageMode != models.StageModeFormOnly && nextStageDef.PersonaID != nil { alreadyIn, _ := h.stores.Channels.IsParticipant(ctx, channelID, "persona", *nextStageDef.PersonaID) if !alreadyIn { _ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{ ChannelID: channelID, ParticipantType: "persona", ParticipantID: *nextStageDef.PersonaID, Role: "member", }) } } // v0.27.0: Assignment + round-robin + WS notifications if nextStageDef.AssignmentTeamID != nil { assignmentID := tools.CreateWorkflowAssignment(ctx, h.stores, channelID, nextStage, *nextStageDef.AssignmentTeamID) // Round-robin auto-assignment if configured assignedTo := h.tryRoundRobin(ctx, nextStageDef, assignmentID) // Notify team members about new assignment h.notifyAssignment(ctx, *nextStageDef.AssignmentTeamID, channelID, nextStageDef.Name, assignedTo) } // v0.27.0: Emit workflow.advanced WS event h.emitWorkflowEvent("workflow.advanced", channelID, map[string]any{ "channel_id": channelID, "workflow_id": workflowID, "stage": nextStage, "stage_name": nextStageDef.Name, }) c.JSON(http.StatusOK, gin.H{ "status": "active", "current_stage": nextStage, "stage": nextStageDef, }) } // ── Reject ────────────────────────────────── // Reject returns the workflow to the previous stage with a reason. // POST /api/v1/channels/:id/workflow/reject func (h *WorkflowInstanceHandler) Reject(c *gin.Context) { ctx := c.Request.Context() channelID := c.Param("id") var body struct { Reason string `json:"reason"` } if err := c.ShouldBindJSON(&body); err != nil || body.Reason == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "reason is required"}) return } _, currentStage, status, err := h.readWorkflowState(ctx, channelID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"}) return } if status != "active" { c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + status + ", cannot reject"}) return } if currentStage <= 0 { c.JSON(http.StatusBadRequest, gin.H{"error": "cannot reject from the first stage"}) return } prevStage := currentStage - 1 err = h.stores.Channels.RejectWorkflowToStage(ctx, channelID, prevStage) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to reject stage"}) return } // Persist rejection as system message if h.stores.Messages != nil { _ = h.stores.Messages.Create(ctx, &models.Message{ ChannelID: channelID, Role: "system", Content: "Stage rejected: " + body.Reason, ParticipantType: "user", ParticipantID: c.GetString("user_id"), }) } c.JSON(http.StatusOK, gin.H{ "status": "active", "current_stage": prevStage, "reason": body.Reason, }) } // ── Cancel (v0.37.15) ──────────────────────── // CancelInstance cancels a workflow instance and all its open assignments. // POST /api/v1/channels/:id/workflow/cancel func (h *WorkflowInstanceHandler) CancelInstance(c *gin.Context) { ctx := c.Request.Context() channelID := c.Param("id") userID := c.GetString("user_id") role, _ := c.Get("role") workflowID, _, status, err := h.readWorkflowState(ctx, channelID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"}) return } if status != "active" { c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + status + ", cannot cancel"}) return } // Auth: instance owner, team admin, or global admin if role != "admin" { ch, err := h.stores.Channels.GetByID(ctx, channelID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) return } isOwner := ch.UserID == userID isTeamAdmin := false if ch.TeamID != nil && h.stores.Teams != nil { isTeamAdmin, _ = h.stores.Teams.IsTeamAdmin(ctx, *ch.TeamID, userID) } if !isOwner && !isTeamAdmin { c.JSON(http.StatusForbidden, gin.H{"error": "only instance owner, team admin, or global admin can cancel"}) return } } // Cancel the workflow instance if err := h.stores.Channels.CancelWorkflow(ctx, channelID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to cancel workflow"}) return } // Cancel all open assignments cancelled, _ := h.stores.Workflows.CancelAssignmentsForChannel(ctx, channelID) // Emit WS event h.emitWorkflowEvent("workflow.cancelled", channelID, map[string]any{ "channel_id": channelID, "workflow_id": workflowID, "cancelled_by": userID, "assignments_cancelled": cancelled, }) c.JSON(http.StatusOK, gin.H{ "cancelled": true, "channel_id": channelID, "assignments_cancelled": cancelled, }) } // CancelTeamInstance cancels a workflow instance via the team-admin monitor. // POST /api/v1/teams/:teamId/workflows/monitor/instances/:channelId/cancel func (h *WorkflowInstanceHandler) CancelTeamInstance(c *gin.Context) { teamID := c.Param("teamId") channelID := c.Param("channelId") userID := c.GetString("user_id") role, _ := c.Get("role") ctx := c.Request.Context() // Auth: team admin or global admin if role != "admin" { if h.stores.Teams == nil { c.JSON(http.StatusForbidden, gin.H{"error": "team admin access required"}) return } isTA, _ := h.stores.Teams.IsTeamAdmin(ctx, teamID, userID) if !isTA { c.JSON(http.StatusForbidden, gin.H{"error": "team admin access required"}) return } } // Verify the channel belongs to this team ch, err := h.stores.Channels.GetByID(ctx, channelID) if err != nil || ch.TeamID == nil || *ch.TeamID != teamID { c.JSON(http.StatusNotFound, gin.H{"error": "workflow instance not found for this team"}) return } _, _, status, err := h.readWorkflowState(ctx, channelID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"}) return } if status != "active" { c.JSON(http.StatusBadRequest, gin.H{"error": "workflow is " + status + ", cannot cancel"}) return } if err := h.stores.Channels.CancelWorkflow(ctx, channelID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to cancel workflow"}) return } cancelled, _ := h.stores.Workflows.CancelAssignmentsForChannel(ctx, channelID) h.emitWorkflowEvent("workflow.cancelled", channelID, map[string]any{ "channel_id": channelID, "cancelled_by": userID, "assignments_cancelled": cancelled, }) c.JSON(http.StatusOK, gin.H{ "cancelled": true, "channel_id": channelID, "assignments_cancelled": cancelled, }) } // ── Helpers ───────────────────────────────── func (h *WorkflowInstanceHandler) readWorkflowState(ctx context.Context, channelID string) (workflowID string, currentStage int, status string, err error) { ws, err := h.stores.Channels.GetWorkflowStatus(ctx, channelID) if err != nil || ws == nil { return "", 0, "", fmt.Errorf("workflow channel not found") } if ws.WorkflowID != nil { workflowID = *ws.WorkflowID } return workflowID, ws.CurrentStage, ws.Status, nil } // emitWorkflowEvent pushes a workflow event to all user participants in the channel. // Uses PublishToUser (not room-scoped Bus.Publish) because room subscriptions are // not yet wired on the client side. See websocket.md § Room Model. func (h *WorkflowInstanceHandler) emitWorkflowEvent(label, channelID string, data map[string]any) { if h.hub == nil { return } payload, _ := json.Marshal(data) evt := events.Event{ Label: label, Payload: payload, Ts: time.Now().UnixMilli(), } // Send to all user participants in the channel pids, err := h.stores.Channels.ListUserParticipantIDs(context.Background(), channelID, "") if err != nil { log.Printf("[ws] %s: failed to query participants for channel %s: %v", label, channelID[:min(8, len(channelID))], err) return } for _, uid := range pids { h.hub.PublishToUser(uid, evt) } } // triggerOnComplete checks if the workflow has an on_complete chain config // and starts the target workflow if so. func (h *WorkflowInstanceHandler) triggerOnComplete(ctx context.Context, workflowID, channelID, mergedData string) { // v0.27.3: Delegate to shared implementation (also handles webhooks) go tools.TriggerWorkflowOnComplete(ctx, h.stores, workflowID, channelID, mergedData) } // tryRoundRobin checks if the stage has auto_assign:"round_robin" in // transition_rules and assigns the newly created assignment to the // least-recently-assigned team member. Returns the assigned user ID or "". func (h *WorkflowInstanceHandler) tryRoundRobin(ctx context.Context, stage models.WorkflowStage, assignmentID string) string { if assignmentID == "" || stage.AssignmentTeamID == nil { return "" } // Check transition_rules for auto_assign var rules struct { AutoAssign string `json:"auto_assign"` } if len(stage.TransitionRules) > 0 { _ = json.Unmarshal(stage.TransitionRules, &rules) } if rules.AutoAssign != "round_robin" { return "" } assignedTo, err := h.stores.Workflows.TryRoundRobin(ctx, *stage.AssignmentTeamID, assignmentID) if err != nil { log.Printf("[workflow] round-robin: failed to auto-assign %s: %v", assignmentID, err) return "" } if assignedTo != "" { log.Printf("[workflow] round-robin: auto-assigned %s to user %s", assignmentID, assignedTo) } return assignedTo } // notifyAssignment sends notifications to team members about a new workflow assignment. func (h *WorkflowInstanceHandler) notifyAssignment(ctx context.Context, teamID, channelID, stageName, assignedTo string) { if h.notifSvc == nil || h.stores.Teams == nil { return } members, err := h.stores.Teams.ListMembers(ctx, teamID) if err != nil { return } for _, m := range members { title := "New workflow assignment" body := "Stage '" + stageName + "' needs review" if assignedTo != "" && m.UserID == assignedTo { title = "Workflow assigned to you" body = "Stage '" + stageName + "' has been assigned to you (round-robin)" } n := &models.Notification{ UserID: m.UserID, Type: "workflow.assigned", Title: title, Body: body, ResourceType: models.ResourceTypeChannel, ResourceID: channelID, } if err := h.notifSvc.Notify(ctx, n); err != nil { log.Printf("[workflow] notify assignment: %v", err) } } // Also emit targeted WS event for immediate UI update if h.hub != nil { payload, _ := json.Marshal(map[string]any{ "channel_id": channelID, "team_id": teamID, "stage_name": stageName, "assigned_to": assignedTo, }) for _, m := range members { h.hub.PublishToUser(m.UserID, events.Event{ Label: "workflow.assigned", Payload: payload, Ts: time.Now().UnixMilli(), }) } } } // createStageNote and mergeStageData are now in tools/workflow.go // (shared between handler and workflow_advance tool).