Agent-System/internal/workflowdef/catalog.go

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
}