Agent-System/internal/workflowdef/service.go

package workflowdef

import (
	"context"
	"fmt"
	"path/filepath"
	"strings"

	"github.com/jackc/pgx/v5/pgxpool"

	"chat-agent-system/internal/storage"
	"chat-agent-system/internal/upload"
)

// Service 工作流定义与模板领域服务。
type Service struct {
	store            *Store
	cas              *storage.CAS
	validateExecutor keyValidator
	validateGate     keyValidator
}

func NewService(pool *pgxpool.Pool, cas *storage.CAS) *Service {
	if pool == nil {
		return &Service{}
	}
	return &Service{store: NewStore(pool), cas: cas}
}

// SetRegistryValidators injects executor/quality-gate key checks from the composition root
// so this package never imports the concrete workflow registry.
func (s *Service) SetRegistryValidators(execValidate, gateValidate keyValidator) {
	if s == nil {
		return
	}
	s.validateExecutor = execValidate
	s.validateGate = gateValidate
}

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), s.validateExecutor, s.validateGate)
	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), s.validateExecutor, s.validateGate)
	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), s.validateExecutor, s.validateGate), 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)
}

// UploadTemplate 上传模板文件并创建 template_version（扩展名白名单 + MIME/魔数校验后标记 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, err := upload.ValidateTemplate(fileName, "", data)
	if err != nil {
		return nil, fmt.Errorf("%w: %v", ErrUploadRejected, err)
	}

	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
	}
}