feat: added support for cd pipelines bulk deletion

api/restHandler/app/DeploymentPipelineRestHandler.go

@@ -34,6 +34,8 @@ type DevtronAppDeploymentRestHandler interface {
 
 	IsReadyToTrigger(w http.ResponseWriter, r *http.Request)
 	FetchCdWorkflowDetails(w http.ResponseWriter, r *http.Request)
+
+	HandleCdPipelineBulkAction(w http.ResponseWriter, r *http.Request)
 }
 
 type DevtronAppDeploymentConfigRestHandler interface {
@@ -1830,3 +1832,79 @@ func (handler PipelineConfigRestHandlerImpl) UpgradeForAllApps(w http.ResponseWr
 	response["failed"] = failedIds
 	common.WriteJsonResp(w, err, response, http.StatusOK)
 }
+
+func (handler PipelineConfigRestHandlerImpl) HandleCdPipelineBulkAction(w http.ResponseWriter, r *http.Request) {
+	decoder := json.NewDecoder(r.Body)
+	userId, err := handler.userAuthService.GetLoggedInUser(r)
+	if userId == 0 || err != nil {
+		common.WriteJsonResp(w, err, "Unauthorized User", http.StatusUnauthorized)
+		return
+	}
+	var cdPipelineBulkActionReq bean.CdBulkActionRequestDto
+	err = decoder.Decode(&cdPipelineBulkActionReq)
+	cdPipelineBulkActionReq.UserId = userId
+	if err != nil {
+		handler.Logger.Errorw("request err, HandleCdPipelineBulkAction", "err", err, "payload", cdPipelineBulkActionReq)
+		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+		return
+	}
+
+	v := r.URL.Query()
+	forceDelete := false
+	forceDeleteParam := v.Get("forceDelete")
+	if len(forceDeleteParam) > 0 {
+		forceDelete, err = strconv.ParseBool(forceDeleteParam)
+		if err != nil {
+			handler.Logger.Errorw("request err, HandleCdPipelineBulkAction", "err", err, "payload", cdPipelineBulkActionReq)
+			common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+			return
+		}
+	}
+	cdPipelineBulkActionReq.ForceDelete = forceDelete
+
+	dryRun := false
+	dryRunParam := v.Get("dryRun")
+	if len(dryRunParam) > 0 {
+		dryRun, err = strconv.ParseBool(dryRunParam)
+		if err != nil {
+			handler.Logger.Errorw("request err, HandleCdPipelineBulkAction", "err", err, "payload", cdPipelineBulkActionReq)
+			common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+			return
+		}
+	}
+	handler.Logger.Infow("request payload, HandleCdPipelineBulkAction", "payload", cdPipelineBulkActionReq)
+	impactedPipelines, err := handler.pipelineBuilder.GetBulkActionImpactedPipelines(&cdPipelineBulkActionReq)
+	if err != nil {
+		handler.Logger.Errorw("service err, GetBulkActionImpactedPipelines", "err", err, "payload", cdPipelineBulkActionReq)
+		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
+		return
+	}
+	token := r.Header.Get("token")
+	for _, impactedPipeline := range impactedPipelines {
+		resourceName := handler.enforcerUtil.GetAppRBACName(impactedPipeline.App.AppName)
+		if ok := handler.enforcer.Enforce(token, casbin.ResourceApplications, casbin.ActionUpdate, resourceName); !ok {
+			common.WriteJsonResp(w, fmt.Errorf("unauthorized user"), "Unauthorized User", http.StatusForbidden)
+			return
+		}
+
+		object := handler.enforcerUtil.GetAppRBACByAppNameAndEnvId(impactedPipeline.App.AppName, impactedPipeline.EnvironmentId)
+		if ok := handler.enforcer.Enforce(token, casbin.ResourceEnvironment, casbin.ActionUpdate, object); !ok {
+			common.WriteJsonResp(w, fmt.Errorf("unauthorized user"), "Unauthorized User", http.StatusForbidden)
+			return
+		}
+	}
+	acdToken, err := handler.argoUserService.GetLatestDevtronArgoCdUserToken()
+	if err != nil {
+		handler.Logger.Errorw("error in getting acd token", "err", err)
+		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
+		return
+	}
+	ctx := context.WithValue(r.Context(), "token", acdToken)
+	resp, err := handler.pipelineBuilder.PerformBulkActionOnCdPipelines(&cdPipelineBulkActionReq, impactedPipelines, ctx, dryRun)
+	if err != nil {
+		handler.Logger.Errorw("service err, HandleCdPipelineBulkAction", "err", err, "payload", cdPipelineBulkActionReq)
+		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
+		return
+	}
+	common.WriteJsonResp(w, nil, resp, http.StatusOK)
+}

api/router/PipelineConfigRouter.go

@@ -70,7 +70,7 @@ func (router PipelineConfigRouterImpl) initPipelineConfigRouter(configRouter *mu
 	configRouter.Path("/cd-pipeline/patch").HandlerFunc(router.restHandler.PatchCdPipeline).Methods("POST")
 	configRouter.Path("/cd-pipeline/{appId}").HandlerFunc(router.restHandler.GetCdPipelines).Methods("GET")
 	configRouter.Path("/cd-pipeline/{appId}/env/{envId}").HandlerFunc(router.restHandler.GetCdPipelinesForAppAndEnv).Methods("GET")
-
+	configRouter.Path("/cd-pipeline/bulk-action").HandlerFunc(router.restHandler.HandleCdPipelineBulkAction).Methods("POST")
 	//save environment specific override
 	configRouter.Path("/env/{appId}/{environmentId}").HandlerFunc(router.restHandler.EnvConfigOverrideCreate).Methods("POST")
 	configRouter.Path("/env").HandlerFunc(router.restHandler.EnvConfigOverrideUpdate).Methods("PUT")

internal/sql/repository/DeploymentGroupRepository.go

@@ -32,6 +32,7 @@ type DeploymentGroupRepository interface {
 	Delete(model *DeploymentGroup) error
 	FindByIdWithApp(id int) (*DeploymentGroup, error)
 	FindByAppIdAndEnvId(envId, appId int) ([]DeploymentGroup, error)
+	GetNamesByAppIdAndEnvId(envId, appId int) ([]string, error)
 }
 
 type DeploymentGroupRepositoryImpl struct {
@@ -123,3 +124,14 @@ func (impl *DeploymentGroupRepositoryImpl) FindByAppIdAndEnvId(envId, appId int)
 	}
 	return models, err
 }
+
+func (impl *DeploymentGroupRepositoryImpl) GetNamesByAppIdAndEnvId(envId, appId int) ([]string, error) {
+	var groupNames []string
+	query := "select dg.name from deployment_group dg INNER JOIN deployment_group_app dga ON dga.deployment_group_id = dg.id where dga.active = ? and dga.app_id = ? and environment_id = ? and dg.active = ?;"
+	_, err := impl.dbConnection.Query(&groupNames, query, true, appId, envId, true)
+	if err != nil {
+		impl.Logger.Errorw("error in fetching group names by appId and envId", "err", err, "appId", appId, "envId", envId)
+		return nil, err
+	}
+	return groupNames, err
+}

internal/sql/repository/appWorkflow/AppWorkflowRepository.go

@@ -43,6 +43,7 @@ type AppWorkflowRepository interface {
 	FindWFCDMappingByCIPipelineId(ciPipelineId int) ([]*AppWorkflowMapping, error)
 	FindWFCDMappingByCDPipelineId(cdPipelineId int) ([]*AppWorkflowMapping, error)
 	DeleteAppWorkflowMapping(appWorkflow *AppWorkflowMapping, tx *pg.Tx) error
+	DeleteAppWorkflowMappingsByCdPipelineId(pipelineId int, tx *pg.Tx) error
 	FindWFCDMappingByCIPipelineIds(ciPipelineIds []int) ([]*AppWorkflowMapping, error)
 	FindWFCDMappingByParentCDPipelineId(cdPipelineId int) ([]*AppWorkflowMapping, error)
 	FindAllWFMappingsByAppId(appId int) ([]*AppWorkflowMapping, error)
@@ -286,3 +287,16 @@ func (impl AppWorkflowRepositoryImpl) FindAllWFMappingsByAppId(appId int) ([]*Ap
 		Select()
 	return appWorkflowsMapping, err
 }
+
+func (impl AppWorkflowRepositoryImpl) DeleteAppWorkflowMappingsByCdPipelineId(pipelineId int, tx *pg.Tx) error {
+	var model AppWorkflowMapping
+	_, err := tx.Model(&model).Set("active = ?", false).
+		Where("component_id = ?", pipelineId).
+		Where("type = ?", CDPIPELINE).
+		Update()
+	if err != nil {
+		impl.Logger.Errorw("error in deleting appWorkflowMapping by cdPipelineId", "err", err, "cdPipelineId", pipelineId)
+		return err
+	}
+	return nil
+}

internal/sql/repository/pipelineConfig/PipelineRepository.go

@@ -96,6 +96,8 @@ type PipelineRepository interface {
 	FindNumberOfAppsWithCdPipeline(appIds []int) (count int, err error)
 	GetAppAndEnvDetailsForDeploymentAppTypePipeline(deploymentAppType string, clusterIds []int) ([]*Pipeline, error)
 	GetPipelineIdsHavingStatusTimelinesPendingAfterKubectlApplyStatus(pendingSinceSeconds int) ([]int, error)
+	FindIdsByAppIdsAndEnvironmentIds(appIds, environmentIds []int) (ids []int, err error)
+	FindIdsByProjectIdsAndEnvironmentIds(projectIds, environmentIds []int) ([]int, error)
 }
 
 type CiArtifactDTO struct {
@@ -456,3 +458,25 @@ func (impl PipelineRepositoryImpl) GetPipelineIdsHavingStatusTimelinesPendingAft
 	}
 	return pipelineIds, nil
 }
+
+func (impl PipelineRepositoryImpl) FindIdsByAppIdsAndEnvironmentIds(appIds, environmentIds []int) ([]int, error) {
+	var pipelineIds []int
+	query := "select id from pipeline where app_id in (?) and environment_id in (?) and deleted = ?;"
+	_, err := impl.dbConnection.Query(&pipelineIds, query, pg.In(appIds), pg.In(environmentIds), false)
+	if err != nil {
+		impl.logger.Errorw("error in getting pipelineIds by appIds and envIds", "err", err, "appIds", appIds, "envIds", environmentIds)
+		return pipelineIds, err
+	}
+	return pipelineIds, err
+}
+
+func (impl PipelineRepositoryImpl) FindIdsByProjectIdsAndEnvironmentIds(projectIds, environmentIds []int) ([]int, error) {
+	var pipelineIds []int
+	query := "select p.id from pipeline p inner join app a on a.id=p.app_id where a.team_id in (?) and p.environment_id in (?) and p.deleted = ? and a.active = ?;"
+	_, err := impl.dbConnection.Query(&pipelineIds, query, pg.In(projectIds), pg.In(environmentIds), false, true)
+	if err != nil {
+		impl.logger.Errorw("error in getting pipelineIds by projectIds and envIds", "err", err, "projectIds", projectIds, "envIds", environmentIds)
+		return pipelineIds, err
+	}
+	return pipelineIds, err
+}

pkg/bean/app.go

@@ -598,3 +598,25 @@ type UpdateProjectBulkAppsRequest struct {
 	TeamId int   `json:"teamId"`
 	UserId int32 `json:"-"`
 }
+
+type CdBulkAction int
+
+const (
+	CD_BULK_DELETE CdBulkAction = iota
+)
+
+type CdBulkActionRequestDto struct {
+	Action      CdBulkAction `json:"action"`
+	EnvIds      []int        `json:"envIds"`
+	AppIds      []int        `json:"appIds"`
+	ProjectIds  []int        `json:"projectIds"`
+	ForceDelete bool         `json:"forceDelete"`
+	UserId      int32        `json:"-"`
+}
+
+type CdBulkActionResponseDto struct {
+	PipelineName    string `json:"pipelineName"`
+	AppName         string `json:"appName"`
+	EnvironmentName string `json:"environmentName"`
+	DeletionResult  string `json:"deletionResult,omitempty"`
+}

pkg/pipeline/PipelineBuilder.go

@@ -108,6 +108,9 @@ type PipelineBuilder interface {
 	FindAllMatchesByAppName(appName string) ([]*AppBean, error)
 	GetEnvironmentByCdPipelineId(pipelineId int) (int, error)
 	PatchRegexCiPipeline(request *bean.CiRegexPatchRequest) (err error)
+
+	GetBulkActionImpactedPipelines(dto *bean.CdBulkActionRequestDto) ([]*pipelineConfig.Pipeline, error)
+	PerformBulkActionOnCdPipelines(dto *bean.CdBulkActionRequestDto, impactedPipelines []*pipelineConfig.Pipeline, ctx context.Context, dryRun bool) ([]*bean.CdBulkActionResponseDto, error)
 }
 
 type PipelineBuilderImpl struct {
@@ -1225,38 +1228,34 @@ func (impl PipelineBuilderImpl) PatchCdPipelines(cdPipelines *bean.CDPatchReques
 		err := impl.updateCdPipeline(ctx, cdPipelines.Pipeline, cdPipelines.UserId)
 		return pipelineRequest, err
 	case bean.CD_DELETE:
-		err := impl.deleteCdPipeline(cdPipelines.Pipeline.Id, cdPipelines.UserId, ctx, cdPipelines.ForceDelete)
+		pipeline, err := impl.pipelineRepository.FindById(cdPipelines.Pipeline.Id)
+		if err != nil {
+			impl.logger.Errorw("error in getting cd pipeline by id", "err", err, "id", cdPipelines.Pipeline.Id)
+			return pipelineRequest, err
+		}
+		err = impl.deleteCdPipeline(pipeline, ctx, cdPipelines.ForceDelete)
 		return pipelineRequest, err
 	default:
 		return nil, &util.ApiError{Code: "404", HttpStatusCode: 404, UserMessage: "operation not supported"}
 	}
 }
 
-func (impl PipelineBuilderImpl) deleteCdPipeline(pipelineId int, userId int32, ctx context.Context, forceDelete bool) (err error) {
+func (impl PipelineBuilderImpl) deleteCdPipeline(pipeline *pipelineConfig.Pipeline, ctx context.Context, forceDelete bool) (err error) {
 	//getting children CD pipeline details
-	appWorkflowMapping, err := impl.appWorkflowRepository.FindWFCDMappingByParentCDPipelineId(pipelineId)
+	appWorkflowMapping, err := impl.appWorkflowRepository.FindWFCDMappingByParentCDPipelineId(pipeline.Id)
 	if err != nil && err != pg.ErrNoRows {
 		impl.logger.Errorw("error in getting children cd details", "err", err)
 		return err
 	} else if len(appWorkflowMapping) > 0 {
 		impl.logger.Debugw("cannot delete cd pipeline, contains children cd")
 		return fmt.Errorf("Please delete children CD pipelines before deleting this pipeline.")
 	}
-	pipeline, err := impl.pipelineRepository.FindById(pipelineId)
-	if err != nil {
-		impl.logger.Errorw("err in fetching pipeline", "id", pipelineId, "err", err)
-		return err
-	}
 	//getting deployment group for this pipeline
-	deploymentGroups, err := impl.deploymentGroupRepository.FindByAppIdAndEnvId(pipeline.EnvironmentId, pipeline.AppId)
+	deploymentGroupNames, err := impl.deploymentGroupRepository.GetNamesByAppIdAndEnvId(pipeline.EnvironmentId, pipeline.AppId)
 	if err != nil && err != pg.ErrNoRows {
-		impl.logger.Errorw("error in getting deployment groups by envId", "err", err)
+		impl.logger.Errorw("error in getting deployment group names by appId and envId", "err", err)
 		return err
-	} else if len(deploymentGroups) > 0 {
-		var deploymentGroupNames []string
-		for _, group := range deploymentGroups {
-			deploymentGroupNames = append(deploymentGroupNames, group.Name)
-		}
+	} else if len(deploymentGroupNames) > 0 {
 		groupNamesByte, err := json.Marshal(deploymentGroupNames)
 		if err != nil {
 			impl.logger.Errorw("error in marshaling deployment group names", "err", err, "deploymentGroupNames", deploymentGroupNames)
@@ -1271,19 +1270,16 @@ func (impl PipelineBuilderImpl) deleteCdPipeline(pipelineId int, userId int32, c
 	}
 	// Rollback tx on error.
 	defer tx.Rollback()
-	if err = impl.dbPipelineOrchestrator.DeleteCdPipeline(pipelineId, tx); err != nil {
+	if err = impl.dbPipelineOrchestrator.DeleteCdPipeline(pipeline.Id, tx); err != nil {
 		impl.logger.Errorw("err in deleting pipeline from db", "id", pipeline, "err", err)
 		return err
 	}
 
 	//delete app workflow mapping
-	appWorkflowMapping, err = impl.appWorkflowRepository.FindWFCDMappingByCDPipelineId(pipelineId)
-	for _, mapping := range appWorkflowMapping {
-		err := impl.appWorkflowRepository.DeleteAppWorkflowMapping(mapping, tx)
-		if err != nil {
-			impl.logger.Errorw("error in deleting workflow mapping", "err", err)
-			return err
-		}
+	err = impl.appWorkflowRepository.DeleteAppWorkflowMappingsByCdPipelineId(pipeline.Id, tx)
+	if err != nil {
+		impl.logger.Errorw("error in deleting workflow mapping", "err", err)
+		return err
 	}
 
 	if pipeline.PreStageConfig != "" {
@@ -1331,7 +1327,7 @@ func (impl PipelineBuilderImpl) deleteCdPipeline(pipelineId int, userId int32, c
 					return err
 				}
 			}
-			impl.logger.Infow("app deleted from argocd", "id", pipelineId, "pipelineName", pipeline.Name, "app", deploymentAppName)
+			impl.logger.Infow("app deleted from argocd", "id", pipeline.Id, "pipelineName", pipeline.Name, "app", deploymentAppName)
 		} else if util.IsHelmApp(pipeline.DeploymentAppType) {
 			appIdentifier := &client.AppIdentifier{
 				ClusterId:   pipeline.Environment.ClusterId,
@@ -2520,3 +2516,77 @@ func (impl PipelineBuilderImpl) updateGitRepoUrlInCharts(appId int, chartGitAttr
 	}
 	return nil
 }
+
+func (impl PipelineBuilderImpl) PerformBulkActionOnCdPipelines(dto *bean.CdBulkActionRequestDto, impactedPipelines []*pipelineConfig.Pipeline, ctx context.Context, dryRun bool) ([]*bean.CdBulkActionResponseDto, error) {
+	switch dto.Action {
+	case bean.CD_BULK_DELETE:
+		bulkDeleteResp := impl.BulkDeleteCdPipelines(impactedPipelines, ctx, dryRun, dto.ForceDelete)
+		return bulkDeleteResp, nil
+	default:
+		return nil, &util.ApiError{Code: "400", HttpStatusCode: 400, UserMessage: "this action is not supported"}
+	}
+}
+
+func (impl PipelineBuilderImpl) BulkDeleteCdPipelines(impactedPipelines []*pipelineConfig.Pipeline, ctx context.Context, dryRun, forceDelete bool) []*bean.CdBulkActionResponseDto {
+	var respDtos []*bean.CdBulkActionResponseDto
+	for _, pipeline := range impactedPipelines {
+		respDto := &bean.CdBulkActionResponseDto{
+			PipelineName:    pipeline.Name,
+			AppName:         pipeline.App.AppName,
+			EnvironmentName: pipeline.Environment.Name,
+		}
+		if !dryRun {
+			err := impl.deleteCdPipeline(pipeline, ctx, forceDelete)
+			if err != nil {
+				impl.logger.Errorw("error in deleting cd pipeline", "err", err, "pipelineId", pipeline.Id)
+				respDto.DeletionResult = fmt.Sprintf("Not able to delete pipeline, %v", err)
+			} else {
+				respDto.DeletionResult = "Pipeline deleted successfully."
+			}
+		}
+		respDtos = append(respDtos, respDto)
+	}
+	return respDtos
+
+}
+
+func (impl PipelineBuilderImpl) GetBulkActionImpactedPipelines(dto *bean.CdBulkActionRequestDto) ([]*pipelineConfig.Pipeline, error) {
+	if len(dto.EnvIds) == 0 || (len(dto.AppIds) == 0 && len(dto.ProjectIds) == 0) {
+		//invalid payload, envIds are must and either of appIds or projectIds are must
+		return nil, &util.ApiError{Code: "400", HttpStatusCode: 400, UserMessage: "invalid payload, can not get pipelines for this filter"}
+	}
+	var pipelineIdsByAppLevel []int
+	var pipelineIdsByProjectLevel []int
+	var err error
+	if len(dto.AppIds) > 0 && len(dto.EnvIds) > 0 {
+		//getting pipeline IDs for app level deletion request
+		pipelineIdsByAppLevel, err = impl.pipelineRepository.FindIdsByAppIdsAndEnvironmentIds(dto.AppIds, dto.EnvIds)
+		if err != nil && err != pg.ErrNoRows {
+			impl.logger.Errorw("error in getting cd pipelines by appIds and envIds", "err", err)
+			return nil, err
+		}
+	}
+	if len(dto.ProjectIds) > 0 && len(dto.EnvIds) > 0 {
+		//getting pipeline IDs for project level deletion request
+		pipelineIdsByProjectLevel, err = impl.pipelineRepository.FindIdsByProjectIdsAndEnvironmentIds(dto.ProjectIds, dto.EnvIds)
+		if err != nil && err != pg.ErrNoRows {
+			impl.logger.Errorw("error in getting cd pipelines by projectIds and envIds", "err", err)
+			return nil, err
+		}
+	}
+	var pipelineIdsMerged []int
+	//it might be possible that pipelineIdsByAppLevel & pipelineIdsByProjectLevel have some same values
+	//we are still appending them to save operation cost of checking same ids as we will get pipelines from
+	//in clause which gives correct results even if some values are repeating
+	pipelineIdsMerged = append(pipelineIdsMerged, pipelineIdsByAppLevel...)
+	pipelineIdsMerged = append(pipelineIdsMerged, pipelineIdsByProjectLevel...)
+	var pipelines []*pipelineConfig.Pipeline
+	if len(pipelineIdsMerged) > 0 {
+		pipelines, err = impl.pipelineRepository.FindByIdsIn(pipelineIdsMerged)
+		if err != nil {
+			impl.logger.Errorw("error in getting cd pipelines by ids", "err", err, "ids", pipelineIdsMerged)
+			return nil, err
+		}
+	}
+	return pipelines, nil
+}