weichenandCursor
033462a4e6
管理台不再把开关值当成能力已接入。lineage 读取必须落在 URL 项目内，并按两端文档可见性裁剪，跨项目或不可见文档只返回空列表。 Co-authored-by: Cursor <cursoragent@cursor.com>
1008 lines
32 KiB
Go
1008 lines
32 KiB
Go
package projectworkflow
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// Store 数据存储层接口
|
|
type Store interface {
|
|
CreateProject(ctx context.Context, p *Project) (*Project, error)
|
|
GetProjectByID(ctx context.Context, id int64) (*Project, error)
|
|
GetProjectByCode(ctx context.Context, code string) (*Project, error)
|
|
ListAllProjects(ctx context.Context) ([]*Project, error)
|
|
ListProjectsForUser(ctx context.Context, userID int64) ([]*Project, error)
|
|
ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error)
|
|
EnsureProjectWorkflows(ctx context.Context, projectID int64, defaultKeys []string) error
|
|
UpdateActiveWorkflow(ctx context.Context, projectID int64, toWorkflowKey string, expectedVersion int, changedBy int64, reason string) (*Project, *WorkflowActivationAudit, error)
|
|
|
|
CreateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error)
|
|
UpdateWorkflowRun(ctx context.Context, runID string, status string, outputDocID string, summary string, errMsg string) error
|
|
GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error)
|
|
ListWorkflowRuns(ctx context.Context, projectID int64, workflowKey string, limit int) ([]*WorkflowRun, error)
|
|
|
|
RecordLineage(ctx context.Context, lineage *DocumentLineage) error
|
|
GetLineageForDownstream(ctx context.Context, docID string, version int) ([]*DocumentLineage, error)
|
|
// GetLineageForProjectDownstream 只返回属于该项目的边，避免用他项目 doc_id 读出血缘。
|
|
GetLineageForProjectDownstream(ctx context.Context, projectID int64, docID string, version int) ([]*DocumentLineage, error)
|
|
GetLineagesForUpstream(ctx context.Context, upstreamDocID string) ([]*DocumentLineage, error)
|
|
MarkDownstreamStale(ctx context.Context, upstreamDocID string, staleReason string) ([]*DocumentLineage, error)
|
|
}
|
|
|
|
// NewStore 根据 pgxpool 是否可用构建 PGStore 或内存 Fallback Store
|
|
func NewStore(pool *pgxpool.Pool) Store {
|
|
if pool != nil {
|
|
return &pgStore{pool: pool}
|
|
}
|
|
return newMemStore()
|
|
}
|
|
|
|
// =============================================================================
|
|
// PostgreSQL 权威持久化实现
|
|
// =============================================================================
|
|
|
|
type pgStore struct {
|
|
pool *pgxpool.Pool
|
|
}
|
|
|
|
func (s *pgStore) CreateProject(ctx context.Context, p *Project) (*Project, error) {
|
|
now := time.Now()
|
|
p.CreatedAt = now
|
|
p.UpdatedAt = now
|
|
if p.Status == "" {
|
|
p.Status = "active"
|
|
}
|
|
if p.Environment == "" {
|
|
p.Environment = "dev"
|
|
}
|
|
if p.CurrentStage == "" {
|
|
p.CurrentStage = "V0.1_triage"
|
|
}
|
|
if p.LifecycleStatus == "" {
|
|
p.LifecycleStatus = "planning"
|
|
}
|
|
if p.WorkflowSelectionVersion <= 0 {
|
|
p.WorkflowSelectionVersion = 1
|
|
}
|
|
|
|
query := `
|
|
INSERT INTO projects (
|
|
code, name, customer_name, device_type, intended_use, target_market,
|
|
product_domain, project_type_key, compliance_profile_key, governance_profile,
|
|
team_mode, execution_mode, data_sensitivity, network_policy, budget_class,
|
|
problem_statement, target_users, target_outcome, product_shape,
|
|
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, $17, $18, $19,
|
|
$20, $21, $22, $23, $24,
|
|
$25, $26, $27,
|
|
$28, $29
|
|
)
|
|
RETURNING id
|
|
`
|
|
err := s.pool.QueryRow(
|
|
ctx, query,
|
|
p.Code, p.Name, p.CustomerName, p.DeviceType, p.IntendedUse, p.TargetMarket,
|
|
p.ProductDomain, p.ProjectTypeKey, p.ComplianceProfileKey, p.GovernanceProfile,
|
|
p.TeamMode, p.ExecutionMode, p.DataSensitivity, p.NetworkPolicy, p.BudgetClass,
|
|
p.ProblemStatement, p.TargetUsers, p.TargetOutcome, p.ProductShape,
|
|
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)
|
|
}
|
|
|
|
return p, nil
|
|
}
|
|
|
|
func (s *pgStore) GetProjectByID(ctx context.Context, id int64) (*Project, error) {
|
|
query := `
|
|
SELECT id, code, name, customer_name, device_type, intended_use, target_market,
|
|
COALESCE(product_domain, ''), COALESCE(project_type_key, ''), COALESCE(compliance_profile_key, ''),
|
|
COALESCE(governance_profile, ''), COALESCE(team_mode, ''), COALESCE(execution_mode, ''),
|
|
COALESCE(data_sensitivity, ''), COALESCE(network_policy, ''), COALESCE(budget_class, ''),
|
|
COALESCE(problem_statement, ''), COALESCE(target_users, ''), COALESCE(target_outcome, ''),
|
|
COALESCE(product_shape, ''),
|
|
current_stage, environment, status, COALESCE(creator_user_id, 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
|
|
`
|
|
p := &Project{}
|
|
err := s.pool.QueryRow(ctx, query, id).Scan(
|
|
&p.ID, &p.Code, &p.Name, &p.CustomerName, &p.DeviceType, &p.IntendedUse, &p.TargetMarket,
|
|
&p.ProductDomain, &p.ProjectTypeKey, &p.ComplianceProfileKey,
|
|
&p.GovernanceProfile, &p.TeamMode, &p.ExecutionMode,
|
|
&p.DataSensitivity, &p.NetworkPolicy, &p.BudgetClass,
|
|
&p.ProblemStatement, &p.TargetUsers, &p.TargetOutcome, &p.ProductShape,
|
|
&p.CurrentStage, &p.Environment, &p.Status, &p.CreatorUserID,
|
|
&p.LifecycleStatus, &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, &p.WorkflowPlanRevision,
|
|
&p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt,
|
|
)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, ErrProjectNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query project failed: %w", err)
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
func (s *pgStore) GetProjectByCode(ctx context.Context, code string) (*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(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
|
|
`
|
|
p := &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.LifecycleStatus, &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, &p.WorkflowPlanRevision,
|
|
&p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt,
|
|
)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, ErrProjectNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query project by code failed: %w", err)
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
func (s *pgStore) ListProjectsForUser(ctx context.Context, userID int64) ([]*Project, error) {
|
|
// 仅创建人或 active 项目成员可见（移除 creator_id=1/NULL 公开规则）
|
|
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, 0),
|
|
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 project_members pm ON pm.project_id = p.id AND pm.user_id = $1 AND pm.status = 'active'
|
|
WHERE (p.creator_user_id = $1 OR pm.user_id IS NOT NULL)
|
|
AND COALESCE(p.status, 'active') <> 'demo_retired'
|
|
ORDER BY p.id DESC
|
|
`
|
|
rows, err := s.pool.Query(ctx, query, userID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list projects failed: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var list []*Project
|
|
for rows.Next() {
|
|
p := &Project{}
|
|
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.LifecycleStatus, &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, &p.WorkflowPlanRevision,
|
|
&p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, p)
|
|
}
|
|
return list, rows.Err()
|
|
}
|
|
|
|
func (s *pgStore) ListAllProjects(ctx context.Context) ([]*Project, error) {
|
|
query := `
|
|
SELECT 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, 0),
|
|
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
|
|
WHERE COALESCE(p.status, 'active') <> 'demo_retired'
|
|
ORDER BY p.id DESC
|
|
`
|
|
rows, err := s.pool.Query(ctx, query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list all projects failed: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var list []*Project
|
|
for rows.Next() {
|
|
p := &Project{}
|
|
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.LifecycleStatus, &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, &p.WorkflowPlanRevision,
|
|
&p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, p)
|
|
}
|
|
return list, rows.Err()
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func (s *pgStore) CreateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error) {
|
|
manifestJSON, _ := json.Marshal(run.InputManifest)
|
|
query := `
|
|
INSERT INTO workflow_runs (
|
|
run_id, project_id, workflow_key, status, started_by, stage_key, task_key,
|
|
input_manifest, output_doc_id, output_summary, error_message, started_at, created_at
|
|
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now())
|
|
RETURNING id, created_at
|
|
`
|
|
now := time.Now()
|
|
run.StartedAt = &now
|
|
err := s.pool.QueryRow(
|
|
ctx, query,
|
|
run.RunID, run.ProjectID, run.WorkflowKey, run.Status, run.StartedBy,
|
|
run.StageKey, run.TaskKey, manifestJSON, run.OutputDocID, run.OutputSummary,
|
|
run.ErrorMessage, run.StartedAt,
|
|
).Scan(&run.ID, &run.CreatedAt)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create workflow run failed: %w", err)
|
|
}
|
|
return run, nil
|
|
}
|
|
|
|
func (s *pgStore) UpdateWorkflowRun(ctx context.Context, runID string, status string, outputDocID string, summary string, errMsg string) error {
|
|
now := time.Now()
|
|
query := `
|
|
UPDATE workflow_runs
|
|
SET status = $1,
|
|
output_doc_id = CASE WHEN $2 <> '' THEN $2 ELSE output_doc_id END,
|
|
output_summary = CASE WHEN $3 <> '' THEN $3 ELSE output_summary END,
|
|
error_message = $4,
|
|
finished_at = $5
|
|
WHERE run_id = $6
|
|
`
|
|
_, err := s.pool.Exec(ctx, query, status, outputDocID, summary, errMsg, now, runID)
|
|
return err
|
|
}
|
|
|
|
func (s *pgStore) GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error) {
|
|
query := `
|
|
SELECT id, run_id, project_id, workflow_key, status, started_by, stage_key, task_key,
|
|
input_manifest, output_doc_id, output_summary, error_message, started_at, finished_at, created_at
|
|
FROM workflow_runs
|
|
WHERE run_id = $1
|
|
`
|
|
run := &WorkflowRun{}
|
|
var manifestRaw []byte
|
|
err := s.pool.QueryRow(ctx, query, runID).Scan(
|
|
&run.ID, &run.RunID, &run.ProjectID, &run.WorkflowKey, &run.Status, &run.StartedBy,
|
|
&run.StageKey, &run.TaskKey, &manifestRaw, &run.OutputDocID, &run.OutputSummary,
|
|
&run.ErrorMessage, &run.StartedAt, &run.FinishedAt, &run.CreatedAt,
|
|
)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, errors.New("workflow run not found")
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(manifestRaw) > 0 {
|
|
_ = json.Unmarshal(manifestRaw, &run.InputManifest)
|
|
}
|
|
return run, nil
|
|
}
|
|
|
|
func (s *pgStore) ListWorkflowRuns(ctx context.Context, projectID int64, workflowKey string, limit int) ([]*WorkflowRun, error) {
|
|
if limit <= 0 {
|
|
limit = 20
|
|
}
|
|
query := `
|
|
SELECT id, run_id, project_id, workflow_key, status, started_by, stage_key, task_key,
|
|
output_doc_id, output_summary, error_message, started_at, finished_at, created_at
|
|
FROM workflow_runs
|
|
WHERE project_id = $1 AND ($2 = '' OR workflow_key = $2)
|
|
ORDER BY id DESC
|
|
LIMIT $3
|
|
`
|
|
rows, err := s.pool.Query(ctx, query, projectID, workflowKey, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var list []*WorkflowRun
|
|
for rows.Next() {
|
|
run := &WorkflowRun{}
|
|
if err := rows.Scan(
|
|
&run.ID, &run.RunID, &run.ProjectID, &run.WorkflowKey, &run.Status, &run.StartedBy,
|
|
&run.StageKey, &run.TaskKey, &run.OutputDocID, &run.OutputSummary,
|
|
&run.ErrorMessage, &run.StartedAt, &run.FinishedAt, &run.CreatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, run)
|
|
}
|
|
return list, rows.Err()
|
|
}
|
|
|
|
func (s *pgStore) RecordLineage(ctx context.Context, lineage *DocumentLineage) error {
|
|
query := `
|
|
INSERT INTO document_lineage (
|
|
project_id, downstream_doc_id, downstream_version, upstream_doc_id, upstream_version,
|
|
upstream_sha256, is_stale, stale_reason, created_at
|
|
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now())
|
|
ON CONFLICT (downstream_doc_id, downstream_version, upstream_doc_id, upstream_version)
|
|
DO UPDATE SET upstream_sha256 = EXCLUDED.upstream_sha256,
|
|
is_stale = EXCLUDED.is_stale,
|
|
stale_reason = EXCLUDED.stale_reason
|
|
`
|
|
_, err := s.pool.Exec(
|
|
ctx, query,
|
|
lineage.ProjectID, lineage.DownstreamDocID, lineage.DownstreamVersion,
|
|
lineage.UpstreamDocID, lineage.UpstreamVersion, lineage.UpstreamSHA256,
|
|
lineage.IsStale, lineage.StaleReason,
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *pgStore) GetLineageForDownstream(ctx context.Context, docID string, version int) ([]*DocumentLineage, error) {
|
|
query := `
|
|
SELECT id, project_id, downstream_doc_id, downstream_version, upstream_doc_id,
|
|
upstream_version, upstream_sha256, is_stale, stale_reason, stale_detected_at, created_at
|
|
FROM document_lineage
|
|
WHERE downstream_doc_id = $1 AND ($2 <= 0 OR downstream_version = $2)
|
|
ORDER BY id DESC
|
|
`
|
|
rows, err := s.pool.Query(ctx, query, docID, version)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var list []*DocumentLineage
|
|
for rows.Next() {
|
|
l := &DocumentLineage{}
|
|
if err := rows.Scan(
|
|
&l.ID, &l.ProjectID, &l.DownstreamDocID, &l.DownstreamVersion, &l.UpstreamDocID,
|
|
&l.UpstreamVersion, &l.UpstreamSHA256, &l.IsStale, &l.StaleReason, &l.StaleDetectedAt, &l.CreatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, l)
|
|
}
|
|
return list, rows.Err()
|
|
}
|
|
|
|
func (s *pgStore) GetLineageForProjectDownstream(ctx context.Context, projectID int64, docID string, version int) ([]*DocumentLineage, error) {
|
|
query := `
|
|
SELECT id, project_id, downstream_doc_id, downstream_version, upstream_doc_id,
|
|
upstream_version, upstream_sha256, is_stale, stale_reason, stale_detected_at, created_at
|
|
FROM document_lineage
|
|
WHERE project_id = $1 AND downstream_doc_id = $2 AND ($3 <= 0 OR downstream_version = $3)
|
|
ORDER BY id DESC
|
|
`
|
|
rows, err := s.pool.Query(ctx, query, projectID, docID, version)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var list []*DocumentLineage
|
|
for rows.Next() {
|
|
l := &DocumentLineage{}
|
|
if err := rows.Scan(
|
|
&l.ID, &l.ProjectID, &l.DownstreamDocID, &l.DownstreamVersion, &l.UpstreamDocID,
|
|
&l.UpstreamVersion, &l.UpstreamSHA256, &l.IsStale, &l.StaleReason, &l.StaleDetectedAt, &l.CreatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, l)
|
|
}
|
|
return list, rows.Err()
|
|
}
|
|
|
|
func (s *pgStore) GetLineagesForUpstream(ctx context.Context, upstreamDocID string) ([]*DocumentLineage, error) {
|
|
query := `
|
|
SELECT id, project_id, downstream_doc_id, downstream_version, upstream_doc_id,
|
|
upstream_version, upstream_sha256, is_stale, stale_reason, stale_detected_at, created_at
|
|
FROM document_lineage
|
|
WHERE upstream_doc_id = $1
|
|
ORDER BY id DESC
|
|
`
|
|
rows, err := s.pool.Query(ctx, query, upstreamDocID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var list []*DocumentLineage
|
|
for rows.Next() {
|
|
l := &DocumentLineage{}
|
|
if err := rows.Scan(
|
|
&l.ID, &l.ProjectID, &l.DownstreamDocID, &l.DownstreamVersion, &l.UpstreamDocID,
|
|
&l.UpstreamVersion, &l.UpstreamSHA256, &l.IsStale, &l.StaleReason, &l.StaleDetectedAt, &l.CreatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, l)
|
|
}
|
|
return list, rows.Err()
|
|
}
|
|
|
|
func (s *pgStore) MarkDownstreamStale(ctx context.Context, upstreamDocID string, staleReason string) ([]*DocumentLineage, error) {
|
|
now := time.Now()
|
|
query := `
|
|
UPDATE document_lineage
|
|
SET is_stale = true,
|
|
stale_reason = $1,
|
|
stale_detected_at = $2
|
|
WHERE upstream_doc_id = $3 AND is_stale = false
|
|
RETURNING id, project_id, downstream_doc_id, downstream_version, upstream_doc_id, upstream_version, upstream_sha256, is_stale, stale_reason, stale_detected_at, created_at
|
|
`
|
|
rows, err := s.pool.Query(ctx, query, staleReason, now, upstreamDocID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var updated []*DocumentLineage
|
|
for rows.Next() {
|
|
l := &DocumentLineage{}
|
|
if err := rows.Scan(
|
|
&l.ID, &l.ProjectID, &l.DownstreamDocID, &l.DownstreamVersion, &l.UpstreamDocID,
|
|
&l.UpstreamVersion, &l.UpstreamSHA256, &l.IsStale, &l.StaleReason, &l.StaleDetectedAt, &l.CreatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
updated = append(updated, l)
|
|
}
|
|
return updated, rows.Err()
|
|
}
|
|
|
|
// =============================================================================
|
|
// 纯内存 Fallback Store 实现（用于极速单元测试及无 Postgres 运行态）
|
|
// =============================================================================
|
|
|
|
type memStore struct {
|
|
mu sync.RWMutex
|
|
projects map[int64]*Project
|
|
workflows map[int64]map[string]*ProjectWorkflow
|
|
runs map[string]*WorkflowRun
|
|
audits []*WorkflowActivationAudit
|
|
lineages []*DocumentLineage
|
|
nextProjID int64
|
|
nextWFID int64
|
|
nextRunID int64
|
|
nextLineageID int64
|
|
}
|
|
|
|
func newMemStore() *memStore {
|
|
m := &memStore{
|
|
projects: make(map[int64]*Project),
|
|
workflows: make(map[int64]map[string]*ProjectWorkflow),
|
|
runs: make(map[string]*WorkflowRun),
|
|
nextProjID: 1000,
|
|
nextWFID: 1,
|
|
nextRunID: 1,
|
|
nextLineageID: 1,
|
|
}
|
|
now := time.Now()
|
|
m.projects[1] = &Project{
|
|
ID: 1,
|
|
Code: "PROJ-001",
|
|
Name: "医用超声高频能量手术集成系统",
|
|
CustomerName: "北医三院创新转化中心",
|
|
DeviceType: "III类医疗器械",
|
|
IntendedUse: "用于外科手术中软组织的切割与血管止血",
|
|
TargetMarket: "中国 NMPA + 欧盟 CE MDR",
|
|
CurrentStage: "stage_1_spec",
|
|
Environment: "dev",
|
|
Status: "active",
|
|
CreatorUserID: 1,
|
|
ActiveWorkflowKey: "main",
|
|
WorkflowSelectionVersion: 1,
|
|
CreatedAt: now,
|
|
UpdatedAt: now,
|
|
}
|
|
m.projects[5201] = &Project{
|
|
ID: 5201,
|
|
Code: "CUSA-NEUROPULSE",
|
|
Name: "神经外科双频超声吸引系统 (CUSA-NeuroPulse)",
|
|
CustomerName: "天坛医院神经外科研究所",
|
|
DeviceType: "III类医疗器械",
|
|
IntendedUse: "用于神经外科颅内脑肿瘤精准超声粉碎吸引与止血",
|
|
TargetMarket: "中国 NMPA + 欧盟 CE MDR",
|
|
CurrentStage: "stage_1_spec",
|
|
Environment: "dev",
|
|
Status: "active",
|
|
CreatorUserID: 1,
|
|
ActiveWorkflowKey: "main",
|
|
WorkflowSelectionVersion: 1,
|
|
CreatedAt: now,
|
|
UpdatedAt: now,
|
|
}
|
|
return m
|
|
}
|
|
|
|
func (m *memStore) CreateProject(ctx context.Context, p *Project) (*Project, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
|
|
m.nextProjID++
|
|
p.ID = m.nextProjID
|
|
if p.Status == "" {
|
|
p.Status = "active"
|
|
}
|
|
if p.LifecycleStatus == "" {
|
|
p.LifecycleStatus = "planning"
|
|
}
|
|
if p.WorkflowSelectionVersion <= 0 {
|
|
p.WorkflowSelectionVersion = 1
|
|
}
|
|
now := time.Now()
|
|
p.CreatedAt = now
|
|
p.UpdatedAt = now
|
|
|
|
cpy := *p
|
|
m.projects[p.ID] = &cpy
|
|
return p, nil
|
|
}
|
|
|
|
func (m *memStore) GetProjectByID(ctx context.Context, id int64) (*Project, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
p, ok := m.projects[id]
|
|
if !ok {
|
|
return nil, ErrProjectNotFound
|
|
}
|
|
cpy := *p
|
|
return &cpy, nil
|
|
}
|
|
|
|
func (m *memStore) GetProjectByCode(ctx context.Context, code string) (*Project, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
for _, p := range m.projects {
|
|
if p.Code == code {
|
|
cpy := *p
|
|
return &cpy, nil
|
|
}
|
|
}
|
|
return nil, ErrProjectNotFound
|
|
}
|
|
|
|
func (m *memStore) ListProjectsForUser(ctx context.Context, userID int64) ([]*Project, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
var list []*Project
|
|
for _, p := range m.projects {
|
|
if p.CreatorUserID == userID {
|
|
cpy := *p
|
|
list = append(list, &cpy)
|
|
}
|
|
}
|
|
return list, nil
|
|
}
|
|
|
|
func (m *memStore) ListAllProjects(ctx context.Context) ([]*Project, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
list := make([]*Project, 0, len(m.projects))
|
|
for _, p := range m.projects {
|
|
cpy := *p
|
|
list = append(list, &cpy)
|
|
}
|
|
return list, nil
|
|
}
|
|
|
|
func (m *memStore) ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
wfs, ok := m.workflows[projectID]
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
var list []*ProjectWorkflow
|
|
for _, pw := range wfs {
|
|
cpy := *pw
|
|
list = append(list, &cpy)
|
|
}
|
|
return list, nil
|
|
}
|
|
|
|
func (m *memStore) EnsureProjectWorkflows(ctx context.Context, projectID int64, defaultKeys []string) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
if _, ok := m.workflows[projectID]; !ok {
|
|
m.workflows[projectID] = make(map[string]*ProjectWorkflow)
|
|
}
|
|
proj := m.projects[projectID]
|
|
activeKey := ""
|
|
if proj != nil {
|
|
activeKey = proj.ActiveWorkflowKey
|
|
}
|
|
|
|
for _, key := range defaultKeys {
|
|
if _, exists := m.workflows[projectID][key]; !exists {
|
|
m.nextWFID++
|
|
status := "ready"
|
|
if key == activeKey {
|
|
status = "active"
|
|
}
|
|
now := time.Now()
|
|
m.workflows[projectID][key] = &ProjectWorkflow{
|
|
ID: m.nextWFID,
|
|
ProjectID: projectID,
|
|
WorkflowKey: key,
|
|
Status: status,
|
|
Enabled: true,
|
|
CreatedAt: now,
|
|
UpdatedAt: now,
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (m *memStore) UpdateActiveWorkflow(
|
|
ctx context.Context,
|
|
projectID int64,
|
|
toWorkflowKey string,
|
|
expectedVersion int,
|
|
changedBy int64,
|
|
reason string,
|
|
) (*Project, *WorkflowActivationAudit, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
|
|
p, ok := m.projects[projectID]
|
|
if !ok {
|
|
return nil, nil, ErrProjectNotFound
|
|
}
|
|
|
|
// 权限检查 (项目创建人、系统管理员 ID=1、演示项目 5201 或 评审项目 >= 6000 允许切换)
|
|
isReviewOrDemo := projectID == 5201 || (projectID >= 6000 && projectID <= 6020)
|
|
if p.CreatorUserID > 0 && changedBy > 0 && p.CreatorUserID != changedBy && changedBy != 1 && !isReviewOrDemo {
|
|
return nil, nil, ErrForbiddenNotCreator
|
|
}
|
|
|
|
// 乐观锁检查
|
|
if expectedVersion > 0 && expectedVersion != p.WorkflowSelectionVersion {
|
|
return nil, nil, ErrConcurrentConflict
|
|
}
|
|
|
|
// 幂等
|
|
if p.ActiveWorkflowKey == toWorkflowKey {
|
|
cpy := *p
|
|
return &cpy, nil, nil
|
|
}
|
|
|
|
curVersion := p.WorkflowSelectionVersion
|
|
nextVersion := curVersion + 1
|
|
now := time.Now()
|
|
|
|
fromKey := p.ActiveWorkflowKey
|
|
p.ActiveWorkflowKey = toWorkflowKey
|
|
p.WorkflowSelectionVersion = nextVersion
|
|
p.ActiveWorkflowChangedAt = &now
|
|
p.ActiveWorkflowChangedBy = &changedBy
|
|
p.UpdatedAt = now
|
|
|
|
// 更新挂载的工作流状态
|
|
if wfs, ok := m.workflows[projectID]; ok {
|
|
for k, wf := range wfs {
|
|
if k == toWorkflowKey {
|
|
wf.Status = "active"
|
|
wf.UpdatedAt = now
|
|
} else if wf.Status == "active" {
|
|
wf.Status = "ready"
|
|
wf.UpdatedAt = now
|
|
}
|
|
}
|
|
}
|
|
|
|
audit := &WorkflowActivationAudit{
|
|
ID: int64(len(m.audits) + 1),
|
|
ProjectID: projectID,
|
|
FromWorkflowKey: fromKey,
|
|
ToWorkflowKey: toWorkflowKey,
|
|
ChangedBy: changedBy,
|
|
VersionBefore: curVersion,
|
|
VersionAfter: nextVersion,
|
|
Reason: reason,
|
|
CreatedAt: now,
|
|
}
|
|
m.audits = append(m.audits, audit)
|
|
|
|
cpy := *p
|
|
return &cpy, audit, nil
|
|
}
|
|
|
|
func (m *memStore) CreateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.nextRunID++
|
|
run.ID = m.nextRunID
|
|
now := time.Now()
|
|
run.CreatedAt = now
|
|
run.StartedAt = &now
|
|
cpy := *run
|
|
m.runs[run.RunID] = &cpy
|
|
return run, nil
|
|
}
|
|
|
|
func (m *memStore) UpdateWorkflowRun(ctx context.Context, runID string, status string, outputDocID string, summary string, errMsg string) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
run, ok := m.runs[runID]
|
|
if !ok {
|
|
return errors.New("workflow run not found")
|
|
}
|
|
now := time.Now()
|
|
run.Status = status
|
|
if outputDocID != "" {
|
|
run.OutputDocID = outputDocID
|
|
}
|
|
if summary != "" {
|
|
run.OutputSummary = summary
|
|
}
|
|
run.ErrorMessage = errMsg
|
|
run.FinishedAt = &now
|
|
return nil
|
|
}
|
|
|
|
func (m *memStore) GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
run, ok := m.runs[runID]
|
|
if !ok {
|
|
return nil, errors.New("workflow run not found")
|
|
}
|
|
cpy := *run
|
|
return &cpy, nil
|
|
}
|
|
|
|
func (m *memStore) ListWorkflowRuns(ctx context.Context, projectID int64, workflowKey string, limit int) ([]*WorkflowRun, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
var list []*WorkflowRun
|
|
for _, r := range m.runs {
|
|
if r.ProjectID == projectID && (workflowKey == "" || r.WorkflowKey == workflowKey) {
|
|
cpy := *r
|
|
list = append(list, &cpy)
|
|
}
|
|
}
|
|
return list, nil
|
|
}
|
|
|
|
func (m *memStore) RecordLineage(ctx context.Context, lineage *DocumentLineage) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.nextLineageID++
|
|
lineage.ID = m.nextLineageID
|
|
lineage.CreatedAt = time.Now()
|
|
cpy := *lineage
|
|
m.lineages = append(m.lineages, &cpy)
|
|
return nil
|
|
}
|
|
|
|
func (m *memStore) GetLineageForDownstream(ctx context.Context, docID string, version int) ([]*DocumentLineage, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
var list []*DocumentLineage
|
|
for _, l := range m.lineages {
|
|
if l.DownstreamDocID == docID && (version <= 0 || l.DownstreamVersion == version) {
|
|
cpy := *l
|
|
list = append(list, &cpy)
|
|
}
|
|
}
|
|
return list, nil
|
|
}
|
|
|
|
func (m *memStore) GetLineageForProjectDownstream(ctx context.Context, projectID int64, docID string, version int) ([]*DocumentLineage, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
var list []*DocumentLineage
|
|
for _, l := range m.lineages {
|
|
if l.ProjectID == projectID && l.DownstreamDocID == docID && (version <= 0 || l.DownstreamVersion == version) {
|
|
cpy := *l
|
|
list = append(list, &cpy)
|
|
}
|
|
}
|
|
return list, nil
|
|
}
|
|
|
|
func (m *memStore) GetLineagesForUpstream(ctx context.Context, upstreamDocID string) ([]*DocumentLineage, error) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
var list []*DocumentLineage
|
|
for _, l := range m.lineages {
|
|
if l.UpstreamDocID == upstreamDocID {
|
|
cpy := *l
|
|
list = append(list, &cpy)
|
|
}
|
|
}
|
|
return list, nil
|
|
}
|
|
|
|
func (m *memStore) MarkDownstreamStale(ctx context.Context, upstreamDocID string, staleReason string) ([]*DocumentLineage, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
now := time.Now()
|
|
var updated []*DocumentLineage
|
|
for _, l := range m.lineages {
|
|
if l.UpstreamDocID == upstreamDocID && !l.IsStale {
|
|
l.IsStale = true
|
|
l.StaleReason = staleReason
|
|
l.StaleDetectedAt = &now
|
|
cpy := *l
|
|
updated = append(updated, &cpy)
|
|
}
|
|
}
|
|
return updated, nil
|
|
}
|