fix(workflows): migrate per-node inventory/environment overrides

db/Migration.go

@@ -131,6 +131,7 @@ func GetMigrations(dialect string) []Migration {
 		{Version: "2.18.15"},
 		{Version: "2.19.2"},
 		{Version: "2.19.11"},
+		{Version: "2.19.12"},
 	}
 
 	return append(initScripts, commonScripts...)

db/sql/migration.go

@@ -261,6 +261,8 @@ func (d *SqlDb) ApplyMigration(migration db.Migration) error {
 		err = migration_2_8_42{db: d}.PostApply(tx)
 	case "2.19.11":
 		err = migration_2_19_11{db: d}.PostApply(tx)
+	case "2.19.12":
+		err = migration_2_19_12{db: d}.PostApply(tx)
 	}
 
 	if err != nil {

db/sql/migration_2_19_11.go

@@ -1,6 +1,7 @@
 package sql
 
 import (
+	"database/sql"
 	"encoding/json"
 
 	"github.com/go-gorp/gorp/v3"
@@ -10,55 +11,98 @@ type migration_2_19_11 struct {
 	db *SqlDb
 }
 
-// PostApply moves per-node run parameters (previously only `limit`) of workflow
-// nodes into project__task_params rows referenced by the freshly added
-// task_params_id column.
+// PostApply moves per-node run parameters (limit, inventory_id, environment_id)
+// of workflow nodes into project__task_params rows referenced by the freshly
+// added task_params_id column.
 //
 // The legacy inventory_id/environment_id/limit columns are intentionally kept
 // (unused): they participate in table-level foreign key definitions, and SQLite
 // can not drop such columns without rebuilding the table.
 func (m migration_2_19_11) PostApply(tx *gorp.Transaction) error {
-	type nodeLimit struct {
-		ID        int    `db:"id"`
-		ProjectID int    `db:"project_id"`
-		Limit     string `db:"limit"`
+	type nodeOverrides struct {
+		ID            int     `db:"id"`
+		ProjectID     int     `db:"project_id"`
+		Limit         *string `db:"limit"`
+		InventoryID   *int    `db:"inventory_id"`
+		EnvironmentID *int    `db:"environment_id"`
 	}
 
-	var nodes []nodeLimit
+	var nodes []nodeOverrides
 	_, err := tx.Select(&nodes, m.db.PrepareQuery(
-		"select n.id as id, wt.project_id as project_id, n.`limit` as `limit` "+
+		"select n.id as id, wt.project_id as project_id, n.`limit` as `limit`, "+
+			"n.inventory_id as inventory_id, n.environment_id as environment_id "+
 			"from project__workflow_node n "+
 			"join project__workflow_template wt on wt.id = n.workflow_template_id "+
-			"where n.`limit` is not null and n.`limit` <> '' and n.`limit` <> '[]'"))
+			"where (n.`limit` is not null and n.`limit` <> '' and n.`limit` <> '[]') "+
+			"or n.inventory_id is not null or n.environment_id is not null"))
 	if err != nil {
 		return err
 	}
 
 	for _, n := range nodes {
-		var limit []string
-		if json.Unmarshal([]byte(n.Limit), &limit) != nil || len(limit) == 0 {
-			continue
+		if err = upsertWorkflowNodeTaskParams(tx, m.db, n.ID, n.ProjectID, n.Limit, n.InventoryID, n.EnvironmentID, nil); err != nil {
+			return err
 		}
+	}
+
+	return nil
+}
+
+// upsertWorkflowNodeTaskParams copies legacy per-node overrides into
+// project__task_params. When existingParamsID is nil and the node already has
+// task_params_id, the existing row is updated instead of creating a duplicate.
+func upsertWorkflowNodeTaskParams(
+	tx *gorp.Transaction,
+	db *SqlDb,
+	nodeID int,
+	projectID int,
+	limit *string,
+	inventoryID *int,
+	environmentID *int,
+	existingParamsID *int64,
+) error {
+	paramsArg, inventoryArg, environment, err := workflowNodeOverrideValues(
+		tx, db, limit, inventoryID, environmentID)
+	if err != nil {
+		return err
+	}
+
+	if paramsArg == nil && inventoryArg == nil && environment == "" {
+		return nil
+	}
 
-		params, err2 := json.Marshal(map[string]any{"limit": limit})
-		if err2 != nil {
-			return err2
+	var paramsID int64
+	if existingParamsID != nil {
+		paramsID = *existingParamsID
+	} else {
+		var currentID sql.NullInt64
+		currentID, err = tx.SelectNullInt(
+			db.PrepareQuery("select task_params_id from project__workflow_node where id = ?"),
+			nodeID)
+		if err != nil {
+			return err
+		}
+		if currentID.Valid {
+			paramsID = currentID.Int64
 		}
+	}
 
-		var paramsID int64
-		// environment and message map to non-pointer string fields — write ''
-		// instead of NULL so the rows scan back like gorp-inserted ones.
-		insertQuery := "insert into project__task_params (project_id, params, environment, message) values (?, ?, '', '')"
-		switch m.db.Sql().Dialect.(type) {
+	if paramsID == 0 {
+		insertQuery := "insert into project__task_params (project_id, params, environment, message, inventory_id) values (?, ?, ?, '', ?)"
+		switch db.Sql().Dialect.(type) {
 		case gorp.PostgresDialect:
-			paramsID, err = tx.SelectInt(m.db.PrepareQuery(insertQuery+" returning id"), n.ProjectID, string(params))
+			paramsID, err = tx.SelectInt(
+				db.PrepareQuery(insertQuery+" returning id"),
+				projectID, paramsArg, environment, inventoryArg)
 			if err != nil {
 				return err
 			}
 		default:
-			res, err2 := tx.Exec(m.db.PrepareQuery(insertQuery), n.ProjectID, string(params))
-			if err2 != nil {
-				return err2
+			res, err := tx.Exec(
+				db.PrepareQuery(insertQuery),
+				projectID, paramsArg, environment, inventoryArg)
+			if err != nil {
+				return err
 			}
 			paramsID, err = res.LastInsertId()
 			if err != nil {
@@ -67,12 +111,54 @@ func (m migration_2_19_11) PostApply(tx *gorp.Transaction) error {
 		}
 
 		_, err = tx.Exec(
-			m.db.PrepareQuery("update project__workflow_node set task_params_id=? where id=?"),
-			paramsID, n.ID)
+			db.PrepareQuery("update project__workflow_node set task_params_id=? where id=?"),
+			paramsID, nodeID)
 		if err != nil {
 			return err
 		}
+		return nil
 	}
 
-	return nil
+	updateQuery := "update project__task_params set params = coalesce(?, params), " +
+		"environment = case when ? <> '' then ? else environment end, " +
+		"inventory_id = coalesce(?, inventory_id) where id = ?"
+	_, err = tx.Exec(
+		db.PrepareQuery(updateQuery),
+		paramsArg, environment, environment, inventoryArg, paramsID)
+	return err
+}
+
+func workflowNodeOverrideValues(
+	tx *gorp.Transaction,
+	db *SqlDb,
+	limit *string,
+	inventoryID *int,
+	environmentID *int,
+) (paramsArg any, inventoryArg any, environment string, err error) {
+	if limit != nil && *limit != "" && *limit != "[]" {
+		var limitVals []string
+		if json.Unmarshal([]byte(*limit), &limitVals) == nil && len(limitVals) > 0 {
+			var params []byte
+			params, err = json.Marshal(map[string]any{"limit": limitVals})
+			if err != nil {
+				return
+			}
+			paramsArg = string(params)
+		}
+	}
+
+	if environmentID != nil {
+		environment, err = tx.SelectStr(
+			db.PrepareQuery("select json from project__environment where id = ?"),
+			*environmentID)
+		if err != nil {
+			return
+		}
+	}
+
+	if inventoryID != nil {
+		inventoryArg = *inventoryID
+	}
+
+	return
 }

db/sql/migration_2_19_11_test.go

@@ -9,9 +9,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestMigration_2_19_11 verifies that workflow node `limit` values are moved
-// into project__task_params rows referenced by the new task_params_id column.
-func TestMigration_2_19_11(t *testing.T) {
+func setupMigrationTestDB(t *testing.T) *SqlDb {
+	t.Helper()
 	util.Config = &util.ConfigType{
 		SQLite: &util.DbConfig{
 			Hostname: ":memory:",
@@ -25,8 +24,14 @@ func TestMigration_2_19_11(t *testing.T) {
 	}
 	store := CreateDb(util.DbDriverSQLite)
 	store.Connect()
+	return store
+}
+
+// TestMigration_2_19_11 verifies that workflow node `limit` values are moved
+// into project__task_params rows referenced by the new task_params_id column.
+func TestMigration_2_19_11(t *testing.T) {
+	store := setupMigrationTestDB(t)
 
-	// Migrate to the schema right before 2.19.11 and seed a node with a limit.
 	target := "2.19.2"
 	require.NoError(t, db.Migrate(store, &target))
 
@@ -58,3 +63,113 @@ func TestMigration_2_19_11(t *testing.T) {
 	assert.Equal(t, proj.ID, taskParams.ProjectID)
 	assert.Equal(t, []any{"web*", "db"}, taskParams.Params["limit"])
 }
+
+// TestMigration_2_19_11_InventoryEnvironment verifies per-node inventory and
+// environment overrides are migrated into project__task_params.
+func TestMigration_2_19_11_InventoryEnvironment(t *testing.T) {
+	store := setupMigrationTestDB(t)
+
+	target := "2.19.2"
+	require.NoError(t, db.Migrate(store, &target))
+
+	proj, err := store.CreateProject(db.Project{Name: "p"})
+	require.NoError(t, err)
+
+	env, err := store.CreateEnvironment(db.Environment{
+		ProjectID: proj.ID,
+		Name:      "env",
+		JSON:      `{"FOO":"bar"}`,
+	})
+	require.NoError(t, err)
+
+	inv, err := store.CreateInventory(db.Inventory{
+		ProjectID: proj.ID,
+		Name:      "inv",
+		Type:      db.InventoryStatic,
+		Inventory: "localhost",
+	})
+	require.NoError(t, err)
+
+	_, err = store.Sql().Exec(
+		"insert into project__workflow_template (project_id, name) values (?, ?)", proj.ID, "wf")
+	require.NoError(t, err)
+	workflowID, err := store.Sql().SelectInt("select id from project__workflow_template where name = 'wf'")
+	require.NoError(t, err)
+
+	_, err = store.Sql().Exec(
+		"insert into project__workflow_node (workflow_template_id, template_id, inventory_id, environment_id) values (?, ?, ?, ?)",
+		workflowID, 0, inv.ID, env.ID)
+	require.NoError(t, err)
+
+	require.NoError(t, db.Migrate(store, nil))
+
+	paramsID, err := store.Sql().SelectNullInt(
+		"select task_params_id from project__workflow_node where workflow_template_id = ?", workflowID)
+	require.NoError(t, err)
+	require.True(t, paramsID.Valid)
+
+	var taskParams db.TaskParams
+	err = store.Sql().SelectOne(&taskParams,
+		"select * from project__task_params where id = ?", paramsID.Int64)
+	require.NoError(t, err)
+	require.NotNil(t, taskParams.InventoryID)
+	assert.Equal(t, inv.ID, *taskParams.InventoryID)
+	assert.Equal(t, env.JSON, taskParams.Environment)
+}
+
+// TestMigration_2_19_12_RepairsPartialMigration simulates the buggy 2.19.11
+// migration that only copied limit, then verifies 2.19.12 backfills inventory.
+func TestMigration_2_19_12_RepairsPartialMigration(t *testing.T) {
+	store := setupMigrationTestDB(t)
+
+	target := "2.19.2"
+	require.NoError(t, db.Migrate(store, &target))
+
+	proj, err := store.CreateProject(db.Project{Name: "p"})
+	require.NoError(t, err)
+
+	inv, err := store.CreateInventory(db.Inventory{
+		ProjectID: proj.ID,
+		Name:      "inv",
+		Type:      db.InventoryStatic,
+		Inventory: "localhost",
+	})
+	require.NoError(t, err)
+
+	_, err = store.Sql().Exec(
+		"insert into project__workflow_template (project_id, name) values (?, ?)", proj.ID, "wf")
+	require.NoError(t, err)
+	workflowID, err := store.Sql().SelectInt("select id from project__workflow_template where name = 'wf'")
+	require.NoError(t, err)
+
+	_, err = store.Sql().Exec(
+		"insert into project__workflow_node (workflow_template_id, template_id, `limit`, inventory_id) values (?, ?, ?, ?)",
+		workflowID, 0, `["web"]`, inv.ID)
+	require.NoError(t, err)
+
+	// Apply schema change from 2.19.11 without its PostApply logic.
+	target = "2.19.11"
+	require.NoError(t, db.Migrate(store, &target))
+
+	// Simulate buggy migration: only limit was copied.
+	res, err := store.Sql().Exec(
+		"insert into project__task_params (project_id, params, environment, message) values (?, ?, '', '')",
+		proj.ID, `{"limit":["web"]}`)
+	require.NoError(t, err)
+	paramsID, err := res.LastInsertId()
+	require.NoError(t, err)
+	_, err = store.Sql().Exec(
+		"update project__workflow_node set task_params_id = ? where workflow_template_id = ?",
+		paramsID, workflowID)
+	require.NoError(t, err)
+
+	require.NoError(t, db.Migrate(store, nil))
+
+	var taskParams db.TaskParams
+	err = store.Sql().SelectOne(&taskParams,
+		"select * from project__task_params where id = ?", paramsID)
+	require.NoError(t, err)
+	assert.Equal(t, []any{"web"}, taskParams.Params["limit"])
+	require.NotNil(t, taskParams.InventoryID)
+	assert.Equal(t, inv.ID, *taskParams.InventoryID)
+}

db/sql/migration_2_19_12.go

@@ -0,0 +1,43 @@
+package sql
+
+import (
+	"github.com/go-gorp/gorp/v3"
+)
+
+type migration_2_19_12 struct {
+	db *SqlDb
+}
+
+// PostApply repairs workflow nodes whose per-node inventory/environment overrides
+// were not copied into project__task_params by the initial 2.19.11 migration.
+// It is idempotent: nodes already fully migrated are left unchanged.
+func (m migration_2_19_12) PostApply(tx *gorp.Transaction) error {
+	type nodeOverrides struct {
+		ID            int     `db:"id"`
+		ProjectID     int     `db:"project_id"`
+		TaskParamsID  *int64  `db:"task_params_id"`
+		Limit         *string `db:"limit"`
+		InventoryID   *int    `db:"inventory_id"`
+		EnvironmentID *int    `db:"environment_id"`
+	}
+
+	var nodes []nodeOverrides
+	_, err := tx.Select(&nodes, m.db.PrepareQuery(
+		"select n.id as id, wt.project_id as project_id, n.task_params_id as task_params_id, "+
+			"n.`limit` as `limit`, n.inventory_id as inventory_id, n.environment_id as environment_id "+
+			"from project__workflow_node n "+
+			"join project__workflow_template wt on wt.id = n.workflow_template_id "+
+			"where n.inventory_id is not null or n.environment_id is not null"))
+	if err != nil {
+		return err
+	}
+
+	for _, n := range nodes {
+		if err = upsertWorkflowNodeTaskParams(
+			tx, m.db, n.ID, n.ProjectID, n.Limit, n.InventoryID, n.EnvironmentID, n.TaskParamsID); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

db/sql/migrations/v2.19.12.err.sql

@@ -0,0 +1 @@
+-- undo repair migration (no schema changes)

db/sql/migrations/v2.19.12.sql

@@ -0,0 +1 @@
+-- repair workflow node inventory/environment overrides missed by 2.19.11