fix(api): clean duplicate hooks (#5094)

Signed-off-by: francois  samin <[email protected]>

Author: fsamin

engine/api/api.go

@@ -699,6 +699,14 @@ func (a *API) Serve(ctx context.Context) error {
 		return migrate.CleanDuplicateNodes(ctx, a.DBConnectionFactory.GetDBMap())
 	}})
 
+	migrate.Add(ctx, sdk.Migration{Name: "ListDuplicateHooks", Release: "0.44.0", Blocker: false, Automatic: true, ExecFunc: func(ctx context.Context) error {
+		return migrate.CleanDuplicateHooks(ctx, a.DBConnectionFactory.GetDBMap(), a.Cache, true)
+	}})
+
+	migrate.Add(ctx, sdk.Migration{Name: "CleanDuplicateHooks", Release: "0.44.0", Blocker: false, Automatic: false, ExecFunc: func(ctx context.Context) error {
+		return migrate.CleanDuplicateHooks(ctx, a.DBConnectionFactory.GetDBMap(), a.Cache, false)
+	}})
+
 	isFreshInstall, errF := version.IsFreshInstall(a.mustDB())
 	if errF != nil {
 		return sdk.WrapError(errF, "Unable to check if it's a fresh installation of CDS")

engine/api/migrate/clean_duplicate_hooks.go

@@ -0,0 +1,133 @@
+package migrate
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/ovh/cds/engine/api/project"
+
+	"github.com/go-gorp/gorp"
+	"github.com/ovh/cds/engine/api/cache"
+	"github.com/ovh/cds/engine/api/workflow"
+	"github.com/ovh/cds/sdk"
+	"github.com/ovh/cds/sdk/log"
+)
+
+func CleanDuplicateHooks(ctx context.Context, db *gorp.DbMap, store cache.Store, dryrun bool) error {
+	var ids []int64
+
+	if _, err := db.Select(&ids, "select id from workflow"); err != nil {
+		return sdk.WrapError(err, "unable to select workflow")
+	}
+
+	var mError = new(sdk.MultiError)
+	for _, id := range ids {
+		if err := cleanDuplicateHooks(ctx, db, store, id, dryrun); err != nil {
+			mError.Append(err)
+			log.Error(ctx, "migrate.CleanDuplicateHooks> unable to clean workflow %d: %v", id, err)
+		}
+	}
+
+	if mError.IsEmpty() {
+		return nil
+	}
+	return mError
+}
+
+func cleanDuplicateHooks(ctx context.Context, db *gorp.DbMap, store cache.Store, workflowID int64, dryrun bool) error {
+	tx, err := db.Begin()
+	if err != nil {
+		return sdk.WithStack(err)
+	}
+
+	defer tx.Rollback() // nolint
+
+	projectID, err := tx.SelectInt("SELECT project_id FROM workflow WHERE id = $1", workflowID)
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil
+		}
+		return sdk.WithStack(err)
+	}
+
+	proj, err := project.LoadByID(tx, store, projectID,
+		project.LoadOptions.WithApplicationWithDeploymentStrategies,
+		project.LoadOptions.WithPipelines,
+		project.LoadOptions.WithEnvironments,
+		project.LoadOptions.WithIntegrations)
+	if err != nil {
+		return sdk.WrapError(err, "unable to load project %d", projectID)
+	}
+
+	w, err := workflow.LoadAndLockByID(ctx, tx, store, *proj, workflowID, workflow.LoadOptions{})
+	if err != nil {
+		if sdk.ErrorIs(err, sdk.ErrNotFound) {
+			return nil
+		}
+		return err
+	}
+
+	if w.FromRepository != "" {
+		return nil
+	}
+
+	if w.FromTemplate != "" {
+		return nil
+	}
+
+	nbHooks := len(w.WorkflowData.Node.Hooks)
+	if nbHooks < 2 {
+		return nil
+	}
+
+	var hookDoublons = []struct {
+		x int
+		y int
+	}{}
+
+	for i, h1 := range w.WorkflowData.Node.Hooks {
+		for j, h2 := range w.WorkflowData.Node.Hooks {
+			if i != j && i < j && h1.Ref() == h2.Ref() {
+				hookDoublons = append(hookDoublons, struct{ x, y int }{i, j})
+			}
+		}
+	}
+
+	if len(hookDoublons) == 0 {
+		return nil
+	}
+
+	var idxToRemove []int64
+	for _, doublon := range hookDoublons {
+		h1 := w.WorkflowData.Node.Hooks[doublon.x]
+		h2 := w.WorkflowData.Node.Hooks[doublon.y]
+		if h1.ID < h2.ID {
+			idxToRemove = append(idxToRemove, int64(doublon.y))
+		} else {
+			idxToRemove = append(idxToRemove, int64(doublon.x))
+		}
+	}
+
+	var newHooks []sdk.NodeHook
+	for i, h := range w.WorkflowData.Node.Hooks {
+		if !sdk.IsInInt64Array(int64(i), idxToRemove) {
+			newHooks = append(newHooks, h)
+		}
+	}
+	w.WorkflowData.Node.Hooks = newHooks
+
+	if err := workflow.Update(ctx, tx, store, *proj, w, workflow.UpdateOptions{DisableHookManagement: dryrun}); err != nil {
+		return err
+	}
+
+	if dryrun {
+		log.Info(ctx, "migrate.cleanDuplicateHooks> workflow %s/%s (%d) should been cleaned", proj.Name, w.Name, w.ID)
+	} else {
+		if err := tx.Commit(); err != nil {
+			return err
+		}
+		log.Info(ctx, "migrate.cleanDuplicateHooks> workflow %s/%s (%d) has been cleaned", proj.Name, w.Name, w.ID)
+	}
+
+	return nil
+}

engine/api/workflow/dao.go

@@ -362,7 +362,24 @@ func Load(ctx context.Context, db gorp.SqlExecutor, store cache.Store, proj sdk.
 	return res, nil
 }
 
-// LoadByID loads a workflow for a given user (ie. checking permissions)
+// LoadAndLockByID loads a workflow
+func LoadAndLockByID(ctx context.Context, db gorp.SqlExecutor, store cache.Store, proj sdk.Project, id int64, opts LoadOptions) (*sdk.Workflow, error) {
+	query := `
+		select *
+		from workflow
+		where id = $1 for update skip locked`
+	res, err := load(ctx, db, proj, opts, query, id)
+	if err != nil {
+		return nil, sdk.WrapError(err, "Unable to load workflow %d", id)
+	}
+
+	if err := IsValid(context.TODO(), store, db, res, proj, opts); err != nil {
+		return nil, sdk.WrapError(err, "Unable to valid workflow")
+	}
+	return res, nil
+}
+
+// LoadByID loads a workflow
 func LoadByID(ctx context.Context, db gorp.SqlExecutor, store cache.Store, proj sdk.Project, id int64, opts LoadOptions) (*sdk.Workflow, error) {
 	query := `
 		select *