Changeset 0.27.1 (#167)

2026-03-10 17:43:29 +00:00
parent 7e4f1581f2
commit 41be9d6081
16 changed files with 780 additions and 57 deletions
--- a/server/handlers/workflow_instances.go
+++ b/server/handlers/workflow_instances.go
@@ -10,7 +10,9 @@ import (
 	"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"
 	"git.gobha.me/xcaliber/chat-switchboard/store"
 	"git.gobha.me/xcaliber/chat-switchboard/tools"
 )
@@ -20,11 +22,13 @@ import (
 // advancing/rejecting stages, and querying status.

 type WorkflowInstanceHandler struct {
-	stores store.Stores
+	stores   store.Stores
+	hub      *events.Hub
+	notifSvc *notifications.Service
 }

-func NewWorkflowInstanceHandler(stores store.Stores) *WorkflowInstanceHandler {
-	return &WorkflowInstanceHandler{stores: stores}
+func NewWorkflowInstanceHandler(stores store.Stores, hub *events.Hub, notifSvc *notifications.Service) *WorkflowInstanceHandler {
+	return &WorkflowInstanceHandler{stores: stores, hub: hub, notifSvc: notifSvc}
 }

 // ── Start ───────────────────────────────────
@@ -198,6 +202,15 @@ func (h *WorkflowInstanceHandler) Advance(c *gin.Context) {
 			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
 	}
@@ -230,13 +243,23 @@ func (h *WorkflowInstanceHandler) Advance(c *gin.Context) {
 		}
 	}

-	// Notify assignment team (stub — fully wired in v0.26.4)
+	// v0.27.0: Assignment + round-robin + WS notifications
 	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)
+		assignmentID := tools.CreateWorkflowAssignment(ctx, 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,
@@ -315,5 +338,236 @@ func (h *WorkflowInstanceHandler) readWorkflowState(ctx context.Context, channel
 	return
 }

+// emitWorkflowEvent pushes a workflow event to all connected clients in the channel's room.
+func (h *WorkflowInstanceHandler) emitWorkflowEvent(label, channelID string, data map[string]any) {
+	if h.hub == nil {
+		return
+	}
+	payload, _ := json.Marshal(data)
+	h.hub.GetBus().Publish(events.Event{
+		Label:   label,
+		Room:    channelID,
+		Payload: payload,
+		Ts:      time.Now().UnixMilli(),
+	})
+}
+
+// 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) {
+	if h.stores.Workflows == nil {
+		return
+	}
+	wf, err := h.stores.Workflows.GetByID(ctx, workflowID)
+	if err != nil || wf == nil || len(wf.OnComplete) == 0 || string(wf.OnComplete) == "null" {
+		return
+	}
+
+	var chain struct {
+		Action     string            `json:"action"`
+		TargetSlug string            `json:"target_slug"`
+		DataMap    map[string]string `json:"data_map"` // source_key → target_key
+	}
+	if err := json.Unmarshal(wf.OnComplete, &chain); err != nil || chain.Action != "start_workflow" || chain.TargetSlug == "" {
+		return
+	}
+
+	// Look up target workflow
+	target, err := h.stores.Workflows.GetBySlug(ctx, wf.TeamID, chain.TargetSlug)
+	if err != nil || target == nil || !target.IsActive {
+		log.Printf("[workflow] on_complete: target workflow '%s' not found or inactive", chain.TargetSlug)
+		return
+	}
+
+	// Map stage_data from completed workflow to initial stage_data for target
+	var sourceData map[string]any
+	if err := json.Unmarshal([]byte(mergedData), &sourceData); err != nil {
+		sourceData = map[string]any{}
+	}
+	targetData := map[string]any{}
+	if len(chain.DataMap) > 0 {
+		for srcKey, tgtKey := range chain.DataMap {
+			if v, ok := sourceData[srcKey]; ok {
+				targetData[tgtKey] = v
+			}
+		}
+	} else {
+		// No mapping = pass all data through
+		targetData = sourceData
+	}
+
+	// Get the channel owner to start the chained workflow as
+	var ownerID string
+	_ = database.DB.QueryRowContext(ctx, database.Q(`
+		SELECT user_id FROM channels WHERE id = $1
+	`), channelID).Scan(&ownerID)
+	if ownerID == "" {
+		return
+	}
+
+	// Create a new workflow channel for the target
+	ver, err := h.stores.Workflows.GetLatestVersion(ctx, target.ID)
+	if err != nil {
+		log.Printf("[workflow] on_complete: target '%s' has no published version", chain.TargetSlug)
+		return
+	}
+	stages, err := h.stores.Workflows.ListStages(ctx, target.ID)
+	if err != nil || len(stages) == 0 {
+		return
+	}
+
+	ch := &models.Channel{
+		UserID:      ownerID,
+		Title:       target.Name + " (chained)",
+		Description: target.Description,
+		Type:        "workflow",
+		TeamID:      target.TeamID,
+	}
+	if err := h.stores.Channels.Create(ctx, ch); err != nil {
+		log.Printf("[workflow] on_complete: failed to create chained channel: %v", err)
+		return
+	}
+
+	initialData, _ := json.Marshal(targetData)
+	_, err = database.DB.ExecContext(ctx, database.Q(`
+		UPDATE channels
+		SET workflow_id = $1, workflow_version = $2, current_stage = 0,
+		    stage_data = $3, workflow_status = 'active',
+		    last_activity_at = $4, ai_mode = 'auto'
+		WHERE id = $5
+	`), target.ID, ver.VersionNumber, string(initialData), time.Now().UTC(), ch.ID)
+	if err != nil {
+		log.Printf("[workflow] on_complete: failed to set workflow columns: %v", err)
+		return
+	}
+
+	_ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{
+		ChannelID: ch.ID, ParticipantType: "user", ParticipantID: ownerID, Role: "owner",
+	})
+	if stages[0].PersonaID != nil {
+		_ = h.stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{
+			ChannelID: ch.ID, ParticipantType: "persona", ParticipantID: *stages[0].PersonaID, Role: "member",
+		})
+	}
+
+	log.Printf("[workflow] on_complete: chained '%s' → '%s' (channel %s → %s)",
+		wf.Slug, target.Slug, channelID, ch.ID)
+}
+
+// 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 ""
+	}
+
+	// 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, MAX(wa.claimed_at) 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 NULLS FIRST
+		LIMIT 1
+	`), *stage.AssignmentTeamID, *stage.AssignmentTeamID)
+	if err == nil {
+		defer rows.Close()
+		if rows.Next() {
+			var uid string
+			var lastClaim *time.Time
+			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)
+	if err != nil {
+		log.Printf("[workflow] round-robin: failed to auto-assign %s to %s: %v", assignmentID, bestUserID, err)
+		return ""
+	}
+
+	log.Printf("[workflow] round-robin: auto-assigned %s to user %s", assignmentID, bestUserID)
+	return bestUserID
+}
+
+// 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.SendToUser(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).