Jeffrey Smith
d91ec02dd7
All checks were successful
Co-authored-by: Jeffrey Smith <jasafpro@gmail.com> Co-committed-by: Jeffrey Smith <jasafpro@gmail.com>
726 lines
24 KiB
Go
726 lines
24 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
|
|
"switchboard-core/models"
|
|
"switchboard-core/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, staleness_timeout_hours, created_by)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
|
|
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.StalenessTimeoutHours, 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,
|
|
staleness_timeout_hours, 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.StalenessTimeoutHours,
|
|
&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,
|
|
staleness_timeout_hours, 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,
|
|
staleness_timeout_hours, 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.StalenessTimeoutHours,
|
|
&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.TeamID != nil {
|
|
add("team_id", *patch.TeamID)
|
|
}
|
|
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 patch.StalenessTimeoutHours != nil {
|
|
add("staleness_timeout_hours", *patch.StalenessTimeoutHours)
|
|
}
|
|
|
|
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,
|
|
staleness_timeout_hours, 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,
|
|
staleness_timeout_hours, 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.StalenessTimeoutHours,
|
|
&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)
|
|
stageConfig := jsonOrEmpty(st.StageConfig)
|
|
branchRules := jsonOrEmpty(st.BranchRules)
|
|
stageMode := st.StageMode
|
|
if stageMode == "" {
|
|
stageMode = models.StageModeForm
|
|
}
|
|
audience := st.Audience
|
|
if audience == "" {
|
|
audience = models.AudienceTeam
|
|
}
|
|
stageType := st.StageType
|
|
if stageType == "" {
|
|
stageType = models.StageTypeSimple
|
|
}
|
|
return DB.QueryRowContext(ctx, `
|
|
INSERT INTO workflow_stages (workflow_id, ordinal, name, assignment_team_id,
|
|
form_template, stage_mode, audience, stage_type,
|
|
auto_transition, stage_config, branch_rules, starlark_hook,
|
|
surface_pkg_id, sla_seconds)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
|
|
RETURNING id, created_at`,
|
|
st.WorkflowID, st.Ordinal, st.Name, st.AssignmentTeamID,
|
|
formTpl, stageMode, audience, stageType,
|
|
st.AutoTransition, stageConfig, branchRules, st.StarlarkHook,
|
|
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, assignment_team_id,
|
|
form_template, stage_mode, audience, stage_type,
|
|
auto_transition, stage_config, branch_rules, starlark_hook,
|
|
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, stageConfig, branchRules []byte
|
|
if err := rows.Scan(&st.ID, &st.WorkflowID, &st.Ordinal, &st.Name,
|
|
&st.AssignmentTeamID, &formTpl, &st.StageMode, &st.Audience, &st.StageType,
|
|
&st.AutoTransition, &stageConfig, &branchRules, &st.StarlarkHook,
|
|
&st.SurfacePkgID, &st.SLASeconds, &st.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
st.FormTemplate = formTpl
|
|
st.StageConfig = stageConfig
|
|
st.BranchRules = branchRules
|
|
result = append(result, st)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *WorkflowStore) UpdateStage(ctx context.Context, st *models.WorkflowStage) error {
|
|
formTpl := jsonOrEmpty(st.FormTemplate)
|
|
stageConfig := jsonOrEmpty(st.StageConfig)
|
|
branchRules := jsonOrEmpty(st.BranchRules)
|
|
stageMode := st.StageMode
|
|
if stageMode == "" {
|
|
stageMode = models.StageModeForm
|
|
}
|
|
audience := st.Audience
|
|
if audience == "" {
|
|
audience = models.AudienceTeam
|
|
}
|
|
stageType := st.StageType
|
|
if stageType == "" {
|
|
stageType = models.StageTypeSimple
|
|
}
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_stages
|
|
SET ordinal = $2, name = $3, assignment_team_id = $4,
|
|
form_template = $5, stage_mode = $6, audience = $7, stage_type = $8,
|
|
auto_transition = $9, stage_config = $10, branch_rules = $11, starlark_hook = $12,
|
|
surface_pkg_id = $13, sla_seconds = $14
|
|
WHERE id = $1`,
|
|
st.ID, st.Ordinal, st.Name, st.AssignmentTeamID,
|
|
formTpl, stageMode, audience, stageType,
|
|
st.AutoTransition, stageConfig, branchRules, st.StarlarkHook,
|
|
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
|
|
}
|
|
|
|
// ── Instances (v0.3.1) ──────────────────────
|
|
|
|
func (s *WorkflowStore) CreateInstance(ctx context.Context, inst *models.WorkflowInstance) error {
|
|
stageData := jsonOrEmpty(inst.StageData)
|
|
metadata := jsonOrEmpty(inst.Metadata)
|
|
if inst.Status == "" {
|
|
inst.Status = models.InstanceStatusActive
|
|
}
|
|
return DB.QueryRowContext(ctx, `
|
|
INSERT INTO workflow_instances (workflow_id, workflow_version, current_stage,
|
|
stage_data, status, started_by, entry_token, metadata)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
|
RETURNING id, stage_entered_at, created_at, updated_at`,
|
|
inst.WorkflowID, inst.WorkflowVersion, inst.CurrentStage,
|
|
stageData, inst.Status, inst.StartedBy, inst.EntryToken, metadata,
|
|
).Scan(&inst.ID, &inst.StageEnteredAt, &inst.CreatedAt, &inst.UpdatedAt)
|
|
}
|
|
|
|
func (s *WorkflowStore) GetInstance(ctx context.Context, id string) (*models.WorkflowInstance, error) {
|
|
inst := &models.WorkflowInstance{}
|
|
var stageData, metadata []byte
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT id, workflow_id, workflow_version, current_stage, stage_data,
|
|
status, started_by, entry_token, metadata,
|
|
stage_entered_at, created_at, updated_at
|
|
FROM workflow_instances WHERE id = $1`, id,
|
|
).Scan(&inst.ID, &inst.WorkflowID, &inst.WorkflowVersion, &inst.CurrentStage,
|
|
&stageData, &inst.Status, &inst.StartedBy, &inst.EntryToken, &metadata,
|
|
&inst.StageEnteredAt, &inst.CreatedAt, &inst.UpdatedAt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
inst.StageData = stageData
|
|
inst.Metadata = metadata
|
|
return inst, nil
|
|
}
|
|
|
|
func (s *WorkflowStore) GetInstanceByToken(ctx context.Context, token string) (*models.WorkflowInstance, error) {
|
|
inst := &models.WorkflowInstance{}
|
|
var stageData, metadata []byte
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT id, workflow_id, workflow_version, current_stage, stage_data,
|
|
status, started_by, entry_token, metadata,
|
|
stage_entered_at, created_at, updated_at
|
|
FROM workflow_instances WHERE entry_token = $1`, token,
|
|
).Scan(&inst.ID, &inst.WorkflowID, &inst.WorkflowVersion, &inst.CurrentStage,
|
|
&stageData, &inst.Status, &inst.StartedBy, &inst.EntryToken, &metadata,
|
|
&inst.StageEnteredAt, &inst.CreatedAt, &inst.UpdatedAt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
inst.StageData = stageData
|
|
inst.Metadata = metadata
|
|
return inst, nil
|
|
}
|
|
|
|
func (s *WorkflowStore) UpdateInstance(ctx context.Context, inst *models.WorkflowInstance) error {
|
|
stageData := jsonOrEmpty(inst.StageData)
|
|
metadata := jsonOrEmpty(inst.Metadata)
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_instances
|
|
SET current_stage = $2, stage_data = $3, status = $4,
|
|
entry_token = $5, metadata = $6, stage_entered_at = $7
|
|
WHERE id = $1`,
|
|
inst.ID, inst.CurrentStage, stageData, inst.Status,
|
|
inst.EntryToken, metadata, inst.StageEnteredAt)
|
|
return err
|
|
}
|
|
|
|
func (s *WorkflowStore) ListInstances(ctx context.Context, workflowID string, status string, opts store.ListOptions) ([]models.WorkflowInstance, error) {
|
|
q := `SELECT id, workflow_id, workflow_version, current_stage, stage_data,
|
|
status, started_by, entry_token, metadata,
|
|
stage_entered_at, created_at, updated_at
|
|
FROM workflow_instances WHERE workflow_id = $1`
|
|
args := []interface{}{workflowID}
|
|
idx := 2
|
|
if status != "" {
|
|
q += fmt.Sprintf(" AND status = $%d", idx)
|
|
args = append(args, status)
|
|
idx++
|
|
}
|
|
q += " ORDER BY 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, `
|
|
UPDATE workflow_instances
|
|
SET current_stage = $2, stage_data = $3, stage_entered_at = NOW()
|
|
WHERE id = $1 AND status = 'active'`,
|
|
id, nextStage, sd)
|
|
return err
|
|
}
|
|
|
|
func (s *WorkflowStore) CompleteInstance(ctx context.Context, id string) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_instances SET status = 'completed' WHERE id = $1 AND status = 'active'`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *WorkflowStore) CancelInstance(ctx context.Context, id string) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_instances SET status = 'cancelled' WHERE id = $1 AND status = 'active'`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *WorkflowStore) MarkInstanceStale(ctx context.Context, id string) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_instances SET status = 'stale' WHERE id = $1 AND status = 'active'`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *WorkflowStore) ListActiveInstances(ctx context.Context) ([]models.WorkflowInstance, error) {
|
|
rows, err := DB.QueryContext(ctx, `
|
|
SELECT id, workflow_id, workflow_version, current_stage, stage_data,
|
|
status, started_by, entry_token, metadata,
|
|
stage_entered_at, created_at, updated_at
|
|
FROM workflow_instances WHERE status = 'active'
|
|
ORDER BY stage_entered_at ASC`)
|
|
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()
|
|
}
|
|
|
|
// ── Assignments (v0.3.1) ────────────────────
|
|
|
|
func (s *WorkflowStore) CreateAssignment(ctx context.Context, a *models.WorkflowAssignment) error {
|
|
reviewData := jsonOrEmpty(a.ReviewData)
|
|
if a.Status == "" {
|
|
a.Status = models.AssignmentStatusUnassigned
|
|
}
|
|
return DB.QueryRowContext(ctx, `
|
|
INSERT INTO workflow_assignments (instance_id, stage, team_id, assigned_to, status, review_data)
|
|
VALUES ($1, $2, $3, $4, $5, $6)
|
|
RETURNING id, created_at`,
|
|
a.InstanceID, a.Stage, a.TeamID, a.AssignedTo, a.Status, reviewData,
|
|
).Scan(&a.ID, &a.CreatedAt)
|
|
}
|
|
|
|
func (s *WorkflowStore) ClaimAssignment(ctx context.Context, id string, userID string) error {
|
|
res, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_assignments
|
|
SET status = 'claimed', assigned_to = $2, claimed_at = NOW()
|
|
WHERE id = $1 AND status = 'unassigned'`, id, userID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n == 0 {
|
|
return fmt.Errorf("assignment %s is not claimable", id)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *WorkflowStore) UnclaimAssignment(ctx context.Context, id string) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_assignments
|
|
SET status = 'unassigned', assigned_to = NULL, claimed_at = NULL
|
|
WHERE id = $1 AND status = 'claimed'`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *WorkflowStore) CompleteAssignment(ctx context.Context, id string, reviewData json.RawMessage) error {
|
|
rd := jsonOrEmpty(reviewData)
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_assignments
|
|
SET status = 'completed', review_data = $2, completed_at = NOW()
|
|
WHERE id = $1 AND status = 'claimed'`, id, rd)
|
|
return err
|
|
}
|
|
|
|
func (s *WorkflowStore) CancelAssignment(ctx context.Context, id string) error {
|
|
_, err := DB.ExecContext(ctx, `
|
|
UPDATE workflow_assignments SET status = 'cancelled'
|
|
WHERE id = $1 AND status IN ('unassigned', 'claimed')`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *WorkflowStore) ListAssignmentsByTeam(ctx context.Context, teamID string, status string) ([]models.WorkflowAssignment, error) {
|
|
q := `SELECT id, instance_id, stage, team_id, assigned_to, status,
|
|
review_data, claimed_at, completed_at, created_at
|
|
FROM workflow_assignments WHERE team_id = $1`
|
|
args := []interface{}{teamID}
|
|
if status != "" {
|
|
q += " AND status = $2"
|
|
args = append(args, status)
|
|
}
|
|
q += " ORDER BY created_at DESC"
|
|
return s.queryAssignments(ctx, q, args...)
|
|
}
|
|
|
|
func (s *WorkflowStore) ListAssignmentsByInstance(ctx context.Context, instanceID string) ([]models.WorkflowAssignment, error) {
|
|
return s.queryAssignments(ctx, `
|
|
SELECT id, instance_id, stage, team_id, assigned_to, status,
|
|
review_data, claimed_at, completed_at, created_at
|
|
FROM workflow_assignments WHERE instance_id = $1
|
|
ORDER BY created_at ASC`, instanceID)
|
|
}
|
|
|
|
func (s *WorkflowStore) ListAssignmentsByUser(ctx context.Context, userID string, status string) ([]models.WorkflowAssignment, error) {
|
|
q := `SELECT id, instance_id, stage, team_id, assigned_to, status,
|
|
review_data, claimed_at, completed_at, created_at
|
|
FROM workflow_assignments WHERE assigned_to = $1`
|
|
args := []interface{}{userID}
|
|
if status != "" {
|
|
q += " AND status = $2"
|
|
args = append(args, status)
|
|
}
|
|
q += " ORDER BY created_at DESC"
|
|
return s.queryAssignments(ctx, q, args...)
|
|
}
|
|
|
|
// ── Signoffs (v0.3.4) ─────────────────────────
|
|
|
|
func (s *WorkflowStore) CreateSignoff(ctx context.Context, so *models.WorkflowSignoff) error {
|
|
return DB.QueryRowContext(ctx, `
|
|
INSERT INTO workflow_signoffs (instance_id, stage, user_id, decision, comment)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
RETURNING id, created_at`,
|
|
so.InstanceID, so.Stage, so.UserID, so.Decision, so.Comment,
|
|
).Scan(&so.ID, &so.CreatedAt)
|
|
}
|
|
|
|
func (s *WorkflowStore) ListSignoffs(ctx context.Context, instanceID, stage string) ([]models.WorkflowSignoff, error) {
|
|
rows, err := DB.QueryContext(ctx, `
|
|
SELECT id, instance_id, stage, user_id, decision, comment, created_at
|
|
FROM workflow_signoffs
|
|
WHERE instance_id = $1 AND stage = $2
|
|
ORDER BY created_at`, instanceID, stage)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []models.WorkflowSignoff
|
|
for rows.Next() {
|
|
var so models.WorkflowSignoff
|
|
if err := rows.Scan(&so.ID, &so.InstanceID, &so.Stage, &so.UserID,
|
|
&so.Decision, &so.Comment, &so.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, so)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *WorkflowStore) CountSignoffs(ctx context.Context, instanceID, stage, decision string) (int, error) {
|
|
var count int
|
|
err := DB.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM workflow_signoffs
|
|
WHERE instance_id = $1 AND stage = $2 AND decision = $3`,
|
|
instanceID, stage, decision).Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
func (s *WorkflowStore) queryAssignments(ctx context.Context, q string, args ...interface{}) ([]models.WorkflowAssignment, error) {
|
|
rows, err := DB.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []models.WorkflowAssignment
|
|
for rows.Next() {
|
|
var a models.WorkflowAssignment
|
|
var reviewData []byte
|
|
if err := rows.Scan(&a.ID, &a.InstanceID, &a.Stage, &a.TeamID,
|
|
&a.AssignedTo, &a.Status, &reviewData,
|
|
&a.ClaimedAt, &a.CompletedAt, &a.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
a.ReviewData = reviewData
|
|
result = append(result, a)
|
|
}
|
|
return result, rows.Err()
|
|
}
|