From edffe7400988508cc19f9c7f16c8176d80e6fa7a Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 12 Sep 2026 02:53:09 +0000 Subject: [PATCH] feat(db): add workflow definitions, versions, and template asset tables Introduce Phase 2 IAM-adjacent schema for versioned workflow definitions, template assets/versions, and workflow_defs permissions seed. Co-authored-by: Cursor --- ...-12-workflow-definitions-and-templates.sql | 99 ++++ internal/bizdev/service.go | 9 +- internal/workflowdef/bootstrap.go | 139 +++++ internal/workflowdef/catalog.go | 142 +++++ internal/workflowdef/errors.go | 14 + internal/workflowdef/publish_test.go | 72 +++ internal/workflowdef/service.go | 159 ++++++ internal/workflowdef/store.go | 527 ++++++++++++++++++ internal/workflowdef/types.go | 122 ++++ internal/workflowdef/validator.go | 88 +++ internal/workflowdef/validator_test.go | 47 ++ 11 files changed, 1417 insertions(+), 1 deletion(-) create mode 100644 deploy/postgres/goose/00016_2026-09-12-workflow-definitions-and-templates.sql create mode 100644 internal/workflowdef/bootstrap.go create mode 100644 internal/workflowdef/catalog.go create mode 100644 internal/workflowdef/errors.go create mode 100644 internal/workflowdef/publish_test.go create mode 100644 internal/workflowdef/service.go create mode 100644 internal/workflowdef/store.go create mode 100644 internal/workflowdef/types.go create mode 100644 internal/workflowdef/validator.go create mode 100644 internal/workflowdef/validator_test.go diff --git a/deploy/postgres/goose/00016_2026-09-12-workflow-definitions-and-templates.sql b/deploy/postgres/goose/00016_2026-09-12-workflow-definitions-and-templates.sql new file mode 100644 index 00000000..674ff87b --- /dev/null +++ b/deploy/postgres/goose/00016_2026-09-12-workflow-definitions-and-templates.sql @@ -0,0 +1,99 @@ +-- +goose Up +-- Phase 2: 工作流定义版本化 + 模板资产库 + +CREATE TABLE IF NOT EXISTS workflow_definitions ( + id BIGSERIAL PRIMARY KEY, + def_key TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'active', + latest_published_version INT NOT NULL DEFAULT 0, + product_domain TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS workflow_versions ( + id BIGSERIAL PRIMARY KEY, + workflow_id BIGINT NOT NULL REFERENCES workflow_definitions(id) ON DELETE CASCADE, + version INT NOT NULL, + status TEXT NOT NULL DEFAULT 'draft', + schema_version INT NOT NULL DEFAULT 1, + definition_json JSONB NOT NULL DEFAULT '{}'::jsonb, + definition_hash TEXT NOT NULL DEFAULT '', + published_by BIGINT REFERENCES users(id) ON DELETE SET NULL, + published_at TIMESTAMPTZ, + created_by BIGINT REFERENCES users(id) ON DELETE SET NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uq_workflow_version UNIQUE (workflow_id, version), + CONSTRAINT chk_workflow_version_status CHECK (status IN ('draft', 'published', 'retired')) +); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_workflow_one_draft + ON workflow_versions(workflow_id) + WHERE status = 'draft'; + +CREATE INDEX IF NOT EXISTS idx_workflow_versions_status ON workflow_versions(status); +CREATE INDEX IF NOT EXISTS idx_workflow_versions_workflow ON workflow_versions(workflow_id, version DESC); + +CREATE TABLE IF NOT EXISTS template_assets ( + id BIGSERIAL PRIMARY KEY, + asset_key TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS template_versions ( + id BIGSERIAL PRIMARY KEY, + asset_id BIGINT NOT NULL REFERENCES template_assets(id) ON DELETE CASCADE, + version INT NOT NULL, + blob_sha256 TEXT NOT NULL, + mime_type TEXT NOT NULL DEFAULT 'application/octet-stream', + size_bytes BIGINT NOT NULL DEFAULT 0, + scan_status TEXT NOT NULL DEFAULT 'pending', + variable_schema JSONB NOT NULL DEFAULT '[]'::jsonb, + created_by BIGINT REFERENCES users(id) ON DELETE SET NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uq_template_asset_version UNIQUE (asset_id, version), + CONSTRAINT chk_template_scan_status CHECK (scan_status IN ('pending', 'ready', 'rejected')) +); + +CREATE INDEX IF NOT EXISTS idx_template_versions_asset ON template_versions(asset_id, version DESC); +CREATE INDEX IF NOT EXISTS idx_template_versions_sha ON template_versions(blob_sha256); + +INSERT INTO system_permissions (perm_key, resource_type, action, risk_level) VALUES + ('workflow_defs.view', 'workflow', 'view', 'normal'), + ('workflow_defs.create', 'workflow', 'create', 'elevated'), + ('workflow_defs.edit_draft', 'workflow', 'edit_draft', 'elevated'), + ('workflow_defs.publish', 'workflow', 'publish', 'critical'), + ('workflow_defs.retire', 'workflow', 'retire', 'elevated'), + ('templates.upload', 'template', 'upload', 'elevated'), + ('templates.download', 'template', 'download', 'normal'), + ('templates.publish', 'template', 'publish', 'elevated') +ON CONFLICT (perm_key) DO NOTHING; + +INSERT INTO system_role_permissions (role_id, permission_id, effect) +SELECT 'role-admin', p.id, 'allow' +FROM system_permissions p +WHERE p.perm_key LIKE 'workflow_defs.%' OR p.perm_key LIKE 'templates.%' +ON CONFLICT DO NOTHING; + +-- +goose Down +DROP TABLE IF EXISTS template_versions; +DROP TABLE IF EXISTS template_assets; +DROP INDEX IF EXISTS uq_workflow_one_draft; +DROP INDEX IF EXISTS idx_workflow_versions_status; +DROP INDEX IF EXISTS idx_workflow_versions_workflow; +DROP TABLE IF EXISTS workflow_versions; +DROP TABLE IF EXISTS workflow_definitions; + +DELETE FROM system_role_permissions +WHERE permission_id IN ( + SELECT id FROM system_permissions + WHERE perm_key LIKE 'workflow_defs.%' OR perm_key LIKE 'templates.%' +); +DELETE FROM system_permissions +WHERE perm_key LIKE 'workflow_defs.%' OR perm_key LIKE 'templates.%'; diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index 1bcc7de7..5f19bc89 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -23,6 +23,7 @@ import ( "chat-agent-system/internal/squad" "chat-agent-system/internal/storage" "chat-agent-system/internal/stream" + "chat-agent-system/internal/workflowdef" "github.com/jackc/pgx/v5/pgxpool" ) @@ -88,7 +89,13 @@ func NewService(cfg *config.Config) (*Service, error) { docGov := docgov.NewGovernanceEngine(pool) freeDoc := freedoc.NewService(pool, ctxMgr) pwStore := projectworkflow.NewStore(pool) - pwSvc := projectworkflow.NewService(pwStore, projectworkflow.NewDefaultCatalog()) + fallbackCatalog := projectworkflow.NewDefaultCatalog() + wfStore := workflowdef.NewStore(pool) + if pool != nil && workflowdef.TablesReady(context.Background(), pool) { + _ = workflowdef.BootstrapSeeds(context.Background(), pool, casStore, "templates/docs", 1) + } + catalog := workflowdef.NewCatalog(context.Background(), wfStore, fallbackCatalog) + pwSvc := projectworkflow.NewService(pwStore, catalog) svc := &Service{ cfg: cfg, diff --git a/internal/workflowdef/bootstrap.go b/internal/workflowdef/bootstrap.go new file mode 100644 index 00000000..aa34e495 --- /dev/null +++ b/internal/workflowdef/bootstrap.go @@ -0,0 +1,139 @@ +package workflowdef + +import ( + "context" + "os" + "path/filepath" + "strings" + + "github.com/jackc/pgx/v5/pgxpool" + + "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/storage" +) + +func legacySingleStage(key, name, desc, stageKey, taskKey, docType, docTitle, upstream string) CreateDefinitionInput { + return CreateDefinitionInput{ + Key: key, + Name: name, + Description: desc, + ProductDomain: "medical_device", + Definition: VersionDefinition{ + SchemaVersion: 1, + RequiredUpstreamWorkflow: upstream, + Stages: []StageDef{{ + Key: stageKey, Name: stageKey, Position: 0, + DocTasks: []DocTaskDef{{ + Key: taskKey, Title: docTitle, Required: true, Blocking: true, DefaultEnabled: true, + OutputDocType: docType, OutputDocTitle: docTitle, + TemplateAssetKey: taskKey, TemplateVersion: 1, + }}, + }}, + }, + } +} + +// BootstrapSeeds 幂等导入静态 catalog、首批行业工作流与 templates/docs。 +func BootstrapSeeds(ctx context.Context, pool *pgxpool.Pool, cas *storage.CAS, templatesDir string, publishedBy int64) error { + if pool == nil || !TablesReady(ctx, pool) { + return nil + } + store := NewStore(pool) + svc := NewService(pool, cas) + + if templatesDir == "" { + templatesDir = "templates/docs" + } + entries, err := os.ReadDir(templatesDir) + if err != nil { + return nil + } + for _, ent := range entries { + if ent.IsDir() || !strings.HasSuffix(strings.ToLower(ent.Name()), ".md") { + continue + } + key := strings.TrimSuffix(ent.Name(), filepath.Ext(ent.Name())) + data, readErr := os.ReadFile(filepath.Join(templatesDir, ent.Name())) + if readErr != nil || len(data) == 0 { + continue + } + _, _ = svc.UploadTemplate(ctx, key, ent.Name(), data, ent.Name(), publishedBy) + } + + seeds := []CreateDefinitionInput{ + legacySingleStage("main", "商务与研发综合主线", "V0.1 分诊到 V0.6 阶段评审", "main", "main", "main_proposal", "商务研发综合方案", ""), + legacySingleStage("product_manager", "智能产品经理", "系统需求规格书生成", "system_reqs", "system_reqs", "system_reqs", "系统需求规格书", ""), + legacySingleStage("test_manager", "智能测试经理", "系统验证方案及计划", "system_test_plan", "test_manager", "system_test_plan", "系统验证方案及计划", "product_manager"), + medicalDeviceWorkflow(), + nonMedicalSoftwareWorkflow(), + } + for _, seed := range seeds { + if err := store.UpsertPublishedSeed(ctx, seed, publishedBy); err != nil { + return err + } + } + return nil +} + +func medicalDeviceWorkflow() CreateDefinitionInput { + return CreateDefinitionInput{ + Key: "medical_device_development", + Name: "医疗器械研发", + Description: "商务摸底 → 需求策划 → 设计开发 → 验证确认 → 设计转移", + ProductDomain: "medical_device", + Definition: VersionDefinition{ + SchemaVersion: 1, + Stages: []StageDef{ + stage("biz_intake", "商务摸底", 0, doc("intake", "客户需求采集", "intake", "客户需求采集表")), + stage("req_planning", "需求与法规策划", 1, doc("system_reqs", "系统需求规格书", "system_reqs", "系统需求规格书")), + stage("design_dev", "设计开发", 2, doc("prelim_design", "初步设计", "prelim_design", "初步设计文档")), + stage("verification", "设计验证/确认", 3, doc("test_manager", "系统验证方案", "system_test_plan", "系统验证方案及计划")), + stage("transfer", "设计转移/发布", 4, doc("feasibility_reg", "法规可行性", "feasibility_reg", "法规可行性评估")), + }, + }, + } +} + +func nonMedicalSoftwareWorkflow() CreateDefinitionInput { + return CreateDefinitionInput{ + Key: "non_medical_software", + Name: "非医疗纯软件", + Description: "机会范围 → 需求 → 架构 → 实现 → 测试验收 → 发布复盘", + ProductDomain: "non_medical_software", + Definition: VersionDefinition{ + SchemaVersion: 1, + Stages: []StageDef{ + stage("opportunity", "机会与范围", 0, doc("intake", "范围说明", "scope_doc", "项目范围说明")), + stage("requirements", "产品需求", 1, doc("user_requirements", "用户需求", "user_requirements", "用户需求说明")), + stage("architecture", "架构与计划", 2, doc("feasibility_dev", "技术可行性", "feasibility_dev", "技术可行性分析")), + stage("implementation", "实现", 3, doc("prelim_design", "架构设计", "architecture_design", "架构设计文档")), + stage("testing", "测试与验收", 4, doc("test_manager", "测试计划", "test_plan", "测试计划")), + stage("release", "发布复盘", 5, doc("research", "发布复盘", "release_review", "发布复盘报告")), + }, + }, + } +} + +func stage(key, name string, pos int, tasks ...DocTaskDef) StageDef { + return StageDef{Key: key, Name: name, Position: pos, DocTasks: tasks} +} + +func doc(key, title, docType, docTitle string) DocTaskDef { + return DocTaskDef{ + Key: key, Title: title, Required: true, Blocking: true, DefaultEnabled: true, + TemplateAssetKey: key, TemplateVersion: 1, + OutputDocType: docType, OutputDocTitle: docTitle, + } +} + +// SeedFromMemoryCatalog 将内存 catalog 转为种子（测试用）。 +func SeedFromMemoryCatalog(cat projectworkflow.Catalog) []CreateDefinitionInput { + if cat == nil { + return nil + } + out := make([]CreateDefinitionInput, 0) + for _, d := range cat.List() { + out = append(out, legacySingleStage(d.Key, d.Name, d.Description, d.StageKey, d.TaskKey, d.OutputDocType, d.OutputDocTitle, d.RequiredUpstreamWorkflow)) + } + return out +} diff --git a/internal/workflowdef/catalog.go b/internal/workflowdef/catalog.go new file mode 100644 index 00000000..e3117d35 --- /dev/null +++ b/internal/workflowdef/catalog.go @@ -0,0 +1,142 @@ +package workflowdef + +import ( + "context" + "sort" + "sync" + + "chat-agent-system/internal/projectworkflow" +) + +// DBCatalog 从已发布工作流版本构建 projectworkflow.Catalog（向后兼容扁平 Definition）。 +type DBCatalog struct { + mu sync.RWMutex + store *Store + items map[string]*projectworkflow.Definition + order []string +} + +func NewDBCatalog(ctx context.Context, store *Store) (*DBCatalog, error) { + c := &DBCatalog{store: store, items: make(map[string]*projectworkflow.Definition)} + if err := c.reload(ctx); err != nil { + return nil, err + } + return c, nil +} + +func (c *DBCatalog) reload(ctx context.Context) error { + versions, err := c.store.ListPublishedCatalogEntries(ctx) + if err != nil { + return err + } + items := make(map[string]*projectworkflow.Definition) + order := make([]string, 0, len(versions)) + for _, v := range versions { + def := flattenVersion(v) + if def == nil { + continue + } + items[v.WorkflowKey] = def + order = append(order, v.WorkflowKey) + } + sort.Strings(order) + c.mu.Lock() + c.items = items + c.order = order + c.mu.Unlock() + return nil +} + +func flattenVersion(v Version) *projectworkflow.Definition { + if len(v.Definition.Stages) == 0 { + return nil + } + st := v.Definition.Stages[0] + taskKey, taskTitle, docType := "", "", "" + if len(st.DocTasks) > 0 { + dt := st.DocTasks[0] + taskKey = dt.Key + taskTitle = dt.OutputDocTitle + docType = dt.OutputDocType + if taskTitle == "" { + taskTitle = dt.Title + } + } + if taskKey == "" { + taskKey = st.Key + } + name := v.WorkflowName + if name == "" { + name = v.WorkflowKey + } + return &projectworkflow.Definition{ + Key: v.WorkflowKey, + Name: name, + Description: v.Description, + StageKey: st.Key, + TaskKey: taskKey, + RequiredUpstreamWorkflow: v.Definition.RequiredUpstreamWorkflow, + OutputDocType: docType, + OutputDocTitle: taskTitle, + Enabled: true, + } +} + +func (c *DBCatalog) HasDefinitions() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return len(c.items) > 0 +} + +func (c *DBCatalog) Refresh(ctx context.Context) error { + return c.reload(ctx) +} + +func (c *DBCatalog) Get(workflowKey string) (*projectworkflow.Definition, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + d, ok := c.items[workflowKey] + if !ok { + return nil, false + } + cpy := *d + return &cpy, true +} + +func (c *DBCatalog) List() []*projectworkflow.Definition { + c.mu.RLock() + defer c.mu.RUnlock() + out := make([]*projectworkflow.Definition, 0, len(c.order)) + for _, key := range c.order { + if d, ok := c.items[key]; ok { + cpy := *d + out = append(out, &cpy) + } + } + return out +} + +func (c *DBCatalog) Validate() error { + c.mu.RLock() + defer c.mu.RUnlock() + for _, def := range c.items { + if def.RequiredUpstreamWorkflow != "" { + if _, ok := c.items[def.RequiredUpstreamWorkflow]; !ok { + return projectworkflow.ErrWorkflowNotFound + } + } + } + return nil +} + +// NewCatalog 优先 DB 已发布 catalog，否则回退静态 catalog。 +func NewCatalog(ctx context.Context, store *Store, fallback projectworkflow.Catalog) projectworkflow.Catalog { + if store == nil || !store.enabled() { + return fallback + } + dbCat, err := NewDBCatalog(ctx, store) + if err != nil || !dbCat.HasDefinitions() { + return fallback + } + return dbCat +} diff --git a/internal/workflowdef/errors.go b/internal/workflowdef/errors.go new file mode 100644 index 00000000..9552c34b --- /dev/null +++ b/internal/workflowdef/errors.go @@ -0,0 +1,14 @@ +package workflowdef + +import "errors" + +var ( + ErrNotAvailable = errors.New("workflowdef: database not available") + ErrNotFound = errors.New("workflowdef: not found") + ErrInvalidInput = errors.New("workflowdef: invalid input") + ErrDraftExists = errors.New("workflowdef: draft version already exists") + ErrNotDraft = errors.New("workflowdef: only draft versions are editable") + ErrPublishedImmutable = errors.New("workflowdef: published version is immutable") + ErrValidationFailed = errors.New("workflowdef: validation failed") + ErrTemplateNotReady = errors.New("workflowdef: template version not ready") +) diff --git a/internal/workflowdef/publish_test.go b/internal/workflowdef/publish_test.go new file mode 100644 index 00000000..e3c239f1 --- /dev/null +++ b/internal/workflowdef/publish_test.go @@ -0,0 +1,72 @@ +package workflowdef_test + +import ( + "context" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "chat-agent-system/internal/workflowdef" +) + +func openPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("TEST_DATABASE_URL") + if dsn == "" { + dsn = "postgres://app:dev_secret_db_pass@127.0.0.1:5433/agent_system_dev?sslmode=disable" + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Skip(err) + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + t.Skip(err) + } + if !workflowdef.TablesReady(ctx, pool) { + pool.Close() + t.Skip("workflowdef tables not migrated") + } + return pool +} + +func TestPublishVersion_ImmutableAfterPublish(t *testing.T) { + pool := openPool(t) + if pool == nil { + return + } + defer pool.Close() + ctx := context.Background() + svc := workflowdef.NewService(pool, nil) + key := "test_wf_" + time.Now().Format("150405") + def, ver, err := svc.CreateDefinition(ctx, workflowdef.CreateDefinitionInput{ + Key: key, Name: "测试工作流", + Definition: workflowdef.VersionDefinition{ + SchemaVersion: 1, + Stages: []workflowdef.StageDef{{ + Key: "s1", Position: 0, + DocTasks: []workflowdef.DocTaskDef{{ + Key: "t1", DefaultEnabled: true, OutputDocType: "d1", OutputDocTitle: "D1", + }}, + }}, + }, + }, 1) + if err != nil { + t.Fatalf("create: %v", err) + } + published, err := svc.PublishVersion(ctx, def.ID, ver.Version, 1) + if err != nil { + t.Fatalf("publish: %v", err) + } + if published.Status != workflowdef.StatusPublished { + t.Fatalf("expected published, got %s", published.Status) + } + _, err = svc.UpdateDraft(ctx, def.ID, ver.Version, published.Definition) + if err == nil { + t.Fatal("expected update on published version to fail") + } +} diff --git a/internal/workflowdef/service.go b/internal/workflowdef/service.go new file mode 100644 index 00000000..f4851585 --- /dev/null +++ b/internal/workflowdef/service.go @@ -0,0 +1,159 @@ +package workflowdef + +import ( + "context" + "path/filepath" + "strings" + + "github.com/jackc/pgx/v5/pgxpool" + + "chat-agent-system/internal/storage" +) + +// Service 工作流定义与模板领域服务。 +type Service struct { + store *Store + cas *storage.CAS +} + +func NewService(pool *pgxpool.Pool, cas *storage.CAS) *Service { + if pool == nil { + return &Service{} + } + return &Service{store: NewStore(pool), cas: cas} +} + +func (s *Service) Available() bool { + return s != nil && s.store != nil && s.store.enabled() +} + +func (s *Service) Store() *Store { + return s.store +} + +func (s *Service) ListDefinitions(ctx context.Context) ([]Definition, error) { + return s.store.ListDefinitions(ctx) +} + +func (s *Service) GetDefinition(ctx context.Context, id int64) (*Definition, error) { + return s.store.GetDefinitionByID(ctx, id) +} + +func (s *Service) CreateDefinition(ctx context.Context, in CreateDefinitionInput, createdBy int64) (*Definition, *Version, error) { + res := ValidateDefinition(in.Definition, s.templateReady(ctx), s.upstreamExists(ctx)) + if !res.Valid { + return nil, nil, ErrValidationFailed + } + return s.store.CreateDefinition(ctx, in, createdBy) +} + +func (s *Service) UpdateDraft(ctx context.Context, workflowID int64, version int, def VersionDefinition) (*Version, error) { + res := ValidateDefinition(def, s.templateReady(ctx), s.upstreamExists(ctx)) + if !res.Valid { + return nil, ErrValidationFailed + } + return s.store.UpdateDraftVersion(ctx, workflowID, version, def) +} + +func (s *Service) ValidateVersion(ctx context.Context, workflowID int64, version int) (ValidationResult, error) { + v, err := s.store.GetVersion(ctx, workflowID, version) + if err != nil { + return ValidationResult{}, err + } + return ValidateDefinition(v.Definition, s.templateReady(ctx), s.upstreamExists(ctx)), nil +} + +func (s *Service) PublishVersion(ctx context.Context, workflowID int64, version int, publishedBy int64) (*Version, error) { + res, err := s.ValidateVersion(ctx, workflowID, version) + if err != nil { + return nil, err + } + if !res.Valid { + return nil, ErrValidationFailed + } + return s.store.PublishVersion(ctx, workflowID, version, publishedBy) +} + +func (s *Service) RetireVersion(ctx context.Context, workflowID int64, version int) (*Version, error) { + return s.store.RetireVersion(ctx, workflowID, version) +} + +func (s *Service) ListVersions(ctx context.Context, workflowID int64) ([]Version, error) { + return s.store.ListVersions(ctx, workflowID) +} + +func (s *Service) ListTemplates(ctx context.Context) ([]TemplateAsset, error) { + return s.store.ListTemplateAssets(ctx) +} + +func (s *Service) ListTemplateVersions(ctx context.Context, assetKey string) ([]TemplateVersion, error) { + return s.store.ListTemplateVersions(ctx, assetKey) +} + +var allowedTemplateExt = map[string]string{ + ".md": "text/markdown", + ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", +} + +// UploadTemplate 上传模板文件并创建 template_version（MVP：扩展名白名单后标记 ready）。 +func (s *Service) UploadTemplate(ctx context.Context, assetKey, name string, data []byte, fileName string, createdBy int64) (*TemplateVersion, error) { + if !s.Available() || s.cas == nil { + return nil, ErrNotAvailable + } + ext := strings.ToLower(filepath.Ext(fileName)) + mime, ok := allowedTemplateExt[ext] + if !ok { + return nil, ErrInvalidInput + } + if len(data) == 0 || len(data) > 32*1024*1024 { + return nil, ErrInvalidInput + } + + ref, err := s.cas.StoreFile(ctx, 0, createdBy, fileName, "template", data, mime) + if err != nil { + return nil, err + } + + assetID, err := s.store.UpsertTemplateAsset(ctx, assetKey, name) + if err != nil { + return nil, err + } + ver, err := s.store.NextTemplateVersion(ctx, assetID) + if err != nil { + return nil, err + } + scanStatus := ScanReady + if ext == ".docx" || ext == ".xlsx" || ext == ".pptx" { + scanStatus = ScanPending + } + tv, err := s.store.CreateTemplateVersion(ctx, assetID, ver, ref.BlobSHA256, mime, int64(len(data)), scanStatus, createdBy) + if err != nil { + return nil, err + } + tv.AssetKey = assetKey + return tv, nil +} + +func (s *Service) templateReady(ctx context.Context) templateReadyChecker { + return func(assetKey string, version int) bool { + if assetKey == "" { + return true + } + if version <= 0 { + version = 1 + } + return s.store.IsTemplateReady(ctx, assetKey, version) + } +} + +func (s *Service) upstreamExists(ctx context.Context) upstreamExistsChecker { + return func(workflowKey string) bool { + if workflowKey == "" { + return true + } + d, err := s.store.GetDefinitionByKey(ctx, workflowKey) + return err == nil && d != nil + } +} diff --git a/internal/workflowdef/store.go b/internal/workflowdef/store.go new file mode 100644 index 00000000..22adf693 --- /dev/null +++ b/internal/workflowdef/store.go @@ -0,0 +1,527 @@ +package workflowdef + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// TablesReady 检查 Phase 2 表是否已迁移。 +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 = 'workflow_definitions' + ) + `).Scan(&ok) + return err == nil && ok +} + +// Store PostgreSQL 工作流定义与模板存储。 +type Store struct { + pool *pgxpool.Pool +} + +func NewStore(pool *pgxpool.Pool) *Store { + return &Store{pool: pool} +} + +func (s *Store) enabled() bool { + return s != nil && s.pool != nil +} + +func hashDefinition(def VersionDefinition) (string, error) { + raw, err := json.Marshal(def) + if err != nil { + return "", err + } + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]), nil +} + +func scanVersion(row pgx.Row) (*Version, error) { + var v Version + var defRaw []byte + var pubBy, createdBy *int64 + err := row.Scan( + &v.ID, &v.WorkflowID, &v.Version, &v.Status, &v.SchemaVersion, + &defRaw, &v.DefinitionHash, &pubBy, &v.PublishedAt, &createdBy, + &v.CreatedAt, &v.UpdatedAt, + ) + if err != nil { + return nil, err + } + v.PublishedBy = pubBy + v.CreatedBy = createdBy + if len(defRaw) > 0 { + _ = json.Unmarshal(defRaw, &v.Definition) + } + return &v, nil +} + +// ListDefinitions 列出全部工作流定义。 +func (s *Store) ListDefinitions(ctx context.Context) ([]Definition, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + rows, err := s.pool.Query(ctx, ` + SELECT id, def_key, name, description, status, latest_published_version, product_domain, created_at, updated_at + FROM workflow_definitions ORDER BY def_key ASC + `) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]Definition, 0) + for rows.Next() { + var d Definition + if err := rows.Scan(&d.ID, &d.Key, &d.Name, &d.Description, &d.Status, &d.LatestPublishedVersion, &d.ProductDomain, &d.CreatedAt, &d.UpdatedAt); err != nil { + return nil, err + } + out = append(out, d) + } + return out, rows.Err() +} + +// GetDefinitionByKey 按 key 查询定义。 +func (s *Store) GetDefinitionByKey(ctx context.Context, key string) (*Definition, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + var d Definition + err := s.pool.QueryRow(ctx, ` + SELECT id, def_key, name, description, status, latest_published_version, product_domain, created_at, updated_at + FROM workflow_definitions WHERE def_key = $1 + `, key).Scan(&d.ID, &d.Key, &d.Name, &d.Description, &d.Status, &d.LatestPublishedVersion, &d.ProductDomain, &d.CreatedAt, &d.UpdatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + return &d, nil +} + +// GetDefinitionByID 按 ID 查询定义。 +func (s *Store) GetDefinitionByID(ctx context.Context, id int64) (*Definition, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + var d Definition + err := s.pool.QueryRow(ctx, ` + SELECT id, def_key, name, description, status, latest_published_version, product_domain, created_at, updated_at + FROM workflow_definitions WHERE id = $1 + `, id).Scan(&d.ID, &d.Key, &d.Name, &d.Description, &d.Status, &d.LatestPublishedVersion, &d.ProductDomain, &d.CreatedAt, &d.UpdatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + return &d, nil +} + +// CreateDefinition 创建定义及首个草稿版本。 +func (s *Store) CreateDefinition(ctx context.Context, in CreateDefinitionInput, createdBy int64) (*Definition, *Version, error) { + if !s.enabled() { + return nil, nil, ErrNotAvailable + } + if in.Key == "" || in.Name == "" { + return nil, nil, ErrInvalidInput + } + defHash, err := hashDefinition(in.Definition) + if err != nil { + return nil, nil, err + } + defRaw, _ := json.Marshal(in.Definition) + + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, nil, err + } + defer tx.Rollback(ctx) + + var defID int64 + err = tx.QueryRow(ctx, ` + INSERT INTO workflow_definitions (def_key, name, description, product_domain) + VALUES ($1, $2, $3, $4) RETURNING id + `, in.Key, in.Name, in.Description, in.ProductDomain).Scan(&defID) + if err != nil { + return nil, nil, fmt.Errorf("insert definition: %w", err) + } + + var ver Version + err = tx.QueryRow(ctx, ` + INSERT INTO workflow_versions (workflow_id, version, status, schema_version, definition_json, definition_hash, created_by) + VALUES ($1, 1, 'draft', $2, $3, $4, $5) + RETURNING id, workflow_id, version, status, schema_version, definition_json, definition_hash, + published_by, published_at, created_by, created_at, updated_at + `, defID, in.Definition.SchemaVersion, defRaw, defHash, createdBy).Scan( + &ver.ID, &ver.WorkflowID, &ver.Version, &ver.Status, &ver.SchemaVersion, + &defRaw, &ver.DefinitionHash, &ver.PublishedBy, &ver.PublishedAt, &ver.CreatedBy, + &ver.CreatedAt, &ver.UpdatedAt, + ) + if err != nil { + return nil, nil, err + } + _ = json.Unmarshal(defRaw, &ver.Definition) + + if err := tx.Commit(ctx); err != nil { + return nil, nil, err + } + d, err := s.GetDefinitionByID(ctx, defID) + return d, &ver, err +} + +// GetVersion 获取指定版本。 +func (s *Store) GetVersion(ctx context.Context, workflowID int64, version int) (*Version, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + row := s.pool.QueryRow(ctx, ` + SELECT id, workflow_id, version, status, schema_version, definition_json, definition_hash, + published_by, published_at, created_by, created_at, updated_at + FROM workflow_versions WHERE workflow_id = $1 AND version = $2 + `, workflowID, version) + v, err := scanVersion(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return v, err +} + +// ListVersions 列出工作流全部版本。 +func (s *Store) ListVersions(ctx context.Context, workflowID int64) ([]Version, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + rows, err := s.pool.Query(ctx, ` + SELECT id, workflow_id, version, status, schema_version, definition_json, definition_hash, + published_by, published_at, created_by, created_at, updated_at + FROM workflow_versions WHERE workflow_id = $1 ORDER BY version DESC + `, workflowID) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]Version, 0) + for rows.Next() { + v, err := scanVersion(rows) + if err != nil { + return nil, err + } + out = append(out, *v) + } + return out, rows.Err() +} + +// UpdateDraftVersion 更新草稿版本定义。 +func (s *Store) UpdateDraftVersion(ctx context.Context, workflowID int64, version int, def VersionDefinition) (*Version, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + existing, err := s.GetVersion(ctx, workflowID, version) + if err != nil { + return nil, err + } + if existing.Status != StatusDraft { + return nil, ErrNotDraft + } + defHash, err := hashDefinition(def) + if err != nil { + return nil, err + } + defRaw, _ := json.Marshal(def) + row := s.pool.QueryRow(ctx, ` + UPDATE workflow_versions SET definition_json = $3, definition_hash = $4, schema_version = $5, updated_at = now() + WHERE workflow_id = $1 AND version = $2 AND status = 'draft' + RETURNING id, workflow_id, version, status, schema_version, definition_json, definition_hash, + published_by, published_at, created_by, created_at, updated_at + `, workflowID, version, defRaw, defHash, def.SchemaVersion) + return scanVersion(row) +} + +// PublishVersion 发布草稿版本（不可变）。 +func (s *Store) PublishVersion(ctx context.Context, workflowID int64, version int, publishedBy int64) (*Version, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + var status string + if err := tx.QueryRow(ctx, `SELECT status FROM workflow_versions WHERE workflow_id = $1 AND version = $2`, workflowID, version).Scan(&status); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return nil, err + } + if status != StatusDraft { + return nil, ErrNotDraft + } + + row := tx.QueryRow(ctx, ` + UPDATE workflow_versions SET status = 'published', published_by = $3, published_at = now(), updated_at = now() + WHERE workflow_id = $1 AND version = $2 AND status = 'draft' + RETURNING id, workflow_id, version, status, schema_version, definition_json, definition_hash, + published_by, published_at, created_by, created_at, updated_at + `, workflowID, version, publishedBy) + v, err := scanVersion(row) + if err != nil { + return nil, err + } + + _, err = tx.Exec(ctx, ` + UPDATE workflow_definitions SET latest_published_version = $2, updated_at = now() WHERE id = $1 + `, workflowID, version) + if err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return v, nil +} + +// RetireVersion 退役已发布版本。 +func (s *Store) RetireVersion(ctx context.Context, workflowID int64, version int) (*Version, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + row := s.pool.QueryRow(ctx, ` + UPDATE workflow_versions SET status = 'retired', updated_at = now() + WHERE workflow_id = $1 AND version = $2 AND status = 'published' + RETURNING id, workflow_id, version, status, schema_version, definition_json, definition_hash, + published_by, published_at, created_by, created_at, updated_at + `, workflowID, version) + v, err := scanVersion(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + return v, err +} + +// ListPublishedCatalogEntries 返回全部 published 版本（catalog 用）。 +func (s *Store) ListPublishedCatalogEntries(ctx context.Context) ([]Version, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + rows, err := s.pool.Query(ctx, ` + SELECT v.id, v.workflow_id, v.version, v.status, v.schema_version, v.definition_json, v.definition_hash, + v.published_by, v.published_at, v.created_by, v.created_at, v.updated_at, + d.def_key, d.name, d.description + FROM workflow_versions v + JOIN workflow_definitions d ON d.id = v.workflow_id + WHERE v.status = 'published' + ORDER BY d.def_key ASC, v.version DESC + `) + if err != nil { + return nil, err + } + defer rows.Close() + seen := map[int64]bool{} + out := make([]Version, 0) + for rows.Next() { + var v Version + var defRaw []byte + var pubBy, createdBy *int64 + var wfKey, wfName, wfDesc string + if err := rows.Scan( + &v.ID, &v.WorkflowID, &v.Version, &v.Status, &v.SchemaVersion, + &defRaw, &v.DefinitionHash, &pubBy, &v.PublishedAt, &createdBy, + &v.CreatedAt, &v.UpdatedAt, &wfKey, &wfName, &wfDesc, + ); err != nil { + return nil, err + } + if seen[v.WorkflowID] { + continue + } + seen[v.WorkflowID] = true + v.WorkflowKey = wfKey + v.WorkflowName = wfName + v.Description = wfDesc + v.PublishedBy = pubBy + v.CreatedBy = createdBy + _ = json.Unmarshal(defRaw, &v.Definition) + out = append(out, v) + } + return out, rows.Err() +} + +// UpsertPublishedSeed 幂等写入已发布种子（bootstrap）。 +func (s *Store) UpsertPublishedSeed(ctx context.Context, in CreateDefinitionInput, publishedBy int64) error { + if !s.enabled() { + return ErrNotAvailable + } + existing, err := s.GetDefinitionByKey(ctx, in.Key) + if err == nil && existing != nil && existing.LatestPublishedVersion > 0 { + return nil + } + if err != nil && !errors.Is(err, ErrNotFound) { + return err + } + + defHash, err := hashDefinition(in.Definition) + if err != nil { + return err + } + defRaw, _ := json.Marshal(in.Definition) + + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + + var defID int64 + if existing != nil { + defID = existing.ID + } else { + err = tx.QueryRow(ctx, ` + INSERT INTO workflow_definitions (def_key, name, description, product_domain, latest_published_version) + VALUES ($1, $2, $3, $4, 1) RETURNING id + `, in.Key, in.Name, in.Description, in.ProductDomain).Scan(&defID) + if err != nil { + return err + } + } + + _, err = tx.Exec(ctx, ` + INSERT INTO workflow_versions (workflow_id, version, status, schema_version, definition_json, definition_hash, published_by, published_at, created_by) + VALUES ($1, 1, 'published', $2, $3, $4, $5, now(), $5) + ON CONFLICT (workflow_id, version) DO NOTHING + `, defID, in.Definition.SchemaVersion, defRaw, defHash, publishedBy) + if err != nil { + return err + } + + _, err = tx.Exec(ctx, ` + UPDATE workflow_definitions SET latest_published_version = 1, updated_at = now() WHERE id = $1 + `, defID) + if err != nil { + return err + } + return tx.Commit(ctx) +} + +// Template store methods + +func (s *Store) UpsertTemplateAsset(ctx context.Context, assetKey, name string) (int64, error) { + if !s.enabled() { + return 0, ErrNotAvailable + } + var id int64 + err := s.pool.QueryRow(ctx, ` + INSERT INTO template_assets (asset_key, name) VALUES ($1, $2) + ON CONFLICT (asset_key) DO UPDATE SET name = EXCLUDED.name, updated_at = now() + RETURNING id + `, assetKey, name).Scan(&id) + return id, err +} + +func (s *Store) CreateTemplateVersion(ctx context.Context, assetID int64, version int, sha256, mime string, size int64, scanStatus string, createdBy int64) (*TemplateVersion, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + var tv TemplateVersion + err := s.pool.QueryRow(ctx, ` + INSERT INTO template_versions (asset_id, version, blob_sha256, mime_type, size_bytes, scan_status, created_by) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (asset_id, version) DO UPDATE SET + blob_sha256 = EXCLUDED.blob_sha256, + mime_type = EXCLUDED.mime_type, + size_bytes = EXCLUDED.size_bytes, + scan_status = EXCLUDED.scan_status + RETURNING id, asset_id, version, blob_sha256, mime_type, size_bytes, scan_status, created_at + `, assetID, version, sha256, mime, size, scanStatus, createdBy).Scan( + &tv.ID, &tv.AssetID, &tv.Version, &tv.BlobSHA256, &tv.MimeType, &tv.SizeBytes, &tv.ScanStatus, &tv.CreatedAt, + ) + if err != nil { + return nil, err + } + return &tv, nil +} + +func (s *Store) IsTemplateReady(ctx context.Context, assetKey string, version int) bool { + if !s.enabled() { + return false + } + var ready bool + _ = s.pool.QueryRow(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM template_versions tv + JOIN template_assets ta ON ta.id = tv.asset_id + WHERE ta.asset_key = $1 AND tv.version = $2 AND tv.scan_status = 'ready' + ) + `, assetKey, version).Scan(&ready) + return ready +} + +func (s *Store) ListTemplateAssets(ctx context.Context) ([]TemplateAsset, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + rows, err := s.pool.Query(ctx, ` + SELECT id, asset_key, name, status, created_at, updated_at FROM template_assets ORDER BY asset_key ASC + `) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]TemplateAsset, 0) + for rows.Next() { + var a TemplateAsset + if err := rows.Scan(&a.ID, &a.AssetKey, &a.Name, &a.Status, &a.CreatedAt, &a.UpdatedAt); err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +func (s *Store) ListTemplateVersions(ctx context.Context, assetKey string) ([]TemplateVersion, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + rows, err := s.pool.Query(ctx, ` + SELECT tv.id, tv.asset_id, ta.asset_key, tv.version, tv.blob_sha256, tv.mime_type, tv.size_bytes, tv.scan_status, tv.created_at + FROM template_versions tv + JOIN template_assets ta ON ta.id = tv.asset_id + WHERE ta.asset_key = $1 ORDER BY tv.version DESC + `, assetKey) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]TemplateVersion, 0) + for rows.Next() { + var tv TemplateVersion + if err := rows.Scan(&tv.ID, &tv.AssetID, &tv.AssetKey, &tv.Version, &tv.BlobSHA256, &tv.MimeType, &tv.SizeBytes, &tv.ScanStatus, &tv.CreatedAt); err != nil { + return nil, err + } + out = append(out, tv) + } + return out, rows.Err() +} + +func (s *Store) NextTemplateVersion(ctx context.Context, assetID int64) (int, error) { + if !s.enabled() { + return 0, ErrNotAvailable + } + var next int + err := s.pool.QueryRow(ctx, `SELECT COALESCE(MAX(version), 0) + 1 FROM template_versions WHERE asset_id = $1`, assetID).Scan(&next) + return next, err +} diff --git a/internal/workflowdef/types.go b/internal/workflowdef/types.go new file mode 100644 index 00000000..f5882f48 --- /dev/null +++ b/internal/workflowdef/types.go @@ -0,0 +1,122 @@ +package workflowdef + +import "time" + +const ( + StatusDraft = "draft" + StatusPublished = "published" + StatusRetired = "retired" + + ScanPending = "pending" + ScanReady = "ready" + ScanRejected = "rejected" +) + +// DocTaskDef 阶段内文档任务定义。 +type DocTaskDef struct { + Key string `json:"key"` + Title string `json:"title"` + Required bool `json:"required"` + Blocking bool `json:"blocking"` + DefaultEnabled bool `json:"default_enabled"` + TemplateAssetKey string `json:"template_asset_key,omitempty"` + TemplateVersion int `json:"template_version,omitempty"` + OutputDocType string `json:"output_doc_type"` + OutputDocTitle string `json:"output_doc_title"` +} + +// StageDef 线性阶段定义。 +type StageDef struct { + Key string `json:"key"` + Name string `json:"name"` + Position int `json:"position"` + Optional bool `json:"optional"` + ManualGate bool `json:"manual_gate"` + DocTasks []DocTaskDef `json:"doc_tasks"` +} + +// VersionDefinition 工作流版本完整快照（JSONB）。 +type VersionDefinition struct { + SchemaVersion int `json:"schema_version"` + RequiredUpstreamWorkflow string `json:"required_upstream_workflow,omitempty"` + Stages []StageDef `json:"stages"` +} + +// Definition 工作流逻辑标识。 +type Definition struct { + ID int64 `json:"id"` + Key string `json:"key"` + Name string `json:"name"` + Description string `json:"description"` + Status string `json:"status"` + ProductDomain string `json:"product_domain,omitempty"` + LatestPublishedVersion int `json:"latest_published_version"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// Version 工作流版本记录。 +type Version struct { + ID int64 `json:"id"` + WorkflowID int64 `json:"workflow_id"` + WorkflowKey string `json:"workflow_key,omitempty"` + WorkflowName string `json:"workflow_name,omitempty"` + Description string `json:"description,omitempty"` + Version int `json:"version"` + Status string `json:"status"` + SchemaVersion int `json:"schema_version"` + Definition VersionDefinition `json:"definition"` + DefinitionHash string `json:"definition_hash"` + PublishedBy *int64 `json:"published_by,omitempty"` + PublishedAt *time.Time `json:"published_at,omitempty"` + CreatedBy *int64 `json:"created_by,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// ValidationIssue 校验问题。 +type ValidationIssue struct { + Code string `json:"code"` + Message string `json:"message"` + Field string `json:"field,omitempty"` +} + +// ValidationResult 校验结果。 +type ValidationResult struct { + Valid bool `json:"valid"` + Issues []ValidationIssue `json:"issues"` +} + +// CreateDefinitionInput 新建工作流定义。 +type CreateDefinitionInput struct { + Key string `json:"key"` + Name string `json:"name"` + Description string `json:"description"` + ProductDomain string `json:"product_domain,omitempty"` + Definition VersionDefinition `json:"definition"` +} + +// TemplateAsset 模板逻辑资产。 +type TemplateAsset struct { + ID int64 `json:"id"` + AssetKey string `json:"asset_key"` + Name string `json:"name"` + Status string `json:"status"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// TemplateVersion 模板版本。 +type TemplateVersion struct { + ID int64 `json:"id"` + AssetID int64 `json:"asset_id"` + AssetKey string `json:"asset_key,omitempty"` + Version int `json:"version"` + BlobSHA256 string `json:"blob_sha256"` + MimeType string `json:"mime_type"` + SizeBytes int64 `json:"size_bytes"` + ScanStatus string `json:"scan_status"` + VariableSchema []string `json:"variable_schema"` + CreatedBy *int64 `json:"created_by,omitempty"` + CreatedAt time.Time `json:"created_at"` +} diff --git a/internal/workflowdef/validator.go b/internal/workflowdef/validator.go new file mode 100644 index 00000000..f6fb6638 --- /dev/null +++ b/internal/workflowdef/validator.go @@ -0,0 +1,88 @@ +package workflowdef + +import ( + "fmt" + "sort" + "strings" +) + +type templateReadyChecker func(assetKey string, version int) bool +type upstreamExistsChecker func(workflowKey string) bool + +// ValidateDefinition 对工作流版本定义做确定性校验。 +func ValidateDefinition(def VersionDefinition, tplReady templateReadyChecker, upstreamExists upstreamExistsChecker) ValidationResult { + issues := make([]ValidationIssue, 0) + if def.SchemaVersion <= 0 { + def.SchemaVersion = 1 + } + if len(def.Stages) == 0 { + issues = append(issues, ValidationIssue{Code: "NO_STAGES", Message: "至少需要一个阶段"}) + return ValidationResult{Valid: false, Issues: issues} + } + + stageKeys := map[string]struct{}{} + taskKeys := map[string]struct{}{} + enabledStageCount := 0 + + sorted := append([]StageDef(nil), def.Stages...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].Position < sorted[j].Position }) + + for i, st := range sorted { + if st.Key == "" { + issues = append(issues, ValidationIssue{Code: "STAGE_KEY_REQUIRED", Message: "阶段 key 不能为空", Field: fmt.Sprintf("stages[%d].key", i)}) + continue + } + if _, dup := stageKeys[st.Key]; dup { + issues = append(issues, ValidationIssue{Code: "DUPLICATE_STAGE_KEY", Message: fmt.Sprintf("重复阶段 key: %s", st.Key), Field: st.Key}) + } + stageKeys[st.Key] = struct{}{} + if st.Position != i { + issues = append(issues, ValidationIssue{Code: "STAGE_POSITION_GAP", Message: fmt.Sprintf("阶段 %s 顺序应为 %d", st.Key, i), Field: st.Key}) + } + + enabledTasks := 0 + for j, dt := range st.DocTasks { + if dt.Key == "" { + issues = append(issues, ValidationIssue{Code: "TASK_KEY_REQUIRED", Message: "文档任务 key 不能为空", Field: fmt.Sprintf("%s.doc_tasks[%d]", st.Key, j)}) + continue + } + if _, dup := taskKeys[dt.Key]; dup { + issues = append(issues, ValidationIssue{Code: "DUPLICATE_TASK_KEY", Message: fmt.Sprintf("重复任务 key: %s", dt.Key), Field: dt.Key}) + } + taskKeys[dt.Key] = struct{}{} + if dt.OutputDocType == "" || dt.OutputDocTitle == "" { + issues = append(issues, ValidationIssue{Code: "TASK_OUTPUT_REQUIRED", Message: fmt.Sprintf("任务 %s 需 output_doc_type/title", dt.Key), Field: dt.Key}) + } + if dt.DefaultEnabled { + enabledTasks++ + } + if dt.TemplateAssetKey != "" && tplReady != nil { + ver := dt.TemplateVersion + if ver <= 0 { + ver = 1 + } + if !tplReady(dt.TemplateAssetKey, ver) { + issues = append(issues, ValidationIssue{Code: "TEMPLATE_NOT_READY", Message: fmt.Sprintf("模板 %s v%d 不可用", dt.TemplateAssetKey, ver), Field: dt.Key}) + } + } + } + if !st.Optional { + if enabledTasks == 0 && !st.ManualGate { + issues = append(issues, ValidationIssue{Code: "STAGE_EMPTY", Message: fmt.Sprintf("阶段 %s 无启用文档且未设 manual_gate", st.Key), Field: st.Key}) + } else { + enabledStageCount++ + } + } + } + + if enabledStageCount == 0 { + issues = append(issues, ValidationIssue{Code: "NO_ENABLED_STAGE", Message: "至少一个非 optional 阶段需有启用文档或 manual_gate"}) + } + + up := strings.TrimSpace(def.RequiredUpstreamWorkflow) + if up != "" && upstreamExists != nil && !upstreamExists(up) { + issues = append(issues, ValidationIssue{Code: "UPSTREAM_NOT_FOUND", Message: fmt.Sprintf("上游工作流 %s 不存在", up), Field: "required_upstream_workflow"}) + } + + return ValidationResult{Valid: len(issues) == 0, Issues: issues} +} diff --git a/internal/workflowdef/validator_test.go b/internal/workflowdef/validator_test.go new file mode 100644 index 00000000..0e9425c1 --- /dev/null +++ b/internal/workflowdef/validator_test.go @@ -0,0 +1,47 @@ +package workflowdef_test + +import ( + "testing" + + "chat-agent-system/internal/workflowdef" +) + +func TestValidateDefinition_RejectsEmptyStages(t *testing.T) { + res := workflowdef.ValidateDefinition(workflowdef.VersionDefinition{}, nil, nil) + if res.Valid { + t.Fatal("expected invalid for empty stages") + } +} + +func TestValidateDefinition_AcceptsMinimalWorkflow(t *testing.T) { + def := workflowdef.VersionDefinition{ + SchemaVersion: 1, + Stages: []workflowdef.StageDef{{ + Key: "s1", Name: "阶段1", Position: 0, + DocTasks: []workflowdef.DocTaskDef{{ + Key: "t1", Title: "任务1", DefaultEnabled: true, + OutputDocType: "doc1", OutputDocTitle: "文档1", + }}, + }}, + } + res := workflowdef.ValidateDefinition(def, func(string, int) bool { return true }, nil) + if !res.Valid { + t.Fatalf("expected valid, got issues: %+v", res.Issues) + } +} + +func TestValidateDefinition_RejectsMissingTemplate(t *testing.T) { + def := workflowdef.VersionDefinition{ + Stages: []workflowdef.StageDef{{ + Key: "s1", Position: 0, + DocTasks: []workflowdef.DocTaskDef{{ + Key: "t1", DefaultEnabled: true, TemplateAssetKey: "missing", + OutputDocType: "d", OutputDocTitle: "D", + }}, + }}, + } + res := workflowdef.ValidateDefinition(def, func(string, int) bool { return false }, nil) + if res.Valid { + t.Fatal("expected invalid when template not ready") + } +}