Feat v0.7.10 workflow handoff + assignment UI
Some checks failed
CI/CD / test-sqlite (pull_request) Has been cancelled
CI/CD / build-and-deploy (pull_request) Has been cancelled
CI/CD / test-runners (pull_request) Has been skipped
CI/CD / test-frontend (pull_request) Has been cancelled
CI/CD / detect-changes (pull_request) Successful in 4s
CI/CD / test-go-pg (pull_request) Has been cancelled
CI/CD / e2e-smoke (pull_request) Has been cancelled

Public→team stage handoff with audience mismatch detection, enriched
assignment API responses (workflow_name, stage_name, sla_breached),
team inbox with claim/assign actions, assignment notifications, team
middleware system-admin bypass fix, and SDK gap closure.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-02 23:15:40 +00:00
parent a9cf71b76d
commit 8f8c1b0e53
17 changed files with 818 additions and 36 deletions

View File

@@ -1,9 +1,11 @@
package handlers
import (
"context"
"database/sql"
"encoding/json"
"net/http"
"time"
"github.com/gin-gonic/gin"
@@ -12,6 +14,15 @@ import (
"armature/workflow"
)
// assignmentView is the enriched response for assignment list endpoints.
type assignmentView struct {
models.WorkflowAssignment
WorkflowID string `json:"workflow_id"`
WorkflowName string `json:"workflow_name"`
StageName string `json:"stage_name"`
SLABreached bool `json:"sla_breached"`
}
// ── Assignment Handlers ─────────────────────
// WorkflowAssignmentHandler manages workflow assignment HTTP endpoints.
@@ -147,7 +158,7 @@ func (h *WorkflowAssignmentHandler) Cancel(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"cancelled": true})
}
// ListByTeam returns assignments for a team.
// ListByTeam returns enriched assignments for a team.
// GET /api/v1/teams/:teamId/assignments?status=
func (h *WorkflowAssignmentHandler) ListByTeam(c *gin.Context) {
teamID := c.Param("teamId")
@@ -156,20 +167,50 @@ func (h *WorkflowAssignmentHandler) ListByTeam(c *gin.Context) {
result, err := h.stores.Workflows.ListAssignmentsByTeam(c.Request.Context(), teamID, status)
if err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusOK, gin.H{"data": []struct{}{}})
c.JSON(http.StatusOK, gin.H{"data": []assignmentView{}})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list assignments"})
return
}
if result == nil {
c.JSON(http.StatusOK, gin.H{"data": []struct{}{}})
c.JSON(http.StatusOK, gin.H{"data": []assignmentView{}})
return
}
c.JSON(http.StatusOK, gin.H{"data": result})
c.JSON(http.StatusOK, gin.H{"data": h.enrichAssignments(c.Request.Context(), result)})
}
// ListMine returns assignments claimed by or assigned to the current user.
// Assign lets a team admin assign an unassigned assignment to a specific user.
// POST /api/v1/assignments/:id/assign
func (h *WorkflowAssignmentHandler) Assign(c *gin.Context) {
assignmentID := c.Param("id")
ctx := c.Request.Context()
var body struct {
UserID string `json:"user_id"`
}
if err := c.ShouldBindJSON(&body); err != nil || body.UserID == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "user_id is required"})
return
}
// Verify the target user exists
targetUser, err := h.stores.Users.GetByID(ctx, body.UserID)
if err != nil || targetUser == nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "user not found"})
return
}
// ClaimAssignment sets assigned_to and status=claimed — works for admin assignment
if err := h.stores.Workflows.ClaimAssignment(ctx, assignmentID, body.UserID); err != nil {
c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"assigned": true, "assigned_to": body.UserID})
}
// ListMine returns enriched assignments claimed by or assigned to the current user.
// GET /api/v1/assignments/mine?status=
func (h *WorkflowAssignmentHandler) ListMine(c *gin.Context) {
userID := c.GetString("user_id")
@@ -181,8 +222,80 @@ func (h *WorkflowAssignmentHandler) ListMine(c *gin.Context) {
return
}
if result == nil {
c.JSON(http.StatusOK, gin.H{"data": []struct{}{}})
c.JSON(http.StatusOK, gin.H{"data": []assignmentView{}})
return
}
c.JSON(http.StatusOK, gin.H{"data": result})
c.JSON(http.StatusOK, gin.H{"data": h.enrichAssignments(c.Request.Context(), result)})
}
// enrichAssignments resolves workflow_name, stage_name, and sla_breached
// for each assignment by looking up instance → workflow → stage snapshot.
func (h *WorkflowAssignmentHandler) enrichAssignments(ctx context.Context, assignments []models.WorkflowAssignment) []assignmentView {
views := make([]assignmentView, 0, len(assignments))
// Cache lookups to avoid repeated DB hits for the same workflow/instance
instanceCache := map[string]*models.WorkflowInstance{}
workflowCache := map[string]*models.Workflow{}
for _, a := range assignments {
v := assignmentView{WorkflowAssignment: a}
// Resolve instance → workflow
inst, ok := instanceCache[a.InstanceID]
if !ok {
inst, _ = h.stores.Workflows.GetInstance(ctx, a.InstanceID)
instanceCache[a.InstanceID] = inst
}
if inst != nil {
v.WorkflowID = inst.WorkflowID
wf, ok := workflowCache[inst.WorkflowID]
if !ok {
wf, _ = h.stores.Workflows.GetByID(ctx, inst.WorkflowID)
workflowCache[inst.WorkflowID] = wf
}
if wf != nil {
v.WorkflowName = wf.Name
}
// Resolve stage name and SLA from the published version snapshot
ver, _ := h.stores.Workflows.GetVersion(ctx, inst.WorkflowID, inst.WorkflowVersion)
if ver != nil {
stages := parseSnapshotStagesForEnrich(ver.Snapshot)
for _, s := range stages {
if s.Name == a.Stage {
v.StageName = s.Name
if s.SLASeconds != nil && *s.SLASeconds > 0 {
elapsed := time.Since(a.CreatedAt)
v.SLABreached = elapsed.Seconds() > float64(*s.SLASeconds)
}
break
}
}
}
}
// Fallback: use stage field as stage_name if not resolved
if v.StageName == "" {
v.StageName = a.Stage
}
views = append(views, v)
}
return views
}
// parseSnapshotStagesForEnrich is a lightweight stage parser for enrichment.
func parseSnapshotStagesForEnrich(snapshot json.RawMessage) []models.WorkflowStage {
var wrapped struct {
Stages []models.WorkflowStage `json:"stages"`
}
if err := json.Unmarshal(snapshot, &wrapped); err == nil && len(wrapped.Stages) > 0 {
return wrapped.Stages
}
var stages []models.WorkflowStage
json.Unmarshal(snapshot, &stages)
return stages
}

