package handlers import ( "context" "encoding/json" "log" "net/http" "time" "github.com/gin-gonic/gin" "git.gobha.me/xcaliber/chat-switchboard/database" "git.gobha.me/xcaliber/chat-switchboard/models" "git.gobha.me/xcaliber/chat-switchboard/store" "git.gobha.me/xcaliber/chat-switchboard/tools" ) // ── Workflow Instance Handler ─────────────── // Manages the runtime lifecycle of workflow channels: starting instances, // advancing/rejecting stages, and querying status. type WorkflowInstanceHandler struct { stores store.Stores } func NewWorkflowInstanceHandler(stores store.Stores) *WorkflowInstanceHandler { return &WorkflowInstanceHandler{stores: stores} } // ── 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" var allowAnonVal interface{} = allowAnon if database.CurrentDialect == database.DialectSQLite { if allowAnon { allowAnonVal = 1 } else { allowAnonVal = 0 } } _, err = database.DB.ExecContext(ctx, database.Q(` UPDATE channels SET workflow_id = $1, workflow_version = $2, current_stage = 0, stage_data = '{}', workflow_status = 'active', last_activity_at = $3, allow_anonymous = $4, ai_mode = 'auto' WHERE id = $5 `), wfID, ver.VersionNumber, time.Now().UTC(), allowAnonVal, ch.ID) if err != nil { log.Printf("Failed to set workflow columns on channel %s: %v", ch.ID, err) } // 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 ────────────────────────────────── // WorkflowChannelStatus holds the runtime state of a workflow instance. type WorkflowChannelStatus struct { WorkflowID *string `json:"workflow_id"` WorkflowVersion *int `json:"workflow_version"` CurrentStage int `json:"current_stage"` StageData json.RawMessage `json:"stage_data"` Status string `json:"status"` LastActivityAt *time.Time `json:"last_activity_at"` } // 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") var ws WorkflowChannelStatus var stageData []byte err := database.DB.QueryRowContext(c.Request.Context(), database.Q(` SELECT workflow_id, workflow_version, current_stage, COALESCE(stage_data, '{}'), COALESCE(workflow_status, 'active'), last_activity_at FROM channels WHERE id = $1 AND type = 'workflow' `), channelID).Scan(&ws.WorkflowID, &ws.WorkflowVersion, &ws.CurrentStage, &stageData, &ws.Status, &ws.LastActivityAt) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "workflow channel not found"}) return } ws.StageData = stageData 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, channelID, body.Data) nextStage := currentStage + 1 if nextStage >= len(stages) { // Workflow complete _, err = database.DB.ExecContext(ctx, database.Q(` UPDATE channels SET current_stage = $1, workflow_status = 'completed', stage_data = $2, last_activity_at = $3, ai_mode = 'off' WHERE id = $4 `), nextStage, mergedData, time.Now().UTC(), channelID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to complete workflow"}) return } tools.CreateWorkflowStageNote(ctx, h.stores, channelID, currentStage, body.Data, "") c.JSON(http.StatusOK, gin.H{"status": "completed", "current_stage": nextStage}) return } // Advance to next stage _, err = database.DB.ExecContext(ctx, database.Q(` UPDATE channels SET current_stage = $1, stage_data = $2, last_activity_at = $3 WHERE id = $4 `), nextStage, mergedData, time.Now().UTC(), channelID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to advance stage"}) return } tools.CreateWorkflowStageNote(ctx, h.stores, channelID, currentStage, body.Data, "") nextStageDef := stages[nextStage] // Bind next persona if 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", }) } } // Notify assignment team (stub — fully wired in v0.26.4) if nextStageDef.AssignmentTeamID != nil { tools.CreateWorkflowAssignment(ctx, channelID, nextStage, *nextStageDef.AssignmentTeamID) log.Printf("Workflow stage '%s' assigned to team %s (channel %s)", nextStageDef.Name, *nextStageDef.AssignmentTeamID, channelID) } 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 = database.DB.ExecContext(ctx, database.Q(` UPDATE channels SET current_stage = $1, last_activity_at = $2 WHERE id = $3 `), prevStage, time.Now().UTC(), channelID) 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, }) } // ── Helpers ───────────────────────────────── func (h *WorkflowInstanceHandler) readWorkflowState(ctx context.Context, channelID string) (workflowID string, currentStage int, status string, err error) { var wfID *string err = database.DB.QueryRowContext(ctx, database.Q(` SELECT workflow_id, COALESCE(current_stage, 0), COALESCE(workflow_status, 'active') FROM channels WHERE id = $1 AND type = 'workflow' `), channelID).Scan(&wfID, ¤tStage, &status) if wfID != nil { workflowID = *wfID } return } // createStageNote and mergeStageData are now in tools/workflow.go // (shared between handler and workflow_advance tool).