simplify LoadYamlChains

Author: pashagolub

internal/log/log_test.go

@@ -35,6 +35,6 @@ func TestPgxLog(*testing.T) {
 	pgxl := log.NewPgxLogger(log.Init(config.LoggingOpts{LogLevel: "trace"}))
 	var level tracelog.LogLevel
 	for level = tracelog.LogLevelNone; level <= tracelog.LogLevelTrace; level++ {
-		pgxl.Log(context.Background(), level, "foo", map[string]interface{}{"func": "TestPgxLog"})
+		pgxl.Log(context.Background(), level, "foo", map[string]any{"func": "TestPgxLog"})
 	}
 }

internal/pgengine/bootstrap.go

@@ -32,9 +32,9 @@ var backoff = retry.WithCappedDuration(maxWaitTime, retry.NewExponential(WaitTim
 // PgxIface is common interface for every pgx class
 type PgxIface interface {
 	Begin(ctx context.Context) (pgx.Tx, error)
-	Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
-	QueryRow(context.Context, string, ...interface{}) pgx.Row
-	Query(ctx context.Context, query string, args ...interface{}) (pgx.Rows, error)
+	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
+	QueryRow(context.Context, string, ...any) pgx.Row
+	Query(ctx context.Context, query string, args ...any) (pgx.Rows, error)
 	Ping(ctx context.Context) error
 	CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
 }
@@ -194,7 +194,7 @@ func (pge *PgEngine) AddLogHook(ctx context.Context) {
 
 // QueryRowIface specifies interface to use QueryRow method
 type QueryRowIface interface {
-	QueryRow(context.Context, string, ...interface{}) pgx.Row
+	QueryRow(context.Context, string, ...any) pgx.Row
 }
 
 // TryLockClientName obtains lock on the server to prevent another client with the same name

internal/pgengine/bootstrap_test.go

@@ -78,10 +78,10 @@ func TestFinalizeConnection(t *testing.T) {
 }
 
 type mockpgrow struct {
-	results []interface{}
+	results []any
 }
 
-func (r *mockpgrow) Scan(dest ...interface{}) error {
+func (r *mockpgrow) Scan(dest ...any) error {
 	if len(r.results) > 0 {
 		if err, ok := r.results[0].(error); ok {
 			r.results = r.results[1:]
@@ -103,7 +103,7 @@ type mockpgconn struct {
 	r pgx.Row
 }
 
-func (m mockpgconn) QueryRow(context.Context, string, ...interface{}) pgx.Row {
+func (m mockpgconn) QueryRow(context.Context, string, ...any) pgx.Row {
 	return m.r
 }
 
@@ -117,15 +117,15 @@ func TestTryLockClientName(t *testing.T) {
 	})
 
 	t.Run("no schema yet", func(t *testing.T) {
-		r := &mockpgrow{results: []interface{}{
+		r := &mockpgrow{results: []any{
 			0, //procoid
 		}}
 		m := mockpgconn{r}
 		assert.NoError(t, pge.TryLockClientName(context.Background(), m))
 	})
 
 	t.Run("locking error", func(t *testing.T) {
-		r := &mockpgrow{results: []interface{}{
+		r := &mockpgrow{results: []any{
 			1,                           //procoid
 			errors.New("locking error"), //error
 		}}
@@ -134,7 +134,7 @@ func TestTryLockClientName(t *testing.T) {
 	})
 
 	t.Run("locking successful", func(t *testing.T) {
-		r := &mockpgrow{results: []interface{}{
+		r := &mockpgrow{results: []any{
 			1,    //procoid
 			true, //locked
 		}}

internal/pgengine/log_hook.go

@@ -140,12 +140,12 @@ func (hook *LogHook) send(cache []logrus.Entry) {
 		pgx.Identifier{"timetable", "log"},
 		[]string{"ts", "client_name", "pid", "log_level", "message", "message_data"},
 		pgx.CopyFromSlice(len(cache),
-			func(i int) ([]interface{}, error) {
+			func(i int) ([]any, error) {
 				jsonData, err := json.Marshal(cache[i].Data)
 				if err != nil {
 					return nil, err
 				}
-				return []interface{}{cache[i].Time,
+				return []any{cache[i].Time,
 					hook.client,
 					hook.pid,
 					adaptEntryLevel(cache[i].Level),

internal/pgengine/transaction.go

@@ -119,7 +119,7 @@ func (pge *PgEngine) ExecAutonomousSQLTask(ctx context.Context, task *ChainTask,
 // ExecuteSQLCommand executes chain command with parameters inside transaction
 func (pge *PgEngine) ExecuteSQLCommand(ctx context.Context, executor executor, command string, paramValues []string) (out string, err error) {
 	var ct pgconn.CommandTag
-	var params []interface{}
+	var params []any
 	if strings.TrimSpace(command) == "" {
 		return "", errors.New("SQL command cannot be empty")
 	}

internal/pgengine/yaml.go

@@ -58,42 +58,12 @@ func (pge *PgEngine) LoadYamlChains(ctx context.Context, filePath string, replac
 			return fmt.Errorf("chain '%s' already exists (use --replace flag to overwrite)", yamlChain.ChainName)
 		}
 
-		// Use existing add_job function for single-task chains
-		if len(yamlChain.Tasks) == 1 {
-			task := yamlChain.Tasks[0]
-			params, _ := task.ToSQLParameters()
-			var paramsValue interface{}
-			if params == "" {
-				paramsValue = nil
-			} else {
-				paramsValue = params
-			}
-
-			_, err := pge.ConfigDb.Exec(ctx, `
-				SELECT timetable.add_job($1, $2, $3, $4::jsonb, $5::timetable.command_kind, $6, $7, $8, $9, $10, $11, $12)`,
-				yamlChain.ChainName,
-				yamlChain.Schedule,
-				task.Command,
-				paramsValue,
-				task.Kind,
-				nullString(yamlChain.ClientName),
-				yamlChain.MaxInstances,
-				yamlChain.Live,
-				yamlChain.SelfDestruct,
-				task.IgnoreError,
-				yamlChain.ExclusiveExecution,
-				nullString(yamlChain.OnError))
-			if err != nil {
-				return fmt.Errorf("failed to create chain %s: %w", yamlChain.ChainName, err)
-			}
-		} else {
-			// Multi-task chain - use direct SQL
-			chainID, err := pge.createChainFromYaml(ctx, &yamlChain)
-			if err != nil {
-				return fmt.Errorf("failed to create multi-task chain %s: %w", yamlChain.ChainName, err)
-			}
-			pge.l.WithField("chain", yamlChain.ChainName).WithField("chain_id", chainID).Info("Created multi-task chain")
+		// Multi-task chain - use direct SQL
+		chainID, err := pge.createChainFromYaml(ctx, &yamlChain)
+		if err != nil {
+			return fmt.Errorf("failed to create multi-task chain %s: %w", yamlChain.ChainName, err)
 		}
+		pge.l.WithField("chain", yamlChain.ChainName).WithField("chain_id", chainID).Info("Created multi-task chain")
 	}
 
 	pge.l.WithField("chains", len(yamlConfig.Chains)).WithField("file", filePath).Info("Successfully imported YAML chains")
@@ -167,7 +137,7 @@ func (pge *PgEngine) createChainFromYaml(ctx context.Context, yamlChain *YamlCha
 }
 
 // nullString returns nil for empty strings, otherwise returns the string
-func nullString(s string) interface{} {
+func nullString(s string) any {
 	if s == "" {
 		return nil
 	}
@@ -185,10 +155,16 @@ func (c *YamlChain) ValidateChain() error {
 	}
 
 	// Validate cron format
-	switch c.Schedule {
-	case "", "@reboot", "@after", "@every":
-		// Valid special schedules
-	default:
+	specialSchedules := []string{"@reboot", "@after", "@every"}
+	isSpecial := false
+	for _, s := range specialSchedules {
+		if strings.HasPrefix(c.Schedule, s) {
+			isSpecial = true
+			break
+		}
+	}
+
+	if !isSpecial {
 		fields := strings.Fields(c.Schedule)
 		if len(fields) != 5 {
 			return fmt.Errorf("invalid cron format: %s (expected 5 fields)", c.Schedule)