From dc629fc4b83bb4038112174185368bc046181e61 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 12 Sep 2026 03:03:34 +0000 Subject: [PATCH] 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 --- internal/projectinstance/errors.go | 19 + internal/projectinstance/hash.go | 31 ++ internal/projectinstance/service.go | 319 +++++++++++++ internal/projectinstance/service_test.go | 140 ++++++ internal/projectinstance/store.go | 564 +++++++++++++++++++++++ internal/projectinstance/types.go | 133 ++++++ internal/projectinstance/validator.go | 213 +++++++++ internal/projectworkflow/errors.go | 3 + internal/projectworkflow/service.go | 13 +- internal/projectworkflow/service_test.go | 7 +- internal/projectworkflow/store.go | 35 +- internal/projectworkflow/types.go | 2 + 12 files changed, 1456 insertions(+), 23 deletions(-) create mode 100644 internal/projectinstance/errors.go create mode 100644 internal/projectinstance/hash.go create mode 100644 internal/projectinstance/service.go create mode 100644 internal/projectinstance/service_test.go create mode 100644 internal/projectinstance/store.go create mode 100644 internal/projectinstance/types.go create mode 100644 internal/projectinstance/validator.go diff --git a/internal/projectinstance/errors.go b/internal/projectinstance/errors.go new file mode 100644 index 00000000..953864f2 --- /dev/null +++ b/internal/projectinstance/errors.go @@ -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") +) diff --git a/internal/projectinstance/hash.go b/internal/projectinstance/hash.go new file mode 100644 index 00000000..8266e794 --- /dev/null +++ b/internal/projectinstance/hash.go @@ -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 +} diff --git a/internal/projectinstance/service.go b/internal/projectinstance/service.go new file mode 100644 index 00000000..4b34ca69 --- /dev/null +++ b/internal/projectinstance/service.go @@ -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 +} diff --git a/internal/projectinstance/service_test.go b/internal/projectinstance/service_test.go new file mode 100644 index 00000000..d701ee7a --- /dev/null +++ b/internal/projectinstance/service_test.go @@ -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 ©V, 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") + } +} diff --git a/internal/projectinstance/store.go b/internal/projectinstance/store.go new file mode 100644 index 00000000..e82f0b60 --- /dev/null +++ b/internal/projectinstance/store.go @@ -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 ©Inst, 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 ©Inst, 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] = ©Inst + return ©Inst, 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 +} diff --git a/internal/projectinstance/types.go b/internal/projectinstance/types.go new file mode 100644 index 00000000..8d53587f --- /dev/null +++ b/internal/projectinstance/types.go @@ -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"` +} diff --git a/internal/projectinstance/validator.go b/internal/projectinstance/validator.go new file mode 100644 index 00000000..24ebabc3 --- /dev/null +++ b/internal/projectinstance/validator.go @@ -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 +} diff --git a/internal/projectworkflow/errors.go b/internal/projectworkflow/errors.go index e0ed3ca8..009b7dba 100644 --- a/internal/projectworkflow/errors.go +++ b/internal/projectworkflow/errors.go @@ -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") ) diff --git a/internal/projectworkflow/service.go b/internal/projectworkflow/service.go index 1e49a9ee..15482539 100644 --- a/internal/projectworkflow/service.go +++ b/internal/projectworkflow/service.go @@ -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) diff --git a/internal/projectworkflow/service_test.go b/internal/projectworkflow/service_test.go index 6b9ea821..a05c8595 100644 --- a/internal/projectworkflow/service_test.go +++ b/internal/projectworkflow/service_test.go @@ -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) diff --git a/internal/projectworkflow/store.go b/internal/projectworkflow/store.go index a7d970f1..59bc2278 100644 --- a/internal/projectworkflow/store.go +++ b/internal/projectworkflow/store.go @@ -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 diff --git a/internal/projectworkflow/types.go b/internal/projectworkflow/types.go index 0216601a..7cbc904b 100644 --- a/internal/projectworkflow/types.go +++ b/internal/projectworkflow/types.go @@ -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"`