Changeset 0.29.0 (#195)

server/handlers/workflow_instances.go

@@ -3,13 +3,13 @@ package handlers
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"log"
 	"net/http"
 	"time"
 
 	"github.com/gin-gonic/gin"
 
-	"git.gobha.me/xcaliber/chat-switchboard/database"
 	"git.gobha.me/xcaliber/chat-switchboard/events"
 	"git.gobha.me/xcaliber/chat-switchboard/models"
 	"git.gobha.me/xcaliber/chat-switchboard/notifications"
@@ -77,24 +77,14 @@ func (h *WorkflowInstanceHandler) Start(c *gin.Context) {
 
 	// 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)
+	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{
@@ -126,34 +116,15 @@ func (h *WorkflowInstanceHandler) Start(c *gin.Context) {
 
 // ── 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  *string         `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 {
+	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
 	}
-	ws.StageData = stageData
 	c.JSON(http.StatusOK, ws)
 }
 
@@ -186,17 +157,12 @@ func (h *WorkflowInstanceHandler) Advance(c *gin.Context) {
 	}
 	_ = c.ShouldBindJSON(&body)
 
-	mergedData := tools.MergeWorkflowStageData(ctx, channelID, body.Data)
+	mergedData := tools.MergeWorkflowStageData(ctx, h.stores.Channels, 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)
+		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
@@ -216,11 +182,7 @@ func (h *WorkflowInstanceHandler) Advance(c *gin.Context) {
 	}
 
 	// 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)
+	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
@@ -245,7 +207,7 @@ func (h *WorkflowInstanceHandler) Advance(c *gin.Context) {
 
 	// v0.27.0: Assignment + round-robin + WS notifications
 	if nextStageDef.AssignmentTeamID != nil {
-		assignmentID := tools.CreateWorkflowAssignment(ctx, channelID, nextStage, *nextStageDef.AssignmentTeamID)
+		assignmentID := tools.CreateWorkflowAssignment(ctx, h.stores, channelID, nextStage, *nextStageDef.AssignmentTeamID)
 
 		// Round-robin auto-assignment if configured
 		assignedTo := h.tryRoundRobin(ctx, nextStageDef, assignmentID)
@@ -298,9 +260,7 @@ func (h *WorkflowInstanceHandler) Reject(c *gin.Context) {
 	}
 
 	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)
+	err = h.stores.Channels.RejectWorkflowToStage(ctx, channelID, prevStage)
 	if err != nil {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to reject stage"})
 		return
@@ -327,15 +287,14 @@ func (h *WorkflowInstanceHandler) Reject(c *gin.Context) {
 // ── 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, &currentStage, &status)
-	if wfID != nil {
-		workflowID = *wfID
+	ws, err := h.stores.Channels.GetWorkflowStatus(ctx, channelID)
+	if err != nil || ws == nil {
+		return "", 0, "", fmt.Errorf("workflow channel not found")
 	}
-	return
+	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.
@@ -353,20 +312,13 @@ func (h *WorkflowInstanceHandler) emitWorkflowEvent(label, channelID string, dat
 	}
 
 	// Send to all user participants in the channel
-	rows, err := database.DB.Query(database.Q(`
-		SELECT participant_id FROM channel_participants
-		WHERE channel_id = $1 AND participant_type = 'user'
-	`), channelID)
+	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
 	}
-	defer rows.Close()
-	for rows.Next() {
-		var uid string
-		if rows.Scan(&uid) == nil {
-			h.hub.SendToUser(uid, evt)
-		}
+	for _, uid := range pids {
+		h.hub.SendToUser(uid, evt)
 	}
 }
 
@@ -396,52 +348,15 @@ func (h *WorkflowInstanceHandler) tryRoundRobin(ctx context.Context, stage model
 		return ""
 	}
 
-	// Get team members
-	members, err := h.stores.Teams.ListMembers(ctx, *stage.AssignmentTeamID)
-	if err != nil || len(members) == 0 {
-		return ""
-	}
-
-	// Find the least-recently-assigned member.
-	// Query: for each member, find their most recent claimed_at in workflow_assignments.
-	// Pick the member with the oldest (or null) claimed_at.
-	var bestUserID string
-	bestUserID = members[0].UserID // fallback to first member
-
-	rows, err := database.DB.QueryContext(ctx, database.Q(`
-		SELECT m.user_id, COALESCE(MAX(wa.claimed_at), '1970-01-01T00:00:00Z') as last_claim
-		FROM team_members m
-		LEFT JOIN workflow_assignments wa ON wa.assigned_to = m.user_id AND wa.team_id = $1
-		WHERE m.team_id = $2
-		GROUP BY m.user_id
-		ORDER BY last_claim ASC
-		LIMIT 1
-	`), *stage.AssignmentTeamID, *stage.AssignmentTeamID)
-	if err == nil {
-		defer rows.Close()
-		if rows.Next() {
-			var uid string
-			var lastClaim string // COALESCE returns TEXT on both dialects
-			if err := rows.Scan(&uid, &lastClaim); err == nil {
-				bestUserID = uid
-			}
-		}
-	}
-
-	// Claim the assignment for this user
-	now := time.Now().UTC()
-	_, err = database.DB.ExecContext(ctx, database.Q(`
-		UPDATE workflow_assignments
-		SET assigned_to = $1, status = 'claimed', claimed_at = $2
-		WHERE id = $3 AND status = 'unassigned'
-	`), bestUserID, now, assignmentID)
+	assignedTo, err := h.stores.Workflows.TryRoundRobin(ctx, *stage.AssignmentTeamID, assignmentID)
 	if err != nil {
-		log.Printf("[workflow] round-robin: failed to auto-assign %s to %s: %v", assignmentID, bestUserID, err)
+		log.Printf("[workflow] round-robin: failed to auto-assign %s: %v", assignmentID, err)
 		return ""
 	}
-
-	log.Printf("[workflow] round-robin: auto-assigned %s to user %s", assignmentID, bestUserID)
-	return bestUserID
+	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.