feat(api): search workflow handler (#5175)

Author: fsamin

engine/api/api_routes.go

@@ -408,6 +408,8 @@ func (api *API) InitRouter() {
 	r.Handle("/group/{permGroupName}/worker/model", Scope(sdk.AuthConsumerScopeWorkerModel), r.GET(api.getWorkerModelsForGroupHandler))
 
 	// Workflows
+
+	r.Handle("/workflow/search", Scope(sdk.AuthConsumerScopeProject), r.GET(api.getSearchWorkflowHandler))
 	r.Handle("/workflow/hook", Scope(sdk.AuthConsumerScopeHooks), r.GET(api.getWorkflowHooksHandler))
 	r.Handle("/workflow/hook/model/{model}", ScopeNone(), r.GET(api.getWorkflowHookModelHandler), r.POST(api.postWorkflowHookModelHandler, NeedAdmin(true)), r.PUT(api.putWorkflowHookModelHandler, NeedAdmin(true)))
 

engine/api/application.go

@@ -161,8 +161,6 @@ func loadApplicationUsage(ctx context.Context, db gorp.SqlExecutor, projKey, app
 	}
 	usage.Workflows = wf
 
-	// TODO: add usage of pipelines and environments
-
 	return usage, nil
 }
 

engine/api/application/dao.go

@@ -5,6 +5,8 @@ import (
 	"database/sql"
 	"time"
 
+	"github.com/lib/pq"
+
 	"github.com/go-gorp/gorp"
 
 	"github.com/ovh/cds/engine/api/database/gorpmapping"
@@ -238,6 +240,16 @@ func LoadAll(db gorp.SqlExecutor, key string, opts ...LoadOptionFunc) ([]sdk.App
 	return getAll(context.Background(), db, opts, query)
 }
 
+// LoadAllByIDs returns all applications
+func LoadAllByIDs(db gorp.SqlExecutor, ids []int64, opts ...LoadOptionFunc) ([]sdk.Application, error) {
+	query := gorpmapping.NewQuery(`
+	SELECT application.*
+	FROM application
+	WHERE application.id = ANY($1)
+	ORDER BY application.name ASC`).Args(pq.Int64Array(ids))
+	return getAll(context.Background(), db, opts, query)
+}
+
 // LoadAllNames returns all application names
 func LoadAllNames(db gorp.SqlExecutor, projID int64) (sdk.IDNames, error) {
 	query := `

engine/api/ascode/dao.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"database/sql"
 
+	"github.com/lib/pq"
+
 	"github.com/go-gorp/gorp"
 
 	"github.com/ovh/cds/engine/api/database/gorpmapping"
@@ -28,6 +30,20 @@ func LoadEventByWorkflowIDAndPullRequest(ctx context.Context, db gorp.SqlExecuto
 	return &event, nil
 }
 
+func LoadEventsByWorkflowIDs(ctx context.Context, db gorp.SqlExecutor, workflowIDs []int64) ([]sdk.AsCodeEvent, error) {
+	query := gorpmapping.NewQuery("SELECT * FROM as_code_events where workflow_id = ANY($1)").Args(pq.Int64Array(workflowIDs))
+	var events []dbAsCodeEvents
+	if err := gorpmapping.GetAll(ctx, db, query, &events); err != nil {
+		return nil, sdk.WrapError(err, "Unable to load as code events")
+	}
+
+	asCodeEvents := make([]sdk.AsCodeEvent, len(events))
+	for i := range events {
+		asCodeEvents[i] = sdk.AsCodeEvent(events[i])
+	}
+	return asCodeEvents, nil
+}
+
 // LoadEventsByWorkflowID returns as code events for the given workflow.
 func LoadEventsByWorkflowID(ctx context.Context, db gorp.SqlExecutor, workflowID int64) ([]sdk.AsCodeEvent, error) {
 	query := gorpmapping.NewQuery(`

engine/api/environment/environment.go

@@ -12,6 +12,45 @@ import (
 	"github.com/ovh/cds/sdk"
 )
 
+// LoadAllByIDs load all environment
+func LoadAllByIDs(db gorp.SqlExecutor, ids []int64) ([]sdk.Environment, error) {
+	var envs []sdk.Environment
+
+	query := `SELECT environment.id, environment.name, environment.last_modified, environment.from_repository, project.projectkey
+		  FROM environment
+		  JOIN project ON project.id = environment.project_id
+		  WHERE environment.id = ANY($1)
+		  ORDER by environment.name`
+	rows, err := db.Query(query, pq.Int64Array(ids))
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return envs, sdk.WithStack(sdk.ErrNoEnvironment)
+		}
+		return envs, sdk.WithStack(err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var env sdk.Environment
+		var lastModified time.Time
+		var projectKey string
+		if err := rows.Scan(&env.ID, &env.Name, &lastModified, &env.FromRepository, &projectKey); err != nil {
+			return envs, sdk.WithStack(err)
+		}
+		env.LastModified = lastModified.Unix()
+		env.ProjectKey = projectKey
+		envs = append(envs, env)
+	}
+	rows.Close()
+
+	for i := range envs {
+		if err := loadDependencies(db, &envs[i]); err != nil {
+			return envs, err
+		}
+	}
+	return envs, nil
+}
+
 // LoadEnvironments load all environment from the given project
 func LoadEnvironments(db gorp.SqlExecutor, projectKey string) ([]sdk.Environment, error) {
 	var envs []sdk.Environment

engine/api/group/gorp_model.go

@@ -99,5 +99,6 @@ func init() {
 		gorpmapping.New(group{}, "group", true, "id"),
 		gorpmapping.New(LinkGroupUser{}, "group_authentified_user", true, "id"),
 		gorpmapping.New(LinkGroupProject{}, "project_group", true, "id"),
+		gorpmapping.New(LinkWorkflowGroupPermission{}, "workflow_perm", false),
 	)
 }

engine/api/group/node_group.go

@@ -5,20 +5,22 @@ import (
 	"strings"
 
 	"github.com/go-gorp/gorp"
+	"github.com/lib/pq"
+
 	"github.com/ovh/cds/sdk"
 )
 
 // LoadGroupsByNode retrieves all groups related to a node
-func LoadGroupsByNode(db gorp.SqlExecutor, nodeID int64) ([]sdk.GroupPermission, error) {
-	query := `SELECT "group".id,"group".name,workflow_node_group.role
+func LoadGroupsByNode(db gorp.SqlExecutor, nodeID []int64) (map[int64][]sdk.GroupPermission, error) {
+	query := `SELECT workflow_node_id, "group".id,"group".name,workflow_node_group.role
 		FROM "group"
-			JOIN project_group ON "group".id = project_group.group_id
-			JOIN workflow_perm ON workflow_perm.project_group_id = project_group.id
-	 		JOIN workflow_node_group ON workflow_node_group.workflow_group_id = workflow_perm.id
-		WHERE workflow_node_group.workflow_node_id = $1
+		JOIN project_group ON "group".id = project_group.group_id
+		JOIN workflow_perm ON workflow_perm.project_group_id = project_group.id
+	 	JOIN workflow_node_group ON workflow_node_group.workflow_group_id = workflow_perm.id
+		WHERE workflow_node_group.workflow_node_id = ANY($1)
 		ORDER BY "group".name ASC`
 
-	rows, err := db.Query(query, nodeID)
+	rows, err := db.Query(query, pq.Int64Array(nodeID))
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, nil
@@ -27,19 +29,23 @@ func LoadGroupsByNode(db gorp.SqlExecutor, nodeID int64) ([]sdk.GroupPermission,
 	}
 	defer rows.Close()
 
-	var groups []sdk.GroupPermission
+	var mapGroups = make(map[int64][]sdk.GroupPermission)
 	for rows.Next() {
 		var group sdk.Group
 		var perm int
-		if err := rows.Scan(&group.ID, &group.Name, &perm); err != nil {
-			return groups, sdk.WithStack(err)
+		var nodeID int64
+		if err := rows.Scan(&nodeID, &group.ID, &group.Name, &perm); err != nil {
+			return nil, sdk.WithStack(err)
 		}
+
+		var groups = mapGroups[nodeID]
 		groups = append(groups, sdk.GroupPermission{
 			Group:      group,
 			Permission: perm,
 		})
+		mapGroups[nodeID] = groups
 	}
-	return groups, nil
+	return mapGroups, nil
 }
 
 // InsertGroupsInNode Link the given groups and the given environment

engine/api/group/workflow_group.go

@@ -6,7 +6,9 @@ import (
 	"strings"
 
 	"github.com/go-gorp/gorp"
+	"github.com/lib/pq"
 
+	"github.com/ovh/cds/engine/api/database/gorpmapping"
 	"github.com/ovh/cds/sdk"
 )
 
@@ -148,14 +150,52 @@ func checkAtLeastOneGroupWithWriteRoleOnWorkflow(db gorp.SqlExecutor, wID int64)
 	return nb > 0, err
 }
 
+type LinkWorkflowGroupPermission struct {
+	WorkflowID int64  `db:"workflow_id"`
+	GroupID    int64  `db:"group_id"`
+	GroupName  string `db:"group_name"`
+	Role       int    `db:"role"`
+}
+
+// LoadWorkflowGroupsByWorkflowIDs returns a map with key: workflowID and value the slite of groups
+func LoadWorkflowGroupsByWorkflowIDs(db gorp.SqlExecutor, workflowIDs []int64) (map[int64][]sdk.GroupPermission, error) {
+	result := make(map[int64][]sdk.GroupPermission, len(workflowIDs))
+	query := gorpmapping.NewQuery(`
+	SELECT workflow_perm.workflow_id, "group".id as "group_id", "group".name as "group_name", workflow_perm.role
+	FROM "group"
+	JOIN project_group ON project_group.group_id = "group".id
+	JOIN workflow_perm ON workflow_perm.project_group_id = project_group.id
+	WHERE workflow_perm.workflow_id = ANY($1)
+	ORDER BY workflow_perm.workflow_id, "group".name ASC
+	`).Args(pq.Int64Array(workflowIDs))
+	var dbResultSet = []LinkWorkflowGroupPermission{}
+	if err := gorpmapping.GetAll(context.Background(), db, query, &dbResultSet); err != nil {
+		return nil, err
+	}
+
+	for _, row := range dbResultSet {
+		perms := result[row.WorkflowID]
+		perms = append(perms, sdk.GroupPermission{
+			Permission: row.Role,
+			Group: sdk.Group{
+				ID:   row.GroupID,
+				Name: row.GroupName,
+			},
+		})
+		result[row.WorkflowID] = perms
+	}
+
+	return result, nil
+}
+
 // LoadWorkflowGroups load groups for a workflow
 func LoadWorkflowGroups(db gorp.SqlExecutor, workflowID int64) ([]sdk.GroupPermission, error) {
 	wgs := []sdk.GroupPermission{}
 
 	query := `SELECT "group".id, "group".name, workflow_perm.role
 		FROM "group"
-			JOIN project_group ON project_group.group_id = "group".id
-			JOIN workflow_perm ON workflow_perm.project_group_id = project_group.id
+		JOIN project_group ON project_group.group_id = "group".id
+		JOIN workflow_perm ON workflow_perm.project_group_id = project_group.id
 		WHERE workflow_perm.workflow_id = $1
 		ORDER BY "group".name ASC`
 	rows, errq := db.Query(query, workflowID)

engine/api/integration/dao_project_integration.go

@@ -3,6 +3,8 @@ package integration
 import (
 	"context"
 
+	"github.com/lib/pq"
+
 	"github.com/go-gorp/gorp"
 
 	"github.com/ovh/cds/engine/api/database/gorpmapping"
@@ -153,6 +155,12 @@ func LoadIntegrationsByProjectID(db gorp.SqlExecutor, id int64) ([]sdk.ProjectIn
 	return loadAll(db, query)
 }
 
+// LoadIntegrationsByIDs load integration integrations by id
+func LoadIntegrationsByIDs(db gorp.SqlExecutor, ids []int64) ([]sdk.ProjectIntegration, error) {
+	query := gorpmapping.NewQuery("SELECT * from project_integration WHERE id = ANY($1)").Args(pq.Int64Array(ids))
+	return loadAll(db, query)
+}
+
 // InsertIntegration inserts a integration
 func InsertIntegration(db gorp.SqlExecutor, pp *sdk.ProjectIntegration) error {
 	oldConfig := pp.Config.Clone()

engine/api/notification/user.go

@@ -120,6 +120,7 @@ func GetUserWorkflowEvents(ctx context.Context, db gorp.SqlExecutor, store cache
 				if err != nil {
 					log.Error(ctx, "notification.GetUserWorkflowEvents> unable to handle event %+v: %v", jn, err)
 				}
+				log.Debug("GetUserWorkflowEvents> will send mail notifications: %+v", notif)
 				go sendMailNotif(ctx, notif)
 			}
 		}

engine/api/permission/dao.go

@@ -6,13 +6,41 @@ import (
 	"strings"
 
 	"github.com/go-gorp/gorp"
+	"github.com/lib/pq"
 
 	"github.com/ovh/cds/engine/api/database/gorpmapping"
 	"github.com/ovh/cds/engine/api/group"
 	"github.com/ovh/cds/engine/api/observability"
 	"github.com/ovh/cds/sdk"
 )
 
+func LoadWorkflowMaxLevelPermissionByWorkflowIDs(ctx context.Context, db gorp.SqlExecutor, workflowIDs []int64, groupIDs []int64) (sdk.EntitiesPermissions, error) {
+	_, end := observability.Span(ctx, "permission.LoadWorkflowMaxLevelPermissionByWorkflowIDs")
+	defer end()
+
+	query := `
+		SELECT workflow.id::text, max(workflow_perm.role)
+		FROM workflow_perm
+		JOIN workflow ON workflow.id = workflow_perm.workflow_id
+		JOIN project ON project.id = workflow.project_id
+		JOIN project_group ON project_group.id = workflow_perm.project_group_id
+		WHERE project_group.project_id = project.id
+		AND workflow.id = ANY($1)
+		AND project_group.group_id = ANY($2)
+		GROUP BY workflow.id, workflow.name`
+
+	rows, err := db.Query(query, pq.Int64Array(workflowIDs), pq.Int64Array(groupIDs))
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, sdk.WithStack(err)
+	}
+	defer rows.Close()
+
+	return scanPermissions(rows)
+}
+
 func LoadWorkflowMaxLevelPermission(ctx context.Context, db gorp.SqlExecutor, projectKey string, workflowNames []string, groupIDs []int64) (sdk.EntitiesPermissions, error) {
 	_, end := observability.Span(ctx, "permission.LoadWorkflowMaxLevelPermission")
 	defer end()

engine/api/pipeline/pipeline.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/go-gorp/gorp"
+	"github.com/lib/pq"
 
 	"github.com/ovh/cds/engine/api/database/gorpmapping"
 	"github.com/ovh/cds/engine/api/group"
@@ -239,6 +240,37 @@ func DeletePipeline(ctx context.Context, db gorp.SqlExecutor, pipelineID int64)
 	return nil
 }
 
+// LoadAllByIDs loads all pipelines
+func LoadAllByIDs(db gorp.SqlExecutor, ids []int64, loadDependencies bool) ([]sdk.Pipeline, error) {
+	var pips []sdk.Pipeline
+	query := `SELECT id, name, description, project_id, last_modified, from_repository
+			  FROM pipeline
+			  WHERE id = ANY($1)
+			  ORDER BY pipeline.name`
+
+	if _, err := db.Select(&pips, query, pq.Int64Array(ids)); err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil
+		}
+		return nil, sdk.WithStack(err)
+	}
+
+	for i := range pips {
+		if loadDependencies {
+			if err := LoadPipelineStage(context.TODO(), db, &pips[i]); err != nil {
+				return nil, err
+			}
+		}
+		params, err := GetAllParametersInPipeline(context.TODO(), db, pips[i].ID)
+		if err != nil {
+			return nil, err
+		}
+		pips[i].Parameter = params
+	}
+
+	return pips, nil
+}
+
 // LoadPipelines loads all pipelines in a project
 func LoadPipelines(db gorp.SqlExecutor, projectID int64, loadDependencies bool) ([]sdk.Pipeline, error) {
 	var pips []sdk.Pipeline

engine/api/repositories_manager.go

@@ -557,7 +557,7 @@ func (api *API) attachRepositoriesManagerHandler() service.Handler {
 				wfDB.WorkflowData.Node.Context.DefaultPayload = defaultPayload
 
 				if err := workflow.Update(ctx, db, api.Cache, *proj, wfDB, workflow.UpdateOptions{DisableHookManagement: true}); err != nil {
-					return sdk.WrapError(err, "cannot update node context %d", wf.WorkflowData.Node.Context.ID)
+					return sdk.WrapError(err, "cannot update node context %d", wfDB.WorkflowData.Node.Context.ID)
 				}
 
 				event.PublishWorkflowUpdate(ctx, proj.Key, *wfDB, *wfOld, getAPIConsumer(ctx))