package handlers import ( "encoding/json" "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/notifications" ) // ── Workflow Assignment Handler ───────────── // Manages the assignment queue for human review stages. type WorkflowAssignmentHandler struct { hub *events.Hub } func NewWorkflowAssignmentHandler(hub ...*events.Hub) *WorkflowAssignmentHandler { h := &WorkflowAssignmentHandler{} if len(hub) > 0 { h.hub = hub[0] } return h } type assignmentRow struct { ID string `json:"id"` ChannelID string `json:"channel_id"` Stage int `json:"stage"` TeamID string `json:"team_id"` AssignedTo *string `json:"assigned_to"` Status string `json:"status"` CreatedAt time.Time `json:"created_at"` ClaimedAt *time.Time `json:"claimed_at"` CompletedAt *time.Time `json:"completed_at"` } // ListForTeam returns unassigned + claimed assignments for a team. // GET /api/v1/teams/:teamId/assignments func (h *WorkflowAssignmentHandler) ListForTeam(c *gin.Context) { teamID := c.Param("teamId") status := c.DefaultQuery("status", "unassigned") rows, err := database.DB.QueryContext(c.Request.Context(), database.Q(` SELECT id, channel_id, stage, team_id, assigned_to, status, created_at, claimed_at, completed_at FROM workflow_assignments WHERE team_id = $1 AND status = $2 ORDER BY created_at ASC `), teamID, status) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list assignments"}) return } defer rows.Close() var result []assignmentRow for rows.Next() { var a assignmentRow if err := rows.Scan(&a.ID, &a.ChannelID, &a.Stage, &a.TeamID, &a.AssignedTo, &a.Status, database.ST(&a.CreatedAt), database.SNT(&a.ClaimedAt), database.SNT(&a.CompletedAt)); err != nil { continue } result = append(result, a) } if result == nil { result = []assignmentRow{} } c.JSON(http.StatusOK, gin.H{"data": result}) } // ListForUser returns assignments claimed by the current user plus // unassigned assignments for teams the user belongs to. // GET /api/v1/workflow-assignments/mine func (h *WorkflowAssignmentHandler) ListMine(c *gin.Context) { userID := c.GetString("user_id") rows, err := database.DB.QueryContext(c.Request.Context(), database.Q(` SELECT DISTINCT wa.id, wa.channel_id, wa.stage, wa.team_id, wa.assigned_to, wa.status, wa.created_at, wa.claimed_at, wa.completed_at FROM workflow_assignments wa LEFT JOIN team_members tm ON tm.team_id = wa.team_id AND tm.user_id = $1 WHERE (wa.assigned_to = $2 AND wa.status = 'claimed') OR (wa.status = 'unassigned' AND tm.user_id IS NOT NULL) ORDER BY wa.created_at DESC `), userID, userID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list assignments"}) return } defer rows.Close() var result []assignmentRow for rows.Next() { var a assignmentRow if err := rows.Scan(&a.ID, &a.ChannelID, &a.Stage, &a.TeamID, &a.AssignedTo, &a.Status, database.ST(&a.CreatedAt), database.SNT(&a.ClaimedAt), database.SNT(&a.CompletedAt)); err != nil { continue } result = append(result, a) } if result == nil { result = []assignmentRow{} } c.JSON(http.StatusOK, gin.H{"data": result}) } // Claim assigns a workflow to the current user via optimistic lock. // POST /api/v1/workflow-assignments/:id/claim func (h *WorkflowAssignmentHandler) Claim(c *gin.Context) { assignmentID := c.Param("id") userID := c.GetString("user_id") now := time.Now().UTC() res, err := database.DB.ExecContext(c.Request.Context(), database.Q(` UPDATE workflow_assignments SET assigned_to = $1, status = 'claimed', claimed_at = $2 WHERE id = $3 AND status = 'unassigned' `), userID, now, assignmentID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to claim assignment"}) return } n, _ := res.RowsAffected() if n == 0 { c.JSON(http.StatusConflict, gin.H{"error": "assignment already claimed or not found"}) return } // Look up channel for this assignment (needed for WS delivery + notification) var channelID string _ = database.DB.QueryRowContext(c.Request.Context(), database.Q(`SELECT channel_id FROM workflow_assignments WHERE id = $1`), assignmentID).Scan(&channelID) // Emit workflow.claimed WS event to all user participants in the channel. // Uses SendToUser (not room-scoped Bus.Publish) because room subscriptions // are not yet wired on the client side. See websocket.md § Room Model. if h.hub != nil && channelID != "" { payload, _ := json.Marshal(map[string]any{ "assignment_id": assignmentID, "claimed_by": userID, "channel_id": channelID, }) evt := events.Event{ Label: "workflow.claimed", Payload: payload, Ts: now.UnixMilli(), } rows, err := database.DB.QueryContext(c.Request.Context(), database.Q(` SELECT participant_id FROM channel_participants WHERE channel_id = $1 AND participant_type = 'user' `), channelID) if err == nil { defer rows.Close() for rows.Next() { var uid string if rows.Scan(&uid) == nil { h.hub.SendToUser(uid, evt) } } } } // Persist notification for bell/inbox (v0.28.2) if svc := notifications.Default(); svc != nil && channelID != "" { notifications.NotifyWorkflowClaimed(svc, userID, assignmentID, channelID) } c.JSON(http.StatusOK, gin.H{"claimed": true, "assignment_id": assignmentID}) } // Complete marks an assignment as completed. // POST /api/v1/workflow-assignments/:id/complete func (h *WorkflowAssignmentHandler) Complete(c *gin.Context) { assignmentID := c.Param("id") now := time.Now().UTC() res, err := database.DB.ExecContext(c.Request.Context(), database.Q(` UPDATE workflow_assignments SET status = 'completed', completed_at = $1 WHERE id = $2 AND status = 'claimed' `), now, assignmentID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to complete assignment"}) return } n, _ := res.RowsAffected() if n == 0 { c.JSON(http.StatusConflict, gin.H{"error": "assignment not in claimed state"}) return } c.JSON(http.StatusOK, gin.H{"completed": true, "assignment_id": assignmentID}) }