View File

@@ -1,17 +1,34 @@
package handlers
import (
"context"
"database/sql"
"encoding/json"
"net/http"
"strconv"
"time"
"github.com/gin-gonic/gin"
"armature/models"
"armature/store"
"armature/workflow"
)
// instanceView is the enriched response for team instance list endpoints.
type instanceView struct {
ID string `json:"id"`
WorkflowID string `json:"workflow_id"`
WorkflowName string `json:"workflow_name"`
CurrentStage string `json:"current_stage"`
StageName string `json:"stage_name"`
Status string `json:"status"`
StartedBy string `json:"started_by"`
AgeSeconds int `json:"age_seconds"`
SLABreached bool `json:"sla_breached"`
CreatedAt string `json:"created_at"`
}
// ── Instance Handlers ───────────────────────
// WorkflowInstanceHandler manages workflow instance HTTP endpoints.
@@ -120,6 +137,123 @@ func (h *WorkflowInstanceHandler) Advance(c *gin.Context) {
c.JSON(http.StatusOK, inst)
}
// ListTeamInstances returns enriched instances for all workflows owned by a team.
// GET /api/v1/teams/:teamId/workflow-instances?status=&limit=&offset=
func (h *WorkflowInstanceHandler) ListTeamInstances(c *gin.Context) {
teamID := c.Param("teamId")
status := c.Query("status")
if status == "" {
status = "active" // default to active for monitoring
}
opts := store.ListOptions{}
if l := c.Query("limit"); l != "" {
opts.Limit, _ = strconv.Atoi(l)
}
if o := c.Query("offset"); o != "" {
opts.Offset, _ = strconv.Atoi(o)
}
if opts.Limit == 0 {
opts.Limit = 50
}
result, err := h.stores.Workflows.ListInstancesByTeam(c.Request.Context(), teamID, status, opts)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list instances"})
return
}
if result == nil {
c.JSON(http.StatusOK, gin.H{"data": []instanceView{}})
return
}
c.JSON(http.StatusOK, gin.H{"data": h.enrichInstances(c.Request.Context(), result)})
}
// enrichInstances resolves workflow_name, stage_name, sla_breached, and age
// for each instance by looking up the workflow definition and version snapshot.
func (h *WorkflowInstanceHandler) enrichInstances(ctx context.Context, instances []models.WorkflowInstance) []instanceView {
views := make([]instanceView, 0, len(instances))
workflowCache := map[string]*models.Workflow{}
for _, inst := range instances {
v := instanceView{
ID: inst.ID,
WorkflowID: inst.WorkflowID,
CurrentStage: inst.CurrentStage,
StageName: inst.CurrentStage, // default
Status: inst.Status,
StartedBy: inst.StartedBy,
AgeSeconds: int(time.Since(inst.CreatedAt).Seconds()),
CreatedAt: inst.CreatedAt.Format(time.RFC3339),
}
wf, ok := workflowCache[inst.WorkflowID]
if !ok {
wf, _ = h.stores.Workflows.GetByID(ctx, inst.WorkflowID)
workflowCache[inst.WorkflowID] = wf
}
if wf != nil {
v.WorkflowName = wf.Name
}
// SLA check from version snapshot
ver, _ := h.stores.Workflows.GetVersion(ctx, inst.WorkflowID, inst.WorkflowVersion)
if ver != nil {
stages := parseSnapshotStagesForView(ver.Snapshot)
for _, s := range stages {
if s.Name == inst.CurrentStage {
v.StageName = s.Name
if s.SLASeconds != nil && *s.SLASeconds > 0 {
elapsed := time.Since(inst.StageEnteredAt)
v.SLABreached = elapsed.Seconds() > float64(*s.SLASeconds)
}
break
}
}
}
views = append(views, v)
}
return views
}
func parseSnapshotStagesForView(snapshot json.RawMessage) []models.WorkflowStage {
var wrapped struct {
Stages []models.WorkflowStage `json:"stages"`
}
if err := json.Unmarshal(snapshot, &wrapped); err == nil && len(wrapped.Stages) > 0 {
return wrapped.Stages
}
var stages []models.WorkflowStage
json.Unmarshal(snapshot, &stages)
return stages
}
// CancelTeamInstance cancels an instance belonging to a team workflow.
// POST /api/v1/teams/:teamId/workflow-instances/:iid/cancel
func (h *WorkflowInstanceHandler) CancelTeamInstance(c *gin.Context) {
userID := c.GetString("user_id")
instanceID := c.Param("iid")
// Verify instance belongs to a workflow owned by this team
inst, err := h.stores.Workflows.GetInstance(c.Request.Context(), instanceID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "instance not found"})
return
}
wf, err := h.stores.Workflows.GetByID(c.Request.Context(), inst.WorkflowID)
if err != nil || wf.TeamID == nil || *wf.TeamID != c.Param("teamId") {
c.JSON(http.StatusForbidden, gin.H{"error": "instance does not belong to this team"})
return
}
if err := h.engine.Cancel(c.Request.Context(), instanceID, userID); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"cancelled": true})
}
// Cancel terminates an active instance.
// POST /api/v1/workflows/:id/instances/:iid/cancel
func (h *WorkflowInstanceHandler) Cancel(c *gin.Context) {

View File

@@ -519,6 +519,7 @@ func main() {
protected.POST("/assignments/:id/unclaim", wfAssignH.Unclaim)
protected.POST("/assignments/:id/complete", wfAssignH.Complete)
protected.POST("/assignments/:id/cancel", middleware.RequirePermission(auth.PermWorkflowCreate, stores), wfAssignH.Cancel)
protected.POST("/assignments/:id/assign", middleware.RequirePermission(auth.PermWorkflowCreate, stores), wfAssignH.Assign)
protected.GET("/assignments/mine", wfAssignH.ListMine)
// Workflow signoffs
@@ -615,7 +616,7 @@ func main() {
// Team admin self-service
teamScoped := protected.Group("/teams/:teamId")
teamScoped.Use(middleware.RequireTeamAdmin(stores.Teams))
teamScoped.Use(middleware.RequireTeamAdmin(stores.Teams, stores))
{
teamScoped.GET("/members", teams.ListMembers)
teamScoped.POST("/members", teams.AddMember)
@@ -677,6 +678,10 @@ func main() {
teamWfAssignH := handlers.NewWorkflowAssignmentHandler(wfEngine, stores)
teamScoped.GET("/assignments", teamWfAssignH.ListByTeam)
// Team workflow instances (cross-workflow)
teamScoped.GET("/workflow-instances", teamWfInstH.ListTeamInstances)
teamScoped.POST("/workflow-instances/:iid/cancel", teamWfInstH.CancelTeamInstance)
// Team workflow signoffs
teamWfSignoffH := handlers.NewWorkflowSignoffHandler(wfEngine, stores)
teamScoped.POST("/instances/:iid/signoffs", teamWfSignoffH.Submit)

View File

@@ -5,20 +5,35 @@ import (
"github.com/gin-gonic/gin"
"armature/auth"
"armature/store"
)
// isSystemAdmin checks if the user has the surface.admin.access permission,
// which grants system-wide admin privileges including team access bypass.
func isSystemAdmin(c *gin.Context, stores store.Stores) bool {
userID := c.GetString("user_id")
if userID == "" {
return false
}
perms, err := resolveAndCachePerms(c, stores, userID)
if err != nil {
return false
}
return perms[auth.PermSurfaceAdminAccess]
}
// RequireTeamAdmin returns middleware that restricts access to team admins.
// System admins are always allowed through.
func RequireTeamAdmin(teams store.TeamStore) gin.HandlerFunc {
func RequireTeamAdmin(teams store.TeamStore, allStores ...store.Stores) gin.HandlerFunc {
return func(c *gin.Context) {
role, _ := c.Get("role")
if role == "admin" {
// System admin bypass via permissions
if len(allStores) > 0 && isSystemAdmin(c, allStores[0]) {
c.Next()
return
}
userID, _ := c.Get("user_id")
userID := c.GetString("user_id")
teamID := c.Param("teamId")
if teamID == "" {
c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{
@@ -27,7 +42,7 @@ func RequireTeamAdmin(teams store.TeamStore) gin.HandlerFunc {
return
}
isAdmin, err := teams.IsTeamAdmin(c.Request.Context(), teamID, userID.(string))
isAdmin, err := teams.IsTeamAdmin(c.Request.Context(), teamID, userID)
if err != nil || !isAdmin {
c.AbortWithStatusJSON(http.StatusForbidden, gin.H{
"error": "team admin access required",
@@ -42,15 +57,15 @@ func RequireTeamAdmin(teams store.TeamStore) gin.HandlerFunc {
// RequireTeamMember returns middleware that restricts access to users who
// belong to the team identified by :teamId (any role).
// System admins are always allowed through.
func RequireTeamMember(teams store.TeamStore) gin.HandlerFunc {
func RequireTeamMember(teams store.TeamStore, allStores ...store.Stores) gin.HandlerFunc {
return func(c *gin.Context) {
role, _ := c.Get("role")
if role == "admin" {
// System admin bypass via permissions
if len(allStores) > 0 && isSystemAdmin(c, allStores[0]) {
c.Next()
return
}
userID, _ := c.Get("user_id")
userID := c.GetString("user_id")
teamID := c.Param("teamId")
if teamID == "" {
c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{
@@ -59,7 +74,7 @@ func RequireTeamMember(teams store.TeamStore) gin.HandlerFunc {
return
}
isMember, _ := teams.IsMember(c.Request.Context(), teamID, userID.(string))
isMember, _ := teams.IsMember(c.Request.Context(), teamID, userID)
if !isMember {
c.AbortWithStatusJSON(http.StatusForbidden, gin.H{
"error": "team membership required",

View File

@@ -646,6 +646,7 @@ type WorkflowPageData struct {
BrandingJSON string
InstanceID string // workflow instance ID (used for API calls)
Status string // pending | active | completed | cancelled | stale
AudienceMismatch bool // true when current stage audience is team/system but visitor is unauthenticated
}
// WorkflowLandingPageData is passed to workflow-landing.html.
@@ -745,6 +746,22 @@ func (e *Engine) RenderWorkflow() gin.HandlerFunc {
entryToken = *inst.EntryToken
}
// Detect audience mismatch: stage requires team/system but visitor is
// unauthenticated. Show a "submitted" screen instead of the stage form.
audienceMismatch := false
userID := c.GetString("user_id")
if userID == "" {
// Visitor is unauthenticated — check the current stage audience
for _, s := range stages {
if s.ID == inst.CurrentStage || s.Name == inst.CurrentStage {
if s.Audience == models.AudienceTeam || s.Audience == models.AudienceSystem {
audienceMismatch = true
}
break
}
}
}
e.Render(c, "workflow.html", PageData{
Surface: "workflow",
InstanceName: instanceName,
@@ -763,6 +780,7 @@ func (e *Engine) RenderWorkflow() gin.HandlerFunc {
BrandingJSON: brandingJSON,
InstanceID: inst.ID,
Status: inst.Status,
AudienceMismatch: audienceMismatch,
},
})
}

View File

@@ -104,6 +104,26 @@
padding: 24px; text-align: center; color: var(--text-2);
}
/* ── Submitted / handoff screen ── */
.wf-submitted {
flex: 1; display: flex; align-items: center; justify-content: center;
padding: 40px 24px; text-align: center;
}
.wf-submitted-inner { max-width: 420px; }
.wf-submitted-icon {
width: 56px; height: 56px; border-radius: 50%;
background: var(--accent); color: #fff;
display: inline-flex; align-items: center; justify-content: center;
font-size: 28px; margin-bottom: 16px;
}
.wf-submitted h3 { font-size: 20px; font-weight: 600; margin-bottom: 8px; }
.wf-submitted p { color: var(--text-2); font-size: 14px; line-height: 1.5; }
.wf-submitted .ref-id {
margin-top: 16px; padding: 8px 16px; background: var(--bg-surface);
border: 1px solid var(--border); border-radius: 6px;
font-size: 12px; color: var(--text-3); display: inline-block;
}
/* ── Branding ────────────────────── */
.wf-branding-logo { max-height: 32px; margin-bottom: 4px; }
.wf-branding-tagline { font-size: 13px; color: var(--text-2); margin-top: 2px; }
@@ -160,7 +180,18 @@
<!-- Stage surface: form, review, delegated (custom surface), or automated -->
{{if .Data.SurfacePkgID}}
{{if .Data.AudienceMismatch}}
<div class="wf-submitted">
<div class="wf-submitted-inner">
<div class="wf-submitted-icon">&#10003;</div>
<h3>Submitted Successfully</h3>
<p>Your submission has been received and is now being reviewed by our team. You can safely close this page.</p>
{{if .Data.EntryToken}}
<div class="ref-id">Reference: {{.Data.EntryToken}}</div>
{{end}}
</div>
</div>
{{else if .Data.SurfacePkgID}}
<div id="customSurfaceMount" style="flex:1;overflow-y:auto"></div>
{{else if eq .Data.StageMode "form"}}
<div class="wf-form" id="formArea"></div>

View File

@@ -517,6 +517,52 @@ func (s *WorkflowStore) ListInstances(ctx context.Context, workflowID string, st
return result, rows.Err()
}
func (s *WorkflowStore) ListInstancesByTeam(ctx context.Context, teamID string, status string, opts store.ListOptions) ([]models.WorkflowInstance, error) {
q := `SELECT i.id, i.workflow_id, i.workflow_version, i.current_stage, i.stage_data,
i.status, i.started_by, i.entry_token, i.metadata,
i.stage_entered_at, i.created_at, i.updated_at
FROM workflow_instances i
JOIN workflows w ON w.id = i.workflow_id
WHERE w.team_id = $1`
args := []interface{}{teamID}
idx := 2
if status != "" {
q += fmt.Sprintf(" AND i.status = $%d", idx)
args = append(args, status)
idx++
}
q += " ORDER BY i.created_at DESC"
if opts.Limit > 0 {
q += fmt.Sprintf(" LIMIT $%d", idx)
args = append(args, opts.Limit)
idx++
}
if opts.Offset > 0 {
q += fmt.Sprintf(" OFFSET $%d", idx)
args = append(args, opts.Offset)
}
rows, err := DB.QueryContext(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var result []models.WorkflowInstance
for rows.Next() {
var inst models.WorkflowInstance
var stageData, metadata []byte
if err := rows.Scan(&inst.ID, &inst.WorkflowID, &inst.WorkflowVersion,
&inst.CurrentStage, &stageData, &inst.Status, &inst.StartedBy,
&inst.EntryToken, &metadata,
&inst.StageEnteredAt, &inst.CreatedAt, &inst.UpdatedAt); err != nil {
return nil, err
}
inst.StageData = stageData
inst.Metadata = metadata
result = append(result, inst)
}
return result, rows.Err()
}
func (s *WorkflowStore) AdvanceStage(ctx context.Context, id string, nextStage string, stageData json.RawMessage) error {
sd := jsonOrEmpty(stageData)
_, err := DB.ExecContext(ctx, `

View File

@@ -534,6 +534,49 @@ func (s *WorkflowStore) ListInstances(ctx context.Context, workflowID string, st
return result, rows.Err()
}
func (s *WorkflowStore) ListInstancesByTeam(ctx context.Context, teamID string, status string, opts store.ListOptions) ([]models.WorkflowInstance, error) {
q := `SELECT i.id, i.workflow_id, i.workflow_version, i.current_stage, i.stage_data,
i.status, i.started_by, i.entry_token, i.metadata,
i.stage_entered_at, i.created_at, i.updated_at
FROM workflow_instances i
JOIN workflows w ON w.id = i.workflow_id
WHERE w.team_id = ?`
args := []interface{}{teamID}
if status != "" {
q += " AND i.status = ?"
args = append(args, status)
}
q += " ORDER BY i.created_at DESC"
if opts.Limit > 0 {
q += " LIMIT ?"
args = append(args, opts.Limit)
}
if opts.Offset > 0 {
q += " OFFSET ?"
args = append(args, opts.Offset)
}
rows, err := DB.QueryContext(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var result []models.WorkflowInstance
for rows.Next() {
var inst models.WorkflowInstance
var stageData, metadata string
if err := rows.Scan(&inst.ID, &inst.WorkflowID, &inst.WorkflowVersion,
&inst.CurrentStage, &stageData, &inst.Status, &inst.StartedBy,
&inst.EntryToken, &metadata,
st(&inst.StageEnteredAt), st(&inst.CreatedAt), st(&inst.UpdatedAt)); err != nil {
return nil, err
}
inst.StageData = json.RawMessage(stageData)
inst.Metadata = json.RawMessage(metadata)
result = append(result, inst)
}
return result, rows.Err()
}
func (s *WorkflowStore) AdvanceStage(ctx context.Context, id string, nextStage string, stageData json.RawMessage) error {
sd := jsonOrEmpty(stageData)
now := time.Now().UTC()

View File

@@ -42,6 +42,7 @@ type WorkflowStore interface {
CancelInstance(ctx context.Context, id string) error
MarkInstanceStale(ctx context.Context, id string) error
ListActiveInstances(ctx context.Context) ([]models.WorkflowInstance, error)
ListInstancesByTeam(ctx context.Context, teamID string, status string, opts ListOptions) ([]models.WorkflowInstance, error)
// Assignments
CreateAssignment(ctx context.Context, a *models.WorkflowAssignment) error

View File

@@ -100,6 +100,8 @@ func (e *Engine) Start(ctx context.Context, workflowID string, initialData json.
}
if err := e.stores.Workflows.CreateAssignment(ctx, a); err != nil {
log.Printf("[workflow-engine] create initial assignment: %v", err)
} else {
e.notifyAssignment(ctx, a, wf.Name, stages[0].Name)
}
}
@@ -247,6 +249,12 @@ func (e *Engine) advanceInternal(ctx context.Context, instanceID string, stageDa
"assignment_id": a.ID,
"team_id": a.TeamID,
})
// Notify team members
wfTitle := ""
if wf, _ := e.stores.Workflows.GetByID(ctx, inst.WorkflowID); wf != nil {
wfTitle = wf.Name
}
e.notifyAssignment(ctx, a, wfTitle, nextStage.Name)
}
}
@@ -477,6 +485,45 @@ func (e *Engine) emit(_ context.Context, label, room string, payload map[string]
})
}
// notifyAssignment creates notifications for team members when a workflow
// assignment is created. If the assignment targets a specific user, only
// that user is notified. Otherwise all team members are notified.
func (e *Engine) notifyAssignment(ctx context.Context, a *models.WorkflowAssignment, workflowTitle, stageName string) {
if e.stores.Notifications == nil || e.stores.Teams == nil {
return
}
var recipients []string
if a.AssignedTo != nil && *a.AssignedTo != "" {
recipients = []string{*a.AssignedTo}
} else {
members, err := e.stores.Teams.ListMembers(ctx, a.TeamID)
if err != nil {
log.Printf("[workflow-engine] list team members for notification: %v", err)
return
}
for _, m := range members {
recipients = append(recipients, m.UserID)
}
}
title := "Workflow assignment: " + stageName
body := workflowTitle
for _, uid := range recipients {
n := &models.Notification{
UserID: uid,
Type: models.NotifTypeWorkflowAssign,
Title: title,
Body: body,
ResourceType: "workflow",
ResourceID: a.InstanceID,
}
if err := e.stores.Notifications.Create(ctx, n); err != nil {
log.Printf("[workflow-engine] create notification: %v", err)
}
}
}
// mergeJSON merges b into a (shallow). Returns a if b is empty.
func mergeJSON(a, b json.RawMessage) json.RawMessage {
if len(b) == 0 || string(b) == "{}" || string(b) == "null" {