package workflowdef import ( "context" "errors" "sort" "sync" ) // ErrCatalogUpstreamMissing is returned when a catalog entry names an unknown upstream. var ErrCatalogUpstreamMissing = errors.New("workflowdef: catalog required upstream missing") // FlatDefinition is the flattened single-stage catalog entry used by legacy catalog consumers. // Owned here so workflowdef does not import projectworkflow; composition-root adapters map to consumer types. type FlatDefinition struct { Key string Name string Description string StageKey string TaskKey string RequiredUpstreamWorkflow string OutputDocType string OutputDocTitle string Enabled bool } // FlatCatalog is the workflowdef-owned catalog port (no projectworkflow import). type FlatCatalog interface { Get(workflowKey string) (*FlatDefinition, bool) List() []*FlatDefinition Validate() error } // CatalogRefresher is an optional FlatCatalog capability for publish/retire hot-reload. type CatalogRefresher interface { Refresh(ctx context.Context) error } // DBCatalog builds a FlatCatalog from published workflow versions. type DBCatalog struct { mu sync.RWMutex store *Store items map[string]*FlatDefinition order []string } func NewDBCatalog(ctx context.Context, store *Store) (*DBCatalog, error) { c := &DBCatalog{store: store, items: make(map[string]*FlatDefinition)} 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]*FlatDefinition) 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) *FlatDefinition { 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 &FlatDefinition{ 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) (*FlatDefinition, 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() []*FlatDefinition { c.mu.RLock() defer c.mu.RUnlock() out := make([]*FlatDefinition, 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 ErrCatalogUpstreamMissing } } } return nil } // NewCatalog prefers DB-published catalog entries; otherwise returns fallback. func NewCatalog(ctx context.Context, store *Store, fallback FlatCatalog) FlatCatalog { if store == nil || !store.enabled() { return fallback } dbCat, err := NewDBCatalog(ctx, store) if err != nil || !dbCat.HasDefinitions() { return fallback } return dbCat }