Agent-System/internal/workflowdef/catalog.go

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
}