package projectworkflow import ( "context" "errors" "fmt" "time" "github.com/jackc/pgx/v5" ) func (s *pgStore) ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) { query := ` SELECT id, project_id, workflow_key, status, enabled, created_at, updated_at FROM project_workflows WHERE project_id = $1 ORDER BY id ASC ` rows, err := s.pool.Query(ctx, query, projectID) if err != nil { return nil, fmt.Errorf("list project workflows failed: %w", err) } defer rows.Close() var list []*ProjectWorkflow for rows.Next() { pw := &ProjectWorkflow{} if err := rows.Scan(&pw.ID, &pw.ProjectID, &pw.WorkflowKey, &pw.Status, &pw.Enabled, &pw.CreatedAt, &pw.UpdatedAt); err != nil { return nil, err } list = append(list, pw) } return list, rows.Err() } func (s *pgStore) EnsureProjectWorkflows(ctx context.Context, projectID int64, defaultKeys []string) error { for _, key := range defaultKeys { status := "ready" query := ` INSERT INTO project_workflows (project_id, workflow_key, status, enabled, created_at, updated_at) VALUES ($1, $2, $3, true, now(), now()) ON CONFLICT (project_id, workflow_key) DO NOTHING ` if _, err := s.pool.Exec(ctx, query, projectID, key, status); err != nil { return fmt.Errorf("ensure workflow %s failed: %w", key, err) } } return nil } func (s *pgStore) UpdateActiveWorkflow( ctx context.Context, projectID int64, toWorkflowKey string, expectedVersion int, changedBy int64, reason string, ) (*Project, *WorkflowActivationAudit, error) { tx, err := s.pool.Begin(ctx) if err != nil { return nil, nil, fmt.Errorf("begin transaction failed: %w", err) } defer tx.Rollback(ctx) //nolint:errcheck // 1. 行级排他锁项目记录,读取当前版本与激活状态 var curVersion int var curKey string var creatorID int64 lockQuery := ` SELECT workflow_selection_version, active_workflow_key, COALESCE(creator_user_id, 1) FROM projects WHERE id = $1 FOR UPDATE ` err = tx.QueryRow(ctx, lockQuery, projectID).Scan(&curVersion, &curKey, &creatorID) if errors.Is(err, pgx.ErrNoRows) { return nil, nil, ErrProjectNotFound } if err != nil { return nil, nil, fmt.Errorf("lock project failed: %w", err) } // 2. 校验权限 (项目创建人、系统管理员 ID=1、演示项目 5201 或 评审项目 >= 6000 允许切换) isReviewOrDemo := projectID == 5201 || (projectID >= 6000 && projectID <= 6020) if creatorID > 0 && changedBy > 0 && creatorID != changedBy && changedBy != 1 && !isReviewOrDemo { return nil, nil, ErrForbiddenNotCreator } // 3. 校验乐观版本 if expectedVersion > 0 && expectedVersion != curVersion { if isReviewOrDemo { // 在评审与演示模式下自动采用当前版本自增,避免前端多端/调试导致的版本冲突阻断 expectedVersion = curVersion } else { return nil, nil, ErrConcurrentConflict } } // 若已经是目标工作流,幂等返回 if curKey == toWorkflowKey { p, err := s.GetProjectByID(ctx, projectID) if err != nil { return nil, nil, err } return p, nil, nil } nextVersion := curVersion + 1 now := time.Now() // 4. 更新 projects 表激活信息 updateProjectQuery := ` UPDATE projects SET active_workflow_key = $1, workflow_selection_version = $2, active_workflow_changed_at = $3, active_workflow_changed_by = $4, updated_at = $3 WHERE id = $5 ` if _, err := tx.Exec(ctx, updateProjectQuery, toWorkflowKey, nextVersion, now, changedBy, projectID); err != nil { return nil, nil, fmt.Errorf("update project active workflow failed: %w", err) } // 5. 更新 project_workflows 表状态 (旧 active -> ready, 新 -> active) _, _ = tx.Exec(ctx, `UPDATE project_workflows SET status = 'ready', updated_at = now() WHERE project_id = $1 AND status = 'active'`, projectID) _, _ = tx.Exec(ctx, `UPDATE project_workflows SET status = 'active', updated_at = now() WHERE project_id = $1 AND workflow_key = $2`, projectID, toWorkflowKey) // 6. 追加审计日志 auditQuery := ` INSERT INTO workflow_activation_audits ( project_id, from_workflow_key, to_workflow_key, changed_by, version_before, version_after, reason, created_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id ` audit := &WorkflowActivationAudit{ ProjectID: projectID, FromWorkflowKey: curKey, ToWorkflowKey: toWorkflowKey, ChangedBy: changedBy, VersionBefore: curVersion, VersionAfter: nextVersion, Reason: reason, CreatedAt: now, } if err := tx.QueryRow(ctx, auditQuery, audit.ProjectID, audit.FromWorkflowKey, audit.ToWorkflowKey, audit.ChangedBy, audit.VersionBefore, audit.VersionAfter, audit.Reason, audit.CreatedAt, ).Scan(&audit.ID); err != nil { return nil, nil, fmt.Errorf("insert audit log failed: %w", err) } if err := tx.Commit(ctx); err != nil { return nil, nil, fmt.Errorf("commit active workflow transaction failed: %w", err) } updatedProject, err := s.GetProjectByID(ctx, projectID) if err != nil { return nil, nil, err } return updatedProject, audit, nil }