package sqlite import ( "context" "database/sql" "encoding/json" "fmt" "time" "switchboard-core/models" "switchboard-core/store" ) // WorkflowStore implements store.WorkflowStore for SQLite. type WorkflowStore struct{} func NewWorkflowStore() *WorkflowStore { return &WorkflowStore{} } // ── Workflow CRUD ─────────────────────────── func (s *WorkflowStore) Create(ctx context.Context, w *models.Workflow) error { w.ID = store.NewID() now := time.Now().UTC() w.CreatedAt = now w.UpdatedAt = now w.Version = 1 branding := jsonOrEmpty(w.Branding) 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, webhook_url, webhook_secret, staleness_timeout_hours, 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.StalenessTimeoutHours, w.CreatedBy, w.CreatedAt.Format(time.RFC3339), w.UpdatedAt.Format(time.RFC3339)) return err } 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, webhook_url, webhook_secret, staleness_timeout_hours, created_by, created_at, updated_at FROM workflows WHERE id = ?`, id) } func (s *WorkflowStore) GetBySlug(ctx context.Context, teamID *string, slug string) (*models.Workflow, error) { if teamID != nil { return s.scanOne(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 = ? AND slug = ?`, *teamID, slug) } return s.scanOne(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 AND slug = ?`, slug) } func (s *WorkflowStore) Update(ctx context.Context, id string, patch models.WorkflowPatch) error { sets := []string{} args := []interface{}{} if patch.Name != nil { sets = append(sets, "name = ?") args = append(args, *patch.Name) } if patch.Description != nil { sets = append(sets, "description = ?") args = append(args, *patch.Description) } if patch.TeamID != nil { sets = append(sets, "team_id = ?") args = append(args, *patch.TeamID) } if patch.Branding != nil { sets = append(sets, "branding = ?") args = append(args, string(*patch.Branding)) } if patch.EntryMode != nil { sets = append(sets, "entry_mode = ?") args = append(args, *patch.EntryMode) } if patch.IsActive != nil { sets = append(sets, "is_active = ?") args = append(args, boolToInt(*patch.IsActive)) } if patch.OnComplete != nil { sets = append(sets, "on_complete = ?") args = append(args, string(*patch.OnComplete)) } if patch.Retention != nil { 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 patch.StalenessTimeoutHours != nil { sets = append(sets, "staleness_timeout_hours = ?") args = append(args, *patch.StalenessTimeoutHours) } if len(sets) == 0 { return nil } sets = append(sets, "version = version + 1") sets = append(sets, "updated_at = ?") args = append(args, time.Now().UTC().Format(time.RFC3339)) q := "UPDATE workflows SET " for i, s := range sets { if i > 0 { q += ", " } q += s } q += " WHERE id = ?" 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 = ?`, 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 = ? 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`) } // ── Stages ────────────────────────────────── func (s *WorkflowStore) CreateStage(ctx context.Context, st *models.WorkflowStage) error { st.ID = store.NewID() st.CreatedAt = time.Now().UTC() 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, ` INSERT INTO workflow_stages (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) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, st.ID, st.WorkflowID, st.Ordinal, st.Name, st.AssignmentTeamID, formTpl, stageMode, audience, stageType, boolToInt(st.AutoTransition), stageConfig, branchRules, st.StarlarkHook, st.SurfacePkgID, st.SLASeconds, st.CreatedAt.Format(time.RFC3339)) return err } 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 = ? ORDER BY ordinal ASC`, workflowID) if err != nil { return nil, err } defer rows.Close() var result []models.WorkflowStage for rows.Next() { var stg models.WorkflowStage var formTpl, stageConfig, branchRules string var autoTrans int if err := rows.Scan(&stg.ID, &stg.WorkflowID, &stg.Ordinal, &stg.Name, &stg.AssignmentTeamID, &formTpl, &stg.StageMode, &stg.Audience, &stg.StageType, &autoTrans, &stageConfig, &branchRules, &stg.StarlarkHook, &stg.SurfacePkgID, &stg.SLASeconds, st(&stg.CreatedAt)); err != nil { return nil, err } stg.FormTemplate = json.RawMessage(formTpl) stg.StageConfig = json.RawMessage(stageConfig) stg.BranchRules = json.RawMessage(branchRules) stg.AutoTransition = autoTrans != 0 result = append(result, stg) } 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 = ?, name = ?, assignment_team_id = ?, form_template = ?, stage_mode = ?, audience = ?, stage_type = ?, auto_transition = ?, stage_config = ?, branch_rules = ?, starlark_hook = ?, surface_pkg_id = ?, sla_seconds = ? WHERE id = ?`, st.Ordinal, st.Name, st.AssignmentTeamID, formTpl, stageMode, audience, stageType, boolToInt(st.AutoTransition), stageConfig, branchRules, st.StarlarkHook, st.SurfacePkgID, st.SLASeconds, st.ID) return err } func (s *WorkflowStore) DeleteStage(ctx context.Context, id string) error { _, err := DB.ExecContext(ctx, `DELETE FROM workflow_stages WHERE id = ?`, 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 = ? WHERE id = ? AND workflow_id = ?`, i, id, workflowID); err != nil { return err } } return tx.Commit() } // ── Versions ──────────────────────────────── func (s *WorkflowStore) Publish(ctx context.Context, v *models.WorkflowVersion) error { v.ID = store.NewID() v.CreatedAt = time.Now().UTC() _, err := DB.ExecContext(ctx, ` INSERT INTO workflow_versions (id, workflow_id, version_number, snapshot, created_at) VALUES (?, ?, ?, ?, ?)`, v.ID, v.WorkflowID, v.VersionNumber, string(v.Snapshot), v.CreatedAt.Format(time.RFC3339)) return err } func (s *WorkflowStore) GetVersion(ctx context.Context, workflowID string, versionNumber int) (*models.WorkflowVersion, error) { v := &models.WorkflowVersion{} var snapshot string err := DB.QueryRowContext(ctx, ` 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, st(&v.CreatedAt)) if err != nil { return nil, err } v.Snapshot = json.RawMessage(snapshot) return v, nil } func (s *WorkflowStore) GetLatestVersion(ctx context.Context, workflowID string) (*models.WorkflowVersion, error) { v := &models.WorkflowVersion{} var snapshot string err := DB.QueryRowContext(ctx, ` SELECT id, workflow_id, version_number, snapshot, created_at FROM workflow_versions WHERE workflow_id = ? ORDER BY version_number DESC LIMIT 1`, workflowID, ).Scan(&v.ID, &v.WorkflowID, &v.VersionNumber, &snapshot, st(&v.CreatedAt)) if err != nil { return nil, err } v.Snapshot = json.RawMessage(snapshot) return v, nil } // ── Helpers ───────────────────────────────── func (s *WorkflowStore) scanOne(ctx context.Context, q string, args ...interface{}) (*models.Workflow, error) { 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, &webhookURL, &webhookSecret, &w.StalenessTimeoutHours, &w.CreatedBy, st(&w.CreatedAt), st(&w.UpdatedAt)) if err != nil { return nil, err } w.Branding = json.RawMessage(branding) w.Retention = json.RawMessage(retention) w.IsActive = isActive != 0 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 } 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 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, &webhookURL, &webhookSecret, &w.StalenessTimeoutHours, &w.CreatedBy, st(&w.CreatedAt), st(&w.UpdatedAt)); err != nil { return nil, err } w.Branding = json.RawMessage(branding) w.Retention = json.RawMessage(retention) w.IsActive = isActive != 0 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() } func boolToInt(b bool) int { if b { return 1 } return 0 } 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 ────────────────────── func (s *WorkflowStore) CreateInstance(ctx context.Context, inst *models.WorkflowInstance) error { inst.ID = store.NewID() now := time.Now().UTC() inst.CreatedAt = now inst.UpdatedAt = now inst.StageEnteredAt = now if inst.Status == "" { inst.Status = models.InstanceStatusActive } stageData := jsonOrEmpty(inst.StageData) metadata := jsonOrEmpty(inst.Metadata) _, err := DB.ExecContext(ctx, ` INSERT INTO workflow_instances (id, workflow_id, workflow_version, current_stage, stage_data, status, started_by, entry_token, metadata, stage_entered_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, inst.ID, inst.WorkflowID, inst.WorkflowVersion, inst.CurrentStage, stageData, inst.Status, inst.StartedBy, inst.EntryToken, metadata, inst.StageEnteredAt.Format(timeFmt), inst.CreatedAt.Format(timeFmt), inst.UpdatedAt.Format(timeFmt)) return err } func (s *WorkflowStore) GetInstance(ctx context.Context, id string) (*models.WorkflowInstance, error) { return s.scanInstance(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 = ?`, id) } func (s *WorkflowStore) GetInstanceByToken(ctx context.Context, token string) (*models.WorkflowInstance, error) { return s.scanInstance(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 = ?`, token) } func (s *WorkflowStore) scanInstance(ctx context.Context, q string, args ...interface{}) (*models.WorkflowInstance, error) { inst := &models.WorkflowInstance{} var stageData, metadata string err := DB.QueryRowContext(ctx, q, args...).Scan( &inst.ID, &inst.WorkflowID, &inst.WorkflowVersion, &inst.CurrentStage, &stageData, &inst.Status, &inst.StartedBy, &inst.EntryToken, &metadata, st(&inst.StageEnteredAt), st(&inst.CreatedAt), st(&inst.UpdatedAt)) if err != nil { return nil, err } inst.StageData = json.RawMessage(stageData) inst.Metadata = json.RawMessage(metadata) return inst, nil } func (s *WorkflowStore) UpdateInstance(ctx context.Context, inst *models.WorkflowInstance) error { stageData := jsonOrEmpty(inst.StageData) metadata := jsonOrEmpty(inst.Metadata) now := time.Now().UTC() _, err := DB.ExecContext(ctx, ` UPDATE workflow_instances SET current_stage = ?, stage_data = ?, status = ?, entry_token = ?, metadata = ?, stage_entered_at = ?, updated_at = ? WHERE id = ?`, inst.CurrentStage, stageData, inst.Status, inst.EntryToken, metadata, inst.StageEnteredAt.Format(timeFmt), now.Format(timeFmt), inst.ID) if err == nil { inst.UpdatedAt = now } 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 = ?` args := []interface{}{workflowID} if status != "" { q += " AND status = ?" args = append(args, status) } q += " ORDER BY created_at DESC" if opts.Limit > 0 { q += " LIMIT ?" args = append(args, opts.Limit) } if opts.Offset > 0 { q += " OFFSET ?" args = append(args, opts.Offset) } rows, err := DB.QueryContext(ctx, q, args...) if err != nil { return nil, err } defer rows.Close() var result []models.WorkflowInstance for rows.Next() { var inst models.WorkflowInstance var stageData, metadata string if err := rows.Scan(&inst.ID, &inst.WorkflowID, &inst.WorkflowVersion, &inst.CurrentStage, &stageData, &inst.Status, &inst.StartedBy, &inst.EntryToken, &metadata, st(&inst.StageEnteredAt), st(&inst.CreatedAt), st(&inst.UpdatedAt)); err != nil { return nil, err } inst.StageData = json.RawMessage(stageData) inst.Metadata = json.RawMessage(metadata) result = append(result, inst) } return result, rows.Err() } func (s *WorkflowStore) AdvanceStage(ctx context.Context, id string, nextStage string, stageData json.RawMessage) error { sd := jsonOrEmpty(stageData) now := time.Now().UTC() _, err := DB.ExecContext(ctx, ` UPDATE workflow_instances SET current_stage = ?, stage_data = ?, stage_entered_at = ?, updated_at = ? WHERE id = ? AND status = 'active'`, nextStage, sd, now.Format(timeFmt), now.Format(timeFmt), id) return err } func (s *WorkflowStore) CompleteInstance(ctx context.Context, id string) error { now := time.Now().UTC() _, err := DB.ExecContext(ctx, ` UPDATE workflow_instances SET status = 'completed', updated_at = ? WHERE id = ? AND status = 'active'`, now.Format(timeFmt), id) return err } func (s *WorkflowStore) CancelInstance(ctx context.Context, id string) error { now := time.Now().UTC() _, err := DB.ExecContext(ctx, ` UPDATE workflow_instances SET status = 'cancelled', updated_at = ? WHERE id = ? AND status = 'active'`, now.Format(timeFmt), id) return err } func (s *WorkflowStore) MarkInstanceStale(ctx context.Context, id string) error { now := time.Now().UTC() _, err := DB.ExecContext(ctx, ` UPDATE workflow_instances SET status = 'stale', updated_at = ? WHERE id = ? AND status = 'active'`, now.Format(timeFmt), 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 string if err := rows.Scan(&inst.ID, &inst.WorkflowID, &inst.WorkflowVersion, &inst.CurrentStage, &stageData, &inst.Status, &inst.StartedBy, &inst.EntryToken, &metadata, st(&inst.StageEnteredAt), st(&inst.CreatedAt), st(&inst.UpdatedAt)); err != nil { return nil, err } inst.StageData = json.RawMessage(stageData) inst.Metadata = json.RawMessage(metadata) result = append(result, inst) } return result, rows.Err() } // ── Assignments ──────────────────── func (s *WorkflowStore) CreateAssignment(ctx context.Context, a *models.WorkflowAssignment) error { a.ID = store.NewID() a.CreatedAt = time.Now().UTC() if a.Status == "" { a.Status = models.AssignmentStatusUnassigned } reviewData := jsonOrEmpty(a.ReviewData) _, err := DB.ExecContext(ctx, ` INSERT INTO workflow_assignments (id, instance_id, stage, team_id, assigned_to, status, review_data, claimed_at, completed_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, a.ID, a.InstanceID, a.Stage, a.TeamID, a.AssignedTo, a.Status, reviewData, nil, nil, a.CreatedAt.Format(timeFmt)) return err } func (s *WorkflowStore) ClaimAssignment(ctx context.Context, id string, userID string) error { now := time.Now().UTC() res, err := DB.ExecContext(ctx, ` UPDATE workflow_assignments SET status = 'claimed', assigned_to = ?, claimed_at = ? WHERE id = ? AND status = 'unassigned'`, userID, now.Format(timeFmt), id) 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 = ? AND status = 'claimed'`, id) return err } func (s *WorkflowStore) CompleteAssignment(ctx context.Context, id string, reviewData json.RawMessage) error { rd := jsonOrEmpty(reviewData) now := time.Now().UTC() _, err := DB.ExecContext(ctx, ` UPDATE workflow_assignments SET status = 'completed', review_data = ?, completed_at = ? WHERE id = ? AND status = 'claimed'`, rd, now.Format(timeFmt), id) return err } func (s *WorkflowStore) CancelAssignment(ctx context.Context, id string) error { _, err := DB.ExecContext(ctx, ` UPDATE workflow_assignments SET status = 'cancelled' WHERE id = ? 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 = ?` args := []interface{}{teamID} if status != "" { q += " AND status = ?" 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 = ? 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 = ?` args := []interface{}{userID} if status != "" { q += " AND status = ?" args = append(args, status) } q += " ORDER BY created_at DESC" return s.queryAssignments(ctx, q, args...) } // ── Signoffs ───────────────────────── func (s *WorkflowStore) CreateSignoff(ctx context.Context, so *models.WorkflowSignoff) error { so.ID = store.NewID() so.CreatedAt = time.Now().UTC() _, err := DB.ExecContext(ctx, ` INSERT INTO workflow_signoffs (id, instance_id, stage, user_id, decision, comment, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`, so.ID, so.InstanceID, so.Stage, so.UserID, so.Decision, so.Comment, so.CreatedAt.Format(timeFmt)) return err } 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 = ? AND stage = ? 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, st(&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 = ? AND stage = ? AND decision = ?`, 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 string if err := rows.Scan(&a.ID, &a.InstanceID, &a.Stage, &a.TeamID, &a.AssignedTo, &a.Status, &reviewData, stN(&a.ClaimedAt), stN(&a.CompletedAt), st(&a.CreatedAt)); err != nil { return nil, err } a.ReviewData = json.RawMessage(reviewData) result = append(result, a) } return result, rows.Err() }