Agent-System/internal/workflowdef/store.go

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
}