feat(service): implement project workflow plan, preview, and activation flow.

New projectinstance package validates three-duty separation, issues confirmation tokens, and activates immutable workflow instances; new projects start in planning mode.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-13 02:50:23 +00:00
committed by weichen
co-authored by Cursor
parent 0f5bc426e8
commit dc629fc4b8
12 changed files with 1456 additions and 23 deletions
+19
View File
@@ -0,0 +1,19 @@
package projectinstance
import "errors"
var (
ErrNotAvailable = errors.New("project instance store not available")
ErrProjectNotFound = errors.New("project not found")
ErrForbiddenNotCreator = errors.New("forbidden: only project creator can configure workflow")
ErrPlanFrozen = errors.New("plan frozen: workflow already active")
ErrWorkflowAlreadyActive = errors.New("workflow already active for project")
ErrPlanChangedReconfirm = errors.New("plan changed: reconfirm required")
ErrInvalidPlan = errors.New("invalid workflow plan")
ErrValidationFailed = errors.New("workflow plan validation failed")
ErrTokenInvalid = errors.New("activation token invalid or expired")
ErrConcurrentConflict = errors.New("conflict: project version mismatch")
ErrPlanRevisionConflict = errors.New("conflict: plan revision mismatch")
ErrWorkflowVersionNotFound = errors.New("workflow version not found or not published")
ErrLegacySwitchBlocked = errors.New("legacy workflow switch blocked: use project settings activation")
)
+31
View File
@@ -0,0 +1,31 @@
package projectinstance
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"sort"
)
func computePlanHash(plan WorkflowPlan) (string, error) {
normalized := normalizePlan(plan)
raw, err := json.Marshal(normalized)
if err != nil {
return "", err
}
sum := sha256.Sum256(raw)
return hex.EncodeToString(sum[:]), nil
}
func normalizePlan(plan WorkflowPlan) WorkflowPlan {
out := plan
sort.Slice(out.Stages, func(i, j int) bool {
return out.Stages[i].StageKey < out.Stages[j].StageKey
})
for i := range out.Stages {
sort.Slice(out.Stages[i].DocTasks, func(a, b int) bool {
return out.Stages[i].DocTasks[a].TaskKey < out.Stages[i].DocTasks[b].TaskKey
})
}
return out
}
+319
View File
@@ -0,0 +1,319 @@
package projectinstance
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"time"
"chat-agent-system/internal/projectworkflow"
"chat-agent-system/internal/workflowdef"
)
// VersionResolver 解析已发布工作流版本。
type VersionResolver interface {
GetDefinitionByKey(ctx context.Context, key string) (*workflowdef.Definition, error)
GetVersion(ctx context.Context, workflowID int64, version int) (*workflowdef.Version, error)
ListDefinitions(ctx context.Context) ([]workflowdef.Definition, error)
}
// Service 项目工作流实例与激活领域服务。
type Service struct {
store Store
wfStore VersionResolver
projects projectworkflow.Store
}
func NewService(store Store, wfStore VersionResolver, projects projectworkflow.Store) *Service {
return &Service{store: store, wfStore: wfStore, projects: projects}
}
func (s *Service) Available() bool {
return s != nil && s.store != nil
}
func (s *Service) HasLiveInstance(ctx context.Context, projectID int64) (bool, error) {
if !s.Available() {
return false, nil
}
return s.store.HasLiveInstance(ctx, projectID)
}
func (s *Service) GetSettings(ctx context.Context, projectID int64, userID int64, isCreator bool) (*ProjectSettings, error) {
if projectID <= 0 {
return nil, ErrProjectNotFound
}
lifecycle, activeKey, selVer, planRev, err := s.store.GetProjectMeta(ctx, projectID)
if err != nil {
return nil, err
}
settings := &ProjectSettings{
ProjectID: projectID,
LifecycleStatus: lifecycle,
ActiveWorkflowKey: activeKey,
WorkflowSelectionVersion: selVer,
WorkflowPlanRevision: planRev,
CanEditPlan: isCreator && lifecycle == LifecyclePlanning,
CanActivate: isCreator && lifecycle == LifecyclePlanning,
}
planInst, err := s.store.GetPlanningInstance(ctx, projectID)
if err != nil {
return nil, err
}
if planInst != nil {
settings.PlanningInstance = planInst
settings.Plan = &planInst.ConfigSnapshot
}
liveInst, err := s.store.GetLiveInstance(ctx, projectID)
if err != nil {
return nil, err
}
if liveInst != nil {
settings.LiveInstance = liveInst
settings.CanEditPlan = false
settings.CanActivate = false
}
if s.wfStore != nil {
defs, listErr := s.wfStore.ListDefinitions(ctx)
if listErr == nil {
for _, d := range defs {
if d.LatestPublishedVersion <= 0 {
continue
}
ver, verErr := s.wfStore.GetVersion(ctx, d.ID, d.LatestPublishedVersion)
if verErr != nil || ver == nil {
continue
}
settings.PublishedWorkflows = append(settings.PublishedWorkflows, PublishedOption{
DefinitionKey: d.Key,
DefinitionName: d.Name,
LatestVersion: d.LatestPublishedVersion,
WorkflowVersion: ver.ID,
ProductDomain: d.ProductDomain,
})
}
}
}
_ = userID
return settings, nil
}
func (s *Service) SaveWorkflowPlan(ctx context.Context, projectID int64, userID int64, isCreator bool, in SavePlanInput) (*ProjectSettings, error) {
if !isCreator {
return nil, ErrForbiddenNotCreator
}
lifecycle, _, selVer, planRev, err := s.store.GetProjectMeta(ctx, projectID)
if err != nil {
return nil, err
}
if lifecycle != LifecyclePlanning {
return nil, ErrPlanFrozen
}
live, err := s.store.HasLiveInstance(ctx, projectID)
if err != nil {
return nil, err
}
if live {
return nil, ErrWorkflowAlreadyActive
}
if in.ExpectedSelectionVersion > 0 && in.ExpectedSelectionVersion != selVer {
return nil, ErrConcurrentConflict
}
if in.ExpectedPlanRevision != planRev {
return nil, ErrPlanRevisionConflict
}
version, err := s.resolvePublishedVersion(ctx, in.Plan)
if err != nil {
return nil, err
}
validation := validatePlan(in.Plan, version)
if !validation.Valid {
return nil, fmt.Errorf("%w: %d issues", ErrValidationFailed, len(validation.Issues))
}
in.Plan.WorkflowVersionID = version.ID
planHash, err := computePlanHash(in.Plan)
if err != nil {
return nil, err
}
inst := &WorkflowInstance{
ProjectID: projectID,
WorkflowVersionID: version.ID,
Status: InstancePlanning,
ConfigRevision: planRev + 1,
PlanHash: planHash,
ConfigSnapshot: in.Plan,
CreatedBy: &userID,
}
if _, err := s.store.UpsertPlanningInstance(ctx, inst); err != nil {
return nil, err
}
if _, err := s.store.BumpPlanRevision(ctx, projectID, planRev); err != nil {
return nil, err
}
return s.GetSettings(ctx, projectID, userID, isCreator)
}
func (s *Service) ActivationPreview(ctx context.Context, projectID int64, userID int64, isCreator bool) (*ActivationPreview, error) {
if !isCreator {
return nil, ErrForbiddenNotCreator
}
lifecycle, _, selVer, planRev, err := s.store.GetProjectMeta(ctx, projectID)
if err != nil {
return nil, err
}
if lifecycle != LifecyclePlanning {
return nil, ErrPlanFrozen
}
planInst, err := s.store.GetPlanningInstance(ctx, projectID)
if err != nil {
return nil, err
}
if planInst == nil || planInst.PlanHash == "" {
return nil, fmt.Errorf("%w: no saved plan", ErrInvalidPlan)
}
version, err := s.resolvePublishedVersion(ctx, planInst.ConfigSnapshot)
if err != nil {
return nil, err
}
validation := validatePlan(planInst.ConfigSnapshot, version)
if !validation.Valid {
return nil, fmt.Errorf("%w", ErrValidationFailed)
}
token, err := randomToken(24)
if err != nil {
return nil, err
}
expiresAt := time.Now().Add(15 * time.Minute)
if err := s.store.SaveActivationToken(ctx, token, projectID, planInst.PlanHash, selVer, planInst.ConfigSnapshot, userID, expiresAt); err != nil {
return nil, err
}
return &ActivationPreview{
ConfirmationToken: token,
PlanHash: planInst.PlanHash,
ExpectedProjectVersion: selVer,
ExpectedPlanRevision: planRev,
Summary: previewSummary(planInst.ConfigSnapshot, version),
Validation: validation,
ExpiresAt: expiresAt,
}, nil
}
func (s *Service) Activate(ctx context.Context, projectID int64, userID int64, isCreator bool, in ActivateInput) (*ActivateResult, error) {
if !isCreator {
return nil, ErrForbiddenNotCreator
}
lifecycle, _, selVer, planRev, err := s.store.GetProjectMeta(ctx, projectID)
if err != nil {
return nil, err
}
if lifecycle != LifecyclePlanning {
return nil, ErrPlanFrozen
}
if in.ExpectedProjectVersion != selVer {
return nil, ErrConcurrentConflict
}
if in.ExpectedPlanRevision != planRev {
return nil, ErrPlanRevisionConflict
}
planInst, err := s.store.GetPlanningInstance(ctx, projectID)
if err != nil {
return nil, err
}
if planInst == nil {
return nil, fmt.Errorf("%w: no saved plan", ErrInvalidPlan)
}
consumed, err := s.store.ConsumeActivationToken(ctx, in.ConfirmationToken, projectID, planInst.PlanHash)
if err != nil {
return nil, err
}
currentHash, err := computePlanHash(*consumed)
if err != nil {
return nil, err
}
if currentHash != planInst.PlanHash {
return nil, ErrPlanChangedReconfirm
}
version, err := s.resolvePublishedVersion(ctx, planInst.ConfigSnapshot)
if err != nil {
return nil, err
}
validation := validatePlan(planInst.ConfigSnapshot, version)
if !validation.Valid {
return nil, ErrValidationFailed
}
activated, err := s.store.ActivateInstance(ctx, projectID, planInst, version.WorkflowKey, selVer, userID)
if err != nil {
return nil, err
}
return &ActivateResult{
ProjectID: projectID,
LifecycleStatus: LifecycleActive,
ActiveWorkflowKey: version.WorkflowKey,
Instance: *activated,
}, nil
}
func (s *Service) BuildDefaultPlan(ctx context.Context, workflowKey string, versionNum int) (*WorkflowPlan, error) {
if s.wfStore == nil {
return nil, ErrNotAvailable
}
def, err := s.wfStore.GetDefinitionByKey(ctx, workflowKey)
if err != nil {
return nil, ErrWorkflowVersionNotFound
}
verNum := versionNum
if verNum <= 0 {
verNum = def.LatestPublishedVersion
}
version, err := s.wfStore.GetVersion(ctx, def.ID, verNum)
if err != nil {
return nil, ErrWorkflowVersionNotFound
}
plan := buildDefaultPlan(version)
return &plan, nil
}
func (s *Service) resolvePublishedVersion(ctx context.Context, plan WorkflowPlan) (*workflowdef.Version, error) {
if s.wfStore == nil {
return nil, ErrNotAvailable
}
def, err := s.wfStore.GetDefinitionByKey(ctx, plan.WorkflowDefinitionKey)
if err != nil {
return nil, ErrWorkflowVersionNotFound
}
verNum := plan.WorkflowVersion
if verNum <= 0 {
verNum = def.LatestPublishedVersion
}
version, err := s.wfStore.GetVersion(ctx, def.ID, verNum)
if err != nil {
return nil, ErrWorkflowVersionNotFound
}
version.WorkflowKey = def.Key
version.WorkflowName = def.Name
return version, nil
}
func randomToken(n int) (string, error) {
b := make([]byte, n)
if _, err := rand.Read(b); err != nil {
return "", err
}
return hex.EncodeToString(b), nil
}
+140
View File
@@ -0,0 +1,140 @@
package projectinstance
import (
"context"
"errors"
"testing"
"chat-agent-system/internal/workflowdef"
)
type stubVersionResolver struct {
versions map[string]*workflowdef.Version
defs map[string]*workflowdef.Definition
}
func (s *stubVersionResolver) GetDefinitionByKey(_ context.Context, key string) (*workflowdef.Definition, error) {
d, ok := s.defs[key]
if !ok {
return nil, workflowdef.ErrNotFound
}
return d, nil
}
func (s *stubVersionResolver) GetVersion(_ context.Context, workflowID int64, version int) (*workflowdef.Version, error) {
for _, v := range s.versions {
if v.WorkflowID == workflowID && v.Version == version {
copyV := *v
return &copyV, nil
}
}
return nil, workflowdef.ErrNotFound
}
func (s *stubVersionResolver) ListDefinitions(_ context.Context) ([]workflowdef.Definition, error) {
out := make([]workflowdef.Definition, 0, len(s.defs))
for _, d := range s.defs {
out = append(out, *d)
}
return out, nil
}
func testResolver() *stubVersionResolver {
ver := &workflowdef.Version{
ID: 100, WorkflowID: 10, Version: 1, Status: workflowdef.StatusPublished,
WorkflowKey: "medical_device_development", WorkflowName: "医疗器械开发",
Definition: workflowdef.VersionDefinition{
Stages: []workflowdef.StageDef{{
Key: "biz_intake", Name: "商务 intake", Optional: false,
DocTasks: []workflowdef.DocTaskDef{{
Key: "intake", Required: true, Blocking: true, DefaultEnabled: true,
}},
}},
},
}
return &stubVersionResolver{
defs: map[string]*workflowdef.Definition{
"medical_device_development": {ID: 10, Key: "medical_device_development", Name: "医疗器械开发", LatestPublishedVersion: 1},
},
versions: map[string]*workflowdef.Version{"medical_device_development": ver},
}
}
func validPlan() WorkflowPlan {
return WorkflowPlan{
WorkflowDefinitionKey: "medical_device_development",
WorkflowVersion: 1,
Stages: []StagePlan{{
StageKey: "biz_intake", Enabled: true,
DocTasks: []DocTaskPlan{{
TaskKey: "intake", Enabled: true,
Duties: DocDuty{AuthorUserID: 1, ReviewerUserID: 2, ApproverUserID: 3},
}},
}},
}
}
func TestService_PlanPreviewActivate(t *testing.T) {
ctx := context.Background()
store := newMemStore()
_ = SeedProjectMeta(store, 42, LifecyclePlanning, "", 1, 0)
svc := NewService(store, testResolver(), nil)
settings, err := svc.SaveWorkflowPlan(ctx, 42, 1, true, SavePlanInput{
Plan: validPlan(), ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1,
})
if err != nil {
t.Fatalf("SaveWorkflowPlan: %v", err)
}
if settings.Plan == nil || settings.Plan.WorkflowDefinitionKey != "medical_device_development" {
t.Fatalf("expected saved plan, got %+v", settings.Plan)
}
preview, err := svc.ActivationPreview(ctx, 42, 1, true)
if err != nil {
t.Fatalf("ActivationPreview: %v", err)
}
if preview.ConfirmationToken == "" || preview.PlanHash == "" {
t.Fatal("expected confirmation token and plan hash")
}
result, err := svc.Activate(ctx, 42, 1, true, ActivateInput{
ConfirmationToken: preview.ConfirmationToken,
ExpectedProjectVersion: preview.ExpectedProjectVersion,
ExpectedPlanRevision: preview.ExpectedPlanRevision,
})
if err != nil {
t.Fatalf("Activate: %v", err)
}
if result.LifecycleStatus != LifecycleActive || result.ActiveWorkflowKey != "medical_device_development" {
t.Fatalf("unexpected activate result: %+v", result)
}
live, _ := store.HasLiveInstance(ctx, 42)
if !live {
t.Fatal("expected live instance after activation")
}
}
func TestService_BlockPlanEditWhenActive(t *testing.T) {
ctx := context.Background()
store := newMemStore()
_ = SeedProjectMeta(store, 7, LifecycleActive, "main", 2, 1)
svc := NewService(store, testResolver(), nil)
_, err := svc.SaveWorkflowPlan(ctx, 7, 1, true, SavePlanInput{
Plan: validPlan(), ExpectedPlanRevision: 1,
})
if !errors.Is(err, ErrPlanFrozen) {
t.Fatalf("expected ErrPlanFrozen, got %v", err)
}
}
func TestService_SeparationOfDutyValidation(t *testing.T) {
bad := validPlan()
bad.Stages[0].DocTasks[0].Duties.ReviewerUserID = 1
res := validatePlan(bad, testResolver().versions["medical_device_development"])
if res.Valid {
t.Fatal("expected separation of duty violation")
}
}
+564
View File
@@ -0,0 +1,564 @@
package projectinstance
import (
"context"
"encoding/json"
"errors"
"fmt"
"sync"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
// TablesReady 检查 Phase 3 表是否已迁移。
func TablesReady(ctx context.Context, pool *pgxpool.Pool) bool {
if pool == nil {
return false
}
var ok bool
err := pool.QueryRow(ctx, `
SELECT EXISTS (
SELECT 1 FROM information_schema.tables
WHERE table_schema = 'public' AND table_name = 'project_workflow_instances'
)
`).Scan(&ok)
return err == nil && ok
}
// Store 项目实例持久化接口。
type Store interface {
GetProjectMeta(ctx context.Context, projectID int64) (lifecycleStatus string, activeWorkflowKey string, selectionVersion int, planRevision int, err error)
UpdateProjectAfterActivate(ctx context.Context, projectID int64, workflowKey string, expectedVersion int, activatedBy int64) error
BumpPlanRevision(ctx context.Context, projectID int64, expectedPlanRevision int) (int, error)
GetPlanningInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error)
GetLiveInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error)
HasLiveInstance(ctx context.Context, projectID int64) (bool, error)
UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error)
ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error)
SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time) error
ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error)
}
// NewStore 构建 PG 或内存 Store。
func NewStore(pool *pgxpool.Pool) Store {
if pool != nil && TablesReady(context.Background(), pool) {
return &pgStore{pool: pool}
}
return newMemStore()
}
type pgStore struct {
pool *pgxpool.Pool
}
func scanInstance(row pgx.Row) (*WorkflowInstance, error) {
var inst WorkflowInstance
var snapRaw []byte
var activatedBy, createdBy *int64
err := row.Scan(
&inst.ID, &inst.ProjectID, &inst.WorkflowVersionID, &inst.InstanceNo, &inst.Status,
&inst.ConfigRevision, &inst.PlanHash, &snapRaw, &activatedBy, &inst.ActivatedAt,
&createdBy, &inst.CreatedAt, &inst.UpdatedAt,
)
if err != nil {
return nil, err
}
inst.ActivatedBy = activatedBy
inst.CreatedBy = createdBy
if len(snapRaw) > 0 {
_ = json.Unmarshal(snapRaw, &inst.ConfigSnapshot)
}
return &inst, nil
}
func (s *pgStore) GetProjectMeta(ctx context.Context, projectID int64) (string, string, int, int, error) {
var lifecycle, activeKey string
var selVer, planRev int
err := s.pool.QueryRow(ctx, `
SELECT COALESCE(lifecycle_status, 'planning'), COALESCE(active_workflow_key, ''),
COALESCE(workflow_selection_version, 1), COALESCE(workflow_plan_revision, 0)
FROM projects WHERE id = $1
`, projectID).Scan(&lifecycle, &activeKey, &selVer, &planRev)
if errors.Is(err, pgx.ErrNoRows) {
return "", "", 0, 0, ErrProjectNotFound
}
return lifecycle, activeKey, selVer, planRev, err
}
func (s *pgStore) UpdateProjectAfterActivate(ctx context.Context, projectID int64, workflowKey string, expectedVersion int, activatedBy int64) error {
tag, err := s.pool.Exec(ctx, `
UPDATE projects SET lifecycle_status = 'active', active_workflow_key = $2,
workflow_selection_version = workflow_selection_version + 1,
active_workflow_changed_at = now(), active_workflow_changed_by = $4,
updated_at = now()
WHERE id = $1 AND workflow_selection_version = $3
`, projectID, workflowKey, expectedVersion, activatedBy)
if err != nil {
return err
}
if tag.RowsAffected() == 0 {
return ErrConcurrentConflict
}
return nil
}
func (s *pgStore) BumpPlanRevision(ctx context.Context, projectID int64, expectedPlanRevision int) (int, error) {
var newRev int
err := s.pool.QueryRow(ctx, `
UPDATE projects SET workflow_plan_revision = workflow_plan_revision + 1, updated_at = now()
WHERE id = $1 AND workflow_plan_revision = $2
RETURNING workflow_plan_revision
`, projectID, expectedPlanRevision).Scan(&newRev)
if errors.Is(err, pgx.ErrNoRows) {
return 0, ErrPlanRevisionConflict
}
return newRev, err
}
func (s *pgStore) GetPlanningInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error) {
row := s.pool.QueryRow(ctx, `
SELECT id, project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
config_snapshot, activated_by, activated_at, created_by, created_at, updated_at
FROM project_workflow_instances WHERE project_id = $1 AND status = 'planning'
ORDER BY id DESC LIMIT 1
`, projectID)
inst, err := scanInstance(row)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
return inst, err
}
func (s *pgStore) GetLiveInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error) {
row := s.pool.QueryRow(ctx, `
SELECT id, project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
config_snapshot, activated_by, activated_at, created_by, created_at, updated_at
FROM project_workflow_instances
WHERE project_id = $1 AND status IN ('active', 'awaiting_stage_confirmation', 'termination_requested')
ORDER BY id DESC LIMIT 1
`, projectID)
inst, err := scanInstance(row)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
return inst, err
}
func (s *pgStore) HasLiveInstance(ctx context.Context, projectID int64) (bool, error) {
var ok bool
err := s.pool.QueryRow(ctx, `
SELECT EXISTS (
SELECT 1 FROM project_workflow_instances
WHERE project_id = $1 AND status IN ('active', 'awaiting_stage_confirmation', 'termination_requested')
)
`, projectID).Scan(&ok)
return ok, err
}
func (s *pgStore) UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) {
snapRaw, _ := json.Marshal(inst.ConfigSnapshot)
now := time.Now()
if inst.CreatedAt.IsZero() {
inst.CreatedAt = now
}
inst.UpdatedAt = now
tx, err := s.pool.Begin(ctx)
if err != nil {
return nil, err
}
defer tx.Rollback(ctx)
var existingID int64
err = tx.QueryRow(ctx, `
SELECT id FROM project_workflow_instances WHERE project_id = $1 AND status = 'planning' LIMIT 1
`, inst.ProjectID).Scan(&existingID)
if errors.Is(err, pgx.ErrNoRows) {
err = tx.QueryRow(ctx, `
INSERT INTO project_workflow_instances (
project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
config_snapshot, created_by, created_at, updated_at
) VALUES ($1, $2, 1, 'planning', $3, $4, $5, $6, $7, $8)
RETURNING id
`, inst.ProjectID, inst.WorkflowVersionID, inst.ConfigRevision, inst.PlanHash, snapRaw,
inst.CreatedBy, inst.CreatedAt, inst.UpdatedAt).Scan(&inst.ID)
if err != nil {
return nil, err
}
} else if err != nil {
return nil, err
} else {
err = tx.QueryRow(ctx, `
UPDATE project_workflow_instances SET
workflow_version_id = $2, config_revision = $3, plan_hash = $4, config_snapshot = $5,
updated_at = $6
WHERE id = $1 AND status = 'planning'
RETURNING id
`, existingID, inst.WorkflowVersionID, inst.ConfigRevision, inst.PlanHash, snapRaw, inst.UpdatedAt).Scan(&inst.ID)
if err != nil {
return nil, err
}
}
if err := s.replaceInstanceChildren(ctx, tx, inst); err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, err
}
return inst, nil
}
func (s *pgStore) replaceInstanceChildren(ctx context.Context, tx pgx.Tx, inst *WorkflowInstance) error {
_, err := tx.Exec(ctx, `
DELETE FROM document_duty_assignments WHERE doc_task_instance_id IN (
SELECT dti.id FROM project_doc_task_instances dti
JOIN project_stage_instances si ON si.id = dti.stage_instance_id
WHERE si.workflow_instance_id = $1
)
`, inst.ID)
if err != nil {
return err
}
_, err = tx.Exec(ctx, `
DELETE FROM project_doc_task_instances WHERE stage_instance_id IN (
SELECT id FROM project_stage_instances WHERE workflow_instance_id = $1
)
`, inst.ID)
if err != nil {
return err
}
_, err = tx.Exec(ctx, `DELETE FROM project_stage_instances WHERE workflow_instance_id = $1`, inst.ID)
if err != nil {
return err
}
for _, st := range inst.ConfigSnapshot.Stages {
var stageID int64
err = tx.QueryRow(ctx, `
INSERT INTO project_stage_instances (workflow_instance_id, stage_key, position, enabled, status)
VALUES ($1, $2, 0, $3, 'pending') RETURNING id
`, inst.ID, st.StageKey, st.Enabled).Scan(&stageID)
if err != nil {
return err
}
for _, dt := range st.DocTasks {
var docTaskID int64
err = tx.QueryRow(ctx, `
INSERT INTO project_doc_task_instances (stage_instance_id, task_key, enabled, status)
VALUES ($1, $2, $3, 'pending') RETURNING id
`, stageID, dt.TaskKey, dt.Enabled).Scan(&docTaskID)
if err != nil {
return err
}
if !dt.Enabled {
continue
}
for role, uid := range map[string]int64{
DutyAuthor: dt.Duties.AuthorUserID, DutyReviewer: dt.Duties.ReviewerUserID, DutyApprover: dt.Duties.ApproverUserID,
} {
if uid <= 0 {
continue
}
_, err = tx.Exec(ctx, `
INSERT INTO document_duty_assignments (doc_task_instance_id, duty_role, user_id, assigned_by)
VALUES ($1, $2, $3, $4)
`, docTaskID, role, uid, inst.CreatedBy)
if err != nil {
return err
}
}
}
}
return nil
}
func (s *pgStore) ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) {
tx, err := s.pool.Begin(ctx)
if err != nil {
return nil, err
}
defer tx.Rollback(ctx)
tag, err := tx.Exec(ctx, `
UPDATE projects SET lifecycle_status = 'active', active_workflow_key = $2,
workflow_selection_version = workflow_selection_version + 1,
active_workflow_changed_at = now(), active_workflow_changed_by = $4,
updated_at = now()
WHERE id = $1 AND workflow_selection_version = $3
`, projectID, workflowKey, expectedVersion, activatedBy)
if err != nil {
return nil, err
}
if tag.RowsAffected() == 0 {
return nil, ErrConcurrentConflict
}
now := time.Now()
row := tx.QueryRow(ctx, `
UPDATE project_workflow_instances SET status = 'active', activated_by = $2, activated_at = $3, updated_at = $3
WHERE id = $1 AND status = 'planning'
RETURNING id, project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
config_snapshot, activated_by, activated_at, created_by, created_at, updated_at
`, inst.ID, activatedBy, now)
activated, err := scanInstance(row)
if err != nil {
return nil, err
}
_, err = tx.Exec(ctx, `
UPDATE project_stage_instances SET status = CASE WHEN enabled THEN 'active' ELSE 'skipped' END, updated_at = now()
WHERE workflow_instance_id = $1 AND stage_key = (
SELECT stage_key FROM project_stage_instances
WHERE workflow_instance_id = $1 AND enabled = true
ORDER BY id ASC LIMIT 1
)
`, inst.ID)
if err != nil {
return nil, err
}
_, err = tx.Exec(ctx, `
UPDATE project_stage_instances SET status = 'pending', updated_at = now()
WHERE workflow_instance_id = $1 AND status <> 'active' AND enabled = true
`, inst.ID)
if err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, err
}
return activated, nil
}
func (s *pgStore) SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time) error {
raw, _ := json.Marshal(snapshot)
_, err := s.pool.Exec(ctx, `
INSERT INTO workflow_activation_tokens (token, project_id, plan_hash, expected_project_version, config_snapshot, created_by, expires_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)
`, token, projectID, planHash, expectedVersion, raw, createdBy, expiresAt)
return err
}
func (s *pgStore) ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) {
var snapRaw []byte
var storedHash string
var usedAt *time.Time
var expiresAt time.Time
err := s.pool.QueryRow(ctx, `
SELECT plan_hash, config_snapshot, used_at, expires_at
FROM workflow_activation_tokens WHERE token = $1 AND project_id = $2
`, token, projectID).Scan(&storedHash, &snapRaw, &usedAt, &expiresAt)
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrTokenInvalid
}
if err != nil {
return nil, err
}
if usedAt != nil || time.Now().After(expiresAt) || storedHash != planHash {
return nil, ErrTokenInvalid
}
tag, err := s.pool.Exec(ctx, `
UPDATE workflow_activation_tokens SET used_at = now() WHERE token = $1 AND used_at IS NULL
`, token)
if err != nil {
return nil, err
}
if tag.RowsAffected() == 0 {
return nil, ErrTokenInvalid
}
var plan WorkflowPlan
_ = json.Unmarshal(snapRaw, &plan)
return &plan, nil
}
// =============================================================================
// 内存 Store(单元测试 / 无 PG 迁移)
// =============================================================================
type memStore struct {
mu sync.Mutex
meta map[int64]projectMeta
plan map[int64]*WorkflowInstance
live map[int64]*WorkflowInstance
tokens map[string]activationTokenRec
}
type projectMeta struct {
lifecycle string
activeKey string
selectionVer int
planRevision int
}
type activationTokenRec struct {
projectID int64
planHash string
expectedVer int
snapshot WorkflowPlan
expiresAt time.Time
used bool
}
func newMemStore() *memStore {
return &memStore{
meta: make(map[int64]projectMeta),
plan: make(map[int64]*WorkflowInstance),
live: make(map[int64]*WorkflowInstance),
tokens: make(map[string]activationTokenRec),
}
}
func (m *memStore) ensureMeta(projectID int64) projectMeta {
if pm, ok := m.meta[projectID]; ok {
return pm
}
pm := projectMeta{lifecycle: LifecyclePlanning, selectionVer: 1}
m.meta[projectID] = pm
return pm
}
func (m *memStore) GetProjectMeta(_ context.Context, projectID int64) (string, string, int, int, error) {
m.mu.Lock()
defer m.mu.Unlock()
pm, ok := m.meta[projectID]
if !ok {
return "", "", 0, 0, ErrProjectNotFound
}
return pm.lifecycle, pm.activeKey, pm.selectionVer, pm.planRevision, nil
}
func (m *memStore) UpdateProjectAfterActivate(_ context.Context, projectID int64, workflowKey string, expectedVersion int, _ int64) error {
m.mu.Lock()
defer m.mu.Unlock()
pm := m.ensureMeta(projectID)
if pm.selectionVer != expectedVersion {
return ErrConcurrentConflict
}
pm.lifecycle = LifecycleActive
pm.activeKey = workflowKey
pm.selectionVer++
m.meta[projectID] = pm
return nil
}
func (m *memStore) BumpPlanRevision(_ context.Context, projectID int64, expectedPlanRevision int) (int, error) {
m.mu.Lock()
defer m.mu.Unlock()
pm := m.ensureMeta(projectID)
if pm.planRevision != expectedPlanRevision {
return 0, ErrPlanRevisionConflict
}
pm.planRevision++
m.meta[projectID] = pm
return pm.planRevision, nil
}
func (m *memStore) GetPlanningInstance(_ context.Context, projectID int64) (*WorkflowInstance, error) {
m.mu.Lock()
defer m.mu.Unlock()
inst := m.plan[projectID]
if inst == nil {
return nil, nil
}
copyInst := *inst
return &copyInst, nil
}
func (m *memStore) GetLiveInstance(_ context.Context, projectID int64) (*WorkflowInstance, error) {
m.mu.Lock()
defer m.mu.Unlock()
inst := m.live[projectID]
if inst == nil {
return nil, nil
}
copyInst := *inst
return &copyInst, nil
}
func (m *memStore) HasLiveInstance(_ context.Context, projectID int64) (bool, error) {
m.mu.Lock()
defer m.mu.Unlock()
return m.live[projectID] != nil, nil
}
func (m *memStore) UpsertPlanningInstance(_ context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) {
m.mu.Lock()
defer m.mu.Unlock()
if inst.ID == 0 {
inst.ID = int64(len(m.plan) + 1)
}
inst.Status = InstancePlanning
now := time.Now()
if inst.CreatedAt.IsZero() {
inst.CreatedAt = now
}
inst.UpdatedAt = now
copyInst := *inst
m.plan[inst.ProjectID] = &copyInst
return &copyInst, nil
}
func (m *memStore) ActivateInstance(_ context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) {
m.mu.Lock()
defer m.mu.Unlock()
pm := m.ensureMeta(projectID)
if pm.selectionVer != expectedVersion {
return nil, ErrConcurrentConflict
}
pm.lifecycle = LifecycleActive
pm.activeKey = workflowKey
pm.selectionVer++
m.meta[projectID] = pm
now := time.Now()
activated := *inst
activated.Status = InstanceActive
activated.ActivatedBy = &activatedBy
activated.ActivatedAt = &now
m.live[projectID] = &activated
delete(m.plan, projectID)
return &activated, nil
}
func (m *memStore) SaveActivationToken(_ context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, _ int64, expiresAt time.Time) error {
m.mu.Lock()
defer m.mu.Unlock()
m.tokens[token] = activationTokenRec{
projectID: projectID, planHash: planHash, expectedVer: expectedVersion,
snapshot: snapshot, expiresAt: expiresAt,
}
return nil
}
func (m *memStore) ConsumeActivationToken(_ context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) {
m.mu.Lock()
defer m.mu.Unlock()
rec, ok := m.tokens[token]
if !ok || rec.used || rec.projectID != projectID || rec.planHash != planHash || time.Now().After(rec.expiresAt) {
return nil, ErrTokenInvalid
}
rec.used = true
m.tokens[token] = rec
plan := rec.snapshot
return &plan, nil
}
// SeedProjectMeta 供测试注入项目元数据。
func SeedProjectMeta(store Store, projectID int64, lifecycle, activeKey string, selectionVer, planRev int) error {
ms, ok := store.(*memStore)
if !ok {
return fmt.Errorf("SeedProjectMeta only supported on memStore")
}
ms.mu.Lock()
defer ms.mu.Unlock()
ms.meta[projectID] = projectMeta{
lifecycle: lifecycle, activeKey: activeKey,
selectionVer: selectionVer, planRevision: planRev,
}
return nil
}
+133
View File
@@ -0,0 +1,133 @@
package projectinstance
import (
"time"
"chat-agent-system/internal/workflowdef"
)
const (
LifecyclePlanning = "planning"
LifecycleActive = "active"
InstancePlanning = "planning"
InstanceActive = "active"
DutyAuthor = "author"
DutyReviewer = "reviewer"
DutyApprover = "approver"
)
// DocDuty 文档三权职责指派。
type DocDuty struct {
AuthorUserID int64 `json:"author_user_id,omitempty"`
ReviewerUserID int64 `json:"reviewer_user_id,omitempty"`
ApproverUserID int64 `json:"approver_user_id,omitempty"`
}
// DocTaskPlan 文档任务裁剪与职责。
type DocTaskPlan struct {
TaskKey string `json:"task_key"`
Enabled bool `json:"enabled"`
Duties DocDuty `json:"duties"`
}
// StagePlan 阶段裁剪与文档任务。
type StagePlan struct {
StageKey string `json:"stage_key"`
Enabled bool `json:"enabled"`
DocTasks []DocTaskPlan `json:"doc_tasks"`
}
// WorkflowPlan 项目工作流规划快照。
type WorkflowPlan struct {
WorkflowDefinitionKey string `json:"workflow_definition_key"`
WorkflowVersion int `json:"workflow_version"`
WorkflowVersionID int64 `json:"workflow_version_id,omitempty"`
Stages []StagePlan `json:"stages"`
}
// WorkflowInstance 项目工作流实例。
type WorkflowInstance struct {
ID int64 `json:"id"`
ProjectID int64 `json:"project_id"`
WorkflowVersionID int64 `json:"workflow_version_id"`
InstanceNo int `json:"instance_no"`
Status string `json:"status"`
ConfigRevision int `json:"config_revision"`
PlanHash string `json:"plan_hash"`
ConfigSnapshot WorkflowPlan `json:"config_snapshot"`
ActivatedBy *int64 `json:"activated_by,omitempty"`
ActivatedAt *time.Time `json:"activated_at,omitempty"`
CreatedBy *int64 `json:"created_by,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// ProjectSettings 项目设置聚合视图。
type ProjectSettings struct {
ProjectID int64 `json:"project_id"`
LifecycleStatus string `json:"lifecycle_status"`
ActiveWorkflowKey string `json:"active_workflow_key"`
WorkflowSelectionVersion int `json:"workflow_selection_version"`
WorkflowPlanRevision int `json:"workflow_plan_revision"`
Plan *WorkflowPlan `json:"plan,omitempty"`
PlanningInstance *WorkflowInstance `json:"planning_instance,omitempty"`
LiveInstance *WorkflowInstance `json:"live_instance,omitempty"`
CanEditPlan bool `json:"can_edit_plan"`
CanActivate bool `json:"can_activate"`
PublishedWorkflows []PublishedOption `json:"published_workflows,omitempty"`
}
// PublishedOption 可选已发布工作流。
type PublishedOption struct {
DefinitionKey string `json:"definition_key"`
DefinitionName string `json:"definition_name"`
LatestVersion int `json:"latest_version"`
WorkflowVersion int64 `json:"workflow_version_id"`
ProductDomain string `json:"product_domain,omitempty"`
}
// SavePlanInput 保存工作流规划。
type SavePlanInput struct {
Plan WorkflowPlan `json:"plan"`
ExpectedPlanRevision int `json:"expected_plan_revision"`
ExpectedSelectionVersion int `json:"expected_selection_version"`
}
// ActivationPreview 激活预览与确认令牌。
type ActivationPreview struct {
ConfirmationToken string `json:"confirmation_token"`
PlanHash string `json:"plan_hash"`
ExpectedProjectVersion int `json:"expected_project_version"`
ExpectedPlanRevision int `json:"expected_plan_revision"`
Summary PreviewSummary `json:"summary"`
Validation workflowdef.ValidationResult `json:"validation"`
ExpiresAt time.Time `json:"expires_at"`
}
// PreviewSummary 激活影响摘要。
type PreviewSummary struct {
WorkflowName string `json:"workflow_name"`
WorkflowVersion int `json:"workflow_version"`
EnabledStages int `json:"enabled_stages"`
EnabledDocTasks int `json:"enabled_doc_tasks"`
AssignedDuties int `json:"assigned_duties"`
FirstStageKey string `json:"first_stage_key,omitempty"`
FirstStageName string `json:"first_stage_name,omitempty"`
}
// ActivateInput 二次确认激活。
type ActivateInput struct {
ConfirmationToken string `json:"confirmation_token"`
ExpectedProjectVersion int `json:"expected_project_version"`
ExpectedPlanRevision int `json:"expected_plan_revision"`
}
// ActivateResult 激活结果。
type ActivateResult struct {
ProjectID int64 `json:"project_id"`
LifecycleStatus string `json:"lifecycle_status"`
ActiveWorkflowKey string `json:"active_workflow_key"`
Instance WorkflowInstance `json:"instance"`
}
+213
View File
@@ -0,0 +1,213 @@
package projectinstance
import (
"fmt"
"chat-agent-system/internal/workflowdef"
)
func validatePlan(plan WorkflowPlan, version *workflowdef.Version) workflowdef.ValidationResult {
issues := make([]workflowdef.ValidationIssue, 0)
if version == nil {
return workflowdef.ValidationResult{
Valid: false,
Issues: []workflowdef.ValidationIssue{{
Code: "WORKFLOW_VERSION_NOT_FOUND", Message: "工作流版本不存在或未发布",
}},
}
}
if version.Status != workflowdef.StatusPublished {
issues = append(issues, workflowdef.ValidationIssue{
Code: "WORKFLOW_NOT_PUBLISHED", Message: "只能绑定已发布的工作流版本",
})
}
if plan.WorkflowDefinitionKey != version.WorkflowKey {
issues = append(issues, workflowdef.ValidationIssue{
Code: "WORKFLOW_KEY_MISMATCH", Field: "workflow_definition_key",
Message: fmt.Sprintf("规划 key %s 与版本 %s 不一致", plan.WorkflowDefinitionKey, version.WorkflowKey),
})
}
if plan.WorkflowVersion != version.Version {
issues = append(issues, workflowdef.ValidationIssue{
Code: "WORKFLOW_VERSION_MISMATCH", Field: "workflow_version",
Message: fmt.Sprintf("规划版本 %d 与已发布版本 %d 不一致", plan.WorkflowVersion, version.Version),
})
}
stageDefMap := make(map[string]workflowdef.StageDef, len(version.Definition.Stages))
for _, st := range version.Definition.Stages {
stageDefMap[st.Key] = st
}
seenStage := map[string]bool{}
for _, st := range plan.Stages {
if seenStage[st.StageKey] {
issues = append(issues, workflowdef.ValidationIssue{
Code: "DUPLICATE_STAGE", Field: st.StageKey, Message: "阶段 key 重复",
})
continue
}
seenStage[st.StageKey] = true
def, ok := stageDefMap[st.StageKey]
if !ok {
issues = append(issues, workflowdef.ValidationIssue{
Code: "UNKNOWN_STAGE", Field: st.StageKey, Message: "阶段不在工作流定义中",
})
continue
}
if !def.Optional && !st.Enabled {
issues = append(issues, workflowdef.ValidationIssue{
Code: "REQUIRED_STAGE_DISABLED", Field: st.StageKey, Message: "必需阶段不能关闭",
})
}
taskDefMap := make(map[string]workflowdef.DocTaskDef, len(def.DocTasks))
for _, dt := range def.DocTasks {
taskDefMap[dt.Key] = dt
}
seenTask := map[string]bool{}
for _, dt := range st.DocTasks {
if seenTask[dt.TaskKey] {
issues = append(issues, workflowdef.ValidationIssue{
Code: "DUPLICATE_DOC_TASK", Field: dt.TaskKey, Message: "文档任务 key 重复",
})
continue
}
seenTask[dt.TaskKey] = true
td, ok := taskDefMap[dt.TaskKey]
if !ok {
issues = append(issues, workflowdef.ValidationIssue{
Code: "UNKNOWN_DOC_TASK", Field: dt.TaskKey, Message: "文档任务不在阶段定义中",
})
continue
}
if (td.Required || td.Blocking) && !dt.Enabled {
issues = append(issues, workflowdef.ValidationIssue{
Code: "REQUIRED_DOC_DISABLED", Field: dt.TaskKey, Message: "必需/阻断文档任务不能关闭",
})
}
if dt.Enabled {
issues = append(issues, validateDuties(dt.Duties, dt.TaskKey)...)
}
}
}
enabledStages := 0
for _, st := range plan.Stages {
if st.Enabled {
enabledStages++
}
}
if enabledStages == 0 {
issues = append(issues, workflowdef.ValidationIssue{
Code: "NO_ENABLED_STAGE", Message: "至少启用一个阶段",
})
}
return workflowdef.ValidationResult{Valid: len(issues) == 0, Issues: issues}
}
func validateDuties(d DocDuty, taskKey string) []workflowdef.ValidationIssue {
issues := make([]workflowdef.ValidationIssue, 0)
if d.AuthorUserID <= 0 {
issues = append(issues, workflowdef.ValidationIssue{
Code: "DUTY_AUTHOR_REQUIRED", Field: taskKey + ".author",
Message: "启用的文档任务必须指定作者",
})
}
if d.ReviewerUserID <= 0 {
issues = append(issues, workflowdef.ValidationIssue{
Code: "DUTY_REVIEWER_REQUIRED", Field: taskKey + ".reviewer",
Message: "启用的文档任务必须指定审核人",
})
}
if d.ApproverUserID <= 0 {
issues = append(issues, workflowdef.ValidationIssue{
Code: "DUTY_APPROVER_REQUIRED", Field: taskKey + ".approver",
Message: "启用的文档任务必须指定批准人",
})
}
ids := []int64{d.AuthorUserID, d.ReviewerUserID, d.ApproverUserID}
for i := 0; i < len(ids); i++ {
if ids[i] <= 0 {
continue
}
for j := i + 1; j < len(ids); j++ {
if ids[j] > 0 && ids[i] == ids[j] {
issues = append(issues, workflowdef.ValidationIssue{
Code: "SEPARATION_OF_DUTY_VIOLATION", Field: taskKey,
Message: "作者、审核人、批准人必须分离",
})
return issues
}
}
}
return issues
}
func buildDefaultPlan(version *workflowdef.Version) WorkflowPlan {
plan := WorkflowPlan{
WorkflowDefinitionKey: version.WorkflowKey,
WorkflowVersion: version.Version,
WorkflowVersionID: version.ID,
Stages: make([]StagePlan, 0, len(version.Definition.Stages)),
}
for _, st := range version.Definition.Stages {
sp := StagePlan{
StageKey: st.Key,
Enabled: !st.Optional,
DocTasks: make([]DocTaskPlan, 0, len(st.DocTasks)),
}
for _, dt := range st.DocTasks {
sp.DocTasks = append(sp.DocTasks, DocTaskPlan{
TaskKey: dt.Key,
Enabled: dt.DefaultEnabled || dt.Required || dt.Blocking,
})
}
plan.Stages = append(plan.Stages, sp)
}
return plan
}
func previewSummary(plan WorkflowPlan, version *workflowdef.Version) PreviewSummary {
sum := PreviewSummary{
WorkflowVersion: plan.WorkflowVersion,
}
if version != nil {
sum.WorkflowName = version.WorkflowName
if sum.WorkflowName == "" {
sum.WorkflowName = version.WorkflowKey
}
}
for _, st := range plan.Stages {
if !st.Enabled {
continue
}
sum.EnabledStages++
if sum.FirstStageKey == "" {
sum.FirstStageKey = st.StageKey
if version != nil {
for _, sd := range version.Definition.Stages {
if sd.Key == st.StageKey {
sum.FirstStageName = sd.Name
break
}
}
}
}
for _, dt := range st.DocTasks {
if !dt.Enabled {
continue
}
sum.EnabledDocTasks++
if dt.Duties.AuthorUserID > 0 {
sum.AssignedDuties++
}
if dt.Duties.ReviewerUserID > 0 {
sum.AssignedDuties++
}
if dt.Duties.ApproverUserID > 0 {
sum.AssignedDuties++
}
}
}
return sum
}
+3
View File
@@ -23,4 +23,7 @@ var (
// ErrInvalidInput 输入参数不合法
ErrInvalidInput = errors.New("invalid parameter: required field missing or malformed")
// ErrLegacySwitchBlocked 已启用实例化工作流治理,禁止旧式即时切换
ErrLegacySwitchBlocked = errors.New("legacy workflow switch blocked: use project settings activation flow")
)
+7 -6
View File
@@ -54,12 +54,11 @@ func (s *Service) CreateProject(ctx context.Context, user *auth.UserAccount, in
code = generateProjectCode(name)
}
initialWorkflow := in.InitialWorkflowKey
if initialWorkflow == "" {
initialWorkflow = "main"
}
if _, ok := s.catalog.Get(initialWorkflow); !ok {
return nil, fmt.Errorf("%w: 初始工作流 %s 不存在", ErrWorkflowNotFound, initialWorkflow)
initialWorkflow := strings.TrimSpace(in.InitialWorkflowKey)
if initialWorkflow != "" {
if _, ok := s.catalog.Get(initialWorkflow); !ok {
return nil, fmt.Errorf("%w: 初始工作流 %s 不存在", ErrWorkflowNotFound, initialWorkflow)
}
}
p := &Project{
@@ -73,8 +72,10 @@ func (s *Service) CreateProject(ctx context.Context, user *auth.UserAccount, in
Environment: "dev",
Status: "active",
CreatorUserID: user.ID,
LifecycleStatus: "planning",
ActiveWorkflowKey: initialWorkflow,
WorkflowSelectionVersion: 1,
WorkflowPlanRevision: 0,
}
created, err := s.store.CreateProject(ctx, p)
+5 -2
View File
@@ -36,8 +36,11 @@ func TestProjectWorkflowService_CreateAndCreatorACL(t *testing.T) {
if proj.CreatorUserID != creator.ID {
t.Errorf("expected CreatorUserID %d, got %d", creator.ID, proj.CreatorUserID)
}
if proj.ActiveWorkflowKey != "main" {
t.Errorf("expected default active workflow 'main', got '%s'", proj.ActiveWorkflowKey)
if proj.LifecycleStatus != "planning" {
t.Errorf("expected lifecycle 'planning', got '%s'", proj.LifecycleStatus)
}
if proj.ActiveWorkflowKey != "" {
t.Errorf("expected empty active workflow in planning mode, got '%s'", proj.ActiveWorkflowKey)
}
if proj.WorkflowSelectionVersion != 1 {
t.Errorf("expected initial version 1, got %d", proj.WorkflowSelectionVersion)
+20 -15
View File
@@ -62,8 +62,8 @@ func (s *pgStore) CreateProject(ctx context.Context, p *Project) (*Project, erro
if p.CurrentStage == "" {
p.CurrentStage = "V0.1_triage"
}
if p.ActiveWorkflowKey == "" {
p.ActiveWorkflowKey = "main"
if p.LifecycleStatus == "" {
p.LifecycleStatus = "planning"
}
if p.WorkflowSelectionVersion <= 0 {
p.WorkflowSelectionVersion = 1
@@ -72,16 +72,18 @@ func (s *pgStore) CreateProject(ctx context.Context, p *Project) (*Project, erro
query := `
INSERT INTO projects (
code, name, customer_name, device_type, intended_use, target_market,
current_stage, environment, status, creator_user_id, active_workflow_key,
workflow_selection_version, created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
current_stage, environment, status, creator_user_id, lifecycle_status,
active_workflow_key, workflow_selection_version, workflow_plan_revision,
created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
RETURNING id
`
err := s.pool.QueryRow(
ctx, query,
p.Code, p.Name, p.CustomerName, p.DeviceType, p.IntendedUse, p.TargetMarket,
p.CurrentStage, p.Environment, p.Status, p.CreatorUserID, p.ActiveWorkflowKey,
p.WorkflowSelectionVersion, p.CreatedAt, p.UpdatedAt,
p.CurrentStage, p.Environment, p.Status, p.CreatorUserID, p.LifecycleStatus,
p.ActiveWorkflowKey, p.WorkflowSelectionVersion, p.WorkflowPlanRevision,
p.CreatedAt, p.UpdatedAt,
).Scan(&p.ID)
if err != nil {
return nil, fmt.Errorf("insert project failed: %w", err)
@@ -94,7 +96,8 @@ func (s *pgStore) GetProjectByID(ctx context.Context, id int64) (*Project, error
query := `
SELECT id, code, name, customer_name, device_type, intended_use, target_market,
current_stage, environment, status, COALESCE(creator_user_id, 1),
COALESCE(active_workflow_key, 'main'), COALESCE(workflow_selection_version, 1),
COALESCE(lifecycle_status, 'planning'), COALESCE(active_workflow_key, ''),
COALESCE(workflow_selection_version, 1), COALESCE(workflow_plan_revision, 0),
active_workflow_changed_at, active_workflow_changed_by, created_at, updated_at
FROM projects
WHERE id = $1
@@ -103,7 +106,7 @@ func (s *pgStore) GetProjectByID(ctx context.Context, id int64) (*Project, error
err := s.pool.QueryRow(ctx, query, id).Scan(
&p.ID, &p.Code, &p.Name, &p.CustomerName, &p.DeviceType, &p.IntendedUse, &p.TargetMarket,
&p.CurrentStage, &p.Environment, &p.Status, &p.CreatorUserID,
&p.ActiveWorkflowKey, &p.WorkflowSelectionVersion,
&p.LifecycleStatus, &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, &p.WorkflowPlanRevision,
&p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt,
)
if errors.Is(err, pgx.ErrNoRows) {
@@ -119,7 +122,8 @@ func (s *pgStore) GetProjectByCode(ctx context.Context, code string) (*Project,
query := `
SELECT id, code, name, customer_name, device_type, intended_use, target_market,
current_stage, environment, status, COALESCE(creator_user_id, 1),
COALESCE(active_workflow_key, 'main'), COALESCE(workflow_selection_version, 1),
COALESCE(lifecycle_status, 'planning'), COALESCE(active_workflow_key, ''),
COALESCE(workflow_selection_version, 1), COALESCE(workflow_plan_revision, 0),
active_workflow_changed_at, active_workflow_changed_by, created_at, updated_at
FROM projects
WHERE code = $1
@@ -128,7 +132,7 @@ func (s *pgStore) GetProjectByCode(ctx context.Context, code string) (*Project,
err := s.pool.QueryRow(ctx, query, code).Scan(
&p.ID, &p.Code, &p.Name, &p.CustomerName, &p.DeviceType, &p.IntendedUse, &p.TargetMarket,
&p.CurrentStage, &p.Environment, &p.Status, &p.CreatorUserID,
&p.ActiveWorkflowKey, &p.WorkflowSelectionVersion,
&p.LifecycleStatus, &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, &p.WorkflowPlanRevision,
&p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt,
)
if errors.Is(err, pgx.ErrNoRows) {
@@ -145,7 +149,8 @@ func (s *pgStore) ListProjectsForUser(ctx context.Context, userID int64) ([]*Pro
query := `
SELECT DISTINCT p.id, p.code, p.name, p.customer_name, p.device_type, p.intended_use, p.target_market,
p.current_stage, p.environment, p.status, COALESCE(p.creator_user_id, 1),
COALESCE(p.active_workflow_key, 'main'), COALESCE(p.workflow_selection_version, 1),
COALESCE(p.lifecycle_status, 'planning'), COALESCE(p.active_workflow_key, ''),
COALESCE(p.workflow_selection_version, 1), COALESCE(p.workflow_plan_revision, 0),
p.active_workflow_changed_at, p.active_workflow_changed_by, p.created_at, p.updated_at
FROM projects p
LEFT JOIN chat_sessions cs ON cs.project_id = p.id AND cs.user_id = $1
@@ -164,7 +169,7 @@ func (s *pgStore) ListProjectsForUser(ctx context.Context, userID int64) ([]*Pro
if err := rows.Scan(
&p.ID, &p.Code, &p.Name, &p.CustomerName, &p.DeviceType, &p.IntendedUse, &p.TargetMarket,
&p.CurrentStage, &p.Environment, &p.Status, &p.CreatorUserID,
&p.ActiveWorkflowKey, &p.WorkflowSelectionVersion,
&p.LifecycleStatus, &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, &p.WorkflowPlanRevision,
&p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt,
); err != nil {
return nil, err
@@ -604,8 +609,8 @@ func (m *memStore) CreateProject(ctx context.Context, p *Project) (*Project, err
if p.Status == "" {
p.Status = "active"
}
if p.ActiveWorkflowKey == "" {
p.ActiveWorkflowKey = "main"
if p.LifecycleStatus == "" {
p.LifecycleStatus = "planning"
}
if p.WorkflowSelectionVersion <= 0 {
p.WorkflowSelectionVersion = 1
+2
View File
@@ -17,8 +17,10 @@ type Project struct {
Environment string `json:"environment"`
Status string `json:"status"`
CreatorUserID int64 `json:"creator_user_id"`
LifecycleStatus string `json:"lifecycle_status"`
ActiveWorkflowKey string `json:"active_workflow_key"`
WorkflowSelectionVersion int `json:"workflow_selection_version"`
WorkflowPlanRevision int `json:"workflow_plan_revision"`
ActiveWorkflowChangedAt *time.Time `json:"active_workflow_changed_at,omitempty"`
ActiveWorkflowChangedBy *int64 `json:"active_workflow_changed_by,omitempty"`
CreatedAt time.Time `json:"created_at"`