feat: Information of Linked CI Pipelines on Parent CI Pipeline (#4786)

* Refactor ArgoUserService.go by adding a TODO

* wip

* updated API spec

* Update repo and service

* Add logging of errors

* modified API spec

* added total count in query

* updated service and rest handler layer

* Update router and resthandler

* set default req params at rest handler

* moved generics to global util

* Update resthandler

* added tracing for new queries

* fixed undefined ctx

* Update handler's rbac

* update error logs

* Update cipipeline repo

* added: comments for changes

* Change handler

* fixed typo and imports

* fixed adapter and constants

* updated test file errors and mock files

* updated API specs

* fixed API specs end points

* fixed: query errors

* fixed: linked cd condition

* fixed: typo

* fixed: search

* fixed nil check for runner query

* fixed: append in adapter

* fix searchkey to lowercase

* Fix duplicated env names

* fix pass env array as empty

* Fix error logs

* fix error

* fixed import

---------

Co-authored-by: komalreddy3 <[email protected]>

api/appbean/AppDetail.go

@@ -5,6 +5,7 @@ import (
 	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig"
 	"github.com/devtron-labs/devtron/pkg/chartRepo/repository"
 	"github.com/devtron-labs/devtron/pkg/pipeline/bean"
+	"github.com/devtron-labs/devtron/pkg/pipeline/bean/CiPipeline"
 )
 
 type AppDetail struct {
@@ -45,11 +46,11 @@ type GitMaterial struct {
 }
 
 type DockerConfig struct {
-	DockerRegistry    string                  `json:"dockerRegistry" validate:"required"`
-	DockerRepository  string                  `json:"dockerRepository" validate:"required"`
-	CiBuildConfig     *bean.CiBuildConfigBean `json:"ciBuildConfig"`
-	DockerBuildConfig *DockerBuildConfig      `json:"dockerBuildConfig,omitempty"` // Deprecated, should use CiBuildConfig for development
-	CheckoutPath      string                  `json:"checkoutPath"`
+	DockerRegistry    string                        `json:"dockerRegistry" validate:"required"`
+	DockerRepository  string                        `json:"dockerRepository" validate:"required"`
+	CiBuildConfig     *CiPipeline.CiBuildConfigBean `json:"ciBuildConfig"`
+	DockerBuildConfig *DockerBuildConfig            `json:"dockerBuildConfig,omitempty"` // Deprecated, should use CiBuildConfig for development
+	CheckoutPath      string                        `json:"checkoutPath"`
 }
 
 type DockerBuildConfig struct {

api/restHandler/CoreAppRestHandler.go

@@ -23,6 +23,7 @@ import (
 	"errors"
 	"fmt"
 	app2 "github.com/devtron-labs/devtron/api/restHandler/app/pipeline/configure"
+	"github.com/devtron-labs/devtron/pkg/pipeline/bean/CiPipeline"
 	"net/http"
 	"strconv"
 	"strings"
@@ -1322,9 +1323,9 @@ func (handler CoreAppRestHandlerImpl) createDockerConfig(appId int, dockerConfig
 	dockerBuildConfig := dockerConfig.DockerBuildConfig
 	if dockerBuildConfig != nil {
 		dockerConfig.CheckoutPath = dockerBuildConfig.GitCheckoutPath
-		dockerConfig.CiBuildConfig = &bean2.CiBuildConfigBean{
-			CiBuildType: bean2.SELF_DOCKERFILE_BUILD_TYPE,
-			DockerBuildConfig: &bean2.DockerBuildConfig{
+		dockerConfig.CiBuildConfig = &CiPipeline.CiBuildConfigBean{
+			CiBuildType: CiPipeline.SELF_DOCKERFILE_BUILD_TYPE,
+			DockerBuildConfig: &CiPipeline.DockerBuildConfig{
 				DockerfilePath:     dockerBuildConfig.DockerfileRelativePath,
 				DockerBuildOptions: dockerBuildConfig.DockerBuildOptions,
 				Args:               dockerBuildConfig.Args,
@@ -1544,7 +1545,7 @@ func (handler CoreAppRestHandlerImpl) createWorkflows(ctx context.Context, appId
 		//Creating CI pipeline starts
 		ciPipeline, err := handler.createCiPipeline(appId, userId, workflowId, workflow.CiPipeline)
 		if err != nil {
-			if err.Error() == bean2.PIPELINE_NAME_ALREADY_EXISTS_ERROR {
+			if err.Error() == CiPipeline.PIPELINE_NAME_ALREADY_EXISTS_ERROR {
 				handler.logger.Errorw("service err, DeleteAppWorkflow ", "err", err)
 				return err, http.StatusBadRequest
 			}
@@ -1672,7 +1673,7 @@ func (handler CoreAppRestHandlerImpl) createCiPipeline(appId int, userId int32,
 			ParentCiPipeline:         ciPipelineData.ParentCiPipeline,
 			ParentAppId:              ciPipelineData.ParentAppId,
 			LinkedCount:              ciPipelineData.LinkedCount,
-			PipelineType:             bean2.PipelineType(ciPipelineData.PipelineType),
+			PipelineType:             CiPipeline.PipelineType(ciPipelineData.PipelineType),
 		},
 	}
 

api/restHandler/app/pipeline/configure/PipelineConfigRestHandler.go

@@ -26,6 +26,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/deployment/manifest/deployedAppMetrics"
 	"github.com/devtron-labs/devtron/pkg/deployment/manifest/deploymentTemplate"
 	"github.com/devtron-labs/devtron/pkg/deployment/manifest/deploymentTemplate/chartRef"
+	bean3 "github.com/devtron-labs/devtron/pkg/pipeline/bean/CiPipeline"
 	"io"
 	"net/http"
 	"strconv"
@@ -39,7 +40,6 @@ import (
 	"github.com/devtron-labs/devtron/pkg/auth/user"
 	"github.com/devtron-labs/devtron/pkg/chart"
 	"github.com/devtron-labs/devtron/pkg/generateManifest"
-	bean3 "github.com/devtron-labs/devtron/pkg/pipeline/bean"
 	resourceGroup2 "github.com/devtron-labs/devtron/pkg/resourceGroup"
 	"github.com/devtron-labs/devtron/util/argo"
 	"github.com/go-pg/pg"
@@ -129,6 +129,7 @@ type PipelineConfigRestHandlerImpl struct {
 	ciArtifactRepository                repository.CiArtifactRepository
 	deployedAppMetricsService           deployedAppMetrics.DeployedAppMetricsService
 	chartRefService                     chartRef.ChartRefService
+	ciCdPipelineOrchestrator            pipeline.CiCdPipelineOrchestrator
 }
 
 func NewPipelineRestHandlerImpl(pipelineBuilder pipeline.PipelineBuilder, Logger *zap.SugaredLogger,
@@ -155,7 +156,8 @@ func NewPipelineRestHandlerImpl(pipelineBuilder pipeline.PipelineBuilder, Logger
 	imageTaggingService pipeline.ImageTaggingService,
 	ciArtifactRepository repository.CiArtifactRepository,
 	deployedAppMetricsService deployedAppMetrics.DeployedAppMetricsService,
-	chartRefService chartRef.ChartRefService) *PipelineConfigRestHandlerImpl {
+	chartRefService chartRef.ChartRefService,
+	ciCdPipelineOrchestrator pipeline.CiCdPipelineOrchestrator) *PipelineConfigRestHandlerImpl {
 	envConfig := &PipelineRestHandlerEnvConfig{}
 	err := env.Parse(envConfig)
 	if err != nil {
@@ -193,6 +195,7 @@ func NewPipelineRestHandlerImpl(pipelineBuilder pipeline.PipelineBuilder, Logger
 		ciArtifactRepository:                ciArtifactRepository,
 		deployedAppMetricsService:           deployedAppMetricsService,
 		chartRefService:                     chartRefService,
+		ciCdPipelineOrchestrator:            ciCdPipelineOrchestrator,
 	}
 }
 
@@ -202,7 +205,7 @@ const (
 	HTTPS_URL_PREFIX = "https://"
 )
 
-func (handler PipelineConfigRestHandlerImpl) DeleteApp(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) DeleteApp(w http.ResponseWriter, r *http.Request) {
 	token := r.Header.Get("token")
 	userId, err := handler.userAuthService.GetLoggedInUser(r)
 	if userId == 0 || err != nil {
@@ -244,7 +247,7 @@ func (handler PipelineConfigRestHandlerImpl) DeleteApp(w http.ResponseWriter, r
 	common.WriteJsonResp(w, err, nil, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) DeleteACDAppWithNonCascade(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) DeleteACDAppWithNonCascade(w http.ResponseWriter, r *http.Request) {
 	token := r.Header.Get("token")
 	userId, err := handler.userAuthService.GetLoggedInUser(r)
 	if userId == 0 || err != nil {
@@ -324,7 +327,7 @@ func (handler PipelineConfigRestHandlerImpl) DeleteACDAppWithNonCascade(w http.R
 	common.WriteJsonResp(w, err, nil, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) CreateApp(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) CreateApp(w http.ResponseWriter, r *http.Request) {
 	token := r.Header.Get("token")
 	decoder := json.NewDecoder(r.Body)
 	userId, err := handler.userAuthService.GetLoggedInUser(r)
@@ -400,7 +403,7 @@ func (handler PipelineConfigRestHandlerImpl) CreateApp(w http.ResponseWriter, r
 	common.WriteJsonResp(w, err, createResp, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) GetApp(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) GetApp(w http.ResponseWriter, r *http.Request) {
 	token := r.Header.Get("token")
 	vars := mux.Vars(r)
 	appId, err := strconv.Atoi(vars["appId"])
@@ -429,7 +432,7 @@ func (handler PipelineConfigRestHandlerImpl) GetApp(w http.ResponseWriter, r *ht
 	common.WriteJsonResp(w, err, ciConf, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) FindAppsByTeamId(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) FindAppsByTeamId(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	teamId, err := strconv.Atoi(vars["teamId"])
 	if err != nil {
@@ -447,7 +450,7 @@ func (handler PipelineConfigRestHandlerImpl) FindAppsByTeamId(w http.ResponseWri
 	common.WriteJsonResp(w, err, project, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) FindAppsByTeamName(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) FindAppsByTeamName(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	teamName := vars["teamName"]
 	handler.Logger.Infow("request payload, FindAppsByTeamName", "teamName", teamName)
@@ -564,7 +567,7 @@ func (handler *PipelineConfigRestHandlerImpl) sendData(event []byte, w http.Resp
 	}
 }
 
-func (handler PipelineConfigRestHandlerImpl) FetchAppWorkflowStatusForTriggerView(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) FetchAppWorkflowStatusForTriggerView(w http.ResponseWriter, r *http.Request) {
 	userId, err := handler.userAuthService.GetLoggedInUser(r)
 	if userId == 0 || err != nil {
 		common.WriteJsonResp(w, err, "Unauthorized User", http.StatusUnauthorized)
@@ -637,7 +640,7 @@ func (handler PipelineConfigRestHandlerImpl) FetchAppWorkflowStatusForTriggerVie
 	common.WriteJsonResp(w, err, triggerWorkflowStatus, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) PipelineNameSuggestion(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) PipelineNameSuggestion(w http.ResponseWriter, r *http.Request) {
 	token := r.Header.Get("token")
 	vars := mux.Vars(r)
 	appId, err := strconv.Atoi(vars["appId"])
@@ -663,7 +666,7 @@ func (handler PipelineConfigRestHandlerImpl) PipelineNameSuggestion(w http.Respo
 	common.WriteJsonResp(w, err, suggestedName, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) FetchAppWorkflowStatusForTriggerViewByEnvironment(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) FetchAppWorkflowStatusForTriggerViewByEnvironment(w http.ResponseWriter, r *http.Request) {
 	token := r.Header.Get("token")
 	userId, err := handler.userAuthService.GetLoggedInUser(r)
 	if userId == 0 || err != nil {
@@ -741,7 +744,7 @@ func (handler PipelineConfigRestHandlerImpl) FetchAppWorkflowStatusForTriggerVie
 	common.WriteJsonResp(w, err, triggerWorkflowStatus, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) GetEnvironmentListWithAppData(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) GetEnvironmentListWithAppData(w http.ResponseWriter, r *http.Request) {
 	v := r.URL.Query()
 	token := r.Header.Get("token")
 	envName := v.Get("envName")
@@ -779,7 +782,7 @@ func (handler PipelineConfigRestHandlerImpl) GetEnvironmentListWithAppData(w htt
 	common.WriteJsonResp(w, err, result, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) GetApplicationsByEnvironment(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) GetApplicationsByEnvironment(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	token := r.Header.Get("token")
 	userId, err := handler.userAuthService.GetLoggedInUser(r)
@@ -834,7 +837,7 @@ func (handler PipelineConfigRestHandlerImpl) GetApplicationsByEnvironment(w http
 	common.WriteJsonResp(w, err, results, http.StatusOK)
 }
 
-func (handler PipelineConfigRestHandlerImpl) FetchAppDeploymentStatusForEnvironments(w http.ResponseWriter, r *http.Request) {
+func (handler *PipelineConfigRestHandlerImpl) FetchAppDeploymentStatusForEnvironments(w http.ResponseWriter, r *http.Request) {
 	token := r.Header.Get("token")
 	userId, err := handler.userAuthService.GetLoggedInUser(r)
 	if userId == 0 || err != nil {

api/router/app/pipeline/configure/PipelineConfigRouter.go

@@ -85,6 +85,9 @@ func (router PipelineConfigRouterImpl) InitPipelineConfigRouter(configRouter *mu
 	configRouter.Path("/ci-pipeline/bulk/branch-update").HandlerFunc(router.restHandler.PatchCiMaterialSourceWithAppIdsAndEnvironmentId).Methods("PUT")
 	configRouter.Path("/ci-pipeline/patch/regex").HandlerFunc(router.restHandler.UpdateBranchCiPipelinesWithRegex).Methods("POST")
 
+	configRouter.Path("/ci-pipeline/{ciPipelineId}/linked-ci/downstream/env").HandlerFunc(router.restHandler.GetSourceCiDownStreamFilters).Methods("GET")
+	configRouter.Path("/ci-pipeline/{ciPipelineId}/linked-ci/downstream/cd").HandlerFunc(router.restHandler.GetSourceCiDownStreamInfo).Methods("GET")
+
 	configRouter.Path("/cd-pipeline/{cd_pipeline_id}/material").HandlerFunc(router.restHandler.GetArtifactsByCDPipeline).Methods("GET")
 	configRouter.Path("/cd-pipeline/{cd_pipeline_id}/material/rollback").HandlerFunc(router.restHandler.GetArtifactsForRollback).Methods("GET")
 

client/cron/CiTriggerCron.go

@@ -6,7 +6,7 @@ import (
 	repository2 "github.com/devtron-labs/devtron/internal/sql/repository"
 	"github.com/devtron-labs/devtron/pkg/bean"
 	"github.com/devtron-labs/devtron/pkg/pipeline"
-	pipelineConfigBean "github.com/devtron-labs/devtron/pkg/pipeline/bean"
+	pipelineConfigBean "github.com/devtron-labs/devtron/pkg/pipeline/bean/CiPipeline"
 	"github.com/devtron-labs/devtron/pkg/pipeline/repository"
 	repository3 "github.com/devtron-labs/devtron/pkg/plugin/repository"
 	cron2 "github.com/devtron-labs/devtron/util/cron"

client/telemetry/TelemetryEventClientExtended.go

@@ -5,6 +5,7 @@ import (
 	cloudProviderIdentifier "github.com/devtron-labs/common-lib/cloud-provider-identifier"
 	client "github.com/devtron-labs/devtron/api/helm-app/gRPC"
 	"github.com/devtron-labs/devtron/pkg/deployment/gitOps/config"
+	"github.com/devtron-labs/devtron/pkg/pipeline/bean/CiPipeline"
 	cron3 "github.com/devtron-labs/devtron/util/cron"
 	"net/http"
 	"time"
@@ -21,7 +22,6 @@ import (
 	"github.com/devtron-labs/devtron/pkg/cluster"
 	moduleRepo "github.com/devtron-labs/devtron/pkg/module/repo"
 	"github.com/devtron-labs/devtron/pkg/pipeline"
-	"github.com/devtron-labs/devtron/pkg/pipeline/bean"
 	serverDataStore "github.com/devtron-labs/devtron/pkg/server/store"
 	util3 "github.com/devtron-labs/devtron/pkg/util"
 	"github.com/devtron-labs/devtron/util"
@@ -345,16 +345,16 @@ func (impl *TelemetryEventClientImplExtended) SendSummaryEvent(eventType string)
 	}
 
 	payload.SelfDockerfileCount = selfDockerfileCount
-	payload.SelfDockerfileSuccessCount = successCount[bean.SELF_DOCKERFILE_BUILD_TYPE]
-	payload.SelfDockerfileFailureCount = failureCount[bean.SELF_DOCKERFILE_BUILD_TYPE]
+	payload.SelfDockerfileSuccessCount = successCount[CiPipeline.SELF_DOCKERFILE_BUILD_TYPE]
+	payload.SelfDockerfileFailureCount = failureCount[CiPipeline.SELF_DOCKERFILE_BUILD_TYPE]
 
 	payload.ManagedDockerfileCount = managedDockerfileCount
-	payload.ManagedDockerfileSuccessCount = successCount[bean.MANAGED_DOCKERFILE_BUILD_TYPE]
-	payload.ManagedDockerfileFailureCount = failureCount[bean.MANAGED_DOCKERFILE_BUILD_TYPE]
+	payload.ManagedDockerfileSuccessCount = successCount[CiPipeline.MANAGED_DOCKERFILE_BUILD_TYPE]
+	payload.ManagedDockerfileFailureCount = failureCount[CiPipeline.MANAGED_DOCKERFILE_BUILD_TYPE]
 
 	payload.BuildPackCount = buildpackCount
-	payload.BuildPackSuccessCount = successCount[bean.BUILDPACK_BUILD_TYPE]
-	payload.BuildPackFailureCount = failureCount[bean.BUILDPACK_BUILD_TYPE]
+	payload.BuildPackSuccessCount = successCount[CiPipeline.BUILDPACK_BUILD_TYPE]
+	payload.BuildPackFailureCount = failureCount[CiPipeline.BUILDPACK_BUILD_TYPE]
 
 	reqBody, err := json.Marshal(payload)
 	if err != nil {
@@ -378,24 +378,24 @@ func (impl *TelemetryEventClientImplExtended) SendSummaryEvent(eventType string)
 
 func (impl *TelemetryEventClientImplExtended) getCiBuildTypeData() (int, int, int) {
 	countByBuildType := impl.ciBuildConfigService.GetCountByBuildType()
-	return countByBuildType[bean.SELF_DOCKERFILE_BUILD_TYPE], countByBuildType[bean.MANAGED_DOCKERFILE_BUILD_TYPE], countByBuildType[bean.BUILDPACK_BUILD_TYPE]
+	return countByBuildType[CiPipeline.SELF_DOCKERFILE_BUILD_TYPE], countByBuildType[CiPipeline.MANAGED_DOCKERFILE_BUILD_TYPE], countByBuildType[CiPipeline.BUILDPACK_BUILD_TYPE]
 }
 
-func (impl *TelemetryEventClientImplExtended) getCiBuildTypeVsStatusVsCount() (successCount map[bean.CiBuildType]int, failureCount map[bean.CiBuildType]int) {
-	successCount = make(map[bean.CiBuildType]int)
-	failureCount = make(map[bean.CiBuildType]int)
+func (impl *TelemetryEventClientImplExtended) getCiBuildTypeVsStatusVsCount() (successCount map[CiPipeline.CiBuildType]int, failureCount map[CiPipeline.CiBuildType]int) {
+	successCount = make(map[CiPipeline.CiBuildType]int)
+	failureCount = make(map[CiPipeline.CiBuildType]int)
 	buildTypeAndStatusVsCount := impl.ciWorkflowRepository.FindBuildTypeAndStatusDataOfLast1Day()
 	for _, buildTypeCount := range buildTypeAndStatusVsCount {
 		if buildTypeCount == nil {
 			continue
 		}
 		if buildTypeCount.Type == "" {
-			buildTypeCount.Type = string(bean.SELF_DOCKERFILE_BUILD_TYPE)
+			buildTypeCount.Type = string(CiPipeline.SELF_DOCKERFILE_BUILD_TYPE)
 		}
 		if buildTypeCount.Status == "Succeeded" {
-			successCount[bean.CiBuildType(buildTypeCount.Type)] = buildTypeCount.Count
+			successCount[CiPipeline.CiBuildType(buildTypeCount.Type)] = buildTypeCount.Count
 		} else {
-			failureCount[bean.CiBuildType(buildTypeCount.Type)] = buildTypeCount.Count
+			failureCount[CiPipeline.CiBuildType(buildTypeCount.Type)] = buildTypeCount.Count
 		}
 	}
 	return successCount, failureCount

internal/sql/repository/pipelineConfig/CdWorfkflowRepository.go

@@ -75,6 +75,7 @@ type CdWorkflowRepository interface {
 
 	FetchArtifactsByCdPipelineId(pipelineId int, runnerType bean.WorkflowType, offset, limit int, searchString string) ([]CdWorkflowRunner, error)
 	GetLatestTriggersOfHelmPipelinesStuckInNonTerminalStatuses(getPipelineDeployedWithinHours int) ([]*CdWorkflowRunner, error)
+	FindLatestRunnerByPipelineIdsAndRunnerType(ctx context.Context, pipelineIds []int, runnerType bean.WorkflowType) ([]CdWorkflowRunner, error)
 }
 
 type CdWorkflowRepositoryImpl struct {
@@ -739,3 +740,26 @@ func (impl *CdWorkflowRepositoryImpl) CheckWorkflowRunnerByReferenceId(reference
 	}
 	return exists, err
 }
+
+func (impl *CdWorkflowRepositoryImpl) FindLatestRunnerByPipelineIdsAndRunnerType(ctx context.Context, pipelineIds []int, runnerType bean.WorkflowType) ([]CdWorkflowRunner, error) {
+	_, span := otel.Tracer("orchestrator").Start(ctx, "FindLatestRunnerByPipelineIdsAndRunnerType")
+	defer span.End()
+	if pipelineIds == nil || len(pipelineIds) == 0 {
+		return nil, pg.ErrNoRows
+	}
+	var latestWfrs []CdWorkflowRunner
+	err := impl.dbConnection.
+		Model(&latestWfrs).
+		Column("cd_workflow_runner.*", "CdWorkflow", "CdWorkflow.Pipeline").
+		ColumnExpr("MAX(cd_workflow_runner.id)").
+		Where("cd_workflow.pipeline_id IN (?)", pg.In(pipelineIds)).
+		Where("cd_workflow_runner.workflow_type = ?", runnerType).
+		Where("cd_workflow__pipeline.deleted = ?", false).
+		Group("cd_workflow_runner.id", "cd_workflow.id", "cd_workflow__pipeline.id").
+		Select()
+	if err != nil {
+		impl.logger.Errorw("error in getting cdWfr by appId, envId and runner type", "pipelineIds", pipelineIds, "runnerType", runnerType)
+		return nil, err
+	}
+	return latestWfrs, err
+}