Feat v0.3.1 instance + assignment tables and store

Add workflow execution layer: workflow_instances tracks running workflows (stage, status, entry token), workflow_assignments tracks per-stage team queue with optimistic-lock claim. 15 new store methods in PG + SQLite. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-27 18:41:03 +00:00
parent 965885a8f7
commit 1bf8422932
6 changed files with 667 additions and 7 deletions
--- a/server/store/sqlite/workflows.go
+++ b/server/store/sqlite/workflows.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"fmt"
 	"time"

 	"switchboard-core/models"
@@ -409,8 +410,246 @@ func nullIfEmpty(s string) interface{} {
 	return s
 }

-// ── Assignments (v0.29.0-cs3) ───────────────────────────────────────────
+// ── Instances (v0.3.1) ──────────────────────

-// ── Lifecycle operations (v0.37.15) ──
+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
+}

-// ── Review Comments (v0.35.0) ───────────────────────────
+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
+}
+
+// ── Assignments (v0.3.1) ────────────────────
+
+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) 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()
+}