package postgres import ( "context" "database/sql" "encoding/json" "fmt" "time" "chat-switchboard/models" "chat-switchboard/store" ) // WorkflowStore implements store.WorkflowStore for Postgres. type WorkflowStore struct{} func NewWorkflowStore() *WorkflowStore { return &WorkflowStore{} } // ── Workflow CRUD ─────────────────────────── 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, 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, 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, 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 } w.Branding = branding w.OnComplete = onComplete w.Retention = retention if webhookURL != nil { w.WebhookURL = *webhookURL } if webhookSecret != nil { w.WebhookSecret = *webhookSecret } return w, nil } func (s *WorkflowStore) GetBySlug(ctx context.Context, teamID *string, slug string) (*models.Workflow, error) { var q string var args []interface{} if teamID != nil { q = `SELECT id, team_id, name, slug, description, branding, entry_mode, is_active, 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, 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 } w.Branding = branding w.OnComplete = onComplete w.Retention = retention if webhookURL != nil { w.WebhookURL = *webhookURL } if webhookSecret != nil { w.WebhookSecret = *webhookSecret } return w, nil } func (s *WorkflowStore) Update(ctx context.Context, id string, patch models.WorkflowPatch) error { // Build dynamic SET clause sets := []string{} args := []interface{}{} idx := 1 add := func(col string, val interface{}) { sets = append(sets, fmt.Sprintf("%s = $%d", col, idx)) args = append(args, val) idx++ } if patch.Name != nil { add("name", *patch.Name) } if patch.Description != nil { add("description", *patch.Description) } if patch.Branding != nil { add("branding", string(*patch.Branding)) } if patch.EntryMode != nil { add("entry_mode", *patch.EntryMode) } if patch.IsActive != nil { add("is_active", *patch.IsActive) } if patch.OnComplete != nil { add("on_complete", string(*patch.OnComplete)) } 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 } // Increment version on any edit sets = append(sets, fmt.Sprintf("version = version + 1")) q := "UPDATE workflows SET " for i, s := range sets { if i > 0 { q += ", " } q += s } q += fmt.Sprintf(" WHERE id = $%d", idx) args = append(args, id) _, err := DB.ExecContext(ctx, q, args...) return err } func (s *WorkflowStore) Delete(ctx context.Context, id string) error { _, err := DB.ExecContext(ctx, `DELETE FROM workflows WHERE id = $1`, id) return err } 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, webhook_url, webhook_secret, created_by, created_at, updated_at FROM workflows WHERE team_id = $1 ORDER BY name ASC`, teamID) } 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, webhook_url, webhook_secret, created_by, created_at, updated_at FROM workflows WHERE team_id IS NULL ORDER BY name ASC`) } func (s *WorkflowStore) queryWorkflows(ctx context.Context, q string, args ...interface{}) ([]models.Workflow, error) { rows, err := DB.QueryContext(ctx, q, args...) if err != nil { return nil, err } defer rows.Close() var result []models.Workflow 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, &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() } // ── Stages ────────────────────────────────── func (s *WorkflowStore) CreateStage(ctx context.Context, st *models.WorkflowStage) error { formTpl := jsonOrEmpty(st.FormTemplate) transRules := jsonOrEmpty(st.TransitionRules) stageMode := st.StageMode if stageMode == "" { stageMode = models.StageModeChatOnly } return DB.QueryRowContext(ctx, ` INSERT INTO workflow_stages (workflow_id, ordinal, name, persona_id, assignment_team_id, form_template, stage_mode, history_mode, auto_transition, transition_rules, surface_pkg_id, sla_seconds) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) RETURNING id, created_at`, st.WorkflowID, st.Ordinal, st.Name, st.PersonaID, st.AssignmentTeamID, formTpl, stageMode, st.HistoryMode, st.AutoTransition, transRules, st.SurfacePkgID, st.SLASeconds, ).Scan(&st.ID, &st.CreatedAt) } func (s *WorkflowStore) ListStages(ctx context.Context, workflowID string) ([]models.WorkflowStage, error) { rows, err := DB.QueryContext(ctx, ` SELECT id, workflow_id, ordinal, name, persona_id, assignment_team_id, form_template, stage_mode, history_mode, auto_transition, transition_rules, surface_pkg_id, sla_seconds, created_at FROM workflow_stages WHERE workflow_id = $1 ORDER BY ordinal ASC`, workflowID) if err != nil { return nil, err } defer rows.Close() var result []models.WorkflowStage for rows.Next() { var st models.WorkflowStage var formTpl, transRules []byte if err := rows.Scan(&st.ID, &st.WorkflowID, &st.Ordinal, &st.Name, &st.PersonaID, &st.AssignmentTeamID, &formTpl, &st.StageMode, &st.HistoryMode, &st.AutoTransition, &transRules, &st.SurfacePkgID, &st.SLASeconds, &st.CreatedAt); err != nil { return nil, err } st.FormTemplate = formTpl st.TransitionRules = transRules result = append(result, st) } return result, rows.Err() } func (s *WorkflowStore) UpdateStage(ctx context.Context, st *models.WorkflowStage) error { formTpl := jsonOrEmpty(st.FormTemplate) transRules := jsonOrEmpty(st.TransitionRules) stageMode := st.StageMode if stageMode == "" { stageMode = models.StageModeChatOnly } _, err := DB.ExecContext(ctx, ` UPDATE workflow_stages SET ordinal = $2, name = $3, persona_id = $4, assignment_team_id = $5, form_template = $6, stage_mode = $7, history_mode = $8, auto_transition = $9, transition_rules = $10, surface_pkg_id = $11, sla_seconds = $12 WHERE id = $1`, st.ID, st.Ordinal, st.Name, st.PersonaID, st.AssignmentTeamID, formTpl, stageMode, st.HistoryMode, st.AutoTransition, transRules, st.SurfacePkgID, st.SLASeconds) return err } func (s *WorkflowStore) DeleteStage(ctx context.Context, id string) error { _, err := DB.ExecContext(ctx, `DELETE FROM workflow_stages WHERE id = $1`, id) return err } func (s *WorkflowStore) ReorderStages(ctx context.Context, workflowID string, orderedIDs []string) error { tx, err := DB.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() for i, id := range orderedIDs { if _, err := tx.ExecContext(ctx, ` UPDATE workflow_stages SET ordinal = $1 WHERE id = $2 AND workflow_id = $3`, i, id, workflowID); err != nil { return err } } return tx.Commit() } // ── Versions ──────────────────────────────── func (s *WorkflowStore) Publish(ctx context.Context, v *models.WorkflowVersion) error { return DB.QueryRowContext(ctx, ` INSERT INTO workflow_versions (workflow_id, version_number, snapshot) VALUES ($1, $2, $3) RETURNING id, created_at`, v.WorkflowID, v.VersionNumber, string(v.Snapshot), ).Scan(&v.ID, &v.CreatedAt) } func (s *WorkflowStore) GetVersion(ctx context.Context, workflowID string, versionNumber int) (*models.WorkflowVersion, error) { v := &models.WorkflowVersion{} var snapshot []byte err := DB.QueryRowContext(ctx, ` SELECT id, workflow_id, version_number, snapshot, created_at FROM workflow_versions WHERE workflow_id = $1 AND version_number = $2`, workflowID, versionNumber, ).Scan(&v.ID, &v.WorkflowID, &v.VersionNumber, &snapshot, &v.CreatedAt) if err != nil { return nil, err } v.Snapshot = snapshot return v, nil } func (s *WorkflowStore) GetLatestVersion(ctx context.Context, workflowID string) (*models.WorkflowVersion, error) { v := &models.WorkflowVersion{} var snapshot []byte err := DB.QueryRowContext(ctx, ` SELECT id, workflow_id, version_number, snapshot, created_at FROM workflow_versions WHERE workflow_id = $1 ORDER BY version_number DESC LIMIT 1`, workflowID, ).Scan(&v.ID, &v.WorkflowID, &v.VersionNumber, &snapshot, &v.CreatedAt) if err != nil { return nil, err } v.Snapshot = snapshot return v, nil } // ── Helpers ───────────────────────────────── func jsonOrEmpty(b json.RawMessage) string { if len(b) == 0 { return "{}" } return string(b) } func jsonOrDefault(b json.RawMessage, def string) string { if len(b) == 0 { return def } return string(b) } func jsonOrNull(b json.RawMessage) interface{} { if len(b) == 0 || string(b) == "null" { return nil } return string(b) } func nullIfEmpty(s string) interface{} { if s == "" { return nil } return s } // ── Assignments (v0.29.0-cs3) ─────────────────────────────────────────── func (s *WorkflowStore) CreateAssignment(ctx context.Context, a *store.WorkflowAssignment) error { a.ID = store.NewID() _, err := DB.ExecContext(ctx, ` INSERT INTO workflow_assignments (id, channel_id, stage, team_id) VALUES ($1, $2, $3, $4) `, a.ID, a.ChannelID, a.Stage, a.TeamID) return err } func (s *WorkflowStore) ListAssignmentsForTeam(ctx context.Context, teamID, status string) ([]store.WorkflowAssignment, error) { rows, err := DB.QueryContext(ctx, ` SELECT id, channel_id, stage, team_id, assigned_to, status, created_at, claimed_at, completed_at FROM workflow_assignments WHERE team_id = $1 AND status = $2 ORDER BY created_at DESC `, teamID, status) if err != nil { return nil, err } defer rows.Close() return scanAssignments(rows) } func (s *WorkflowStore) ListAssignmentsMine(ctx context.Context, userID string) ([]store.WorkflowAssignment, error) { rows, err := DB.QueryContext(ctx, ` 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 { return nil, err } defer rows.Close() return scanAssignments(rows) } func (s *WorkflowStore) ClaimAssignment(ctx context.Context, assignmentID, userID string) (int64, error) { res, err := DB.ExecContext(ctx, ` UPDATE workflow_assignments SET assigned_to = $1, status = 'claimed', claimed_at = $2 WHERE id = $3 AND status = 'unassigned' `, userID, time.Now().UTC(), assignmentID) if err != nil { return 0, err } return res.RowsAffected() } func (s *WorkflowStore) CompleteAssignment(ctx context.Context, assignmentID string) (int64, error) { res, err := DB.ExecContext(ctx, ` UPDATE workflow_assignments SET status = 'completed', completed_at = $1 WHERE id = $2 AND status = 'claimed' `, time.Now().UTC(), assignmentID) if err != nil { return 0, err } return res.RowsAffected() } func (s *WorkflowStore) GetAssignmentChannelID(ctx context.Context, assignmentID string) (string, error) { var channelID string err := DB.QueryRowContext(ctx, `SELECT channel_id FROM workflow_assignments WHERE id = $1`, assignmentID).Scan(&channelID) return channelID, err } func (s *WorkflowStore) TryRoundRobin(ctx context.Context, teamID, assignmentID string) (string, error) { // Find least-recently-assigned team member rows, err := DB.QueryContext(ctx, ` SELECT m.user_id, COALESCE(MAX(wa.claimed_at), '1970-01-01T00:00:00Z') as last_claim FROM team_members m LEFT JOIN workflow_assignments wa ON wa.assigned_to = m.user_id AND wa.team_id = $1 WHERE m.team_id = $2 GROUP BY m.user_id ORDER BY last_claim ASC LIMIT 1 `, teamID, teamID) if err != nil { return "", err } defer rows.Close() if !rows.Next() { return "", nil // no team members } var userID, lastClaim string if err := rows.Scan(&userID, &lastClaim); err != nil { return "", err } // Claim for that user _, err = DB.ExecContext(ctx, ` UPDATE workflow_assignments SET assigned_to = $1, status = 'claimed', claimed_at = $2 WHERE id = $3 AND status = 'unassigned' `, userID, time.Now().UTC(), assignmentID) if err != nil { return "", err } return userID, nil } func scanAssignments(rows *sql.Rows) ([]store.WorkflowAssignment, error) { var result []store.WorkflowAssignment for rows.Next() { var a store.WorkflowAssignment if err := rows.Scan(&a.ID, &a.ChannelID, &a.Stage, &a.TeamID, &a.AssignedTo, &a.Status, &a.CreatedAt, &a.ClaimedAt, &a.CompletedAt); err != nil { return nil, err } result = append(result, a) } if result == nil { result = []store.WorkflowAssignment{} } return result, rows.Err() } // ── Review Comments (v0.35.0) ─────────────────────────── func (s *WorkflowStore) GetAssignmentByID(ctx context.Context, id string) (*store.WorkflowAssignment, error) { var a store.WorkflowAssignment var rc []byte err := DB.QueryRowContext(ctx, ` SELECT id, channel_id, stage, team_id, assigned_to, status, review_comments, created_at, claimed_at, completed_at FROM workflow_assignments WHERE id = $1 `, id).Scan(&a.ID, &a.ChannelID, &a.Stage, &a.TeamID, &a.AssignedTo, &a.Status, &rc, &a.CreatedAt, &a.ClaimedAt, &a.CompletedAt) if err != nil { return nil, err } a.ReviewComments = rc return &a, nil } func (s *WorkflowStore) AddReviewComment(ctx context.Context, assignmentID string, comment store.ReviewComment) error { commentJSON, err := json.Marshal(comment) if err != nil { return err } _, err = DB.ExecContext(ctx, ` UPDATE workflow_assignments SET review_comments = review_comments || $1::jsonb WHERE id = $2 `, "["+string(commentJSON)+"]", assignmentID) return err }