Changeset 0.26.0 (#165)
This commit is contained in:
319
server/handlers/workflow_instances.go
Normal file
319
server/handlers/workflow_instances.go
Normal file
@@ -0,0 +1,319 @@
|
||||
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).
|
||||
Reference in New Issue
Block a user