fix(workflowdef): heal retired seeds and fall back published_by
启动种子不再依赖固定用户 id=1,并把白名单工作流的 retired 状态恢复为 active。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -8,6 +8,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -125,7 +126,9 @@ func NewService(cfg *config.Config) (*Service, error) {
|
||||
fallbackCatalog := projectworkflow.NewDefaultCatalog()
|
||||
wfStore := workflowdef.NewStore(pool)
|
||||
if pool != nil && workflowdef.TablesReady(context.Background(), pool) {
|
||||
_ = workflowdef.BootstrapSeeds(context.Background(), pool, casStore, "templates/docs", 1)
|
||||
if seedErr := workflowdef.BootstrapSeeds(context.Background(), pool, casStore, "templates/docs", 1); seedErr != nil {
|
||||
slog.Warn("workflow seed bootstrap failed", "error", seedErr)
|
||||
}
|
||||
}
|
||||
catalog := workflowdef.NewCatalog(context.Background(), wfStore, fallbackCatalog)
|
||||
pwSvc := projectworkflow.NewService(pwStore, catalog)
|
||||
|
||||
@@ -2,10 +2,13 @@ package workflowdef
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"chat-agent-system/internal/storage"
|
||||
@@ -48,11 +51,49 @@ var officialSeedMeta = map[string]struct {
|
||||
},
|
||||
}
|
||||
|
||||
// resolveSeedActor 返回可作为 workflow_versions.created_by 外键的用户 id。
|
||||
// 请求的 publishedBy 不存在时回落到库内最小 id;没有任何用户时返回 0。
|
||||
func resolveSeedActor(ctx context.Context, pool *pgxpool.Pool, publishedBy int64) (int64, error) {
|
||||
if pool == nil {
|
||||
return 0, nil
|
||||
}
|
||||
if publishedBy > 0 {
|
||||
var exists bool
|
||||
if err := pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM users WHERE id = $1)`, publishedBy).Scan(&exists); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if exists {
|
||||
return publishedBy, nil
|
||||
}
|
||||
}
|
||||
var id int64
|
||||
err := pool.QueryRow(ctx, `SELECT id FROM users ORDER BY id ASC LIMIT 1`).Scan(&id)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return 0, nil
|
||||
}
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// BootstrapSeeds 幂等导入模板与正式 v2 工作流;非正式定义标记为不可选。
|
||||
func BootstrapSeeds(ctx context.Context, pool *pgxpool.Pool, cas *storage.CAS, templatesDir string, publishedBy int64) error {
|
||||
if pool == nil || !TablesReady(ctx, pool) {
|
||||
return nil
|
||||
}
|
||||
actor, err := resolveSeedActor(ctx, pool, publishedBy)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if actor == 0 {
|
||||
slog.Warn("workflow seed skipped: no users available for published_by")
|
||||
return nil
|
||||
}
|
||||
if actor != publishedBy {
|
||||
slog.Warn("workflow seed published_by fallback", "requested", publishedBy, "actor", actor)
|
||||
}
|
||||
publishedBy = actor
|
||||
store := NewStore(pool)
|
||||
svc := NewService(pool, cas)
|
||||
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
package workflowdef
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"chat-agent-system/internal/db"
|
||||
)
|
||||
|
||||
func openSeedPool(t *testing.T) *pgxpool.Pool {
|
||||
t.Helper()
|
||||
dsn := db.SkipIfNoTestDatabaseURL(t)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Skip(err)
|
||||
}
|
||||
if err := pool.Ping(ctx); err != nil {
|
||||
pool.Close()
|
||||
t.Skip(err)
|
||||
}
|
||||
if !TablesReady(ctx, pool) {
|
||||
pool.Close()
|
||||
t.Skip("workflowdef tables not migrated")
|
||||
}
|
||||
return pool
|
||||
}
|
||||
|
||||
func TestResolveSeedActor_FallsBackWhenUserMissing(t *testing.T) {
|
||||
pool := openSeedPool(t)
|
||||
defer pool.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
var minID int64
|
||||
if err := pool.QueryRow(ctx, `SELECT id FROM users ORDER BY id ASC LIMIT 1`).Scan(&minID); err != nil {
|
||||
t.Skip("no users to fall back to")
|
||||
}
|
||||
|
||||
got, err := resolveSeedActor(ctx, pool, minID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got != minID {
|
||||
t.Fatalf("existing user: got %d want %d", got, minID)
|
||||
}
|
||||
|
||||
const missing int64 = 9_000_000_001
|
||||
got, err = resolveSeedActor(ctx, pool, missing)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got != minID {
|
||||
t.Fatalf("missing user fallback: got %d want %d", got, minID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBootstrapSeeds_RestoresRetiredOfficialStatus(t *testing.T) {
|
||||
pool := openSeedPool(t)
|
||||
defer pool.Close()
|
||||
ctx := context.Background()
|
||||
|
||||
var exists bool
|
||||
if err := pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM workflow_definitions WHERE def_key = 'rdev')`).Scan(&exists); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !exists {
|
||||
t.Skip("rdev definition not present")
|
||||
}
|
||||
|
||||
if _, err := pool.Exec(ctx, `UPDATE workflow_definitions SET status = 'retired' WHERE def_key = 'rdev'`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := BootstrapSeeds(ctx, pool, nil, t.TempDir(), 9_000_000_001); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var status string
|
||||
if err := pool.QueryRow(ctx, `SELECT status FROM workflow_definitions WHERE def_key = 'rdev'`).Scan(&status); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if status != "active" {
|
||||
t.Fatalf("rdev status = %q, want active", status)
|
||||
}
|
||||
}
|
||||
@@ -55,6 +55,7 @@ func (s *Store) RetireNonOfficialDefinitions(ctx context.Context) error {
|
||||
_, err = s.pool.Exec(ctx, `
|
||||
UPDATE workflow_definitions
|
||||
SET project_selectable = true,
|
||||
status = 'active',
|
||||
catalog_order = CASE def_key
|
||||
WHEN 'main' THEN 10
|
||||
WHEN 'product_manager' THEN 20
|
||||
|
||||
@@ -401,7 +401,7 @@ func (s *Store) UpsertPublishedSeed(ctx context.Context, in CreateDefinitionInpu
|
||||
if latestHash == defHash {
|
||||
_, _ = tx.Exec(ctx, `
|
||||
UPDATE workflow_definitions SET name = $2, description = $3, project_selectable = true,
|
||||
catalog_order = $4, updated_at = now()
|
||||
catalog_order = $4, status = 'active', updated_at = now()
|
||||
WHERE id = $1
|
||||
`, defID, in.Name, in.Description, OfficialCatalogOrder(in.Key))
|
||||
return tx.Commit(ctx)
|
||||
@@ -410,7 +410,7 @@ func (s *Store) UpsertPublishedSeed(ctx context.Context, in CreateDefinitionInpu
|
||||
}
|
||||
_, err = tx.Exec(ctx, `
|
||||
UPDATE workflow_definitions SET name = $2, description = $3, product_domain = $4,
|
||||
project_selectable = true, catalog_order = $5, updated_at = now()
|
||||
project_selectable = true, catalog_order = $5, status = 'active', updated_at = now()
|
||||
WHERE id = $1
|
||||
`, defID, in.Name, in.Description, in.ProductDomain, OfficialCatalogOrder(in.Key))
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user