switchboard/core
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Releases 9 Wiki Activity

Changeset 0.28.0.1 (#173)

Browse Source
This commit is contained in:
xcaliber
2026-03-11 14:45:37 +00:00
parent 93c72daadf
commit 58313f7e31
57 changed files with 5139 additions and 3206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
server/database/migrations/sqlite/002_teams.sql
View File

@@ -42,6 +42,9 @@ CREATE TABLE IF NOT EXISTS groups (
updated_at TEXT DEFAULT (datetime('now'))
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_groups_name_scope
ON groups(name, COALESCE(team_id, ''));
CREATE TABLE IF NOT EXISTS group_members (
id TEXT PRIMARY KEY,
group_id TEXT NOT NULL REFERENCES groups(id) ON DELETE CASCADE,

23
server/database/migrations/sqlite/016_v023_multiuser.sql
View File

@@ -1,23 +0,0 @@
-- ==========================================
-- Chat Switchboard — 006 Multi-User: DMs + Channels (SQLite)
-- ==========================================
-- SQLite does not support ALTER CONSTRAINT or DROP CONSTRAINT.
-- The type check is enforced by application logic for SQLite.
-- ==========================================
-- ── ai_mode ──────────────────────────────────────────────────────────
-- SQLite ignores CHECK constraints on existing rows; we add the column
-- and rely on app-layer validation for the enum values.
ALTER TABLE channels ADD COLUMN ai_mode TEXT NOT NULL DEFAULT 'auto';
-- ── topic ─────────────────────────────────────────────────────────────
ALTER TABLE channels ADD COLUMN topic TEXT;
-- ── Presence ──────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS user_presence (
user_id TEXT PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
last_seen DATETIME NOT NULL DEFAULT (datetime('now')),
status TEXT NOT NULL DEFAULT 'online'
);
CREATE INDEX IF NOT EXISTS idx_user_presence_last_seen ON user_presence(last_seen DESC);

3
server/database/migrations/sqlite/017_unread.sql
View File

@@ -1,3 +0,0 @@
-- Chat Switchboard — 017 Unread Counts (SQLite)
ALTER TABLE channel_participants ADD COLUMN last_read_message_id TEXT;

8
server/database/migrations/sqlite/018_auth_abstraction.sql
View File

@@ -1,8 +0,0 @@
-- Chat Switchboard — 018 Auth Abstraction (SQLite) (v0.24.0)
ALTER TABLE users ADD COLUMN auth_source TEXT NOT NULL DEFAULT 'builtin';
ALTER TABLE users ADD COLUMN external_id TEXT;
ALTER TABLE users ADD COLUMN handle TEXT;
UPDATE users SET handle = LOWER(REPLACE(REPLACE(username, ' ', '-'), '_', '-'))
WHERE handle IS NULL;
CREATE UNIQUE INDEX IF NOT EXISTS idx_users_handle ON users(handle COLLATE NOCASE);
CREATE UNIQUE INDEX IF NOT EXISTS idx_users_external_id ON users(auth_source, external_id);

13
server/database/migrations/sqlite/019_auth_providers.sql
View File

@@ -1,13 +0,0 @@
-- Chat Switchboard — 019 Auth Providers (SQLite) (v0.24.1)
CREATE TABLE IF NOT EXISTS oidc_auth_state (
state TEXT PRIMARY KEY,
nonce TEXT NOT NULL,
redirect_to TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_oidc_state_created
ON oidc_auth_state(created_at);
ALTER TABLE groups ADD COLUMN source TEXT NOT NULL DEFAULT 'manual';

25
server/database/migrations/sqlite/020_permissions.sql
View File

@@ -1,25 +0,0 @@
-- Chat Switchboard — 020 Permissions (SQLite) (v0.24.2)
-- SQLite cannot drop NOT NULL inline; recreating the constraint isn't needed
-- since SQLite doesn't enforce CHECK constraints strictly and the column
-- already accepts NULL in practice. The store layer guards source='system'.
ALTER TABLE groups ADD COLUMN permissions TEXT NOT NULL DEFAULT '[]';
ALTER TABLE groups ADD COLUMN token_budget_daily INTEGER;
ALTER TABLE groups ADD COLUMN token_budget_monthly INTEGER;
ALTER TABLE groups ADD COLUMN allowed_models TEXT;
-- Seed the Everyone group with a stable well-known ID.
INSERT OR IGNORE INTO groups
(id, name, description, scope, created_by, source, permissions, created_at, updated_at)
VALUES (
'00000000-0000-0000-0000-000000000001',
'Everyone',
'Implicit group — all authenticated users receive these permissions.',
'global',
NULL,
'system',
'["model.use","kb.read","channel.create"]',
datetime('now'),
datetime('now')
);

20
server/database/migrations/sqlite/021_sessions.sql
View File

@@ -1,20 +0,0 @@
-- 021_sessions.sql (SQLite)
-- Anonymous / session participants for workflow channels (v0.24.3)
CREATE TABLE IF NOT EXISTS session_participants (
id TEXT PRIMARY KEY,
session_token TEXT NOT NULL UNIQUE,
channel_id TEXT NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
display_name TEXT NOT NULL DEFAULT 'Visitor',
fingerprint TEXT,
created_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_session_participants_channel
ON session_participants(channel_id);
CREATE INDEX IF NOT EXISTS idx_session_participants_token
ON session_participants(session_token);
-- Allow channels to opt in to anonymous session access.
ALTER TABLE channels ADD COLUMN allow_anonymous INTEGER NOT NULL DEFAULT 0;

56
server/database/migrations/sqlite/023_workflows.sql
View File

@@ -1,56 +0,0 @@
-- 023_workflows.sql (SQLite)
-- Workflow definitions, stages, and version snapshots (v0.26.1)
CREATE TABLE IF NOT EXISTS workflows (
id TEXT PRIMARY KEY,
team_id TEXT REFERENCES teams(id) ON DELETE CASCADE,
name TEXT NOT NULL,
slug TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
branding TEXT NOT NULL DEFAULT '{}',
entry_mode TEXT NOT NULL DEFAULT 'public_link'
CHECK (entry_mode IN ('public_link', 'team_only')),
is_active INTEGER NOT NULL DEFAULT 0,
version INTEGER NOT NULL DEFAULT 1,
on_complete TEXT,
retention TEXT NOT NULL DEFAULT '{"mode": "archive"}',
created_by TEXT NOT NULL REFERENCES users(id),
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_workflows_team_slug
ON workflows(team_id, slug) WHERE team_id IS NOT NULL;
CREATE UNIQUE INDEX IF NOT EXISTS idx_workflows_global_slug
ON workflows(slug) WHERE team_id IS NULL;
CREATE TABLE IF NOT EXISTS workflow_stages (
id TEXT PRIMARY KEY,
workflow_id TEXT NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
ordinal INTEGER NOT NULL,
name TEXT NOT NULL,
persona_id TEXT REFERENCES personas(id) ON DELETE SET NULL,
assignment_team_id TEXT REFERENCES teams(id) ON DELETE SET NULL,
form_template TEXT NOT NULL DEFAULT '{}',
history_mode TEXT NOT NULL DEFAULT 'full'
CHECK (history_mode IN ('full', 'summary', 'fresh')),
auto_transition INTEGER NOT NULL DEFAULT 0,
transition_rules TEXT NOT NULL DEFAULT '{}',
created_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_workflow_stages_workflow
ON workflow_stages(workflow_id, ordinal);
CREATE TABLE IF NOT EXISTS workflow_versions (
id TEXT PRIMARY KEY,
workflow_id TEXT NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
version_number INTEGER NOT NULL,
snapshot TEXT NOT NULL,
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
UNIQUE (workflow_id, version_number)
);
CREATE INDEX IF NOT EXISTS idx_workflow_versions_workflow
ON workflow_versions(workflow_id, version_number);

13
server/database/migrations/sqlite/024_workflow_instances.sql
View File

@@ -1,13 +0,0 @@
-- 024_workflow_instances.sql (SQLite)
-- Workflow instance columns on channels (v0.26.2)
ALTER TABLE channels ADD COLUMN workflow_id TEXT REFERENCES workflows(id) ON DELETE SET NULL;
ALTER TABLE channels ADD COLUMN workflow_version INTEGER;
ALTER TABLE channels ADD COLUMN current_stage INTEGER DEFAULT 0;
ALTER TABLE channels ADD COLUMN stage_data TEXT DEFAULT '{}';
ALTER TABLE channels ADD COLUMN workflow_status TEXT DEFAULT 'active'
CHECK (workflow_status IN ('active', 'completed', 'stale', 'cancelled'));
ALTER TABLE channels ADD COLUMN last_activity_at DATETIME DEFAULT (datetime('now'));
CREATE INDEX IF NOT EXISTS idx_channels_workflow
ON channels(workflow_id) WHERE workflow_id IS NOT NULL;

21
server/database/migrations/sqlite/025_workflow_assignments.sql
View File

@@ -1,21 +0,0 @@
-- 025_workflow_assignments.sql (SQLite)
-- Assignment queue for workflow stages (v0.26.4)
CREATE TABLE IF NOT EXISTS workflow_assignments (
id TEXT PRIMARY KEY,
channel_id TEXT NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
stage INTEGER NOT NULL,
team_id TEXT NOT NULL REFERENCES teams(id) ON DELETE CASCADE,
assigned_to TEXT REFERENCES users(id) ON DELETE SET NULL,
status TEXT NOT NULL DEFAULT 'unassigned'
CHECK (status IN ('unassigned', 'claimed', 'completed')),
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
claimed_at DATETIME,
completed_at DATETIME
);
CREATE INDEX IF NOT EXISTS idx_workflow_assignments_channel
ON workflow_assignments(channel_id, stage);
CREATE INDEX IF NOT EXISTS idx_workflow_assignments_team_status
ON workflow_assignments(team_id, status);

63
server/database/migrations/sqlite/026_tasks.sql
View File

@@ -1,63 +0,0 @@
-- 026_tasks.sql — Task scheduling for autonomous agents (v0.27.1) — SQLite
-- Note: SQLite CHECK on channels.type is in the CREATE TABLE (005),
-- which cannot be altered. The 'service' type is accepted by convention;
-- the Go layer validates.
CREATE TABLE IF NOT EXISTS tasks (
id TEXT PRIMARY KEY,
owner_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
team_id TEXT REFERENCES teams(id) ON DELETE SET NULL,
name TEXT NOT NULL,
description TEXT DEFAULT '',
scope TEXT NOT NULL DEFAULT 'personal',
task_type TEXT NOT NULL DEFAULT 'prompt',
persona_id TEXT REFERENCES personas(id) ON DELETE SET NULL,
model_id TEXT,
system_prompt TEXT DEFAULT '',
user_prompt TEXT DEFAULT '',
workflow_id TEXT REFERENCES workflows(id) ON DELETE SET NULL,
tool_grants TEXT,
schedule TEXT NOT NULL,
timezone TEXT NOT NULL DEFAULT 'UTC',
is_active INTEGER NOT NULL DEFAULT 1,
max_tokens INTEGER NOT NULL DEFAULT 4096,
max_tool_calls INTEGER NOT NULL DEFAULT 10,
max_wall_clock INTEGER NOT NULL DEFAULT 300,
output_mode TEXT NOT NULL DEFAULT 'channel',
output_channel_id TEXT REFERENCES channels(id) ON DELETE SET NULL,
webhook_url TEXT,
provider_config_id TEXT REFERENCES provider_configs(id) ON DELETE SET NULL,
notify_on_complete INTEGER NOT NULL DEFAULT 0,
notify_on_failure INTEGER NOT NULL DEFAULT 1,
last_run_at TEXT,
next_run_at TEXT,
run_count INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_tasks_next_run ON tasks (next_run_at);
CREATE INDEX IF NOT EXISTS idx_tasks_owner ON tasks (owner_id);
CREATE INDEX IF NOT EXISTS idx_tasks_team ON tasks (team_id);
CREATE TABLE IF NOT EXISTS task_runs (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
channel_id TEXT REFERENCES channels(id) ON DELETE SET NULL,
status TEXT NOT NULL DEFAULT 'running',
started_at TEXT NOT NULL DEFAULT (datetime('now')),
completed_at TEXT,
tokens_used INTEGER DEFAULT 0,
tool_calls INTEGER DEFAULT 0,
wall_clock INTEGER DEFAULT 0,
error TEXT
);
CREATE INDEX IF NOT EXISTS idx_task_runs_task ON task_runs (task_id);

5
server/database/migrations/sqlite/027_webhooks.sql
View File

@@ -1,5 +0,0 @@
-- 027_webhooks.sql — Webhook secrets + workflow webhook support (v0.27.3) — SQLite
ALTER TABLE tasks ADD COLUMN webhook_secret TEXT;
ALTER TABLE workflows ADD COLUMN webhook_url TEXT;
ALTER TABLE workflows ADD COLUMN webhook_secret TEXT;

50
server/handlers/workflow_assignments.go
View File

@@ -1,21 +1,29 @@
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"
)
// ── Workflow Assignment Handler ─────────────
// Manages the assignment queue for human review stages.
type WorkflowAssignmentHandler struct{}
type WorkflowAssignmentHandler struct {
hub *events.Hub
}
func NewWorkflowAssignmentHandler() *WorkflowAssignmentHandler {
return &WorkflowAssignmentHandler{}
func NewWorkflowAssignmentHandler(hub ...*events.Hub) *WorkflowAssignmentHandler {
h := &WorkflowAssignmentHandler{}
if len(hub) > 0 {
h.hub = hub[0]
}
return h
}
type assignmentRow struct {
@@ -53,7 +61,7 @@ func (h *WorkflowAssignmentHandler) ListForTeam(c *gin.Context) {
for rows.Next() {
var a assignmentRow
if err := rows.Scan(&a.ID, &a.ChannelID, &a.Stage, &a.TeamID,
&a.AssignedTo, &a.Status, &a.CreatedAt, &a.ClaimedAt, &a.CompletedAt); err != nil {
&a.AssignedTo, &a.Status, database.ST(&a.CreatedAt), database.SNT(&a.ClaimedAt), database.SNT(&a.CompletedAt)); err != nil {
continue
}
result = append(result, a)
@@ -64,17 +72,20 @@ func (h *WorkflowAssignmentHandler) ListForTeam(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"data": result})
}
// ListForUser returns assignments claimed by the current user.
// 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 id, channel_id, stage, team_id, assigned_to, status,
created_at, claimed_at, completed_at
FROM workflow_assignments
WHERE assigned_to = $1 AND status = 'claimed'
ORDER BY claimed_at DESC
`), userID)
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
@@ -85,7 +96,7 @@ func (h *WorkflowAssignmentHandler) ListMine(c *gin.Context) {
for rows.Next() {
var a assignmentRow
if err := rows.Scan(&a.ID, &a.ChannelID, &a.Stage, &a.TeamID,
&a.AssignedTo, &a.Status, &a.CreatedAt, &a.ClaimedAt, &a.CompletedAt); err != nil {
&a.AssignedTo, &a.Status, database.ST(&a.CreatedAt), database.SNT(&a.ClaimedAt), database.SNT(&a.CompletedAt)); err != nil {
continue
}
result = append(result, a)
@@ -117,6 +128,21 @@ func (h *WorkflowAssignmentHandler) Claim(c *gin.Context) {
c.JSON(http.StatusConflict, gin.H{"error": "assignment already claimed or not found"})
return
}
// Emit workflow.claimed WS event
if h.hub != nil {
payload, _ := json.Marshal(map[string]any{
"assignment_id": assignmentID,
"claimed_by": userID,
})
h.hub.GetBus().Publish(events.Event{
Label: "workflow.claimed",
Room: assignmentID,
Payload: payload,
Ts: now.UnixMilli(),
})
}
c.JSON(http.StatusOK, gin.H{"claimed": true, "assignment_id": assignmentID})
}

10
server/handlers/workflow_entry.go
View File

@@ -77,13 +77,17 @@ func (h *WorkflowEntryHandler) StartVisitor(c *gin.Context) {
}
// Set workflow columns
allowAnonVal := interface{}(true)
if database.CurrentDialect == database.DialectSQLite {
allowAnonVal = 1
}
_, 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 = true, ai_mode = 'auto'
WHERE id = $4
`), wf.ID, ver.VersionNumber, time.Now().UTC(), ch.ID)
last_activity_at = $3, allow_anonymous = $4, ai_mode = 'auto'
WHERE id = $5
`), wf.ID, ver.VersionNumber, time.Now().UTC(), allowAnonVal, ch.ID)
if err != nil {
log.Printf("Failed to set workflow columns: %v", err)
}

844
server/handlers/workflow_instance_test.go Normal file
View File

@@ -0,0 +1,844 @@
package handlers
import (
"encoding/json"
"fmt"
"net/http"
"testing"
"github.com/gin-gonic/gin"
"git.gobha.me/xcaliber/chat-switchboard/config"
authpkg "git.gobha.me/xcaliber/chat-switchboard/auth"
"git.gobha.me/xcaliber/chat-switchboard/database"
"git.gobha.me/xcaliber/chat-switchboard/middleware"
"git.gobha.me/xcaliber/chat-switchboard/store"
postgres "git.gobha.me/xcaliber/chat-switchboard/store/postgres"
sqlite "git.gobha.me/xcaliber/chat-switchboard/store/sqlite"
)
// ═══════════════════════════════════════════════════════════
// Extended harness: CRUD + instances + assignments + entry
// ═══════════════════════════════════════════════════════════
type workflowInstanceHarness struct {
*testHarness
stores store.Stores
adminToken string
adminID string
}
func setupWorkflowInstanceHarness(t *testing.T) *workflowInstanceHarness {
t.Helper()
database.RequireTestDB(t)
database.TruncateAll(t)
cfg := &config.Config{
JWTSecret: testJWTSecret,
BasePath: "",
}
var stores store.Stores
if database.IsSQLite() {
stores = sqlite.NewStores(database.TestDB)
} else {
stores = postgres.NewStores(database.TestDB)
}
r := gin.New()
api := r.Group("/api/v1")
auth := NewAuthHandler(cfg, stores, nil, authpkg.NewBuiltinProvider())
api.POST("/auth/login", auth.Login)
api.POST("/auth/register", auth.Register)
protected := api.Group("")
protected.Use(middleware.Auth(cfg))
// Workflow CRUD
wfH := NewWorkflowHandler(stores)
protected.GET("/workflows", wfH.List)
protected.POST("/workflows", wfH.Create)
protected.GET("/workflows/:id", wfH.Get)
protected.PATCH("/workflows/:id", wfH.Update)
protected.DELETE("/workflows/:id", wfH.Delete)
protected.GET("/workflows/:id/stages", wfH.ListStages)
protected.POST("/workflows/:id/stages", wfH.CreateStage)
protected.PUT("/workflows/:id/stages/:sid", wfH.UpdateStage)
protected.DELETE("/workflows/:id/stages/:sid", wfH.DeleteStage)
protected.PATCH("/workflows/:id/stages/reorder", wfH.ReorderStages)
protected.POST("/workflows/:id/publish", wfH.Publish)
protected.GET("/workflows/:id/versions/:version", wfH.GetVersion)
// Workflow instances
wfInstH := NewWorkflowInstanceHandler(stores, nil, nil)
protected.POST("/workflows/:id/start", wfInstH.Start)
protected.GET("/channels/:id/workflow/status", wfInstH.GetStatus)
protected.POST("/channels/:id/workflow/advance", wfInstH.Advance)
protected.POST("/channels/:id/workflow/reject", wfInstH.Reject)
// Assignments
wfAssignH := NewWorkflowAssignmentHandler()
protected.GET("/workflow-assignments/mine", wfAssignH.ListMine)
protected.POST("/workflow-assignments/:id/claim", wfAssignH.Claim)
protected.POST("/workflow-assignments/:id/complete", wfAssignH.Complete)
// Teams (admin routes for creating teams + adding members)
teamH := NewTeamHandler(nil)
admin := api.Group("/admin")
admin.Use(middleware.Auth(cfg), middleware.RequireAdmin())
admin.POST("/teams", teamH.CreateTeam)
admin.POST("/teams/:id/members", teamH.AddMember)
// Team-scoped assignment listing (mirrors main.go teamScoped)
teamScoped := protected.Group("/teams/:teamId")
teamAssignH := NewWorkflowAssignmentHandler()
teamScoped.GET("/assignments", teamAssignH.ListForTeam)
// Visitor entry
wfEntry := NewWorkflowEntryHandler(stores)
r.POST("/api/v1/workflow-entry/:scope/:slug", wfEntry.StartVisitor)
// Seed admin user
adminID := seedInsertReturningID(t,
`INSERT INTO users (username, email, password_hash, role, handle, auth_source) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`,
"wfinst-admin", "wfinst-admin@test.com", "$2a$10$test", "admin", "wfinst-admin", "builtin",
)
token := makeToken(adminID, "wfinst-admin@test.com", "admin")
return &workflowInstanceHarness{
testHarness: &testHarness{router: r, t: t},
stores: stores,
adminToken: token,
adminID: adminID,
}
}
// createPublishedWorkflow is a helper that creates a workflow with N stages,
// activates it, and publishes it. Returns workflow ID + stage IDs.
func (h *workflowInstanceHarness) createPublishedWorkflow(name string, numStages int) (string, []string) {
h.t.Helper()
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": name,
"entry_mode": "public_link",
})
if resp.Code != http.StatusCreated {
h.t.Fatalf("create workflow: %d: %s", resp.Code, resp.Body.String())
}
var wf struct{ ID string `json:"id"` }
json.Unmarshal(resp.Body.Bytes(), &wf)
stageIDs := make([]string, numStages)
for i := 0; i < numStages; i++ {
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": fmt.Sprintf("Stage %d", i),
"ordinal": i,
"history_mode": "full",
})
if resp.Code != http.StatusCreated {
h.t.Fatalf("create stage %d: %d: %s", i, resp.Code, resp.Body.String())
}
var st struct{ ID string `json:"id"` }
json.Unmarshal(resp.Body.Bytes(), &st)
stageIDs[i] = st.ID
}
// Activate + publish
h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"is_active": true,
})
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/publish", h.adminToken, nil)
if resp.Code != http.StatusCreated {
h.t.Fatalf("publish: %d: %s", resp.Code, resp.Body.String())
}
return wf.ID, stageIDs
}
// ═══════════════════════════════════════════════════════════
// #15 — Instance Lifecycle (currently untested)
// ═══════════════════════════════════════════════════════════
// TestWorkflowInstanceLifecycle exercises: start → status → advance → complete.
func TestWorkflowInstanceLifecycle(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Lifecycle Test", 2)
// ── Start ───────────────────────────────
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
if resp.Code != http.StatusCreated {
t.Fatalf("Start: got %d, body: %s", resp.Code, resp.Body.String())
}
var startResp struct {
ChannelID string `json:"channel_id"`
WorkflowVersion int `json:"workflow_version"`
CurrentStage int `json:"current_stage"`
}
json.Unmarshal(resp.Body.Bytes(), &startResp)
if startResp.ChannelID == "" {
t.Fatal("Start: channel_id empty")
}
if startResp.CurrentStage != 0 {
t.Errorf("Start: current_stage = %d, want 0", startResp.CurrentStage)
}
chID := startResp.ChannelID
// ── Status ──────────────────────────────
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("Status: got %d, body: %s", resp.Code, resp.Body.String())
}
var status struct {
WorkflowID *string `json:"workflow_id"`
Status string `json:"status"`
Stage int `json:"current_stage"`
StageData json.RawMessage `json:"stage_data"`
}
json.Unmarshal(resp.Body.Bytes(), &status)
if status.Status != "active" {
t.Errorf("Status: got %q, want \"active\"", status.Status)
}
if status.WorkflowID == nil || *status.WorkflowID != wfID {
t.Errorf("Status: workflow_id mismatch")
}
// ── Advance (stage 0 → 1) ───────────────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"name": "Jane", "email": "jane@test.com"},
})
if resp.Code != http.StatusOK {
t.Fatalf("Advance 0→1: got %d, body: %s", resp.Code, resp.Body.String())
}
var advResp struct {
Status string `json:"status"`
CurrentStage int `json:"current_stage"`
}
json.Unmarshal(resp.Body.Bytes(), &advResp)
if advResp.Status != "active" {
t.Errorf("Advance 0→1: status = %q, want \"active\"", advResp.Status)
}
if advResp.CurrentStage != 1 {
t.Errorf("Advance 0→1: current_stage = %d, want 1", advResp.CurrentStage)
}
// ── Verify merged data persisted ────────
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
json.Unmarshal(resp.Body.Bytes(), &status)
var data map[string]string
json.Unmarshal(status.StageData, &data)
if data["name"] != "Jane" || data["email"] != "jane@test.com" {
t.Errorf("Stage data after advance: got %v", data)
}
// ── Advance (stage 1 → complete) ────────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"resolution": "approved"},
})
if resp.Code != http.StatusOK {
t.Fatalf("Advance 1→complete: got %d, body: %s", resp.Code, resp.Body.String())
}
json.Unmarshal(resp.Body.Bytes(), &advResp)
if advResp.Status != "completed" {
t.Errorf("Completion: status = %q, want \"completed\"", advResp.Status)
}
// ── Status after completion ──────────────
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
json.Unmarshal(resp.Body.Bytes(), &status)
if status.Status != "completed" {
t.Errorf("Post-complete status: got %q, want \"completed\"", status.Status)
}
// ── Advance on completed should fail ────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"extra": "data"},
})
if resp.Code != http.StatusBadRequest {
t.Errorf("Advance after complete: got %d, want 400", resp.Code)
}
}
// TestWorkflowInstanceStart_InactiveWorkflow verifies start fails on inactive workflow.
func TestWorkflowInstanceStart_InactiveWorkflow(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
// Create workflow with a stage but don't activate
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Inactive WF",
})
var wf struct{ ID string `json:"id"` }
json.Unmarshal(resp.Body.Bytes(), &wf)
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "Stage 0", "ordinal": 0, "history_mode": "full",
})
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/start", h.adminToken, nil)
if resp.Code != http.StatusBadRequest {
t.Errorf("Start inactive: got %d, want 400", resp.Code)
}
}
// TestWorkflowInstanceStart_NoPublishedVersion verifies start fails without a published version.
func TestWorkflowInstanceStart_NoPublishedVersion(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Unpublished WF",
})
var wf struct{ ID string `json:"id"` }
json.Unmarshal(resp.Body.Bytes(), &wf)
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "Stage 0", "ordinal": 0, "history_mode": "full",
})
// Activate but don't publish
h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"is_active": true,
})
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/start", h.adminToken, nil)
if resp.Code != http.StatusBadRequest {
t.Errorf("Start unpublished: got %d, want 400", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #15 — Reject validation
// ═══════════════════════════════════════════════════════════
// TestWorkflowReject exercises reject validation: reason required, stage-0 guard.
func TestWorkflowReject(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Reject Test", 3)
// Start
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
chID := start.ChannelID
// ── Reject from stage 0 should fail ─────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/reject", h.adminToken, map[string]interface{}{
"reason": "want to go back",
})
if resp.Code != http.StatusBadRequest {
t.Errorf("Reject at stage 0: got %d, want 400", resp.Code)
}
// ── Reject without reason should fail ───
// Advance to stage 1 first
h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"step": "0"},
})
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/reject", h.adminToken, nil)
if resp.Code != http.StatusBadRequest {
t.Errorf("Reject no reason: got %d, want 400", resp.Code)
}
// ── Reject with empty reason should fail ─
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/reject", h.adminToken, map[string]interface{}{
"reason": "",
})
if resp.Code != http.StatusBadRequest {
t.Errorf("Reject empty reason: got %d, want 400", resp.Code)
}
// ── Valid reject ────────────────────────
resp = h.request("POST", "/api/v1/channels/"+chID+"/workflow/reject", h.adminToken, map[string]interface{}{
"reason": "Missing email field",
})
if resp.Code != http.StatusOK {
t.Fatalf("Valid reject: got %d, body: %s", resp.Code, resp.Body.String())
}
var rejResp struct {
CurrentStage int `json:"current_stage"`
Reason string `json:"reason"`
}
json.Unmarshal(resp.Body.Bytes(), &rejResp)
if rejResp.CurrentStage != 0 {
t.Errorf("After reject: current_stage = %d, want 0", rejResp.CurrentStage)
}
if rejResp.Reason != "Missing email field" {
t.Errorf("Reject reason: got %q", rejResp.Reason)
}
}
// ═══════════════════════════════════════════════════════════
// #1 — Advance body field is "data" (not "stage_data")
// ═══════════════════════════════════════════════════════════
// TestWorkflowAdvance_BodyFieldName verifies the advance endpoint reads "data"
// from the request body, not "stage_data" (which the old ICD incorrectly documented).
func TestWorkflowAdvance_BodyFieldName(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Field Name Test", 2)
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
chID := start.ChannelID
// Send data under the correct "data" key
h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"collected": "yes"},
})
// Verify data was merged
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
var status struct{ StageData json.RawMessage `json:"stage_data"` }
json.Unmarshal(resp.Body.Bytes(), &status)
var data map[string]string
json.Unmarshal(status.StageData, &data)
if data["collected"] != "yes" {
t.Errorf("Data sent under 'data' key not merged: got %v", data)
}
}
// ═══════════════════════════════════════════════════════════
// #2 — GetStatus response uses "status" (not "workflow_status")
// ═══════════════════════════════════════════════════════════
// TestWorkflowGetStatus_ResponseShape verifies the exact JSON field names
// returned by the status endpoint.
func TestWorkflowGetStatus_ResponseShape(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Status Shape", 1)
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
resp = h.request("GET", "/api/v1/channels/"+start.ChannelID+"/workflow/status", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("GetStatus: %d", resp.Code)
}
// Parse as raw map to check exact field names
var raw map[string]interface{}
json.Unmarshal(resp.Body.Bytes(), &raw)
if _, ok := raw["status"]; !ok {
t.Error("Response missing 'status' field — should be 'status', not 'workflow_status'")
}
if _, ok := raw["workflow_status"]; ok {
t.Error("Response has 'workflow_status' — should be 'status' per ICD")
}
if _, ok := raw["workflow_id"]; !ok {
t.Error("Response missing 'workflow_id'")
}
if _, ok := raw["current_stage"]; !ok {
t.Error("Response missing 'current_stage'")
}
if _, ok := raw["last_activity_at"]; !ok {
t.Error("Response missing 'last_activity_at'")
}
}
// ═══════════════════════════════════════════════════════════
// #4 — StartVisitor response shape
// ═══════════════════════════════════════════════════════════
// TestWorkflowVisitorEntry_ResponseShape verifies the visitor start endpoint
// returns the correct field names (session_id + redirect_to, not session_token).
func TestWorkflowVisitorEntry_ResponseShape(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Visitor Test", 1)
// Read slug
resp := h.request("GET", "/api/v1/workflows/"+wfID, h.adminToken, nil)
var wf struct{ Slug string `json:"slug"` }
json.Unmarshal(resp.Body.Bytes(), &wf)
// Visitor start (no auth, no body)
resp = h.request("POST", "/api/v1/workflow-entry/global/"+wf.Slug, "", nil)
if resp.Code != http.StatusCreated {
t.Fatalf("Visitor start: got %d, body: %s", resp.Code, resp.Body.String())
}
var raw map[string]interface{}
json.Unmarshal(resp.Body.Bytes(), &raw)
if _, ok := raw["channel_id"]; !ok {
t.Error("Response missing 'channel_id'")
}
if _, ok := raw["session_id"]; !ok {
t.Error("Response missing 'session_id' — should be 'session_id', not 'session_token'")
}
if _, ok := raw["redirect_to"]; !ok {
t.Error("Response missing 'redirect_to'")
}
// These should NOT be present
if _, ok := raw["session_token"]; ok {
t.Error("Response has 'session_token' — field was renamed to 'session_id'")
}
if _, ok := raw["workflow"]; ok {
t.Error("Response has 'workflow' object — not in actual response")
}
}
// TestWorkflowVisitorEntry_TeamOnlyBlocked verifies team_only workflows reject visitors.
func TestWorkflowVisitorEntry_TeamOnlyBlocked(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Team Only WF",
"entry_mode": "team_only",
})
var wf struct {
ID string `json:"id"`
Slug string `json:"slug"`
}
json.Unmarshal(resp.Body.Bytes(), &wf)
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "S0", "ordinal": 0, "history_mode": "full",
})
h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"is_active": true,
})
h.request("POST", "/api/v1/workflows/"+wf.ID+"/publish", h.adminToken, nil)
resp = h.request("POST", "/api/v1/workflow-entry/global/"+wf.Slug, "", nil)
if resp.Code != http.StatusForbidden {
t.Errorf("Visitor on team_only: got %d, want 403", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #6/#7 — webhook_url/webhook_secret round-trip
// EXPECTED TO FAIL until stores are fixed to SELECT/INSERT these columns.
// ═══════════════════════════════════════════════════════════
// TestWorkflowWebhookFields verifies webhook_url and webhook_secret survive
// create → get round-trip. This WILL FAIL until the Postgres and SQLite
// workflow stores are updated to include these columns in their queries.
func TestWorkflowWebhookFields(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Webhook WF",
"webhook_url": "https://hooks.example.com/wf",
})
if resp.Code != http.StatusCreated {
t.Fatalf("Create with webhook_url: %d: %s", resp.Code, resp.Body.String())
}
var wf struct {
ID string `json:"id"`
WebhookURL string `json:"webhook_url"`
}
json.Unmarshal(resp.Body.Bytes(), &wf)
// BUG #6: webhook_url is not included in the store's INSERT/SELECT,
// so it will be empty on the create response.
if wf.WebhookURL != "https://hooks.example.com/wf" {
t.Errorf("Create response webhook_url: got %q, want %q (BUG #6: store doesn't persist webhook_url)",
wf.WebhookURL, "https://hooks.example.com/wf")
}
// Also verify via GET
resp = h.request("GET", "/api/v1/workflows/"+wf.ID, h.adminToken, nil)
json.Unmarshal(resp.Body.Bytes(), &wf)
if wf.WebhookURL != "https://hooks.example.com/wf" {
t.Errorf("GET webhook_url: got %q, want %q (BUG #6: store doesn't SELECT webhook_url)",
wf.WebhookURL, "https://hooks.example.com/wf")
}
// BUG #7: PATCH webhook_url should work
resp = h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"webhook_url": "https://hooks.example.com/updated",
})
if resp.Code != http.StatusOK {
t.Fatalf("PATCH webhook_url: %d: %s", resp.Code, resp.Body.String())
}
json.Unmarshal(resp.Body.Bytes(), &wf)
if wf.WebhookURL != "https://hooks.example.com/updated" {
t.Errorf("PATCH webhook_url: got %q, want %q (BUG #7: Update() ignores webhook_url)",
wf.WebhookURL, "https://hooks.example.com/updated")
}
}
// ═══════════════════════════════════════════════════════════
// #18 — ListMine should include unassigned-for-my-teams
// EXPECTED TO FAIL until ListMine query is fixed.
// ═══════════════════════════════════════════════════════════
// TestWorkflowAssignment_ListMineIncludesTeamUnassigned verifies that
// GET /workflow-assignments/mine returns both claimed assignments AND
// unassigned assignments for the user's teams.
func TestWorkflowAssignment_ListMineIncludesTeamUnassigned(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
// Create a team and add a member
memberID := seedInsertReturningID(t,
`INSERT INTO users (username, email, password_hash, role, handle, auth_source) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`,
"wf-member", "wf-member@test.com", "$2a$10$test", "user", "wf-member", "builtin",
)
memberToken := makeToken(memberID, "wf-member@test.com", "user")
resp := h.request("POST", "/api/v1/admin/teams", h.adminToken, map[string]string{
"name": "Review Team", "description": "Test team",
})
if resp.Code != http.StatusCreated {
t.Fatalf("Create team: %d: %s", resp.Code, resp.Body.String())
}
var team map[string]interface{}
decode(resp, &team)
teamID := team["id"].(string)
h.request("POST", fmt.Sprintf("/api/v1/admin/teams/%s/members", teamID), h.adminToken,
map[string]string{"user_id": memberID, "role": "member"})
// Create a workflow with an assignment stage
wfResp := h.request("POST", "/api/v1/workflows", h.adminToken, map[string]interface{}{
"name": "Assignment WF",
"entry_mode": "team_only",
})
var wf struct{ ID string `json:"id"` }
json.Unmarshal(wfResp.Body.Bytes(), &wf)
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "Intake", "ordinal": 0, "history_mode": "full",
})
h.request("POST", "/api/v1/workflows/"+wf.ID+"/stages", h.adminToken, map[string]interface{}{
"name": "Review",
"ordinal": 1,
"history_mode": "full",
"assignment_team_id": teamID,
})
h.request("PATCH", "/api/v1/workflows/"+wf.ID, h.adminToken, map[string]interface{}{
"is_active": true,
})
h.request("POST", "/api/v1/workflows/"+wf.ID+"/publish", h.adminToken, nil)
// Start instance and advance to stage 1 (creates an assignment)
resp = h.request("POST", "/api/v1/workflows/"+wf.ID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
h.request("POST", "/api/v1/channels/"+start.ChannelID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"intake": "done"},
})
// BUG #18: ListMine should include the unassigned assignment for
// the member's team, but currently only returns claimed assignments.
resp = h.request("GET", "/api/v1/workflow-assignments/mine", memberToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("ListMine: %d: %s", resp.Code, resp.Body.String())
}
var listResp struct {
Data []struct {
ID string `json:"id"`
Status string `json:"status"`
TeamID string `json:"team_id"`
To *string `json:"assigned_to"`
} `json:"data"`
}
json.Unmarshal(resp.Body.Bytes(), &listResp)
if len(listResp.Data) == 0 {
t.Error("ListMine returned 0 assignments — BUG #18: should include unassigned assignments for user's teams")
}
// If we do get results, verify the shape
for _, a := range listResp.Data {
if a.TeamID != teamID {
t.Errorf("Assignment team_id: got %q, want %q", a.TeamID, teamID)
}
}
}
// ═══════════════════════════════════════════════════════════
// Assignment Claim + Complete (basic happy path)
// ═══════════════════════════════════════════════════════════
// TestWorkflowAssignment_ClaimAndComplete exercises the claim → complete flow.
func TestWorkflowAssignment_ClaimAndComplete(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
// Seed assignment directly (simulates the advance handler creating one)
wfID, _ := h.createPublishedWorkflow("Assign WF", 2)
// Start + advance to create assignment stage
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
// Manually insert an assignment (since our stages don't have assignment_team_id)
teamID := seedInsertReturningID(t,
`INSERT INTO teams (name, description, created_by) VALUES ($1, $2, $3) RETURNING id`,
"Claim Team", "Test", h.adminID,
)
assignID := seedInsertReturningID(t,
`INSERT INTO workflow_assignments (channel_id, stage, team_id) VALUES ($1, $2, $3) RETURNING id`,
start.ChannelID, 0, teamID,
)
// Claim
resp = h.request("POST", "/api/v1/workflow-assignments/"+assignID+"/claim", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("Claim: got %d, body: %s", resp.Code, resp.Body.String())
}
var claimResp struct {
Claimed bool `json:"claimed"`
AssignmentID string `json:"assignment_id"`
}
json.Unmarshal(resp.Body.Bytes(), &claimResp)
if !claimResp.Claimed {
t.Error("Claim: claimed should be true")
}
// Double-claim should conflict
resp = h.request("POST", "/api/v1/workflow-assignments/"+assignID+"/claim", h.adminToken, nil)
if resp.Code != http.StatusConflict {
t.Errorf("Double claim: got %d, want 409", resp.Code)
}
// Complete
resp = h.request("POST", "/api/v1/workflow-assignments/"+assignID+"/complete", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("Complete: got %d, body: %s", resp.Code, resp.Body.String())
}
// Double-complete should conflict
resp = h.request("POST", "/api/v1/workflow-assignments/"+assignID+"/complete", h.adminToken, nil)
if resp.Code != http.StatusConflict {
t.Errorf("Double complete: got %d, want 409", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #15 — Status on non-workflow channel returns 404
// ═══════════════════════════════════════════════════════════
func TestWorkflowGetStatus_NonWorkflowChannel(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("GET", "/api/v1/channels/00000000-0000-0000-0000-000000000000/workflow/status", h.adminToken, nil)
if resp.Code != http.StatusNotFound {
t.Errorf("Status on non-existent channel: got %d, want 404", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #15 — Data merge across multiple advances
// ═══════════════════════════════════════════════════════════
// TestWorkflowAdvance_DataMerge verifies that stage data accumulates
// across multiple advance calls (merge, not replace).
func TestWorkflowAdvance_DataMerge(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
wfID, _ := h.createPublishedWorkflow("Merge Test", 3)
resp := h.request("POST", "/api/v1/workflows/"+wfID+"/start", h.adminToken, nil)
var start struct{ ChannelID string `json:"channel_id"` }
json.Unmarshal(resp.Body.Bytes(), &start)
chID := start.ChannelID
// Advance stage 0 → 1 with key "a"
h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"a": "1"},
})
// Advance stage 1 → 2 with key "b"
h.request("POST", "/api/v1/channels/"+chID+"/workflow/advance", h.adminToken, map[string]interface{}{
"data": map[string]string{"b": "2"},
})
// Check accumulated data contains both keys
resp = h.request("GET", "/api/v1/channels/"+chID+"/workflow/status", h.adminToken, nil)
var status struct{ StageData json.RawMessage `json:"stage_data"` }
json.Unmarshal(resp.Body.Bytes(), &status)
var merged map[string]string
json.Unmarshal(status.StageData, &merged)
if merged["a"] != "1" {
t.Errorf("Merged data missing key 'a': got %v", merged)
}
if merged["b"] != "2" {
t.Errorf("Merged data missing key 'b': got %v", merged)
}
}
// ═══════════════════════════════════════════════════════════
// #17 — PATCH non-existent workflow
// ═══════════════════════════════════════════════════════════
// TestWorkflowPatchNonExistent verifies PATCH to a non-existent ID returns 404.
// Documents the confusing code path: Update affects 0 rows (no error),
// then GetByID returns sql.ErrNoRows → 404.
func TestWorkflowPatchNonExistent(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
resp := h.request("PATCH", "/api/v1/workflows/00000000-0000-0000-0000-000000000000", h.adminToken, map[string]interface{}{
"name": "ghost",
})
if resp.Code != http.StatusNotFound {
t.Errorf("PATCH non-existent: got %d, want 404", resp.Code)
}
}
// ═══════════════════════════════════════════════════════════
// #19 — ListForTeam with ?status= filter
// ═══════════════════════════════════════════════════════════
// TestWorkflowAssignment_ListForTeamStatusFilter verifies the ?status=
// query param on GET /teams/:teamId/assignments.
func TestWorkflowAssignment_ListForTeamStatusFilter(t *testing.T) {
h := setupWorkflowInstanceHarness(t)
// Create team
resp := h.request("POST", "/api/v1/admin/teams", h.adminToken, map[string]string{
"name": "Filter Team", "description": "Test",
})
if resp.Code != http.StatusCreated {
t.Fatalf("Create team: %d: %s", resp.Code, resp.Body.String())
}
var team map[string]interface{}
decode(resp, &team)
teamID := team["id"].(string)
// Add self as member (needed for access)
h.request("POST", fmt.Sprintf("/api/v1/admin/teams/%s/members", teamID), h.adminToken,
map[string]string{"user_id": h.adminID, "role": "admin"})
// Create real channels for FK satisfaction
ch1ID := database.SeedTestChannel(t, h.adminID, "filter-ch-1")
ch2ID := database.SeedTestChannel(t, h.adminID, "filter-ch-2")
// Seed two assignments: one unassigned, one claimed
seedExec(t,
`INSERT INTO workflow_assignments (channel_id, stage, team_id, status) VALUES ($1, $2, $3, $4)`,
ch1ID, 0, teamID, "unassigned",
)
seedExec(t,
`INSERT INTO workflow_assignments (channel_id, stage, team_id, status, assigned_to) VALUES ($1, $2, $3, $4, $5)`,
ch2ID, 1, teamID, "claimed", h.adminID,
)
// Default (unassigned)
resp = h.request("GET", "/api/v1/teams/"+teamID+"/assignments", h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("ListForTeam default: %d: %s", resp.Code, resp.Body.String())
}
var lr struct {
Data []struct{ Status string `json:"status"` } `json:"data"`
}
json.Unmarshal(resp.Body.Bytes(), &lr)
if len(lr.Data) != 1 {
t.Errorf("Default filter: expected 1 unassigned, got %d", len(lr.Data))
} else if lr.Data[0].Status != "unassigned" {
t.Errorf("Default filter: expected status 'unassigned', got %q", lr.Data[0].Status)
}
// Explicit ?status=claimed
resp = h.request("GET", "/api/v1/teams/"+teamID+"/assignments?status=claimed", h.adminToken, nil)
json.Unmarshal(resp.Body.Bytes(), &lr)
if len(lr.Data) != 1 {
t.Errorf("?status=claimed: expected 1, got %d", len(lr.Data))
} else if lr.Data[0].Status != "claimed" {
t.Errorf("?status=claimed: expected status 'claimed', got %q", lr.Data[0].Status)
}
}

8
server/handlers/workflow_instances.go
View File

@@ -133,7 +133,7 @@ type WorkflowChannelStatus struct {
CurrentStage int `json:"current_stage"`
StageData json.RawMessage `json:"stage_data"`
Status string `json:"status"`
LastActivityAt *time.Time `json:"last_activity_at"`
LastActivityAt *string `json:"last_activity_at"`
}
// GetStatus returns the workflow state for a channel.
@@ -391,19 +391,19 @@ func (h *WorkflowInstanceHandler) tryRoundRobin(ctx context.Context, stage model
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
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 NULLS FIRST
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 *time.Time
var lastClaim string // COALESCE returns TEXT on both dialects
if err := rows.Scan(&uid, &lastClaim); err == nil {
bestUserID = uid
}

2
server/handlers/workflow_test.go
View File

@@ -53,7 +53,7 @@ func TestWorkflowCRUD(t *testing.T) {
resp = h.request("GET", "/api/v1/workflows/"+wf.ID, h.adminToken, nil)
if resp.Code != http.StatusOK {
t.Fatalf("Get workflow: got %d", resp.Code)
t.Fatalf("Get workflow: got %d, body: %s", resp.Code, resp.Body.String())
}
// ── List workflows ──────────────────────

12
server/handlers/workflows.go
View File

@@ -61,7 +61,11 @@ func (h *WorkflowHandler) List(c *gin.Context) {
func (h *WorkflowHandler) Get(c *gin.Context) {
w, err := h.stores.Workflows.GetByID(c.Request.Context(), c.Param("id"))
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"})
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get workflow: " + err.Error()})
return
}
stages, _ := h.stores.Workflows.ListStages(c.Request.Context(), w.ID)
@@ -251,7 +255,11 @@ func (h *WorkflowHandler) Publish(c *gin.Context) {
wfID := c.Param("id")
w, err := h.stores.Workflows.GetByID(c.Request.Context(), wfID)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"})
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get workflow: " + err.Error()})
return
}

4
server/main.go
View File

@@ -605,7 +605,7 @@ func main() {
protected.POST("/channels/:id/workflow/reject", wfInstH.Reject)
// Workflow assignments (v0.26.4 — team assignment queue)
wfAssignH := handlers.NewWorkflowAssignmentHandler()
wfAssignH := handlers.NewWorkflowAssignmentHandler(hub)
protected.GET("/workflow-assignments/mine", wfAssignH.ListMine)
protected.POST("/workflow-assignments/:id/claim", wfAssignH.Claim)
protected.POST("/workflow-assignments/:id/complete", wfAssignH.Complete)
@@ -907,7 +907,7 @@ func main() {
teamScoped.DELETE("/roles/:role", teamRoles.DeleteTeamRole)
// Team workflow assignments (v0.26.4)
teamAssignH := handlers.NewWorkflowAssignmentHandler()
teamAssignH := handlers.NewWorkflowAssignmentHandler(hub)
teamScoped.GET("/assignments", teamAssignH.ListForTeam)
// Team tasks — admin CRUD (v0.27.5)

3
server/models/workflow.go
View File

@@ -39,7 +39,8 @@ type WorkflowPatch struct {
IsActive *bool `json:"is_active,omitempty"`
OnComplete *json.RawMessage `json:"on_complete,omitempty"`
Retention *json.RawMessage `json:"retention,omitempty"`
WebhookURL *string `json:"webhook_url,omitempty"`
WebhookURL *string `json:"webhook_url,omitempty"`
WebhookSecret *string `json:"webhook_secret,omitempty"`
}
// ── Workflow Stage ──────────────────────────

61
server/store/postgres/workflows.go
View File

@@ -19,23 +19,27 @@ func (s *WorkflowStore) Create(ctx context.Context, w *models.Workflow) error {
branding := jsonOrEmpty(w.Branding)
retention := jsonOrDefault(w.Retention, `{"mode":"archive"}`)
return DB.QueryRowContext(ctx, `
INSERT INTO workflows (team_id, name, slug, description, branding, entry_mode, is_active, on_complete, retention, created_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
INSERT INTO workflows (team_id, name, slug, description, branding, entry_mode, is_active, on_complete, retention, webhook_url, webhook_secret, created_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
RETURNING id, version, created_at, updated_at`,
w.TeamID, w.Name, w.Slug, w.Description, branding, w.EntryMode,
w.IsActive, jsonOrNull(w.OnComplete), retention, w.CreatedBy,
w.IsActive, jsonOrNull(w.OnComplete), retention,
nullIfEmpty(w.WebhookURL), nullIfEmpty(w.WebhookSecret), w.CreatedBy,
).Scan(&w.ID, &w.Version, &w.CreatedAt, &w.UpdatedAt)
}
func (s *WorkflowStore) GetByID(ctx context.Context, id string) (*models.Workflow, error) {
w := &models.Workflow{}
var branding, retention, onComplete []byte
var webhookURL, webhookSecret *string
err := DB.QueryRowContext(ctx, `
SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE id = $1`, id,
).Scan(&w.ID, &w.TeamID, &w.Name, &w.Slug, &w.Description, &branding,
&w.EntryMode, &w.IsActive, &w.Version, &onComplete, &retention,
&webhookURL, &webhookSecret,
&w.CreatedBy, &w.CreatedAt, &w.UpdatedAt)
if err != nil {
return nil, err
@@ -43,6 +47,12 @@ func (s *WorkflowStore) GetByID(ctx context.Context, id string) (*models.Workflo
w.Branding = branding
w.OnComplete = onComplete
w.Retention = retention
if webhookURL != nil {
w.WebhookURL = *webhookURL
}
if webhookSecret != nil {
w.WebhookSecret = *webhookSecret
}
return w, nil
}
@@ -51,20 +61,24 @@ func (s *WorkflowStore) GetBySlug(ctx context.Context, teamID *string, slug stri
var args []interface{}
if teamID != nil {
q = `SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE team_id = $1 AND slug = $2`
args = []interface{}{*teamID, slug}
} else {
q = `SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE team_id IS NULL AND slug = $1`
args = []interface{}{slug}
}
w := &models.Workflow{}
var branding, retention, onComplete []byte
var webhookURL, webhookSecret *string
err := DB.QueryRowContext(ctx, q, args...).Scan(
&w.ID, &w.TeamID, &w.Name, &w.Slug, &w.Description, &branding,
&w.EntryMode, &w.IsActive, &w.Version, &onComplete, &retention,
&webhookURL, &webhookSecret,
&w.CreatedBy, &w.CreatedAt, &w.UpdatedAt)
if err != nil {
return nil, err
@@ -72,6 +86,12 @@ func (s *WorkflowStore) GetBySlug(ctx context.Context, teamID *string, slug stri
w.Branding = branding
w.OnComplete = onComplete
w.Retention = retention
if webhookURL != nil {
w.WebhookURL = *webhookURL
}
if webhookSecret != nil {
w.WebhookSecret = *webhookSecret
}
return w, nil
}
@@ -108,6 +128,12 @@ func (s *WorkflowStore) Update(ctx context.Context, id string, patch models.Work
if patch.Retention != nil {
add("retention", string(*patch.Retention))
}
if patch.WebhookURL != nil {
add("webhook_url", *patch.WebhookURL)
}
if patch.WebhookSecret != nil {
add("webhook_secret", *patch.WebhookSecret)
}
if len(sets) == 0 {
return nil
@@ -138,7 +164,8 @@ func (s *WorkflowStore) Delete(ctx context.Context, id string) error {
func (s *WorkflowStore) ListForTeam(ctx context.Context, teamID string) ([]models.Workflow, error) {
return s.queryWorkflows(ctx, `
SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE team_id = $1
ORDER BY name ASC`, teamID)
}
@@ -146,7 +173,8 @@ func (s *WorkflowStore) ListForTeam(ctx context.Context, teamID string) ([]model
func (s *WorkflowStore) ListGlobal(ctx context.Context) ([]models.Workflow, error) {
return s.queryWorkflows(ctx, `
SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE team_id IS NULL
ORDER BY name ASC`)
}
@@ -161,14 +189,22 @@ func (s *WorkflowStore) queryWorkflows(ctx context.Context, q string, args ...in
for rows.Next() {
var w models.Workflow
var branding, retention, onComplete []byte
var webhookURL, webhookSecret *string
if err := rows.Scan(&w.ID, &w.TeamID, &w.Name, &w.Slug, &w.Description,
&branding, &w.EntryMode, &w.IsActive, &w.Version, &onComplete,
&retention, &w.CreatedBy, &w.CreatedAt, &w.UpdatedAt); err != nil {
&retention, &webhookURL, &webhookSecret,
&w.CreatedBy, &w.CreatedAt, &w.UpdatedAt); err != nil {
return nil, err
}
w.Branding = branding
w.OnComplete = onComplete
w.Retention = retention
if webhookURL != nil {
w.WebhookURL = *webhookURL
}
if webhookSecret != nil {
w.WebhookSecret = *webhookSecret
}
result = append(result, w)
}
return result, rows.Err()
@@ -314,3 +350,10 @@ func jsonOrNull(b json.RawMessage) interface{} {
}
return string(b)
}
func nullIfEmpty(s string) interface{} {
if s == "" {
return nil
}
return s
}

4
server/store/sqlite/routing_policy.go
View File

@@ -56,7 +56,7 @@ func (s *RoutingPolicyStore) GetByID(ctx context.Context, id string) (*models.Ro
err := DB.QueryRowContext(ctx, `
SELECT id, name, scope, team_id, priority, policy_type, config, is_active, created_at, updated_at
FROM routing_policies WHERE id = ?
`, id).Scan(&p.ID, &p.Name, &p.Scope, &p.TeamID, &p.Priority, &p.Type, &configJSON, &isActive, &p.CreatedAt, &p.UpdatedAt)
`, id).Scan(&p.ID, &p.Name, &p.Scope, &p.TeamID, &p.Priority, &p.Type, &configJSON, &isActive, st(&p.CreatedAt), st(&p.UpdatedAt))
if err != nil {
return nil, err
}
@@ -103,7 +103,7 @@ func (s *RoutingPolicyStore) query(ctx context.Context, q string, args ...interf
var p models.RoutingPolicy
var configJSON string
var isActive int
if err := rows.Scan(&p.ID, &p.Name, &p.Scope, &p.TeamID, &p.Priority, &p.Type, &configJSON, &isActive, &p.CreatedAt, &p.UpdatedAt); err != nil {
if err := rows.Scan(&p.ID, &p.Name, &p.Scope, &p.TeamID, &p.Priority, &p.Type, &configJSON, &isActive, st(&p.CreatedAt), st(&p.UpdatedAt)); err != nil {
return nil, err
}
p.IsActive = isActive != 0

8
server/store/sqlite/sessions.go
View File

@@ -28,7 +28,7 @@ func (s *SessionStore) Create(ctx context.Context, sp *models.SessionParticipant
return err
}
return DB.QueryRowContext(ctx, `SELECT created_at FROM session_participants WHERE id = ?`, sp.ID).
Scan(&sp.CreatedAt)
Scan(st(&sp.CreatedAt))
}
func (s *SessionStore) GetByToken(ctx context.Context, token string) (*models.SessionParticipant, error) {
@@ -36,7 +36,7 @@ func (s *SessionStore) GetByToken(ctx context.Context, token string) (*models.Se
err := DB.QueryRowContext(ctx, `
SELECT id, session_token, channel_id, display_name, COALESCE(fingerprint, ''), created_at
FROM session_participants WHERE session_token = ?`, token,
).Scan(&sp.ID, &sp.SessionToken, &sp.ChannelID, &sp.DisplayName, &sp.Fingerprint, &sp.CreatedAt)
).Scan(&sp.ID, &sp.SessionToken, &sp.ChannelID, &sp.DisplayName, &sp.Fingerprint, st(&sp.CreatedAt))
if err != nil {
return nil, err
}
@@ -48,7 +48,7 @@ func (s *SessionStore) GetByID(ctx context.Context, id string) (*models.SessionP
err := DB.QueryRowContext(ctx, `
SELECT id, session_token, channel_id, display_name, COALESCE(fingerprint, ''), created_at
FROM session_participants WHERE id = ?`, id,
).Scan(&sp.ID, &sp.SessionToken, &sp.ChannelID, &sp.DisplayName, &sp.Fingerprint, &sp.CreatedAt)
).Scan(&sp.ID, &sp.SessionToken, &sp.ChannelID, &sp.DisplayName, &sp.Fingerprint, st(&sp.CreatedAt))
if err != nil {
return nil, err
}
@@ -68,7 +68,7 @@ func (s *SessionStore) ListForChannel(ctx context.Context, channelID string) ([]
var result []models.SessionParticipant
for rows.Next() {
var sp models.SessionParticipant
if err := rows.Scan(&sp.ID, &sp.SessionToken, &sp.ChannelID, &sp.DisplayName, &sp.Fingerprint, &sp.CreatedAt); err != nil {
if err := rows.Scan(&sp.ID, &sp.SessionToken, &sp.ChannelID, &sp.DisplayName, &sp.Fingerprint, st(&sp.CreatedAt)); err != nil {
return nil, err
}
result = append(result, sp)

10
server/store/sqlite/tasks.go
View File

@@ -47,7 +47,7 @@ func (s *TaskStore) GetByID(ctx context.Context, id string) (*models.Task, error
&t.MaxTokens, &t.MaxToolCalls, &t.MaxWallClock, &t.OutputMode,
&t.OutputChannelID, &t.WebhookURL, &t.ProviderConfigID,
&t.NotifyOnComplete, &t.NotifyOnFailure,
&t.LastRunAt, &t.NextRunAt, &t.RunCount, &t.CreatedAt, &t.UpdatedAt)
stN(&t.LastRunAt), stN(&t.NextRunAt), &t.RunCount, st(&t.CreatedAt), st(&t.UpdatedAt))
if err != nil {
return nil, err
}
@@ -131,7 +131,7 @@ func (s *TaskStore) GetActiveRun(ctx context.Context, taskID string) (*models.Ta
err := DB.QueryRowContext(ctx, `
SELECT id, task_id, channel_id, status, started_at
FROM task_runs WHERE task_id = ? AND status = 'running'
LIMIT 1`, taskID).Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.StartedAt)
LIMIT 1`, taskID).Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, st(&r.StartedAt))
if err != nil {
return nil, err
}
@@ -151,8 +151,8 @@ func (s *TaskStore) ListRuns(ctx context.Context, taskID string, limit int) ([]m
var runs []models.TaskRun
for rows.Next() {
var r models.TaskRun
if err := rows.Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.StartedAt,
&r.CompletedAt, &r.TokensUsed, &r.ToolCalls, &r.WallClock, &r.Error); err != nil {
if err := rows.Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, st(&r.StartedAt),
stN(&r.CompletedAt), &r.TokensUsed, &r.ToolCalls, &r.WallClock, &r.Error); err != nil {
continue
}
runs = append(runs, r)
@@ -183,7 +183,7 @@ func (s *TaskStore) list(ctx context.Context, where string, args ...interface{})
&t.WorkflowID, &t.Schedule, &t.Timezone, &t.IsActive,
&t.MaxTokens, &t.MaxToolCalls, &t.MaxWallClock, &t.OutputMode,
&t.NotifyOnComplete, &t.NotifyOnFailure,
&t.LastRunAt, &t.NextRunAt, &t.RunCount, &t.CreatedAt, &t.UpdatedAt); err != nil {
stN(&t.LastRunAt), stN(&t.NextRunAt), &t.RunCount, st(&t.CreatedAt), st(&t.UpdatedAt)); err != nil {
continue
}
tasks = append(tasks, t)

76
server/store/sqlite/workflows.go
View File

@@ -27,10 +27,12 @@ func (s *WorkflowStore) Create(ctx context.Context, w *models.Workflow) error {
retention := jsonOrDefault(w.Retention, `{"mode":"archive"}`)
_, err := DB.ExecContext(ctx, `
INSERT INTO workflows (id, team_id, name, slug, description, branding, entry_mode,
is_active, version, on_complete, retention, created_by, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
is_active, version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
w.ID, w.TeamID, w.Name, w.Slug, w.Description, branding, w.EntryMode,
boolToInt(w.IsActive), w.Version, jsonOrNull(w.OnComplete), retention,
nullIfEmpty(w.WebhookURL), nullIfEmpty(w.WebhookSecret),
w.CreatedBy, w.CreatedAt.Format(time.RFC3339), w.UpdatedAt.Format(time.RFC3339))
return err
}
@@ -38,7 +40,8 @@ func (s *WorkflowStore) Create(ctx context.Context, w *models.Workflow) error {
func (s *WorkflowStore) GetByID(ctx context.Context, id string) (*models.Workflow, error) {
return s.scanOne(ctx, `
SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE id = ?`, id)
}
@@ -46,12 +49,14 @@ func (s *WorkflowStore) GetBySlug(ctx context.Context, teamID *string, slug stri
if teamID != nil {
return s.scanOne(ctx, `
SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE team_id = ? AND slug = ?`, *teamID, slug)
}
return s.scanOne(ctx, `
SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE team_id IS NULL AND slug = ?`, slug)
}
@@ -87,6 +92,14 @@ func (s *WorkflowStore) Update(ctx context.Context, id string, patch models.Work
sets = append(sets, "retention = ?")
args = append(args, string(*patch.Retention))
}
if patch.WebhookURL != nil {
sets = append(sets, "webhook_url = ?")
args = append(args, *patch.WebhookURL)
}
if patch.WebhookSecret != nil {
sets = append(sets, "webhook_secret = ?")
args = append(args, *patch.WebhookSecret)
}
if len(sets) == 0 {
return nil
@@ -118,7 +131,8 @@ func (s *WorkflowStore) Delete(ctx context.Context, id string) error {
func (s *WorkflowStore) ListForTeam(ctx context.Context, teamID string) ([]models.Workflow, error) {
return s.queryWorkflows(ctx, `
SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE team_id = ?
ORDER BY name ASC`, teamID)
}
@@ -126,7 +140,8 @@ func (s *WorkflowStore) ListForTeam(ctx context.Context, teamID string) ([]model
func (s *WorkflowStore) ListGlobal(ctx context.Context) ([]models.Workflow, error) {
return s.queryWorkflows(ctx, `
SELECT id, team_id, name, slug, description, branding, entry_mode, is_active,
version, on_complete, retention, created_by, created_at, updated_at
version, on_complete, retention, webhook_url, webhook_secret,
created_by, created_at, updated_at
FROM workflows WHERE team_id IS NULL
ORDER BY name ASC`)
}
@@ -160,18 +175,18 @@ func (s *WorkflowStore) ListStages(ctx context.Context, workflowID string) ([]mo
defer rows.Close()
var result []models.WorkflowStage
for rows.Next() {
var st models.WorkflowStage
var stg models.WorkflowStage
var formTpl, transRules string
var autoTrans int
if err := rows.Scan(&st.ID, &st.WorkflowID, &st.Ordinal, &st.Name,
&st.PersonaID, &st.AssignmentTeamID, &formTpl, &st.HistoryMode,
&autoTrans, &transRules, &st.CreatedAt); err != nil {
if err := rows.Scan(&stg.ID, &stg.WorkflowID, &stg.Ordinal, &stg.Name,
&stg.PersonaID, &stg.AssignmentTeamID, &formTpl, &stg.HistoryMode,
&autoTrans, &transRules, st(&stg.CreatedAt)); err != nil {
return nil, err
}
st.FormTemplate = json.RawMessage(formTpl)
st.TransitionRules = json.RawMessage(transRules)
st.AutoTransition = autoTrans != 0
result = append(result, st)
stg.FormTemplate = json.RawMessage(formTpl)
stg.TransitionRules = json.RawMessage(transRules)
stg.AutoTransition = autoTrans != 0
result = append(result, stg)
}
return result, rows.Err()
}
@@ -231,7 +246,7 @@ func (s *WorkflowStore) GetVersion(ctx context.Context, workflowID string, versi
SELECT id, workflow_id, version_number, snapshot, created_at
FROM workflow_versions WHERE workflow_id = ? AND version_number = ?`,
workflowID, versionNumber,
).Scan(&v.ID, &v.WorkflowID, &v.VersionNumber, &snapshot, &v.CreatedAt)
).Scan(&v.ID, &v.WorkflowID, &v.VersionNumber, &snapshot, st(&v.CreatedAt))
if err != nil {
return nil, err
}
@@ -247,7 +262,7 @@ func (s *WorkflowStore) GetLatestVersion(ctx context.Context, workflowID string)
FROM workflow_versions WHERE workflow_id = ?
ORDER BY version_number DESC LIMIT 1`,
workflowID,
).Scan(&v.ID, &v.WorkflowID, &v.VersionNumber, &snapshot, &v.CreatedAt)
).Scan(&v.ID, &v.WorkflowID, &v.VersionNumber, &snapshot, st(&v.CreatedAt))
if err != nil {
return nil, err
}
@@ -261,11 +276,13 @@ func (s *WorkflowStore) scanOne(ctx context.Context, q string, args ...interface
w := &models.Workflow{}
var branding, retention string
var onComplete sql.NullString
var webhookURL, webhookSecret sql.NullString
var isActive int
err := DB.QueryRowContext(ctx, q, args...).Scan(
&w.ID, &w.TeamID, &w.Name, &w.Slug, &w.Description, &branding,
&w.EntryMode, &isActive, &w.Version, &onComplete, &retention,
&w.CreatedBy, &w.CreatedAt, &w.UpdatedAt)
&webhookURL, &webhookSecret,
&w.CreatedBy, st(&w.CreatedAt), st(&w.UpdatedAt))
if err != nil {
return nil, err
}
@@ -275,6 +292,12 @@ func (s *WorkflowStore) scanOne(ctx context.Context, q string, args ...interface
if onComplete.Valid {
w.OnComplete = json.RawMessage(onComplete.String)
}
if webhookURL.Valid {
w.WebhookURL = webhookURL.String
}
if webhookSecret.Valid {
w.WebhookSecret = webhookSecret.String
}
return w, nil
}
@@ -289,10 +312,12 @@ func (s *WorkflowStore) queryWorkflows(ctx context.Context, q string, args ...in
var w models.Workflow
var branding, retention string
var onComplete sql.NullString
var webhookURL, webhookSecret sql.NullString
var isActive int
if err := rows.Scan(&w.ID, &w.TeamID, &w.Name, &w.Slug, &w.Description,
&branding, &w.EntryMode, &isActive, &w.Version, &onComplete,
&retention, &w.CreatedBy, &w.CreatedAt, &w.UpdatedAt); err != nil {
&retention, &webhookURL, &webhookSecret,
&w.CreatedBy, st(&w.CreatedAt), st(&w.UpdatedAt)); err != nil {
return nil, err
}
w.Branding = json.RawMessage(branding)
@@ -301,6 +326,12 @@ func (s *WorkflowStore) queryWorkflows(ctx context.Context, q string, args ...in
if onComplete.Valid {
w.OnComplete = json.RawMessage(onComplete.String)
}
if webhookURL.Valid {
w.WebhookURL = webhookURL.String
}
if webhookSecret.Valid {
w.WebhookSecret = webhookSecret.String
}
result = append(result, w)
}
return result, rows.Err()
@@ -326,3 +357,10 @@ func jsonOrNull(b json.RawMessage) interface{} {
}
return string(b)
}
func nullIfEmpty(s string) interface{} {
if s == "" {
return nil
}
return s
}

11
server/tools/workflow.go
View File

@@ -352,11 +352,12 @@ func TriggerWorkflowOnComplete(ctx context.Context, stores store.Stores, workflo
// Bind first stage persona
if len(stages) > 0 && stages[0].PersonaID != nil {
_, _ = database.DB.ExecContext(ctx, database.Q(`
INSERT INTO channel_participants (channel_id, participant_type, participant_id)
VALUES ($1, 'persona', $2)
ON CONFLICT DO NOTHING
`), ch.ID, *stages[0].PersonaID)
_ = stores.Channels.AddParticipant(ctx, &models.ChannelParticipant{
ChannelID: ch.ID,
ParticipantType: "persona",
ParticipantID: *stages[0].PersonaID,
Role: "member",
})
}
log.Printf("[workflow] on_complete: started chained workflow %s → %s (channel %s)",