feat: making pre,post,deploy triggers flows idempotent

api/restHandler/PipelineTriggerRestHandler.go

@@ -129,7 +129,7 @@ func (handler PipelineTriggerRestHandlerImpl) OverrideConfig(w http.ResponseWrit
 	}
 	ctx := context.WithValue(r.Context(), "token", acdToken)
 	_, span := otel.Tracer("orchestrator").Start(ctx, "workflowDagExecutor.ManualCdTrigger")
-	mergeResp, err := handler.workflowDagExecutor.ManualCdTrigger(&overrideRequest, ctx)
+	mergeResp, err := handler.workflowDagExecutor.ManualCdTrigger(ctx, &overrideRequest, nil)
 	span.End()
 	if err != nil {
 		handler.logger.Errorw("request err, OverrideConfig", "err", err, "payload", overrideRequest)
@@ -224,7 +224,7 @@ func (handler PipelineTriggerRestHandlerImpl) StartStopApp(w http.ResponseWriter
 		return
 	}
 	ctx := context.WithValue(r.Context(), "token", acdToken)
-	mergeResp, err := handler.workflowDagExecutor.StopStartApp(&overrideRequest, ctx)
+	mergeResp, err := handler.workflowDagExecutor.StopStartApp(ctx, &overrideRequest, nil)
 	if err != nil {
 		handler.logger.Errorw("service err, StartStopApp", "err", err, "payload", overrideRequest)
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)

api/router/pubsub/ApplicationStatusHandler.go

@@ -74,7 +74,7 @@ func NewApplicationStatusHandlerImpl(logger *zap.SugaredLogger, pubsubClient *pu
 	}
 	err := appStatusUpdateHandlerImpl.Subscribe()
 	if err != nil {
-		//logger.Error("err", err)
+		// logger.Error("err", err)
 		return nil
 	}
 	err = appStatusUpdateHandlerImpl.SubscribeDeleteStatus()
@@ -109,10 +109,10 @@ func (impl *ApplicationStatusHandlerImpl) Subscribe() error {
 		_, err = impl.pipelineRepository.GetArgoPipelineByArgoAppName(app.ObjectMeta.Name)
 		if err != nil && err == pg.ErrNoRows {
 			impl.logger.Infow("this app not found in pipeline table looking in installed_apps table", "appName", app.ObjectMeta.Name)
-			//if not found in pipeline table then search in installed_apps table
+			// if not found in pipeline table then search in installed_apps table
 			gitOpsDeployedAppNames, err := impl.installedAppRepository.GetAllGitOpsDeploymentAppName()
 			if err != nil && err == pg.ErrNoRows {
-				//no installed_apps found
+				// no installed_apps found
 				impl.logger.Errorw("no installed apps found", "err", err)
 				return
 			} else if err != nil {
@@ -127,17 +127,17 @@ func (impl *ApplicationStatusHandlerImpl) Subscribe() error {
 				devtronGitOpsAppName = app.ObjectMeta.Name
 			}
 			if slices.Contains(gitOpsDeployedAppNames, devtronGitOpsAppName) {
-				//app found in installed_apps table hence setting flag to true
+				// app found in installed_apps table hence setting flag to true
 				isAppStoreApplication = true
 			} else {
-				//app neither found in installed_apps nor in pipeline table hence returning
+				// app neither found in installed_apps nor in pipeline table hence returning
 				return
 			}
 		}
 		isSucceeded, pipelineOverride, err := impl.appService.UpdateDeploymentStatusAndCheckIsSucceeded(app, applicationDetail.StatusTime, isAppStoreApplication)
 		if err != nil {
 			impl.logger.Errorw("error on application status update", "err", err, "msg", string(msg.Data))
-			//TODO - check update for charts - fix this call
+			// TODO - check update for charts - fix this call
 			if err == pg.ErrNoRows {
 				// if not found in charts (which is for devtron apps) try to find in installed app (which is for devtron charts)
 				_, err := impl.installedAppService.UpdateInstalledAppVersionStatus(app)
@@ -153,7 +153,7 @@ func (impl *ApplicationStatusHandlerImpl) Subscribe() error {
 		// invoke DagExecutor, for cd success which will trigger post stage if exist.
 		if isSucceeded {
 			impl.logger.Debugw("git hash history", "list", app.Status.History)
-			err = impl.workflowDagExecutor.HandleDeploymentSuccessEvent(pipelineOverride)
+			err = impl.workflowDagExecutor.HandleDeploymentSuccessEvent(pipelineOverride, msg.MsgId)
 			if err != nil {
 				impl.logger.Errorw("deployment success event error", "pipelineOverride", pipelineOverride, "err", err)
 				return
@@ -162,7 +162,12 @@ func (impl *ApplicationStatusHandlerImpl) Subscribe() error {
 		impl.logger.Debugw("application status update completed", "app", app.Name)
 	}
 
-	err := impl.pubsubClient.Subscribe(pubsub.APPLICATION_STATUS_UPDATE_TOPIC, callback)
+	loggerFunc := func(msg *model.PubSubMsg) {
+		impl.logger.Debugw("APP_STATUS_UPDATE_REQ", "topic", pubsub.APPLICATION_STATUS_UPDATE_TOPIC, "msgId", msg.MsgId, "data", msg.Data)
+	}
+
+	validations := impl.workflowDagExecutor.GetTriggerValidateFuncs()
+	err := impl.pubsubClient.Subscribe(pubsub.APPLICATION_STATUS_UPDATE_TOPIC, callback, loggerFunc, validations...)
 	if err != nil {
 		impl.logger.Error(err)
 		return err
@@ -191,7 +196,10 @@ func (impl *ApplicationStatusHandlerImpl) SubscribeDeleteStatus() error {
 			impl.logger.Errorw("error in updating pipeline delete status", "err", err, "appName", app.Name)
 		}
 	}
-	err := impl.pubsubClient.Subscribe(pubsub.APPLICATION_STATUS_DELETE_TOPIC, callback)
+	loggerFunc := func(msg *model.PubSubMsg) {
+		impl.logger.Debugw("APP_STATUS_DELETE_REQ", "topic", pubsub.APPLICATION_STATUS_DELETE_TOPIC, "msgId", msg.MsgId, "data", msg.Data)
+	}
+	err := impl.pubsubClient.Subscribe(pubsub.APPLICATION_STATUS_DELETE_TOPIC, callback, loggerFunc)
 	if err != nil {
 		impl.logger.Errorw("error in subscribing to argo application status delete topic", "err", err)
 		return err
@@ -210,7 +218,7 @@ func (impl *ApplicationStatusHandlerImpl) updateArgoAppDeleteStatus(app *v1alpha
 		return errors.New("invalid nats message, pipeline already deleted")
 	}
 	if err == pg.ErrNoRows {
-		//Helm app deployed using argocd
+		// Helm app deployed using argocd
 		var gitHash string
 		if app.Operation != nil && app.Operation.Sync != nil {
 			gitHash = app.Operation.Sync.Revision
@@ -229,7 +237,7 @@ func (impl *ApplicationStatusHandlerImpl) updateArgoAppDeleteStatus(app *v1alpha
 			impl.logger.Errorw("App not found in database", "installedAppId", model.InstalledAppId, "err", err)
 			return fmt.Errorf("app not found in database %s", err)
 		} else if installedApp.DeploymentAppDeleteRequest == false {
-			//TODO 4465 remove app from log after final RCA
+			// TODO 4465 remove app from log after final RCA
 			impl.logger.Infow("Deployment delete not requested for app, not deleting app from DB", "appName", app.Name, "app", app)
 			return nil
 		}

api/router/pubsub/CiEventHandler.go

@@ -50,10 +50,11 @@ type CiEventHandler interface {
 }
 
 type CiEventHandlerImpl struct {
-	logger         *zap.SugaredLogger
-	pubsubClient   *pubsub.PubSubClientServiceImpl
-	webhookService pipeline.WebhookService
-	ciEventConfig  *CiEventConfig
+	logger              *zap.SugaredLogger
+	pubsubClient        *pubsub.PubSubClientServiceImpl
+	webhookService      pipeline.WebhookService
+	workflowDagExecutor pipeline.WorkflowDagExecutor
+	ciEventConfig       *CiEventConfig
 }
 
 type ImageDetailsFromCR struct {
@@ -80,12 +81,13 @@ type CiCompleteEvent struct {
 	PluginArtifactStage           string                   `json:"pluginArtifactStage"`
 }
 
-func NewCiEventHandlerImpl(logger *zap.SugaredLogger, pubsubClient *pubsub.PubSubClientServiceImpl, webhookService pipeline.WebhookService, ciEventConfig *CiEventConfig) *CiEventHandlerImpl {
+func NewCiEventHandlerImpl(logger *zap.SugaredLogger, pubsubClient *pubsub.PubSubClientServiceImpl, webhookService pipeline.WebhookService, ciEventConfig *CiEventConfig, workflowDagExecutor pipeline.WorkflowDagExecutor) *CiEventHandlerImpl {
 	ciEventHandlerImpl := &CiEventHandlerImpl{
-		logger:         logger,
-		pubsubClient:   pubsubClient,
-		webhookService: webhookService,
-		ciEventConfig:  ciEventConfig,
+		logger:              logger,
+		pubsubClient:        pubsubClient,
+		webhookService:      webhookService,
+		ciEventConfig:       ciEventConfig,
+		workflowDagExecutor: workflowDagExecutor,
 	}
 	err := ciEventHandlerImpl.Subscribe()
 	if err != nil {
@@ -98,7 +100,7 @@ func NewCiEventHandlerImpl(logger *zap.SugaredLogger, pubsubClient *pubsub.PubSu
 func (impl *CiEventHandlerImpl) Subscribe() error {
 	callback := func(msg *model.PubSubMsg) {
 		impl.logger.Debugw("ci complete event received")
-		//defer msg.Ack()
+		// defer msg.Ack()
 		ciCompleteEvent := CiCompleteEvent{}
 		err := json.Unmarshal([]byte(string(msg.Data)), &ciCompleteEvent)
 		if err != nil {
@@ -133,7 +135,7 @@ func (impl *CiEventHandlerImpl) Subscribe() error {
 						impl.logger.Error("Error while creating request for pipelineID", "pipelineId", ciCompleteEvent.PipelineId, "err", err)
 						return
 					}
-					resp, err := impl.webhookService.HandleCiSuccessEvent(ciCompleteEvent.PipelineId, request, detail.ImagePushedAt)
+					resp, err := impl.webhookService.HandleCiSuccessEvent(ciCompleteEvent.PipelineId, request, detail.ImagePushedAt, msg.MsgId)
 					if err != nil {
 						impl.logger.Error("Error while sending event for CI success for pipelineID", "pipelineId",
 							ciCompleteEvent.PipelineId, "request", request, "err", err)
@@ -145,7 +147,7 @@ func (impl *CiEventHandlerImpl) Subscribe() error {
 
 		} else {
 			util.TriggerCIMetrics(ciCompleteEvent.Metrics, impl.ciEventConfig.ExposeCiMetrics, ciCompleteEvent.PipelineName, ciCompleteEvent.AppName)
-			resp, err := impl.webhookService.HandleCiSuccessEvent(ciCompleteEvent.PipelineId, req, &time.Time{})
+			resp, err := impl.webhookService.HandleCiSuccessEvent(ciCompleteEvent.PipelineId, req, &time.Time{}, msg.MsgId)
 			if err != nil {
 				impl.logger.Error("Error while sending event for CI success for pipelineID: ",
 					ciCompleteEvent.PipelineId, "request: ", req, "error: ", err)
@@ -154,7 +156,12 @@ func (impl *CiEventHandlerImpl) Subscribe() error {
 			impl.logger.Debug(resp)
 		}
 	}
-	err := impl.pubsubClient.Subscribe(pubsub.CI_COMPLETE_TOPIC, callback)
+	loggerFunc := func(msg *model.PubSubMsg) {
+		impl.logger.Debugw("CI_COMPLETE_EVENT", "topic", pubsub.CI_COMPLETE_TOPIC, "msgId", msg.MsgId, "data", msg.Data)
+	}
+
+	validations := impl.workflowDagExecutor.GetTriggerValidateFuncs()
+	err := impl.pubsubClient.Subscribe(pubsub.CI_COMPLETE_TOPIC, callback, loggerFunc, validations...)
 	if err != nil {
 		impl.logger.Error(err)
 		return err

api/router/pubsub/GitWebhookHandler.go

@@ -68,7 +68,12 @@ func (impl *GitWebhookHandlerImpl) Subscribe() error {
 			return
 		}
 	}
-	err := impl.pubsubClient.Subscribe(pubsub.NEW_CI_MATERIAL_TOPIC, callback)
+
+	loggerFunc := func(msg *model.PubSubMsg) {
+		impl.logger.Debugw("NEW_CI_MATERIAL", "topic", pubsub.NEW_CI_MATERIAL_TOPIC, "msgId", msg.MsgId, "data", msg.Data)
+	}
+
+	err := impl.pubsubClient.Subscribe(pubsub.NEW_CI_MATERIAL_TOPIC, callback, loggerFunc)
 	if err != nil {
 		impl.logger.Error("err", err)
 		return err

api/router/pubsub/WorkflowStatusUpdateHandler.go

@@ -93,7 +93,12 @@ func (impl *WorkflowStatusUpdateHandlerImpl) Subscribe() error {
 		}
 
 	}
-	err := impl.pubsubClient.Subscribe(pubsub.WORKFLOW_STATUS_UPDATE_TOPIC, callback)
+
+	loggerFunc := func(msg *model.PubSubMsg) {
+		impl.logger.Debugw("WORKFLOW_STATUS_UPDATE", "topic", pubsub.WORKFLOW_STATUS_UPDATE_TOPIC, "msgId", msg.MsgId, "data", msg.Data)
+	}
+
+	err := impl.pubsubClient.Subscribe(pubsub.WORKFLOW_STATUS_UPDATE_TOPIC, callback, loggerFunc)
 
 	if err != nil {
 		impl.logger.Error("err", err)
@@ -170,7 +175,11 @@ func (impl *WorkflowStatusUpdateHandlerImpl) SubscribeCD() error {
 			}
 		}
 	}
-	err := impl.pubsubClient.Subscribe(pubsub.CD_WORKFLOW_STATUS_UPDATE, callback)
+	loggerFunc := func(msg *model.PubSubMsg) {
+		impl.logger.Debugw("CD_WORKFLOW_STATUS_UPDATE", "topic", pubsub.WORKFLOW_STATUS_UPDATE_TOPIC, "msgId", msg.MsgId, "data", msg.Data)
+	}
+
+	err := impl.pubsubClient.Subscribe(pubsub.CD_WORKFLOW_STATUS_UPDATE, callback, loggerFunc)
 	if err != nil {
 		impl.logger.Error("err", err)
 		return err

client/cron/CdApplicationStatusUpdateHandler.go

@@ -144,13 +144,19 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) Subscribe() error {
 			}
 		}
 
-		err, _ = impl.CdHandler.UpdatePipelineTimelineAndStatusByLiveApplicationFetch(cdPipeline, installedApp, statusUpdateEvent.UserId)
+		err, _ = impl.CdHandler.UpdatePipelineTimelineAndStatusByLiveApplicationFetch(cdPipeline, installedApp, statusUpdateEvent.UserId, msg.MsgId)
 		if err != nil {
 			impl.logger.Errorw("error on argo pipeline status update", "err", err, "msg", string(msg.Data))
 			return
 		}
 	}
-	err := impl.pubsubClient.Subscribe(pubsub.ARGO_PIPELINE_STATUS_UPDATE_TOPIC, callback)
+
+	loggerFunc := func(msg *model.PubSubMsg) {
+		impl.logger.Debugw("ARGO_PIPELINE_STATUS_UPDATE", "topic", pubsub.ARGO_PIPELINE_STATUS_UPDATE_TOPIC, "msgId", msg.MsgId, "data", msg.Data)
+	}
+
+	validations := impl.workflowDagExecutor.GetTriggerValidateFuncs()
+	err := impl.pubsubClient.Subscribe(pubsub.ARGO_PIPELINE_STATUS_UPDATE_TOPIC, callback, loggerFunc, validations...)
 	if err != nil {
 		impl.logger.Errorw("error in subscribing to argo application status update topic", "err", err)
 		return err
@@ -182,7 +188,7 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) ArgoApplicationStatusUpdate()
 	defer func() {
 		middleware.DeploymentStatusCronDuration.WithLabelValues(pipeline.DEVTRON_APP_ARGO_PIPELINE_STATUS_UPDATE_CRON).Observe(time.Since(cronProcessStartTime).Seconds())
 	}()
-	//TODO: remove below cron with division of cron for argo pipelines of devtron-apps and helm-apps
+	// TODO: remove below cron with division of cron for argo pipelines of devtron-apps and helm-apps
 	defer func() {
 		middleware.DeploymentStatusCronDuration.WithLabelValues(pipeline.HELM_APP_ARGO_PIPELINE_STATUS_UPDATE_CRON).Observe(time.Since(cronProcessStartTime).Seconds())
 	}()
@@ -227,7 +233,7 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) SyncPipelineStatusForResourceT
 }
 
 func (impl *CdApplicationStatusUpdateHandlerImpl) SyncPipelineStatusForAppStoreForResourceTreeCall(installedAppVersion *repository2.InstalledAppVersions) error {
-	//find installed app version history using parameter obj
+	// find installed app version history using parameter obj
 	installedAppVersionHistory, err := impl.installedAppVersionHistoryRepository.GetLatestInstalledAppVersionHistory(installedAppVersion.Id)
 	if err != nil {
 		impl.logger.Errorw("error in getting latest installedAppVersionHistory by installedAppVersionId", "err", err, "installedAppVersionId", installedAppVersion.Id)
@@ -263,7 +269,7 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) ManualSyncPipelineStatus(appId
 		cdPipeline = cdPipelines[0]
 	}
 
-	err, isTimelineUpdated := impl.CdHandler.UpdatePipelineTimelineAndStatusByLiveApplicationFetch(cdPipeline, installedApp, userId)
+	err, isTimelineUpdated := impl.CdHandler.UpdatePipelineTimelineAndStatusByLiveApplicationFetch(cdPipeline, installedApp, userId, nil)
 	if err != nil {
 		impl.logger.Errorw("error on argo pipeline status update", "err", err)
 		return nil

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/deckarep/golang-set v1.8.0
 	github.com/devtron-labs/authenticator v0.4.32
-	github.com/devtron-labs/common-lib v0.0.9-0.20231226070212-c47f7a07ebf5
+	github.com/devtron-labs/common-lib v0.0.9-0.20240103054524-5c4f393b38e3
 	github.com/devtron-labs/protos v0.0.0-20230503113602-282404f70fd2
 	github.com/evanphx/json-patch v5.6.0+incompatible
 	github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32

go.sum

@@ -223,8 +223,8 @@ github.com/denisenkom/go-mssqldb v0.0.0-20190707035753-2be1aa521ff4 h1:YcpmyvADG
 github.com/denisenkom/go-mssqldb v0.0.0-20190707035753-2be1aa521ff4/go.mod h1:zAg7JM8CkOJ43xKXIj7eRO9kmWm/TW578qo+oDO6tuM=
 github.com/devtron-labs/authenticator v0.4.32 h1:JAIJ0WqTXWj2nW7b8so9wunNICQn7O1Qpkk8INpatcs=
 github.com/devtron-labs/authenticator v0.4.32/go.mod h1:ozNfT8WcruiSgnUbyp48WVfc41++W6xYXhKFp67lNTU=
-github.com/devtron-labs/common-lib v0.0.9-0.20231226070212-c47f7a07ebf5 h1:+Nh2SMzAdgBr1tgdKAlF5cN0CvTPUj1V/sI5aRUrZnE=
-github.com/devtron-labs/common-lib v0.0.9-0.20231226070212-c47f7a07ebf5/go.mod h1:pBThgympEjsza6GShqNNGCPBFXNDx0DGMc7ID/VHTAw=
+github.com/devtron-labs/common-lib v0.0.9-0.20240103054524-5c4f393b38e3 h1:Yc6RJTrzCQyhKFbb5Onv+TC3x4ODd0TnNv7Bvll1Zwg=
+github.com/devtron-labs/common-lib v0.0.9-0.20240103054524-5c4f393b38e3/go.mod h1:pBThgympEjsza6GShqNNGCPBFXNDx0DGMc7ID/VHTAw=
 github.com/devtron-labs/protos v0.0.0-20230503113602-282404f70fd2 h1:/IEIsJTxDZ3hv8uOoCaqdWCXqcv7nCAgX9AP/v84dUY=
 github.com/devtron-labs/protos v0.0.0-20230503113602-282404f70fd2/go.mod h1:l85jxWHlcSo910hdUfRycL40yGzC6glE93V1sVxVPto=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=

internal/sql/repository/CiArtifactRepository.go

@@ -59,10 +59,10 @@ type CiArtifactWithExtraData struct {
 type CiArtifact struct {
 	tableName             struct{}  `sql:"ci_artifact" pg:",discard_unknown_columns"`
 	Id                    int       `sql:"id,pk"`
-	PipelineId            int       `sql:"pipeline_id"` //id of the ci pipeline from which this webhook was triggered
+	PipelineId            int       `sql:"pipeline_id"` // id of the ci pipeline from which this webhook was triggered
 	Image                 string    `sql:"image,notnull"`
 	ImageDigest           string    `sql:"image_digest,notnull"`
-	MaterialInfo          string    `sql:"material_info"`       //git material metadata json array string
+	MaterialInfo          string    `sql:"material_info"`       // git material metadata json array string
 	DataSource            string    `sql:"data_source,notnull"` // possible values -> (CI_RUNNER,ext,post_ci,pre_cd,post_cd) CI_runner is for normal build ci
 	WorkflowId            *int      `sql:"ci_workflow_id"`
 	ParentCiArtifact      int       `sql:"parent_ci_artifact"`
@@ -315,15 +315,15 @@ func (impl CiArtifactRepositoryImpl) GetArtifactsByCDPipelineV3(listingFilterOpt
 }
 
 func (impl CiArtifactRepositoryImpl) setDeployedDataInArtifacts(pipelineId int, artifacts []*CiArtifact) ([]*CiArtifact, error) {
-	//processing
+	// processing
 	artifactsMap := make(map[int]*CiArtifact)
 	artifactsIds := make([]int, 0, len(artifacts))
 	for _, artifact := range artifacts {
 		artifactsMap[artifact.Id] = artifact
 		artifactsIds = append(artifactsIds, artifact.Id)
 	}
 
-	//(this will fetch all the artifacts that were deployed on the given pipeline atleast once in new->old deployed order)
+	// (this will fetch all the artifacts that were deployed on the given pipeline atleast once in new->old deployed order)
 	artifactsDeployed := make([]*CiArtifact, 0, len(artifactsIds))
 	query := " SELECT cia.id,pco.created_on AS created_on " +
 		" FROM ci_artifact cia" +
@@ -337,7 +337,7 @@ func (impl CiArtifactRepositoryImpl) setDeployedDataInArtifacts(pipelineId int,
 		return artifacts, nil
 	}
 
-	//set deployed time and latest deployed artifact
+	// set deployed time and latest deployed artifact
 	for _, deployedArtifact := range artifactsDeployed {
 		artifactId := deployedArtifact.Id
 		if _, ok := artifactsMap[artifactId]; ok {
@@ -392,8 +392,8 @@ func (impl CiArtifactRepositoryImpl) GetArtifactsByCDPipelineAndRunnerType(cdPip
 		" INNER JOIN cd_workflow_runner wfr on wfr.cd_workflow_id = wf.id" +
 		" WHERE p.id= ? and wfr.workflow_type = ? GROUP BY cia.id, cia.data_source, cia.image ORDER BY cia.id DESC"*/
 
-	//this query gets details for status = Succeeded, this status is only valid
-	//for pre stages & post stages, for deploy stage status will be healthy, degraded, aborted, missing etc
+	// this query gets details for status = Succeeded, this status is only valid
+	// for pre stages & post stages, for deploy stage status will be healthy, degraded, aborted, missing etc
 	queryFetchArtifacts = "SELECT cia.id, cia.data_source, cia.image, cia.image_digest FROM cd_workflow_runner wfr" +
 		" INNER JOIN cd_workflow wf on wf.id=wfr.cd_workflow_id" +
 		" INNER JOIN pipeline p on p.id = wf.pipeline_id" +
@@ -444,7 +444,7 @@ func (impl CiArtifactRepositoryImpl) GetArtifactsByCDPipelineAndRunnerType(cdPip
 		return nil, err
 	}
 
-	//find latest deployed entry
+	// find latest deployed entry
 	latestObj := Object{}
 	latestDeployedQuery := "SELECT cia.id FROM ci_artifact cia" +
 		" INNER JOIN pipeline_config_override pco ON pco.ci_artifact_id=cia.id" +
@@ -651,7 +651,7 @@ func (impl CiArtifactRepositoryImpl) FinDByParentCiArtifactAndCiId(parentCiArtif
 }
 
 func (impl CiArtifactRepositoryImpl) GetLatest(cdPipelineId int) (int, error) {
-	//find latest deployed entry
+	// find latest deployed entry
 	type Object struct {
 		Id           int    `json:"id"`
 		MaterialInfo string `json:"material_info"`